Open raw ↗"""Alert routing engine for HAD Digital MVP.
Grade 1 = routine (timeline only)
Grade 2 = urgent (notification within configurable delay)
Grade 3+ = emergency (immediate notification)
"""
from datetime import datetime, timezone, timedelta
from database import get_db
from audit_logger import log_action
# Role-based alert routing rules
ALERT_ROUTING = {
1: {"type": "routine", "severity": "low", "roles": ["had_nurse"]},
2: {"type": "urgent", "severity": "medium", "roles": ["had_nurse", "oncologist"]},
3: {"type": "emergency", "severity": "high", "roles": ["oncologist", "had_nurse"]},
4: {"type": "emergency", "severity": "critical", "roles": ["oncologist", "had_nurse"]},
5: {"type": "emergency", "severity": "critical", "roles": ["oncologist", "had_nurse"]},
}
class AlertEngine:
"""Routes toxicity grades to appropriate alerts and notifications."""
def __init__(self):
from config_manager import config
self.urgent_delay = config.get("alerting", "urgent_delay_minutes", default=30)
self.emergency_delay = config.get("alerting", "emergency_delay_minutes", default=0)
def create_alert_for_grade(
self,
grade_id: int,
patient_id: int,
report_id: int,
grade: int,
symptom_id: str,
display_name: str,
reporter_id: int,
) -> dict | None:
"""Create an alert based on toxicity grade.
Returns the created alert dict, or None for grade 1 (routine - timeline only).
"""
if grade < 1 or grade > 5:
return None
routing = ALERT_ROUTING.get(grade)
if routing is None:
return None
db = get_db()
# Grade 1: timeline only, no alert
if grade == 1:
self._add_timeline_event(
patient_id=patient_id,
event_type="system",
title=f"Grade 1 {display_name} reported",
description=f"Routine grade 1 toxicity for {symptom_id}. No alert generated.",
created_by=reporter_id,
)
return None
# Build alert
severity = routing["severity"]
alert_type = routing["type"]
if alert_type == "emergency":
title = f"EMERGENCY: Grade {grade} {display_name}"
message = (
f"Patient has grade {grade} {display_name} toxicity. "
f"Immediate clinical assessment required."
)
else:
title = f"URGENT: Grade {grade} {display_name}"
message = (
f"Patient has grade {grade} {display_name} toxicity. "
f"Clinical assessment needed within {self.urgent_delay} minutes."
)
# Determine assigned role
assigned_role = routing["roles"][0] # Primary role
# Find a user of the assigned role for this patient
assigned_user = self._find_assigned_user(patient_id, routing["roles"])
# Create the alert
cursor = db.execute(
"""INSERT INTO alerts
(patient_id, grade_id, report_id, alert_type, severity, title, message,
assigned_to, assigned_role, status)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending')""",
(
patient_id, grade_id, report_id, alert_type, severity,
title, message, assigned_user, assigned_role,
),
)
db.commit()
alert_id = cursor.lastrowid
# Add timeline event
self._add_timeline_event(
patient_id=patient_id,
event_type="alert",
title=title,
description=message,
reference_id=alert_id,
reference_table="alerts",
created_by=reporter_id,
)
# Send notification messages to assigned roles
self._send_alert_notifications(
patient_id=patient_id,
alert_id=alert_id,
title=title,
message=message,
roles=routing["roles"],
alert_type=alert_type,
)
alert = db.fetchone("SELECT * FROM alerts WHERE id = ?", (alert_id,))
return dict(alert) if alert else None
def _find_assigned_user(self, patient_id: int, roles: list[str]) -> int | None:
"""Find a user of the given roles associated with the patient."""
db = get_db()
# First check primary oncologist
patient = db.fetchone(
"SELECT primary_oncologist_id FROM patients WHERE id = ?",
(patient_id,),
)
if patient and patient["primary_oncologist_id"]:
user = db.fetchone(
"SELECT id FROM users WHERE id = ? AND role IN ({})".format(
",".join("?" * len(roles))
),
(patient["primary_oncologist_id"], *roles),
)
if user:
return user["id"]
# Find any active user with matching role
placeholders = ",".join("?" * len(roles))
user = db.fetchone(
f"SELECT id FROM users WHERE role IN ({placeholders}) AND active = 1 LIMIT 1",
tuple(roles),
)
return user["id"] if user else None
def _send_alert_notifications(
self,
patient_id: int,
alert_id: int,
title: str,
message: str,
roles: list[str],
alert_type: str,
):
"""Send notification messages to users of specified roles."""
db = get_db()
placeholders = ",".join("?" * len(roles))
users = db.fetchall(
f"SELECT id FROM users WHERE role IN ({placeholders}) AND active = 1",
tuple(roles),
)
# Use first matching user as sender for system alert
sender_id = users[0]['id'] if users else None
for user in users:
db.execute(
"""INSERT INTO messages
(sender_id, recipient_id, patient_id, subject, body, message_type)
VALUES (?, ?, ?, ?, ?, 'alert_notification')""",
(sender_id, user["id"], patient_id, title, message),
)
db.commit()
def _add_timeline_event(
self,
patient_id: int,
event_type: str,
title: str,
description: str = None,
reference_id: int = None,
reference_table: str = None,
created_by: int = None,
):
"""Add an event to the patient timeline."""
db = get_db()
db.execute(
"""INSERT INTO timeline_events
(patient_id, event_type, title, description, reference_id, reference_table, created_by)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(patient_id, event_type, title, description, reference_id, reference_table, created_by),
)
db.commit()
def acknowledge_alert(self, alert_id: int, user_id: int) -> bool:
"""Acknowledge an alert."""
db = get_db()
alert = db.fetchone("SELECT * FROM alerts WHERE id = ?", (alert_id,))
if alert is None:
return False
db.execute(
"""UPDATE alerts
SET status = 'acknowledged', acknowledged_by = ?,
acknowledged_at = datetime('now'), updated_at = datetime('now')
WHERE id = ?""",
(user_id, alert_id),
)
db.commit()
log_action(
user_id=user_id,
action="acknowledge_alert",
resource_type="alert",
resource_id=alert_id,
details=f"Alert acknowledged for patient {alert['patient_id']}",
)
return True
def resolve_alert(self, alert_id: int, user_id: int) -> bool:
"""Resolve an alert."""
db = get_db()
db.execute(
"""UPDATE alerts
SET status = 'resolved', resolved_at = datetime('now'), updated_at = datetime('now')
WHERE id = ?""",
(alert_id,),
)
db.commit()
return True
def get_alerts(
self,
patient_id: int = None,
status: str = None,
assigned_to: int = None,
limit: int = 50,
) -> list[dict]:
"""Get alerts with optional filters."""
db = get_db()
conditions = []
params = []
if patient_id:
conditions.append("patient_id = ?")
params.append(patient_id)
if status:
conditions.append("status = ?")
params.append(status)
if assigned_to:
conditions.append("assigned_to = ?")
params.append(assigned_to)
where_clause = " AND ".join(conditions) if conditions else "1=1"
rows = db.fetchall(
f"SELECT * FROM alerts WHERE {where_clause} ORDER BY created_at DESC LIMIT ?",
tuple(params) + (limit,),
)
return [dict(r) for r in rows]
def get_pending_alerts_for_user(self, user_id: int) -> list[dict]:
"""Get pending alerts assigned to a user or their role."""
db = get_db()
user = db.fetchone("SELECT role FROM users WHERE id = ?", (user_id,))
if user is None:
return []
rows = db.fetchall(
"""SELECT * FROM alerts
WHERE status = 'pending' AND (assigned_to = ? OR assigned_role = ?)
ORDER BY
CASE severity WHEN 'critical' THEN 0 WHEN 'high' THEN 1
WHEN 'medium' THEN 2 ELSE 3 END,
created_at DESC""",
(user_id, user["role"]),
)
return [dict(r) for r in rows]
# Global singleton
_alert_engine = None
def get_alert_engine() -> AlertEngine:
"""Get the global alert engine instance."""
global _alert_engine
if _alert_engine is None:
_alert_engine = AlertEngine()
return _alert_engine