Spaces:
Configuration error
Configuration error
File size: 8,912 Bytes
2567e7e | 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 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 | 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
|