Create app.py

app.py

@@ -0,0 +1,141 @@
+from starlette.applications import Starlette
+from starlette.routing import Route
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+from pydantic import BaseModel, Field
+from typing import List, Optional
+from datetime import datetime
+import uvicorn
+import numpy as np
+import json
+
+app = Starlette(debug=True)
+
+# In-memory store for historical scores (replace with database for production)
+historical_scores = []
+
+# Define the request body structure
+class VendorLog(BaseModel):
+    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 weights (configurable)
+WEIGHTS = {
+    "quality": 0.3,
+    "timeliness": 0.3,
+    "safety": 0.2,
+    "communication": 0.2
+}
+
+async def calculate_score(request: Request):
+    try:
+        # Parse JSON body
+        body = await request.json()
+        logs = VendorLog(**body)
+
+        # 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}"
+
+        # Prepare response
+        response = ScoreResponse(
+            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
+        )
+
+        return JSONResponse(content=response.dict())
+
+    except ValueError as e:
+        return JSONResponse(
+            content={"detail": f"Invalid input: {str(e)}"},
+            status_code=400
+        )
+    except json.JSONDecodeError:
+        return JSONResponse(
+            content={"detail": "Invalid JSON format"},
+            status_code=400
+        )
+    except Exception as e:
+        return JSONResponse(
+            content={"detail": f"Internal server error: {str(e)}"},
+            status_code=500
+        )
+
+async def health_check(request: Request):
+    return JSONResponse(content={"status": "healthy"})
+
+# Define routes
+app.routes.append(Route("/score", endpoint=calculate_score, methods=["POST"]))
+app.routes.append(Route("/health", endpoint=health_check, methods=["GET"]))
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=8001)