84 lines
3 KiB
Python
84 lines
3 KiB
Python
|
"""Heuristische Notfall-Erkennung und Eskalation (Senioren-Kontext).
|
||
|
|
|
||
|
|
WICHTIG: Schluesselwort-Heuristik, KEIN Ersatz fuer eine echte Klassifikation.
|
||
|
|
Sie kann Notlagen verpassen oder Fehlalarme ausloesen. Die erkannten Textauszuege
|
||
|
|
sind hochsensibel und werden bewusst protokolliert (DSGVO beachten: Einwilligung,
|
||
|
|
Aufbewahrung, Zugriff).
|
||
|
|
"""
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
|
||
|
|
from app.config import settings, Settings
|
||
|
|
from app.metrics import metrics
|
||
|
|
|
||
|
|
# Phrasen je Kategorie (de/en), bewusst eher spezifisch gegen Fehlalarme.
|
||
|
|
_PATTERNS: dict[str, list[str]] = {
|
||
|
|
"medical": [
|
||
|
|
"brustschmerz", "schmerzen in der brust", "kann nicht atmen", "keine luft",
|
||
|
|
"atemnot", "herzinfarkt", "schlaganfall", "bewusstlos", "gestuerzt", "gestürzt",
|
||
|
|
"gefallen und komme nicht hoch", "starke blutung",
|
||
|
|
"chest pain", "can't breathe", "cannot breathe", "heart attack", "stroke",
|
||
|
|
"i fell and can't", "bleeding badly",
|
||
|
|
],
|
||
|
|
"self_harm": [
|
||
|
|
"nicht mehr leben", "mich umbringen", "selbstmord", "suizid", "will sterben",
|
||
|
|
"kill myself", "end my life", "suicide", "want to die",
|
||
|
|
],
|
||
|
|
"help": [
|
||
|
|
"notruf", "notarzt", "krankenwagen", "ruf einen arzt", "es brennt",
|
||
|
|
"call an ambulance", "call 911", "call 112",
|
||
|
|
],
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def detect(text: str):
|
||
|
|
"""Liefert (category, matched_phrase) oder None."""
|
||
|
|
if not text:
|
||
|
|
return None
|
||
|
|
low = text.lower()
|
||
|
|
for category, phrases in _PATTERNS.items():
|
||
|
|
for phrase in phrases:
|
||
|
|
if phrase in low:
|
||
|
|
return category, phrase
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
async def _fire_webhook(url: str, user, category: str, snippet: str) -> None:
|
||
|
|
try:
|
||
|
|
async with httpx.AsyncClient(timeout=5) as client:
|
||
|
|
await client.post(
|
||
|
|
url,
|
||
|
|
json={
|
||
|
|
"user_id": user.id,
|
||
|
|
"display_name": user.display_name,
|
||
|
|
"category": category,
|
||
|
|
"text": snippet,
|
||
|
|
},
|
||
|
|
)
|
||
|
|
except Exception: # noqa: BLE001 - best effort, darf den Chat nicht brechen
|
||
|
|
metrics.inc("emergency_webhook_error_total")
|
||
|
|
|
||
|
|
|
||
|
|
def handle_emergency(user, text: str, store, cfg: Settings = settings):
|
||
|
|
"""Erkennt, protokolliert und eskaliert ein Notfall-Signal.
|
||
|
|
|
||
|
|
Gibt {"category", "matched"} zurueck, wenn etwas erkannt wurde, sonst None.
|
||
|
|
Der Webhook (falls konfiguriert) wird nicht-blockierend ausgeloest.
|
||
|
|
"""
|
||
|
|
match = detect(text)
|
||
|
|
if not match:
|
||
|
|
return None
|
||
|
|
category, phrase = match
|
||
|
|
snippet = text[:500]
|
||
|
|
store.log_emergency(user.id, category, snippet)
|
||
|
|
metrics.inc("emergency_total", {"category": category})
|
||
|
|
if cfg.emergency_webhook_url:
|
||
|
|
try:
|
||
|
|
asyncio.get_running_loop().create_task(
|
||
|
|
_fire_webhook(cfg.emergency_webhook_url, user, category, snippet)
|
||
|
|
)
|
||
|
|
except RuntimeError:
|
||
|
|
pass # kein laufender Event-Loop (z. B. im Test) -> Webhook ueberspringen
|
||
|
|
return {"category": category, "matched": phrase}
|