Update app.py

app.py

@@ -1,50 +1,77 @@
+import gradio as gr
+from fastapi import FastAPI, Request
+from transformers import pipeline
+import uvicorn
 import json
-from datetime import datetime
-
-def calculate_scores(logs):
-    quality_scores = []
-    timeliness_scores = []
-    safety_scores = []
-    communication_scores = []
-    
+
+# Initialize sentiment analysis pipeline
+sentiment_analyzer = pipeline("sentiment-analysis", model="distilbert-base-uncased-finetuned-sst-2-english")
+
+# FastAPI app for API endpoint
+app = FastAPI()
+
+@app.post("/calculate_scores")
+async def calculate_scores(request: Request):
+    logs = await request.json()
+    quality_score = 0
+    timeliness_score = 0
+    safety_score = 0
+    communication_score = 0
+    quality_count = 0
+    delay_count = 0
+    incident_count = 0
+    feedback_count = 0
+
     for log in logs:
-        if log['log_type'] == 'Quality':
-            quality_scores.append(log['quality_score'] != null and log['quality_score'] or 50)
-        elif log['log_type'] == 'Delay':
-            timeliness_scores.append(100 - (log['delay_percentage'] != null and log['delay_percentage'] or 0))
-        elif log['log_type'] == 'Incident':
-            safety_scores.append(log['safety_compliance'] and 100 or 0)
-        elif log['log_type'] == 'Work Completion':
-            communication_scores.append(log['feedback'] != null and len(log['feedback']) > 0 and 80 or 20)
-    
-    quality_avg = sum(quality_scores) / len(quality_scores) if quality_scores else 50
-    timeliness_avg = sum(timeliness_scores) / len(timeliness_scores) if timeliness_scores else 50
-    safety_avg = sum(safety_scores) / len(safety_scores) if safety_scores else 50
-    communication_avg = sum(communication_scores) / len(communication_scores) if communication_scores else 50
-    
-    final_score = (quality_avg * 0.3) + (timeliness_avg * 0.3) + (safety_avg * 0.2) + (communication_avg * 0.2)
-    alert_flag = final_score < 60
-    
+        log_type = log.get("log_type")
+        if log_type == "Quality" and log.get("quality_score"):
+            quality_score += float(log["quality_score"])
+            quality_count += 1
+        if log_type == "Delay" and log.get("delay_percentage"):
+            timeliness_score += (100 - float(log["delay_percentage"]))
+            delay_count += 1
+        if log_type == "Incident":
+            safety_score += (100 if log.get("safety_compliance", False) else 0)
+            incident_count += 1
+        if log.get("feedback"):
+            feedback = log["feedback"]
+            sentiment = sentiment_analyzer(feedback)[0]
+            sentiment_score = sentiment["score"] * 100 if sentiment["label"] == "POSITIVE" else 50
+            base_score = 90 if len(feedback) > 50 else 70
+            communication_score += base_score * (sentiment_score / 100)
+            feedback_count += 1
+
+    # Calculate averages
+    quality_score = quality_score / quality_count if quality_count > 0 else 0
+    timeliness_score = timeliness_score / delay_count if delay_count > 0 else 0
+    safety_score = safety_score / incident_count if incident_count > 0 else 0
+    communication_score = communication_score / feedback_count if feedback_count > 0 else 0
+
     return {
-        'quality_score': round(quality_avg, 2),
-        'timeliness_score': round(timeliness_avg, 2),
-        'safety_score': round(safety_avg, 2),
-        'communication_score': round(communication_avg, 2),
-        'final_score': round(final_score, 2),
-        'alert_flag': alert_flag,
-        'trend': 'stable'  # Simplified; add trend detection later
+        "quality_score": round(quality_score, 2),
+        "timeliness_score": round(timeliness_score, 2),
+        "safety_score": round(safety_score, 2),
+        "communication_score": round(communication_score, 2),
+        "alert_flag": quality_score < 50 or timeliness_score < 50
     }
 
-def main():
-    # Sample input for testing
-    sample_logs = [
-        {'log_type': 'Quality', 'quality_score': 85},
-        {'log_type': 'Delay', 'delay_percentage': 10},
-        {'log_type': 'Incident', 'safety_compliance': True},
-        {'log_type': 'Work Completion', 'feedback': 'Good communication'}
-    ]
-    result = calculate_scores(sample_logs)
-    print(json.dumps(result, indent=2))
-
-if __name__ == '__main__':
-    main()
+# Gradio interface for testing
+def test_api(logs_json):
+    try:
+        logs = json.loads(logs_json)
+        result = calculate_scores(logs)
+        return json.dumps(result, indent=2)
+    except Exception as e:
+        return f"Error: {str(e)}"
+
+# Launch Gradio UI
+gr.Interface(
+    fn=test_api,
+    inputs=gr.Textbox(label="Enter Logs JSON", placeholder='[{"log_type": "Quality", "quality_score": 80, "feedback": "Great work!"}]'),
+    outputs=gr.Textbox(label="Scores"),
+    title="Subcontractor Scoring API Tester"
+).launch()
+
+# Run FastAPI with Uvicorn (Gradio handles this automatically)
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=7860)