Create utils.py

utils.py

@@ -0,0 +1,71 @@
+from simple_salesforce import Salesforce
+from reportlab.lib.pagesizes import letter
+from reportlab.platypus import SimpleDocTemplate, Table, TableStyle, Paragraph
+from reportlab.lib import colors
+from reportlab.lib.styles import getSampleStyleSheet
+import pandas as pd
+from datetime import datetime
+import time
+
+def fetch_salesforce_data(sf: Salesforce, query: str, retries=3) -> list:
+    """Fetch data from Salesforce using SOQL query with retry logic."""
+    for attempt in range(retries):
+        try:
+            result = sf.query_all(query)
+            return result["records"]
+        except Exception as e:
+            if attempt == retries - 1:
+                print(f"Error fetching Salesforce data after {retries} attempts: {e}")
+                return []
+            time.sleep(2)  # Wait before retrying
+    return []
+
+def detect_anomalies(log_text: str, anomaly_detector) -> str:
+    """Detect anomalies in log text using Hugging Face model."""
+    try:
+        result = anomaly_detector(log_text, clean_up_tokenization_spaces=True)
+        return result[0]["label"]  # Returns 'POSITIVE' for anomaly, 'NEGATIVE' for normal
+    except Exception as e:
+        print(f"Error detecting anomaly: {e}")
+        return "NEGATIVE"
+
+def generate_pdf_report(df: pd.DataFrame, lab_site: str, equipment_type: str, date_range: list) -> str:
+    """Generate a PDF report from dashboard data."""
+    pdf_file = "LabOps_Report.pdf"
+    doc = SimpleDocTemplate(pdf_file, pagesize=letter)
+    elements = []
+    styles = getSampleStyleSheet()
+    
+    # Title
+    elements.append(Paragraph(f"LabOps Dashboard Report - {datetime.now().strftime('%Y-%m-%d')}", styles["Title"]))
+    
+    # Filters
+    elements.append(Paragraph(f"Lab: {lab_site} | Equipment: {equipment_type} | Date Range: {date_range[0]} to {date_range[1]}", styles["Normal"]))
+    
+    # Data Table
+    data = [["Equipment", "Timestamp", "Status", "Usage Count", "Anomaly"]]
+    for _, row in df.iterrows():
+        timestamp = row["Log_Timestamp__c"].strftime('%Y-%m-%d %H:%M:%S') if pd.notnull(row["Log_Timestamp__c"]) else "N/A"
+        data.append([
+            row["Equipment__c"],
+            timestamp,
+            row["Status__c"],
+            str(row["Usage_Count__c"]),
+            row["Anomaly"]
+        ])
+    
+    table = Table(data)
+    table.setStyle(TableStyle([
+        ("BACKGROUND", (0, 0), (-1, 0), colors.grey),
+        ("TEXTCOLOR", (0, 0), (-1, 0), colors.whitesmoke),
+        ("ALIGN", (0, 0), (-1, -1), "CENTER"),
+        ("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
+        ("FONTSIZE", (0, 0), (-1, 0), 14),
+        ("BOTTOMPADDING", (0, 0), (-1, 0), 12),
+        ("BACKGROUND", (0, 1), (-1, -1), colors.beige),
+        ("GRID", (0, 0), (-1, -1), 1, colors.black)
+    ]))
+    elements.append(table)
+    
+    doc.build(elements)
+    return pdf_file