Update app.py

app.py

@@ -8,18 +8,18 @@ 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  # Import the io module
+import io
+from simple_salesforce import Salesforce
+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
+model_path = 'your_model.pkl'  # Make sure this path is correct
 scaler_path = 'scaler.pkl'  # If you used a scaler
-# encoder_path = 'encoder.pkl'  # If you used an encoder
 
 try:
     model = joblib.load(model_path)
     scaler = joblib.load(scaler_path) if scaler_path else None
-    # encoder = joblib.load(encoder_path) if encoder_path else None
 except Exception as e:
     print(f"Error loading model or artifacts: {e}")
     model = None
@@ -27,12 +27,10 @@ except Exception as e:
 # 2. Preprocessing Function
 def preprocess_data(data: Dict[str, Any]) -> pd.DataFrame:
     """
-    Preprocesses input data for the model.  This version is adapted to use
-    the Vendor_Log__c fields directly.
+    Preprocesses input data for the model.
 
     Args:
-        data (dict): A dictionary containing the input data from the Gradio interface,
-                      matching Vendor_Log__c fields.
+        data (dict): A dictionary containing the input data.
 
     Returns:
         pandas.DataFrame: A DataFrame ready for model prediction.
@@ -40,36 +38,30 @@ def preprocess_data(data: Dict[str, Any]) -> pd.DataFrame:
     try:
         df = pd.DataFrame([data])
 
-        # Expected fields based on Vendor_Log__c.  Adjust *EXACTLY* to match your Salesforce field names.
         expected_fields = [
-            'Quality_Report__c',  # Long Text Area
-            'Delay_Days__c',      # Number
-            'Incident_Log__c',    # Long Text Area
-            'Vendor__c',          # Lookup (Important: Send the Salesforce ID, not the name)
-            'Work_Details__c',    # Long Text Area
-            'Work_Completion_Date__c',  # Date
-            'Actual_Completion_Date__c'  # Date
+            'Quality_Report__c',
+            'Delay_Days__c',
+            'Incident_Log__c',
+            'Vendor__c',
+            'Work_Details__c',
+            'Work_Completion_Date__c',
+            'Actual_Completion_Date__c'
         ]
 
-        # 1. Check for missing fields
         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)}")
 
-        # 2. Data Transformations (Adapt to your needs)
-        # Example: Scaling
         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']])
 
-        # 3. Handle Dates (Example -  you might need more complex logic)
         if 'Work_Completion_Date__c' in df.columns:
             df['Work_Completion_Date__c'] = pd.to_datetime(df['Work_Completion_Date__c'])
-            #  Example: Extract year, month, day.  Include if your model uses these.
             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'])  # remove original date
+            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'])
@@ -78,30 +70,20 @@ def preprocess_data(data: Dict[str, Any]) -> pd.DataFrame:
             df['Actual_Completion_Day'] = df['Actual_Completion_Date__c'].dt.day
             df = df.drop(columns=['Actual_Completion_Date__c'])
 
-        # 4.  Text Handling (Example -  for long text areas)
-        # if 'Work_Details__c' in df.columns:
-        #    df['Work_Details__c'] = df['Work_Details__c'].fillna('')
-        #     df['Work_Details_Length'] = df['Work_Details__c'].apply(len) #example feature
-
         df = df[expected_fields]
         return df
     except ValueError as ve:
         print(f"Error in preprocess_data: {ve}")
-        return pd.DataFrame()  # Return empty DataFrame on error
+        return pd.DataFrame()
     except Exception as e:
         print(f"Error in preprocess_data: {e}")
         return pd.DataFrame()
 
+
 # 3. Prediction Function
-def predict_vendor_score(*args: Any) -> Tuple[List[Tuple[str, float]], bytes]:
+def predict_vendor_score() -> Tuple[List[Tuple[str, float]], bytes]:
     """
-    Predicts the vendor performance score based on the input data from Vendor_Log__c.
-    Returns a list of tuples, where each tuple contains the score name and its value,
-    and the PDF data as bytes.
-
-    Args:
-        *args: Input values from the Gradio interface, in the order matching
-                the Vendor_Log__c fields.
+    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.
@@ -116,37 +98,65 @@ def predict_vendor_score(*args: Any) -> Tuple[List[Tuple[str, float]], bytes]:
             ("Final Score (%)", 0.0),
         ], generate_pdf(error_message)
 
-    # 1. Prepare input data as a dictionary, mapping to Vendor_Log__c fields
-    input_data = {
-        'Quality_Report__c': args[0],
-        'Delay_Days__c': args[1],
-        'Incident_Log__c': args[2],
-        'Vendor__c': args[3],
-        'Work_Details__c': args[4],
-        'Work_Completion_Date__c': args[5],
-        'Actual_Completion_Date__c': args[6]
-    }
-
-    # 2. Preprocess the data
-    processed_df = preprocess_data(input_data)
-    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)  # Return empty PDF
-
-    # 3. Make Prediction
     try:
-        predictions = model.predict(processed_df)[0]  # Get the first element
-
-        # 4.  Output:  Map model output to Subcontractor_Performance_Score__c fields
-        #   IMPORTANT:  Adjust this mapping based on your model's output.
-        #   This is a *crucial* step where you tell Gradio how to interpret
-        #   the numbers coming from your model.
+        # 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),
@@ -158,8 +168,8 @@ def predict_vendor_score(*args: Any) -> Tuple[List[Tuple[str, float]], bytes]:
         return output_data, pdf_data
 
     except Exception as e:
-        error_message = f"Error during prediction: {e}"
-        print(f"Error during prediction: {e}")
+        error_message = f"Error: {e}"
+        print(error_message)
         return [
             ("Quality Score (%)", 0.0),
             ("Timeliness Score (%)", 0.0),
@@ -168,6 +178,9 @@ def predict_vendor_score(*args: Any) -> Tuple[List[Tuple[str, float]], bytes]:
             ("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.
 
@@ -176,48 +189,40 @@ def generate_pdf(scores: List[Tuple[str, float]]) -> bytes:
     Returns:
         bytes: The PDF data as bytes.
     """
-    buffer = io.BytesIO()  # Use BytesIO for in-memory PDF generation
+    buffer = io.BytesIO()
     doc = SimpleDocTemplate(buffer, pagesize=letter)
     styles = getSampleStyleSheet()
     Story = []
-
-    # Add a title
     Story.append(Paragraph("Subcontractor Performance Report", styles['Title']))
     Story.append(Spacer(1, 0.2 * inch))
-
-    if isinstance(scores, str):  # Check if it is an error message
+    if isinstance(scores, str):
         Story.append(Paragraph(scores, styles['Normal']))
     else:
-        # Add the scores to the PDF
         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
 
-# 4. Gradio Interface
+# 5. Gradio Interface
 iface = gr.Interface(
     fn=predict_vendor_score,
-    inputs=[
-        gr.Textbox(label="Quality Report (Long Text)", type="text"),
-        gr.Number(label="Delay Days (Number)"),
-        gr.Textbox(label="Incident Log (Long Text)", type="text"),
-        gr.Textbox(label="Vendor ID (Text)"),  # Send ID
-        gr.Textbox(label="Work Details (Long Text)", type="text"),
-        gr.DateTime(label="Work Completion Date", type="datetime"),  # Corrected to "datetime"
-        gr.DateTime(label="Actual Completion Date", type="datetime"),  # Corrected to "datetime"
-    ],
-    outputs=[  # Changed to a list
+    inputs=[],
+    outputs=[
         gr.HighlightedText(label="Performance Scores"),
         gr.File(label="Download PDF Report"),
     ],
     title="Subcontractor Performance Score Calculator",
-    description="Enter Vendor Log details to calculate performance scores and generate a PDF report.",
+    description="Click the button to retrieve Vendor Log details from Salesforce, calculate performance scores, and generate a PDF report.",
+    buttons=[gr.Button("Get Scores from Salesforce")]
 )
 
-# 5. Launch the Interface
+# 6. Launch the Interface
 if __name__ == "__main__":
-    iface.launch()
+    # 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()