app.py

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)