Update app.py

app.py

@@ -5,108 +5,121 @@ from reportlab.pdfgen import canvas
 import base64
 import os
 import logging
-from huggingface_hub import InferenceClient
-import numpy as np
+from datetime import datetime
+from fastapi.responses import HTMLResponse
 
-# Set up logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
 app = FastAPI()
 
-# Load API key from environment variable
 API_KEY = os.getenv("HUGGINGFACE_API_KEY")
 if not API_KEY:
     logger.error("HUGGINGFACE_API_KEY environment variable not set")
     raise ValueError("HUGGINGFACE_API_KEY environment variable not set")
 
-# Initialize Hugging Face Inference Client
-client = InferenceClient(token=API_KEY)
-
 class VendorLog(BaseModel):
     vendorLogId: str
     vendorId: str
-    workDetails: str
-    workCompletionDate: str  # ISO format (e.g., "2025-05-12")
-    actualCompletionDate: str  # ISO format
-    incidentLog: str
-    qualityReport: str
+    workCompletionPercentage: float
+    qualityPercentage: float
+    incidentSeverity: str
+    workCompletionDate: str
+    actualCompletionDate: str
     vendorLogName: str
-
-def analyze_sentiment(text: str) -> float:
-    """Analyze text sentiment using Hugging Face Inference API."""
-    try:
-        if not text:
-            return 50.0  # Neutral score for empty text
-        result = client.text_classification(text, model="distilbert-base-uncased-finetuned-sst-2-english")
-        score = result[0]['score'] if result[0]['label'] == 'POSITIVE' else 1 - result[0]['score']
-        return score * 100  # Convert to percentage
-    except Exception as e:
-        logger.error(f"Sentiment analysis error: {e}")
-        return 50.0  # Fallback score
+    delayDays: int
 
 def calculate_scores(log: VendorLog):
-    # Sentiment analysis for work details, quality report, incident log
-    work_completion_score = analyze_sentiment(log.workDetails)
-    quality_score = analyze_sentiment(log.qualityReport)
-    incident_score = analyze_sentiment(log.incidentLog) if log.incidentLog else 100.0  # No incidents = high score
+    quality_score = log.qualityPercentage
 
-    # Timeliness score based on date difference
-    from datetime import datetime
-    work_date = datetime.fromisoformat(log.workCompletionDate.replace('Z', '+00:00'))
-    actual_date = datetime.fromisoformat(log.actualCompletionDate.replace('Z', '+00:00'))
-    delay_days = (actual_date - work_date).days
-    timeliness_score = 100.0 if delay_days <= 0 else 80.0 if delay_days <= 3 else 60.0 if delay_days <= 7 else 40.0
+    timeliness_score = 100.0 if log.delayDays <= 0 else 80.0 if log.delayDays <= 3 else 60.0 if log.delayDays <= 7 else 40.0
 
-    # Safety score (inverse of incident sentiment)
-    safety_score = 100.0 - (incident_score if incident_score < 100.0 else 0.0)
+    severity_map = {'None': 100.0, 'Low': 80.0, 'Medium': 50.0, 'High': 20.0}
+    safety_score = severity_map.get(log.incidentSeverity, 100.0)
 
-    # Communication score (average of work and quality sentiment)
-    communication_score = (work_completion_score + quality_score) / 2
+    communication_score = (quality_score * 0.33 + timeliness_score * 0.33 + safety_score * 0.33)
+
+    final_score = (quality_score + timeliness_score + safety_score + communication_score) / 4
 
     return {
         'qualityScore': round(quality_score, 2),
         'timelinessScore': round(timeliness_score, 2),
         'safetyScore': round(safety_score, 2),
         'communicationScore': round(communication_score, 2),
-        'finalScore': round((quality_score + timeliness_score + safety_score + communication_score) / 4, 2)
+        'finalScore': round(final_score, 2)
     }
 
-def generate_pdf(vendor_id: str, scores: dict):
+def get_feedback(score: float, metric: str) -> str:
+    if score >= 90:
+        return "Excellent: Maintain this standard"
+    elif score >= 70:
+        return "Good: Keep up the good work"
+    elif score >= 50:
+        if metric == 'Timeliness':
+            return "Needs Improvement: Maintain schedules to complete tasks on time"
+        elif metric == 'Safety':
+            return "Needs Improvement: Implement stricter safety protocols"
+        elif metric == 'Quality':
+            return "Needs Improvement: Focus on improving work quality"
+        else:
+            return "Needs Improvement: Enhance coordination with project teams"
+    else:
+        if metric == 'Timeliness':
+            return "Poor: Significant delays detected"
+        elif metric == 'Safety':
+            return "Poor: Critical safety issues identified"
+        elif metric == 'Quality':
+            return "Poor: Quality standards not met"
+        else:
+            return "Poor: Communication issues detected"
+
+def generate_pdf(vendor_id: str, vendor_log_name: str, scores: dict):
     filename = f'report_{vendor_id}.pdf'
     c = canvas.Canvas(filename, pagesize=letter)
     c.setFont('Helvetica', 12)
-    c.drawString(100, 750, 'Vendor Performance Report')
+    c.drawString(100, 750, 'Subcontractor Performance Report')
     c.drawString(100, 730, f'Vendor ID: {vendor_id}')
-    y = 700
-    for metric, score in scores.items():
-        c.drawString(100, y, f'{metric.replace("Score", " Score")}: {score}%')
-        y -= 20
+    c.drawString(100, 710, f'Quality Score: {scores["qualityScore"]}% ({get_feedback(scores["qualityScore"], "Quality")})')
+    c.drawString(100, 690, f'Timeliness Score: {scores["timelinessScore"]}% ({get_feedback(scores["timelinessScore"], "Timeliness")})')
+    c.drawString(100, 670, f'Safety Score: {scores["safetyScore"]}% ({get_feedback(scores["safetyScore"], "Safety")})')
+    c.drawString(100, 650, f'Communication Score: {scores["communicationScore"]}% ({get_feedback(scores["communicationScore"], "Communication")})')
+    c.drawString(100, 630, f'Final Score: {scores["finalScore"]}%')
     c.save()
-    
+
     with open(filename, 'rb') as f:
         pdf_content = f.read()
     os.remove(filename)
     return pdf_content
 
-def send_alert(vendor_id: str, final_score: float):
-    """Placeholder for alert logic (e.g., email or webhook)."""
+def determine_alert_flag(final_score: float, all_scores: list):
     if final_score < 50:
-        logger.info(f"Alert: Vendor {vendor_id} has low final score ({final_score}%)")
-        # Implement email or webhook notification here
-        return {"vendorId": vendor_id, "message": f"Low performance score ({final_score}%) detected"}
-    return None
+        return True
+    above_50 = any(score['scores']['finalScore'] >= 50 for score in all_scores)
+    if not above_50:
+        lowest_score = min([score['scores']['finalScore'] for score in all_scores])
+        return final_score == lowest_score
+    return False
+
+vendor_logs = []
 
 @app.post('/score')
 async def score_vendor(log: VendorLog, authorization: str = Header(...)):
     if authorization != f'Bearer {API_KEY}':
         raise HTTPException(status_code=401, detail='Invalid API key')
-    
+
     scores = calculate_scores(log)
-    pdf_content = generate_pdf(log.vendorId, scores)
+    pdf_content = generate_pdf(log.vendorId, log.vendorLogName, scores)
     pdf_base64 = base64.b64encode(pdf_content).decode('utf-8')
-    alert = send_alert(log.vendorId, scores['finalScore'])
-    
+    alert_flag = determine_alert_flag(scores['finalScore'], vendor_logs)
+
+    vendor_logs.append({
+        'vendorLogId': log.vendorLogId,
+        'vendorId': log.vendorId,
+        'vendorLogName': log.vendorLogName,
+        'scores': scores,
+        'extracted': True
+    })
+
     return {
         'vendorLogId': log.vendorLogId,
         'vendorId': log.vendorId,
@@ -118,9 +131,78 @@ async def score_vendor(log: VendorLog, authorization: str = Header(...)):
         'finalScore': scores['finalScore'],
         'pdfContent': pdf_base64,
         'pdfUrl': f'/files/report_{log.vendorId}.pdf',
-        'alert': alert
+        'alertFlag': alert_flag
     }
 
+@app.get('/', response_class=HTMLResponse)
+async def get_dashboard():
+    html_content = """
+    <html>
+        <head>
+            <title>Subcontractor Performance Score App</title>
+            <style>
+                body { font-family: Arial, sans-serif; margin: 20px; }
+                table { width: 100%; border-collapse: collapse; margin-top: 20px; }
+                th, td { border: 1px solid #ddd; padding: 8px; text-align: left; }
+                th { background-color: #f2f2f2; }
+                h1, h2 { text-align: center; }
+            </style>
+        </head>
+        <body>
+            <h1>Subcontractor Performance Score App Generator</h1>
+            <h2>Vendor Logs</h2>
+            <table>
+                <tr>
+                    <th>Vendor ID</th>
+                    <th>Details Extracted</th>
+                    <th>Final Score</th>
+                </tr>
+    """
+    for log in vendor_logs:
+        html_content += f"""
+                <tr>
+                    <td>{log['vendorId']}</td>
+                    <td>{'YES' if log['extracted'] else 'NO'}</td>
+                    <td>{log['scores']['finalScore'] if log['extracted'] else 'NOT CALCULATED'}</td>
+                </tr>
+        """
+    html_content += """
+            </table>
+            <h2>Subcontractor Performance Scores</h2>
+            <table>
+                <tr>
+                    <th>Vendor ID</th>
+                    <th>Vendor Log Name</th>
+                    <th>Quality Score</th>
+                    <th>Timeliness Score</th>
+                    <th>Safety Score</th>
+                    <th>Communication Score</th>
+                    <th>Final Score</th>
+                    <th>Alert Flag</th>
+                </tr>
+    """
+    for log in vendor_logs:
+        scores = log['scores']
+        alert_flag = determine_alert_flag(scores['finalScore'], vendor_logs)
+        html_content += f"""
+                <tr>
+                    <td>{log['vendorId']}</td>
+                    <td>{log['vendorLogName']}</td>
+                    <td>{scores['qualityScore']}%</td>
+                    <td>{scores['timelinessScore']}%</td>
+                    <td>{scores['safetyScore']}%</td>
+                    <td>{scores['communicationScore']}%</td>
+                    <td>{scores['finalScore']}%</td>
+                    <td>{'Checked' if alert_flag else 'Unchecked'}</td>
+                </tr>
+        """
+    html_content += """
+            </table>
+        </body>
+    </html>
+    """
+    return HTMLResponse(content=html_content)
+
 if __name__ == "__main__":
     import uvicorn
     uvicorn.run(app, host="0.0.0.0", port=7860)