174 lines
6.0 KiB
Python
174 lines
6.0 KiB
Python
"""
|
|
Urgency analyzer for alerts.
|
|
|
|
Assesses time sensitivity, deadlines, and determines if action can wait.
|
|
"""
|
|
|
|
from typing import Dict, Any
|
|
from datetime import datetime, timedelta, timezone
|
|
import structlog
|
|
|
|
logger = structlog.get_logger()
|
|
|
|
|
|
class UrgencyAnalyzer:
|
|
"""Analyze urgency from event metadata"""
|
|
|
|
def analyze(self, event_type: str, metadata: Dict[str, Any]) -> dict:
|
|
"""
|
|
Analyze urgency for an event.
|
|
|
|
Returns dict with:
|
|
- hours_until_consequence: Time until impact occurs
|
|
- can_wait_until_tomorrow: Boolean
|
|
- deadline_utc: ISO datetime if deadline exists
|
|
- peak_hour_relevant: Boolean
|
|
- hours_pending: Age of alert
|
|
"""
|
|
|
|
urgency = {
|
|
"hours_until_consequence": 24, # Default: 24 hours
|
|
"can_wait_until_tomorrow": True,
|
|
"deadline_utc": None,
|
|
"peak_hour_relevant": False,
|
|
"hours_pending": 0
|
|
}
|
|
|
|
# Calculate based on event type
|
|
if "critical" in event_type or "urgent" in event_type:
|
|
urgency["hours_until_consequence"] = 2
|
|
urgency["can_wait_until_tomorrow"] = False
|
|
|
|
elif "production" in event_type:
|
|
urgency.update(self._analyze_production_urgency(metadata))
|
|
|
|
elif "stock" in event_type or "shortage" in event_type:
|
|
urgency.update(self._analyze_stock_urgency(metadata))
|
|
|
|
elif "delivery" in event_type or "overdue" in event_type:
|
|
urgency.update(self._analyze_delivery_urgency(metadata))
|
|
|
|
elif "po_approval" in event_type:
|
|
urgency.update(self._analyze_po_approval_urgency(metadata))
|
|
|
|
# Check for explicit deadlines
|
|
if "required_delivery_date" in metadata:
|
|
urgency.update(self._calculate_deadline_urgency(metadata["required_delivery_date"]))
|
|
|
|
if "production_date" in metadata:
|
|
urgency.update(self._calculate_deadline_urgency(metadata["production_date"]))
|
|
|
|
if "expected_date" in metadata:
|
|
urgency.update(self._calculate_deadline_urgency(metadata["expected_date"]))
|
|
|
|
return urgency
|
|
|
|
def _analyze_production_urgency(self, metadata: Dict[str, Any]) -> dict:
|
|
"""Analyze urgency for production alerts"""
|
|
urgency = {}
|
|
|
|
delay_minutes = metadata.get("delay_minutes", 0)
|
|
|
|
if delay_minutes > 120:
|
|
urgency["hours_until_consequence"] = 1
|
|
urgency["can_wait_until_tomorrow"] = False
|
|
elif delay_minutes > 60:
|
|
urgency["hours_until_consequence"] = 4
|
|
urgency["can_wait_until_tomorrow"] = False
|
|
else:
|
|
urgency["hours_until_consequence"] = 8
|
|
|
|
# Production is peak-hour sensitive
|
|
urgency["peak_hour_relevant"] = True
|
|
|
|
return urgency
|
|
|
|
def _analyze_stock_urgency(self, metadata: Dict[str, Any]) -> dict:
|
|
"""Analyze urgency for stock alerts"""
|
|
urgency = {}
|
|
|
|
# Hours until needed
|
|
if "hours_until" in metadata:
|
|
urgency["hours_until_consequence"] = metadata["hours_until"]
|
|
urgency["can_wait_until_tomorrow"] = urgency["hours_until_consequence"] > 24
|
|
|
|
# Days until expiry
|
|
elif "days_until_expiry" in metadata:
|
|
days = metadata["days_until_expiry"]
|
|
if days <= 1:
|
|
urgency["hours_until_consequence"] = days * 24
|
|
urgency["can_wait_until_tomorrow"] = False
|
|
else:
|
|
urgency["hours_until_consequence"] = days * 24
|
|
|
|
return urgency
|
|
|
|
def _analyze_delivery_urgency(self, metadata: Dict[str, Any]) -> dict:
|
|
"""Analyze urgency for delivery alerts"""
|
|
urgency = {}
|
|
|
|
days_overdue = metadata.get("days_overdue", 0)
|
|
|
|
if days_overdue > 3:
|
|
urgency["hours_until_consequence"] = 2
|
|
urgency["can_wait_until_tomorrow"] = False
|
|
elif days_overdue > 1:
|
|
urgency["hours_until_consequence"] = 8
|
|
urgency["can_wait_until_tomorrow"] = False
|
|
|
|
return urgency
|
|
|
|
def _analyze_po_approval_urgency(self, metadata: Dict[str, Any]) -> dict:
|
|
"""
|
|
Analyze urgency for PO approval alerts.
|
|
|
|
Uses stockout time (when you run out of stock) instead of delivery date
|
|
to determine true urgency.
|
|
"""
|
|
urgency = {}
|
|
|
|
# Extract min_depletion_hours from reasoning_data.parameters
|
|
reasoning_data = metadata.get("reasoning_data", {})
|
|
parameters = reasoning_data.get("parameters", {})
|
|
min_depletion_hours = parameters.get("min_depletion_hours")
|
|
|
|
if min_depletion_hours is not None:
|
|
urgency["hours_until_consequence"] = max(0, round(min_depletion_hours, 1))
|
|
urgency["can_wait_until_tomorrow"] = min_depletion_hours > 24
|
|
|
|
# Set deadline_utc to when stock runs out
|
|
now = datetime.now(timezone.utc)
|
|
stockout_time = now + timedelta(hours=min_depletion_hours)
|
|
urgency["deadline_utc"] = stockout_time.isoformat()
|
|
|
|
logger.info(
|
|
"po_approval_urgency_calculated",
|
|
min_depletion_hours=min_depletion_hours,
|
|
stockout_deadline=urgency["deadline_utc"],
|
|
can_wait=urgency["can_wait_until_tomorrow"]
|
|
)
|
|
|
|
return urgency
|
|
|
|
def _calculate_deadline_urgency(self, deadline_str: str) -> dict:
|
|
"""Calculate urgency based on deadline"""
|
|
try:
|
|
if isinstance(deadline_str, str):
|
|
deadline = datetime.fromisoformat(deadline_str.replace('Z', '+00:00'))
|
|
else:
|
|
deadline = deadline_str
|
|
|
|
now = datetime.now(timezone.utc)
|
|
time_until = deadline - now
|
|
|
|
hours_until = time_until.total_seconds() / 3600
|
|
|
|
return {
|
|
"deadline_utc": deadline.isoformat(),
|
|
"hours_until_consequence": max(0, round(hours_until, 1)),
|
|
"can_wait_until_tomorrow": hours_until > 24
|
|
}
|
|
except Exception as e:
|
|
logger.warning("deadline_parse_failed", deadline=deadline_str, error=str(e))
|
|
return {}
|