Files

101 lines
4.2 KiB
Python

"""SLA engine — priority-based deadlines, escalation triggers, and status checks."""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from app.models.ticket import Ticket
# ── Priority SLA Targets (from PRD §11) ──────────────────────────────
# Each entry: (response_window_minutes, resolution_window_hours)
SLA_TARGETS: dict[str, tuple[int, int]] = {
"urgent": (15, 4), # respond 10-15min, resolve 2-4h → use upper bound
"high": (30, 24), # respond 30min, resolve 8-24h
"medium": (240, 72), # respond 4h (240min), resolve 2-3d (72h)
"low": (1440, 168), # respond 1d (1440min), resolve 5-7d (168h)
}
def compute_sla_deadline(priority: str, created_at: datetime | None = None) -> datetime:
"""Return the resolution deadline for a given priority.
Uses the upper-bound resolution target.
"""
now = created_at or datetime.now(timezone.utc)
_response_min, resolution_hours = SLA_TARGETS.get(priority, (240, 72))
return now + timedelta(hours=resolution_hours)
def get_sla_response_window(priority: str) -> timedelta:
"""Return the response-window timedelta for a given priority."""
minutes, _hours = SLA_TARGETS.get(priority, (240, 72))
return timedelta(minutes=minutes)
def get_sla_resolution_window(priority: str) -> timedelta:
"""Return the resolution-window timedelta for a given priority."""
_minutes, hours = SLA_TARGETS.get(priority, (240, 72))
return timedelta(hours=hours)
# ── Escalation detection ─────────────────────────────────────────────
def should_escalate_on_response(ticket: Ticket) -> bool:
"""Return True if the ticket's response SLA has been breached.
A ticket is in breach if its status is still in the pre-acknowledgement
states (New, Logged, Triage, Assigned) beyond the response window.
"""
if not ticket.priority or ticket.created_at is None:
return False
pre_ack_states = {"New", "Logged", "Triage", "Assigned"}
if ticket.status not in pre_ack_states:
return False
response_window = get_sla_response_window(ticket.priority)
deadline = ticket.created_at.replace(tzinfo=timezone.utc) + response_window
return datetime.now(timezone.utc) > deadline
def is_sla_breached(ticket: Ticket) -> bool:
"""Return True if the ticket's resolution SLA deadline has passed."""
if ticket.sla_deadline is None or ticket.status in ("Closed", "Completed"):
return False
deadline = ticket.sla_deadline
if deadline.tzinfo is None:
deadline = deadline.replace(tzinfo=timezone.utc)
return datetime.now(timezone.utc) > deadline
# ── API helper ───────────────────────────────────────────────────────
async def get_sla_status(ticket: Ticket) -> dict:
"""Return a structured SLA status dict for a ticket."""
now = datetime.now(timezone.utc)
created = ticket.created_at
if created and created.tzinfo is None:
created = created.replace(tzinfo=timezone.utc)
response_deadline = None
resolution_deadline = ticket.sla_deadline
if ticket.priority and created:
response_deadline = created + get_sla_response_window(ticket.priority)
response_breached = False
if response_deadline and ticket.status in {"New", "Logged", "Triage", "Assigned"}:
response_breached = now > response_deadline
resolution_breached = False
if resolution_deadline:
rd = resolution_deadline
if rd.tzinfo is None:
rd = rd.replace(tzinfo=timezone.utc)
resolution_breached = now > rd if ticket.status not in ("Closed", "Completed") else False
return {
"priority": ticket.priority,
"response_deadline": response_deadline.isoformat() if response_deadline else None,
"resolution_deadline": resolution_deadline.isoformat() if resolution_deadline else None,
"response_breached": response_breached,
"resolution_breached": resolution_breached,
"current_status": ticket.status,
}