Update app.py

app.py

@@ -1,40 +1,118 @@
-from fastapi import FastAPI
-from pydantic import BaseModel
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel, Field
+from typing import List, Optional
+from datetime import datetime
 import uvicorn
+import numpy as np
 
-app = FastAPI()
+app = FastAPI(title="Vendor Performance Scoring API", description="API for calculating vendor performance scores with trend detection")
+
+# In-memory store for historical scores (replace with database for production)
+historical_scores = []
 
 # Define the request body structure
 class VendorLog(BaseModel):
-    work_logs: int
-    delay_logs: int
-    quality_logs: int
-    incident_logs: int
+    vendor_id: str = Field(..., description="Unique identifier for the vendor")
+    delay_logs: int = Field(..., ge=0, description="Number of delay incidents")
+    qa_incidents: int = Field(..., ge=0, description="Number of quality assurance incidents")
+    safety_compliance: int = Field(..., ge=0, description="Number of safety compliance issues")
+    feedback_logs: int = Field(..., ge=0, description="Number of feedback logs")
+    work_logs: int = Field(..., ge=1, description="Total number of work logs")
+    month: str = Field(..., description="Month of the score (YYYY-MM format)")
+
+# Define the response structure
+class ScoreResponse(BaseModel):
+    vendor_id: str
+    month: str
+    final_score: float
+    quality_score: float
+    timeliness_score: float
+    safety_score: float
+    communication_score: float
+    alert_flag: bool
+    trend_deviation: Optional[float]
+    certification_url: Optional[str]
 
-# Scoring logic
-@app.post("/score")
+# Scoring weights (configurable)
+WEIGHTS = {
+    "quality": 0.3,
+    "timeliness": 0.3,
+    "safety": 0.2,
+    "communication": 0.2
+}
+
+@app.post("/score", response_model=ScoreResponse)
 async def calculate_score(logs: VendorLog):
-    # Simple scoring logic (you can customize this)
-    quality_score = (logs.quality_logs / (logs.work_logs + 1)) * 100
-    timeliness_score = (1 - logs.delay_logs / (logs.work_logs + 1)) * 100
-    safety_score = (1 - logs.incident_logs / (logs.work_logs + 1)) * 100
-    communication_score = 50  # Placeholder, adjust based on your logic
-
-    # Final score (average of individual scores)
-    final_score = (quality_score + timeliness_score + safety_score + communication_score) / 4
-
-    # Alert flag logic (example: flag if final score < 50)
-    alert_flag = final_score < 50
-
-    return {
-        "final_score": round(final_score, 2),
-        "alert_flag": alert_flag,
-        "quality_score": round(quality_score, 2),
-        "timeliness_score": round(timeliness_score, 2),
-        "safety_score": round(safety_score, 2),
-        "communication_score": round(communication_score, 2)
-    }
-
-# Run the server
+    try:
+        # Validate month format
+        datetime.strptime(logs.month, "%Y-%m")
+
+        # Calculate individual scores
+        quality_score = (1 - logs.qa_incidents / logs.work_logs) * 100
+        timeliness_score = (1 - logs.delay_logs / logs.work_logs) * 100
+        safety_score = (1 - logs.safety_compliance / logs.work_logs) * 100
+        communication_score = (logs.feedback_logs / logs.work_logs) * 50  # Adjusted logic
+
+        # Ensure scores are within 0-100
+        quality_score = max(0, min(100, quality_score))
+        timeliness_score = max(0, min(100, timeliness_score))
+        safety_score = max(0, min(100, safety_score))
+        communication_score = max(0, min(100, communication_score))
+
+        # Calculate weighted final score
+        final_score = (
+            WEIGHTS["quality"] * quality_score +
+            WEIGHTS["timeliness"] * timeliness_score +
+            WEIGHTS["safety"] * safety_score +
+            WEIGHTS["communication"] * communication_score
+        )
+
+        # Alert flag logic
+        alert_flag = final_score < 50
+
+        # Trend detection (compare with historical scores for this vendor)
+        vendor_history = [score for score in historical_scores if score["vendor_id"] == logs.vendor_id]
+        trend_deviation = None
+        if len(vendor_history) >= 2:
+            previous_scores = [score["final_score"] for score in vendor_history[-2:]]
+            trend_deviation = final_score - np.mean(previous_scores)
+
+        # Store the current score
+        historical_scores.append({
+            "vendor_id": logs.vendor_id,
+            "month": logs.month,
+            "final_score": final_score,
+            "quality_score": quality_score,
+            "timeliness_score": timeliness_score,
+            "safety_score": safety_score,
+            "communication_score": communication_score
+        })
+
+        # Mock certification URL (replace with actual logic)
+        certification_url = f"https://example.com/cert/{logs.vendor_id}"
+
+        return {
+            "vendor_id": logs.vendor_id,
+            "month": logs.month,
+            "final_score": round(final_score, 2),
+            "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": alert_flag,
+            "trend_deviation": round(trend_deviation, 2) if trend_deviation is not None else None,
+            "certification_url": certification_url
+        }
+
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=f"Invalid month format: {str(e)}")
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Internal server error: {str(e)}")
+
+# Health check endpoint
+@app.get("/health")
+async def health_check():
+    return {"status": "healthy"}
+
 if __name__ == "__main__":
     uvicorn.run(app, host="0.0.0.0", port=8000)