root
65622de7a6
- Ticket CRUD with full 15-status lifecycle enforcement
- Auto-generated ticket numbers (PAV-YYYY-NNNNN)
- Status transitions validated and auto-logged to timeline
- Reopen with 7-day window + auto-escalation to Ama
- SLA engine: priority-based deadlines and breach detection
- Photo uploads to uploads/ with multipart support
- Category system seeded from PRD §7 taxonomy (20 top-level)
- Flat and tree category listing endpoints
- SLA status check endpoint (/api/tickets/{id}/sla)
- Paginated ticket listing with filters
104 lines
4.3 KiB
Python
104 lines
4.3 KiB
Python
"""SLA engine — priority-based deadlines, escalation triggers, and status checks."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
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,
|
|
}
|