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 import logging # Configure logging logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") logger = logging.getLogger(__name__) def fetch_salesforce_data(sf: Salesforce, query: str, retries=3) -> list: """Fetch data from Salesforce using SOQL query with retry logic.""" logger.info(f"Executing SOQL query: {query}") for attempt in range(retries): try: result = sf.query_all(query) logger.info(f"Successfully fetched {len(result['records'])} records.") return result["records"] except Exception as e: logger.error(f"Attempt {attempt + 1}/{retries} failed: {e}") if attempt == retries - 1: logger.error(f"Failed to fetch Salesforce data after {retries} attempts.") return [] time.sleep(2) return [] def detect_anomalies(log_text: str, anomaly_detector) -> str: """Detect anomalies in log text using Hugging Face model.""" logger.info(f"Detecting anomaly for text: {log_text}") try: result = anomaly_detector(log_text, clean_up_tokenization_spaces=True) label = result[0]["label"] logger.info(f"Anomaly detection result: {label}") return label except Exception as e: logger.error(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.""" logger.info("Generating PDF report...") try: pdf_file = "/tmp/LabOps_Report.pdf" doc = SimpleDocTemplate(pdf_file, pagesize=letter) elements = [] styles = getSampleStyleSheet() elements.append(Paragraph(f"LabOps Dashboard Report - {datetime.now().strftime('%Y-%m-%d')}", styles["Title"])) elements.append(Paragraph(f"Lab: {lab_site} | Equipment: {equipment_type} | Date Range: {date_range[0]} to {date_range[1]}", styles["Normal"])) 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) logger.info(f"PDF report generated: {pdf_file}") return pdf_file except Exception as e: logger.error(f"Failed to generate PDF report: {e}") raise