feat(1b): Automatische Notfallerkennung raus → manueller Notruf-Knopf

Die unzuverlässige/intransparente Auto-Erkennung (Stichwörter + LLM) ist
ersatzlos entfernt: emergency.py neu (nur manueller, vom Nutzer bestätigter
Notruf), llm_classifier.py gelöscht, Aufrufe in chat.py/ws.py raus,
emergency_llm_* aus der Config.

Neu: prominenter "🆘 Hilfe rufen"-Knopf (Bestätigung) → POST /api/emergency
→ protokolliert + liefert Hinweis in Nutzersprache. Eskalation (Benachrichtigung
an Angehörige) folgt später; vorerst Platzhalter "ACHTUNG: Es ist noch keine
Benachrichtigung eingebaut." (10 Sprachen). Admin-Tab "Notfälle" → "Notrufe".

Tests: test_emergency_llm.py entfernt, test_quota_safety.py auf Notruf-Endpunkt
umgeschrieben. 208 passed. Live verifiziert.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Dieter Schlüter 2026-06-25 03:02:52 +02:00
commit 3389f75af0
10 changed files with 112 additions and 433 deletions

View file

@ -1,140 +1,41 @@
"""Notfall-Erkennung und Eskalation (Senioren-Kontext).
"""Manuell ausgelöster Notruf — vom Nutzer bewusst bestätigt.
Zweistufig:
1. Schnelle Stichwort-Heuristik (`detect` / `handle_emergency`) im Hot-Path -> 0 Latenz.
2. LLM-Klassifikation (`schedule_llm_emergency_check`) als Hintergrund-Task, der NUR
laeuft, wenn die Heuristik nichts fand -> faengt verpasste Formulierungen, ohne die
Antwortlatenz zu erhoehen.
Es gibt KEINE automatische Notfallerkennung (keine Stichwörter, kein LLM): die
wäre unzuverlässig (Fehlalarme wie verpasste Notfälle) und intransparent. Ein
Notruf entsteht nur durch eine ausdrückliche Nutzeraktion (Knopf + Bestätigung).
Die erkannten Textauszuege sind hochsensibel und werden bewusst protokolliert
(DSGVO beachten: Einwilligung, Aufbewahrung, Zugriff).
Die tatsächliche Benachrichtigung von Angehörigen (SMS/Anruf/Webhook) ist noch
NICHT implementiert. Bis dahin wird der Notruf protokolliert und der Nutzer
erhält einen klaren Hinweis in seiner Sprache.
"""
import asyncio
import logging
import httpx
from app.config import settings, Settings
from app.metrics import metrics
logger = logging.getLogger(__name__)
# Referenzen auf laufende Hintergrund-Tasks halten (sonst GC-gefaehrdet).
_pending: set[asyncio.Task] = set()
# 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",
],
# Platzhalter-Hinweis je Sprache, solange die Eskalation noch nicht gebaut ist.
_NO_NOTIFY_NOTICE: dict[str, str] = {
"de": "ACHTUNG: Es ist noch keine Benachrichtigung eingebaut.",
"en": "ATTENTION: No notification has been set up yet.",
"fr": "ATTENTION : aucune notification n'est encore configurée.",
"es": "ATENCIÓN: todavía no se ha configurado ninguna notificación.",
"it": "ATTENZIONE: non è ancora stata configurata alcuna notifica.",
"pt": "ATENÇÃO: ainda não há nenhuma notificação configurada.",
"pl": "UWAGA: Nie skonfigurowano jeszcze żadnego powiadomienia.",
"ar": "تنبيه: لم يتم إعداد أي إشعار بعد.",
"ru": "ВНИМАНИЕ: Уведомление пока не настроено.",
"zh": "注意:尚未设置任何通知。",
}
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
def notify_notice(language: str | None) -> str:
"""Liefert den Platzhalter-Hinweis in der Nutzersprache (Fallback Deutsch)."""
return _NO_NOTIFY_NOTICE.get((language or "de").lower(), _NO_NOTIFY_NOTICE["de"])
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 record_manual_emergency(user, store, language: str | None = None) -> dict:
"""Protokolliert einen vom Nutzer bestätigten Notruf und liefert den Hinweis.
def _escalate(user, category: str, snippet: str, store, cfg: Settings, source: str) -> None:
"""Protokolliert + eskaliert einen erkannten Notfall (Log, Metrik, Webhook)."""
store.log_emergency(user.id, category, snippet)
metrics.inc("emergency_total", {"category": category, "source": source})
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
def handle_emergency(user, text: str, store, cfg: Settings = settings):
"""Stufe 1: Stichwort-Heuristik. Erkennt, protokolliert und eskaliert sofort.
Gibt {"category", "matched"} zurueck, wenn etwas erkannt wurde, sonst None.
Der Webhook (falls konfiguriert) wird nicht-blockierend ausgeloest.
Eskalation (Benachrichtigung) folgt später; hier wird nur protokolliert.
"""
match = detect(text)
if not match:
return None
category, phrase = match
_escalate(user, category, text[:500], store, cfg, source="keyword")
return {"category": category, "matched": phrase}
async def _llm_check(user, text: str, store, cfg: Settings, on_emergency) -> None:
"""Hintergrund: LLM-Klassifikation + Eskalation (best-effort)."""
try:
from app.safety.llm_classifier import classify_emergency
result = await classify_emergency(text, cfg)
if not result:
return
category = result["category"]
_escalate(user, category, text[:500], store, cfg, source="llm")
logger.info(
"llm-emergency: %s (conf=%.2f) fuer %s", category, result["confidence"], user.id
)
if on_emergency is not None:
await on_emergency(category)
except Exception: # best-effort: darf den Turn nie brechen
logger.exception("llm-emergency-check fehlgeschlagen (ignoriert)")
def schedule_llm_emergency_check(user, text: str, store, keyword_hit,
cfg: Settings = settings, on_emergency=None):
"""Stufe 2: plant die LLM-Klassifikation als Hintergrund-Task.
Laeuft NUR, wenn die Stichwort-Heuristik nichts fand (`keyword_hit` ist None) und
die LLM-Stufe aktiviert ist. `on_emergency(category)` (async) wird bei Treffer
aufgerufen (z. B. WS-Event). Gibt den Task zurueck oder None - blockiert nie.
"""
if keyword_hit is not None or not cfg.emergency_llm_enabled:
return None
if not text or not text.strip():
return None
try:
task = asyncio.get_running_loop().create_task(
_llm_check(user, text, store, cfg, on_emergency)
)
except RuntimeError:
return None # kein laufender Event-Loop (z. B. Test) -> ueberspringen
_pending.add(task)
task.add_done_callback(_pending.discard)
return task
store.log_emergency(user.id, "manual", "Vom Nutzer ausgelöster Notruf")
metrics.inc("emergency_total", {"category": "manual", "source": "user"})
return {"category": "manual", "notice": notify_notice(language)}