utils.py

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