"""Resolution tracking — audit trail for leak resolution actions.
Tracks who resolved what, when, and with what notes. Also monitors
for recurring leaks (same detector + source reappearing after resolution).
Public API:
record_resolution(leak_id, user_id, notes)
get_resolution_history(leak_id)
get_recurring_leaks(company_id)
get_resolution_stats(company_id)
"""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional
from app.models import RevenueLeak
__all__ = [
"record_resolution",
"get_resolution_history",
"get_recurring_leaks",
"get_resolution_stats",
]
def record_resolution(
leak_id: str,
user_id: str,
notes: str,
) -> Dict[str, Any]:
"""Record a resolution action for a leak.
Appends to the resolution_history in metadata_json without overwriting
existing entries. Returns the updated leak data.
"""
from app.models import db
leak = db.session.get(RevenueLeak, leak_id)
if leak is None:
return {"error": "Leak not found"}
metadata = leak.metadata_json or {}
history = metadata.get("resolution_history") or []
# Append new resolution entry
entry = {
"resolved_at": datetime.now(timezone.utc).isoformat(),
"resolved_by": user_id,
"notes": notes,
}
history.append(entry)
metadata["resolution_history"] = history
# Update leak fields
leak.metadata_json = metadata
leak.resolved = True
leak.resolved_at = datetime.now(timezone.utc)
leak.resolution_notes = notes
db.session.commit()
return {
"success": True,
"leak_id": leak_id,
"resolution_count": len(history),
}
def get_resolution_history(leak_id: str) -> List[Dict[str, Any]]:
"""Get the full resolution history for a leak."""
from app.models import db
leak = db.session.get(RevenueLeak, leak_id)
if leak is None:
return []
return (
(leak.metadata_json or {}).get("resolution_history") or []
)
def get_recurring_leaks(
company_id: str,
days: int = 90,
) -> List[Dict[str, Any]]:
"""Find leaks that have reappeared after being resolved.
A 'recurring leak' is when the same detector_id + source combo
shows up multiple times within the time window.
"""
from app.models import db
cutoff = datetime.now(timezone.utc) - timedelta(days=days)
leaks = RevenueLeak.query.filter(
RevenueLeak.company_id == company_id,
RevenueLeak.detection_type == "auto",
RevenueLeak.detected_at >= cutoff,
).all()
# Group by detector_id + source
groups: Dict[str, List[RevenueLeak]] = {}
for leak in leaks:
key = f"{leak.detector_id}:{leak.source}"
groups.setdefault(key, []).append(leak)
# Find recurring (2+ occurrences)
recurring = []
for key, group_leaks in groups.items():
if len(group_leaks) >= 2:
# Count resolved vs unresolved
resolved_count = sum(1 for l in group_leaks if l.resolved)
unresolved_count = len(group_leaks) - resolved_count
total_loss = sum(
l.estimated_loss or 0 for l in group_leaks
)
# Determine highest severity in the group
severity_order = {"critical": 4, "high": 3, "medium": 2, "low": 1}
worst_severity = max(
(l.severity for l in group_leaks if l.severity),
key=lambda s: severity_order.get(s, 0),
default="medium",
)
recurring.append({
"detector_id": group_leaks[0].detector_id,
"source": group_leaks[0].source,
"severity": worst_severity,
"count": len(group_leaks),
"occurrence_count": len(group_leaks),
"resolved_count": resolved_count,
"unresolved_count": unresolved_count,
"total_loss": total_loss,
"first_detected": group_leaks[0].detected_at.isoformat(),
"last_detected": group_leaks[-1].detected_at.isoformat(),
"leak_ids": [l.id for l in group_leaks],
})
# Sort by occurrence count descending
recurring.sort(key=lambda r: r["occurrence_count"], reverse=True)
return recurring
def get_resolution_stats(
company_id: str,
days: int = 30,
) -> Dict[str, Any]:
"""Get resolution statistics for a company.
Returns:
- total_detected: total leaks detected in period
- resolved_count: leaks marked resolved
- avg_resolution_time: average hours from detection to resolution
- recurring_count: unique detector+source combos with 2+ occurrences
- resolution_rate: percentage of leaks resolved
- top_detectors: most frequent detectors
"""
from app.models import db
from datetime import timedelta
cutoff = datetime.now(timezone.utc) - timedelta(days=days)
leaks = RevenueLeak.query.filter(
RevenueLeak.company_id == company_id,
RevenueLeak.detected_at >= cutoff,
).all()
total = len(leaks)
if total == 0:
return {
"total_detected": 0,
"resolved_count": 0,
"resolution_rate": 0.0,
"avg_resolution_time_hours": None,
"recurring_count": 0,
"top_detectors": [],
"severity_breakdown": {},
}
# Resolved count
resolved = [l for l in leaks if l.resolved]
resolved_count = len(resolved)
# Average resolution time
resolution_times = []
for l in resolved:
if l.resolved_at and l.detected_at:
delta = l.resolved_at - l.detected_at
resolution_times.append(delta.total_seconds() / 3600)
avg_resolution_hours = None
if resolution_times:
avg_resolution_hours = round(
sum(resolution_times) / len(resolution_times), 1
)
# Recurring leaks
recurring = get_recurring_leaks(company_id, days)
recurring_count = len(recurring)
# Top detectors
detector_counts: Dict[str, int] = {}
for l in leaks:
det_id = l.detector_id or "manual"
detector_counts[det_id] = detector_counts.get(det_id, 0) + 1
top_detectors = sorted(
detector_counts.items(), key=lambda x: x[1], reverse=True
)[:5]
# Severity breakdown
severity_counts: Dict[str, int] = {}
for l in leaks:
sev = l.severity or "unknown"
severity_counts[sev] = severity_counts.get(sev, 0) + 1
return {
"total_detected": total,
"resolved_count": resolved_count,
"resolution_rate": round(resolved_count / total * 100, 1) if total > 0 else 0,
"avg_resolution_time_hours": avg_resolution_hours,
"recurring_count": recurring_count,
"top_detectors": top_detectors,
"severity_breakdown": severity_counts,
}