|
|
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 |
|
|
|
|
|
|
|
|
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 |
