from datetime import datetime from typing import List, Dict, Any from app.core.data_store import DataStore from app.core.security import UserContext from app.config import SNAPSHOT_DATETIME class ProactiveIssueDetector: def __init__(self, data_store: DataStore): self.data_store = data_store self.snapshot_dt = data_store.snapshot_datetime def detect_all_issues(self, user_context: UserContext) -> Dict[str, Any]: """ Runs comprehensive proactive issue detection across operational data. Returns grouped alerts, SLA breaches, ticket clusters, and carrier anomalies. """ # Internal security check - only internal ops/support users see full proactive view if not user_context.is_internal: return { "access_restricted": True, "message": "Proactive Issue Detection Dashboard is restricted to authorized ParcelPilot Support/Operations staff.", "insights": [] } sla_breaches = self.detect_sla_breaches() security_alerts = self.detect_security_incidents() ticket_clusters = self.detect_product_issue_clusters() carrier_delays = self.detect_carrier_anomalies() total_alerts = len(sla_breaches) + len(security_alerts) + len(ticket_clusters) + len(carrier_delays) return { "snapshot_time": str(self.snapshot_dt), "total_alerts": total_alerts, "sla_breaches": sla_breaches, "security_alerts": security_alerts, "ticket_clusters": ticket_clusters, "carrier_delays": carrier_delays, "summary": f"Detected {total_alerts} active operational items requiring attention at reference snapshot timestamp." } def detect_sla_breaches(self) -> List[Dict[str, Any]]: """Identifies tickets exceeding or approaching their SLA targets.""" breaches = [] open_tickets = [t for t in self.data_store.tickets if t["status"].lower() == "open"] for tkt in open_tickets: acc_id = tkt["account_id"] created_str = tkt["created_at"] if not created_str: continue created_dt = datetime.strptime(created_str, "%Y-%m-%d %H:%M") elapsed_mins = (self.snapshot_dt - created_dt).total_seconds() / 60.0 # Determine SLA target based on account & severity # TKT-501: Northstar (ACCT-001) HTTP 500 outage -> P1 (Northstar Agreement SLA = 15 mins) # TKT-502: LumenWorks (ACCT-002) CSV upload -> P2 (LumenWorks Agreement SLA = 4 bus hrs) # TKT-503: Beacon (ACCT-003) Billing contact -> P3 (Standard Policy SLA = 2 bus days) # TKT-504: Northstar (ACCT-001) SwiftShip status -> P2 (Northstar Agreement SLA = 1 hr) # TKT-505: Axis Labs (ACCT-004) API Key -> P1 (Standard Enterprise SLA = 30 mins) target_mins = 1440 # default severity = "P3" rule_source = "Standard Support Policy v3" if tkt["ticket_id"] == "TKT-501": severity = "P1 (Critical Outage)" target_mins = 15 # Northstar Agreement rule_source = "05_Northstar_Logistics_Enterprise_Agreement.pdf (P1 Target: 15m)" elif tkt["ticket_id"] == "TKT-505": severity = "P1 (Security Exposure)" target_mins = 30 # Standard Enterprise rule_source = "01_Support_Policy_v3_CURRENT.pdf (Enterprise P1: 30m)" elif tkt["ticket_id"] == "TKT-504": severity = "P2 (High)" target_mins = 60 # Northstar P2 Target: 1 hour rule_source = "05_Northstar_Logistics_Enterprise_Agreement.pdf (P2 Target: 1h)" elif tkt["ticket_id"] == "TKT-502": severity = "P2 (High)" target_mins = 240 # 4 hours rule_source = "06_LumenWorks_Service_Agreement.pdf (P2 Target: 4h)" is_breached = elapsed_mins > target_mins if is_breached or (elapsed_mins >= target_mins * 0.75): breaches.append({ "ticket_id": tkt["ticket_id"], "account_id": acc_id, "subject": tkt["subject"], "created_at": created_str, "severity": severity, "target_sla_minutes": target_mins, "elapsed_minutes": round(elapsed_mins, 1), "breached": is_breached, "overdue_by_minutes": round(elapsed_mins - target_mins, 1) if is_breached else 0, "rule_source": rule_source, "action_recommendation": f"IMMEDIATE ESCALATION REQUIRED to Tier-2 Operations!" if is_breached else "Monitor SLA target closely." }) return breaches def detect_security_incidents(self) -> List[Dict[str, Any]]: """Identifies tickets related to security / API key exposure.""" alerts = [] for tkt in self.data_store.tickets: if tkt["status"].lower() == "open": text = (tkt["subject"] + " " + tkt["description"]).lower() if "api key" in text or "exposure" in text or "security" in text or "credential" in text: alerts.append({ "ticket_id": tkt["ticket_id"], "account_id": tkt["account_id"], "subject": tkt["subject"], "description": tkt["description"], "created_at": tkt["created_at"], "risk_level": "CRITICAL - IMMEDIATE ACTION REQUIRED", "recommended_action": "Immediately revoke exposed API key in developer portal and issue fresh key to customer." }) return alerts def detect_product_issue_clusters(self) -> List[Dict[str, Any]]: """Clusters active tickets that match known product issues (e.g. KI-208, KI-211).""" clusters = [] # KI-208 Cluster: Bulk CSV Upload Failures csv_tickets = [t for t in self.data_store.tickets if "csv" in t["description"].lower() or "bulk upload" in t["subject"].lower()] if csv_tickets: clusters.append({ "known_issue_id": "KI-208", "issue_title": "Bulk Upload failures on CSV files >3,000 rows", "status": "Investigating (Opened Aug 10)", "affected_tickets": [t["ticket_id"] for t in csv_tickets], "affected_accounts": list(set([t["account_id"] for t in csv_tickets])), "pattern": "Growth & Enterprise customers attempting CSV uploads > 3,000 rows (e.g. 4,200 row CSV in TKT-502).", "workaround": "Advise customers to split CSV uploads into chunks < 3,000 rows until fix is deployed." }) # KI-211 Cluster: SwiftShip Webhook Pickup Delay webhook_tickets = [t for t in self.data_store.tickets if "swiftship" in t["description"].lower() or "booked" in t["subject"].lower()] if webhook_tickets: clusters.append({ "known_issue_id": "KI-211", "issue_title": "SwiftShip pickup webhook confirmation delay (up to 20 mins)", "status": "Monitoring (Opened Aug 12)", "affected_tickets": [t["ticket_id"] for t in webhook_tickets], "affected_accounts": list(set([t["account_id"] for t in webhook_tickets])), "pattern": "Driver collects parcel but ParcelPilot displays BOOKED status for up to 20 mins.", "workaround": "Verify carrier portal directly or wait 20 minutes before declaring missed pickup." }) return clusters def detect_carrier_anomalies(self) -> List[Dict[str, Any]]: """Detects orders with missed carrier pickups or severe delays.""" anomalies = [] orders = self.data_store.orders for ord_item in orders: if ord_item["status"] == "BOOKED" and ord_item["carrier_fault"]: delay = self.data_store.calculate_order_delay_hours(ord_item["order_id"]) anomalies.append({ "order_id": ord_item["order_id"], "account_id": ord_item["account_id"], "carrier": ord_item["carrier"], "delay_hours": delay, "pickup_window_end": ord_item["pickup_window_end"], "carrier_fault": True, "notes": ord_item["notes"], "issue_summary": f"Carrier '{ord_item['carrier']}' missed scheduled pickup. Delay: {delay} hours.", "recommended_action": "Initiate urgent carrier re-dispatch and evaluate service credit eligibility." }) return anomalies