File size: 2,840 Bytes
114cd9c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 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 |