Update app.py

app.py

@@ -2,225 +2,91 @@ import gradio as gr
 import pandas as pd
 import joblib
 import numpy as np
-from typing import Dict, Any, List, Tuple
+from fastapi import FastAPI, HTTPException, Header
+from pydantic import BaseModel
 from reportlab.lib.pagesizes import letter
 from reportlab.pdfgen import canvas
-from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer
-from reportlab.lib.styles import getSampleStyleSheet
-from reportlab.lib.units import inch
-import io
-from simple_salesforce import Salesforce
+import base64
 import os
 
-# 1. Load the Model and Artifacts
-# Model and artifact paths - adjust as necessary
-model_path = 'your_model.pkl'  # Make sure this path is correct
-scaler_path = 'scaler.pkl'  # If you used a scaler
-
-try:
-    model = joblib.load(model_path)
-    scaler = joblib.load(scaler_path) if scaler_path else None
-except Exception as e:
-    print(f"Error loading model or artifacts: {e}")
-    model = None
-
-# 2. Preprocessing Function
-def preprocess_data(data: Dict[str, Any]) -> pd.DataFrame:
-    """
-    Preprocesses input data for the model.
-
-    Args:
-        data (dict): A dictionary containing the input data.
-
-    Returns:
-        pandas.DataFrame: A DataFrame ready for model prediction.
-    """
-    try:
-        df = pd.DataFrame([data])
-
-        expected_fields = [
-            'Quality_Report__c',
-            'Delay_Days__c',
-            'Incident_Log__c',
-            'Vendor__c',
-            'Work_Details__c',
-            'Work_Completion_Date__c',
-            'Actual_Completion_Date__c'
-        ]
-
-        missing_fields = [field for field in expected_fields if field not in df.columns]
-        if missing_fields:
-            raise ValueError(f"Missing required fields: {', '.join(missing_fields)}")
-
-        if 'Delay_Days__c' in df.columns and scaler:
-            df['Delay_Days__c'] = df['Delay_Days__c'].fillna(0)
-            df[['Delay_Days__c']] = scaler.transform(df[['Delay_Days__c']])
-
-        if 'Work_Completion_Date__c' in df.columns:
-            df['Work_Completion_Date__c'] = pd.to_datetime(df['Work_Completion_Date__c'])
-            df['Work_Completion_Year'] = df['Work_Completion_Date__c'].dt.year
-            df['Work_Completion_Month'] = df['Work_Completion_Date__c'].dt.month
-            df['Work_Completion_Day'] = df['Work_Completion_Date__c'].dt.day
-            df = df.drop(columns=['Work_Completion_Date__c'])
-
-        if 'Actual_Completion_Date__c' in df.columns:
-            df['Actual_Completion_Date__c'] = pd.to_datetime(df['Actual_Completion_Date__c'])
-            df['Actual_Completion_Year'] = df['Actual_Completion_Date__c'].dt.year
-            df['Actual_Completion_Month'] = df['Actual_Completion_Date__c'].dt.month
-            df = df.drop(columns=['Actual_Completion_Date__c'])
-
-        df = df[expected_fields]
-        return df
-    except ValueError as ve:
-        print(f"Error in preprocess_data: {ve}")
-        return pd.DataFrame()
-    except Exception as e:
-        print(f"Error in preprocess_data: {e}")
-        return pd.DataFrame()
-
-
-# 3. Prediction Function
-def predict_vendor_score() -> Tuple[List[Tuple[str, float]], bytes]:
-    """
-    Retrieves data from Salesforce, predicts vendor performance scores, and generates a PDF report.
-
-    Returns:
-        Tuple[List[Tuple[str, float]], bytes]: A tuple containing the score data and the PDF data.
-    """
-    if model is None:
-        error_message = "Model not loaded.  Check server logs."
-        return [
-            ("Quality Score (%)", 0.0),
-            ("Timeliness Score (%)", 0.0),
-            ("Safety Score (%)", 0.0),
-            ("Communication Score (%)", 0.0),
-            ("Final Score (%)", 0.0),
-        ], generate_pdf(error_message)
-
-    try:
-        # 1. Connect to Salesforce
-        sf = Salesforce(
-            username=os.environ.get('SALESFORCE_USERNAME'),  # Use environment variables
-            password=os.environ.get('SALESFORCE_PASSWORD'),
-            security_token=os.environ.get('SALESFORCE_SECURITY_TOKEN'),
-            domain='login'  # or 'test' for a sandbox
-        )
-
-        # 2. SOQL Query (Adapt to your needs)
-        query = """
-            SELECT
-                Quality_Report__c,
-                Delay_Days__c,
-                Incident_Log__c,
-                Vendor__c,
-                Work_Details__c,
-                Work_Completion_Date__c,
-                Actual_Completion_Date__c
-            FROM
-                Vendor_Log__c
-            WHERE
-                # Add any filtering criteria here (e.g., date range)
-                CreatedDate >= LAST_MONTH
-        """
-        results = sf.query(query)
-
-        # 3. Data Transformation
-        records = results['records']
-        df = pd.DataFrame(records)
-        df = df.rename(columns={
-            'Quality_Report__c': 'Quality_Report__c',
-            'Delay_Days__c': 'Delay_Days__c',
-            'Incident_Log__c': 'Incident_Log__c',
-            'Vendor__c': 'Vendor__c',
-            'Work_Details__c': 'Work_Details__c',
-            'Work_Completion_Date__c': 'Work_Completion_Date__c',
-            'Actual_Completion_Date__c': 'Actual_Completion_Date__c'
-        })
-        df = df.drop(columns=['attributes'], errors='ignore')
-
-        # 4. Preprocess Data
-        processed_df = preprocess_data(df.iloc[0].to_dict())  # pass the first row as dict
-
-
-        if processed_df.empty:
-            error_message = "Error in input data. Check logs."
-            return [
-                ("Quality Score (%)", 0.0),
-                ("Timeliness Score (%)", 0.0),
-                ("Safety Score (%)", 0.0),
-                ("Communication Score (%)", 0.0),
-                ("Final Score (%)", 0.0),
-            ], generate_pdf(error_message)
-
-        # 5. Make Prediction
-        predictions = model.predict(processed_df)[0]
-
-        # 6.  Output
-        output_data = [
-            ("Quality Score (%)", predictions[0] * 100),
-            ("Timeliness Score (%)", predictions[1] * 100),
-            ("Safety Score (%)", predictions[2] * 100),
-            ("Communication Score (%)", predictions[3] * 100),
-            ("Final Score (%)", (predictions[0] + predictions[1] + predictions[2] + predictions[3]) / 4 * 100),
-        ]
-        pdf_data = generate_pdf(output_data)
-        return output_data, pdf_data
-
-    except Exception as e:
-        error_message = f"Error: {e}"
-        print(error_message)
-        return [
-            ("Quality Score (%)", 0.0),
-            ("Timeliness Score (%)", 0.0),
-            ("Safety Score (%)", 0.0),
-            ("Communication Score (%)", 0.0),
-            ("Final Score (%)", 0.0),
-        ], generate_pdf(error_message)
-
-
-
-# 4. PDF Generation
-def generate_pdf(scores: List[Tuple[str, float]]) -> bytes:
-    """Generates a PDF report of the subcontractor performance scores.
-
-    Args:
-        scores (List[Tuple[str, float]]): A list of (score_name, score_value) tuples.
-    Returns:
-        bytes: The PDF data as bytes.
-    """
-    buffer = io.BytesIO()
-    doc = SimpleDocTemplate(buffer, pagesize=letter)
-    styles = getSampleStyleSheet()
-    Story = []
-    Story.append(Paragraph("Subcontractor Performance Report", styles['Title']))
-    Story.append(Spacer(1, 0.2 * inch))
-    if isinstance(scores, str):
-        Story.append(Paragraph(scores, styles['Normal']))
+app = FastAPI()
+
+API_KEY = "hf_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"  # Replace with your Hugging Face API key
+
+class VendorLog(BaseModel):
+    vendorLogId: str
+    vendorId: str
+    workCompletionPercentage: float
+    qualityPercentage: float
+    incidentSeverity: str
+    delayDays: int
+
+def calculate_scores(log: VendorLog):
+    quality_score = log.qualityPercentage
+    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 = 100.0 if log.incidentSeverity == 'None' else 80.0 if log.incidentSeverity == 'Minor' else 50.0 if log.incidentSeverity == 'Major' else 20.0
+    communication_score = (quality_score + timeliness_score + safety_score) / 3
+    return {
+        'qualityScore': quality_score,
+        'timelinessScore': timeliness_score,
+        'safetyScore': safety_score,
+        'communicationScore': communication_score
+    }
+
+def generate_feedback(score: float, metric: str):
+    if score >= 90:
+        return 'Excellent: Maintain this standard.'
+    elif score >= 70:
+        return 'Good: Continue to improve consistency.'
+    elif score >= 50:
+        return f'Needs Improvement: {metric_feedback(metric)}'
     else:
-        for score_name, score_value in scores:
-            Story.append(Paragraph(f"{score_name}: {score_value:.2f}", styles['Normal']))
-            Story.append(Spacer(1, 0.1 * inch))
-    doc.build(Story)
-    pdf_data = buffer.getvalue()
-    buffer.close()
-    return pdf_data
-
-# 5. Gradio Interface
-iface = gr.Interface(
-    fn=predict_vendor_score,
-    inputs=[],
-    outputs=[
-        gr.HighlightedText(label="Performance Scores"),
-        gr.File(label="Download PDF Report"),
-    ],
-    title="Subcontractor Performance Score Calculator",
-    description="Click the button to retrieve Vendor Log details from Salesforce, calculate performance scores, and generate a PDF report.",
-)
-
-# 6. Launch the Interface
-if __name__ == "__main__":
-    # Set up environment variables for Salesforce credentials
-    os.environ['SALESFORCE_USERNAME'] = 'scores@app.com'  # Replace with your username
-    os.environ['SALESFORCE_PASSWORD'] = 'Internal@1'  # Replace with your password
-    os.environ['SALESFORCE_SECURITY_TOKEN'] = 'NbUKcTx45azba5HEdntE9YAh'  # Replace with your security token
-    iface.launch()
+        return f'Poor: {metric_feedback(metric, True)}'
+
+def metric_feedback(metric: str, poor: bool = False):
+    if metric == 'Timeliness':
+        return 'Significant delays detected; prioritize timely completion.' if poor else 'Maintain schedules to complete tasks on time.'
+    elif metric == 'Safety':
+        return 'Critical safety issues; review safety procedures.' if poor else 'Implement stricter safety protocols.'
+    elif metric == 'Quality':
+        return 'Low quality output; improve quality control.' if poor else 'Enhance work quality standards.'
+    else:
+        return 'Poor communication; improve team interactions.' if poor else 'Enhance coordination with project teams.'
+
+def generate_pdf(vendor_id: str, scores: dict):
+    filename = f'report_{vendor_id}.pdf'
+    c = canvas.Canvas(filename, pagesize=letter)
+    c.setFont('Helvetica', 12)
+    c.drawString(100, 750, f'Vendor Performance Report')
+    c.drawString(100, 730, f'Vendor ID: {vendor_id}')
+    y = 700
+    for metric, score in scores.items():
+        feedback = generate_feedback(score, metric.replace('Score', ''))
+        c.drawString(100, y, f'{metric.replace("Score", " Score")}: {score}% ({feedback})')
+        y -= 20
+    c.drawString(100, y, f'Final Score: {sum(scores.values()) / 4}%')
+    c.save()
+    
+    with open(filename, 'rb') as f:
+        pdf_content = f.read()
+    os.remove(filename)
+    return pdf_content
+
+@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_base64 = base64.b64encode(pdf_content).decode('utf-8')
+    
+    return {
+        'vendorLogId': log.vendorLogId,
+        'qualityScore': scores['qualityScore'],
+        'timelinessScore': scores['timelinessScore'],
+        'safetyScore': scores['safetyScore'],
+        'communicationScore': scores['communicationScore'],
+        'pdfContent': pdf_base64,
+        'pdfUrl': f'/files/report_{log.vendorId}.pdf'
+    }