"""Manuell ausgelöster Notruf — vom Nutzer bewusst bestätigt. KEINE automatische Erkennung (zu unzuverlässig/intransparent). Ein Notruf entsteht nur durch eine ausdrückliche Nutzeraktion (Knopf + Bestätigung). Benachrichtigung von Kontaktpersonen über zwei unabhängige Kanäle (best effort): - E-Mail per SMTP (settings.smtp_*, Empfänger settings.emergency_contact_email). - SMS/Anruf per provider-agnostischem Webhook (settings.emergency_webhook_url). Die Meldungen werden mit den Profildaten der auslösenden Person (Name, Adresse, Telefon, med. Hinweise) und — falls vom Gerät mitgeschickt — der Geoposition angereichert. Kanal-Rollen: die E-Mail trägt das Voll-Dossier, die SMS den Karten-Link, der Anruf nennt Name + Wohnadresse gesprochen und verweist für den genauen Standort auf SMS/E-Mail (Koordinaten werden NICHT vorgelesen). Schlägt die Geo-Erfassung fehl, wird der Grund (geo_error) in E-Mail und Log ausgewiesen. """ import asyncio import logging import smtplib from datetime import datetime, timezone from email.message import EmailMessage import httpx from app.config import settings from app.metrics import metrics logger = logging.getLogger(__name__) # Hinweis je Sprache, wenn (noch) KEINE Benachrichtigung rausging. _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": "注意：尚未设置任何通知。", } # Hinweis je Sprache, wenn mindestens ein Kanal die Meldung rausgeschickt hat. _SENT_NOTICE: dict[str, str] = { "de": "Hilfe wurde verständigt. Eine Kontaktperson wurde benachrichtigt.", "en": "Help has been alerted. A contact person was notified.", "fr": "Les secours ont été alertés. Un proche a été prévenu.", "es": "Se ha avisado a ayuda. Se notificó a un contacto.", "it": "L'aiuto è stato allertato. Una persona di contatto è stata avvisata.", "pt": "A ajuda foi avisada. Uma pessoa de contato foi notificada.", "pl": "Pomoc została powiadomiona. Osoba kontaktowa została powiadomiona.", "ar": "تم تنبيه المساعدة. تم إخطار شخص الاتصال.", "ru": "Помощь оповещена. Контактное лицо уведомлено.", "zh": "已通知求助。已通知联系人。", } # Hinweis je Sprache, wenn GAR KEIN Kanal nutzbar ist (keine Kontakte/kein Webhook # konfiguriert) — niemand wurde erreicht. Handlungsleitend: selbst den Notruf rufen. _NO_RECIPIENTS_NOTICE: dict[str, str] = { "de": "Es konnte niemand benachrichtigt werden – es sind keine Notfall-Kontakte eingerichtet. Bitte rufen Sie selbst den Notruf 112 an.", "en": "No one could be notified – no emergency contacts are set up. Please call your emergency number (112) yourself.", "fr": "Personne n'a pu être prévenu – aucun contact d'urgence n'est configuré. Veuillez appeler vous-même le 112.", "es": "No se pudo avisar a nadie: no hay contactos de emergencia configurados. Por favor, llame usted mismo al 112.", "it": "Non è stato possibile avvisare nessuno: nessun contatto di emergenza è configurato. Per favore chiami lei stesso il 112.", "pt": "Não foi possível avisar ninguém – não há contactos de emergência configurados. Por favor, ligue você mesmo para o 112.", "pl": "Nie udało się nikogo powiadomić – nie skonfigurowano kontaktów alarmowych. Proszę samodzielnie zadzwonić pod numer 112.", "ar": "تعذّر إخطار أي شخص – لا توجد جهات اتصال للطوارئ. يرجى الاتصال بنفسك برقم الطوارئ 112.", "ru": "Никого не удалось уведомить — экстренные контакты не настроены. Пожалуйста, позвоните в службу спасения 112 сами.", "zh": "无法通知任何人——尚未设置紧急联系人。请您自行拨打急救电话 112。", } # Hinweis je Sprache, wenn Kanäle KONFIGURIERT waren, der Versand aber scheiterte # (SMTP-/Webhook-Fehler). Unterscheidet den echten Zustellfehler vom Fehlen von Kontakten. _DELIVERY_FAILED_NOTICE: dict[str, str] = { "de": "Die Benachrichtigung konnte nicht zugestellt werden. Bitte versuchen Sie es erneut oder rufen Sie selbst den Notruf 112 an.", "en": "The notification could not be delivered. Please try again or call your emergency number (112) yourself.", "fr": "La notification n'a pas pu être transmise. Veuillez réessayer ou appeler vous-même le 112.", "es": "No se pudo entregar la notificación. Por favor, inténtelo de nuevo o llame usted mismo al 112.", "it": "Non è stato possibile recapitare la notifica. Per favore riprovi o chiami lei stesso il 112.", "pt": "Não foi possível entregar a notificação. Por favor, tente novamente ou ligue você mesmo para o 112.", "pl": "Nie udało się dostarczyć powiadomienia. Proszę spróbować ponownie lub samodzielnie zadzwonić pod numer 112.", "ar": "تعذّر تسليم الإشعار. يرجى المحاولة مرة أخرى أو الاتصال بنفسك برقم الطوارئ 112.", "ru": "Не удалось доставить уведомление. Пожалуйста, повторите попытку или позвоните в службу спасения 112 сами.", "zh": "通知发送失败。请重试或您自行拨打急救电话 112。", } # Status -> Hinweis-Tabelle. "cooldown" nutzt bewusst den "gesendet"-Text (beruhigen). _NOTICE_TABLES: dict[str, dict[str, str]] = { "alerted": _SENT_NOTICE, "no_recipients": _NO_RECIPIENTS_NOTICE, "delivery_failed": _DELIVERY_FAILED_NOTICE, } def emergency_notice(language: str | None, status: str) -> str: """Hinweis in Nutzersprache (Fallback Deutsch) passend zum Notruf-Status.""" table = _NOTICE_TABLES.get(status, _NO_NOTIFY_NOTICE) return table.get((language or "de").lower(), table["de"]) def notify_notice(language: str | None, sent: bool) -> str: """Hinweis in Nutzersprache (Fallback Deutsch), je nachdem ob etwas rausging.""" table = _SENT_NOTICE if sent else _NO_NOTIFY_NOTICE return table.get((language or "de").lower(), table["de"]) def _csv(raw: str | None) -> list[str]: """Trennt eine CSV-/Semikolon-Liste in bereinigte Einzelwerte.""" return [v.strip() for v in (raw or "").replace(";", ",").split(",") if v.strip()] # --- Profil, Standort, Gerät ---------------------------------------------- def _profile(user) -> dict: """Liest die Notfall-Profildaten der Person aus user.prefs (mit Fallbacks).""" prefs = getattr(user, "prefs", None) prefs = prefs if isinstance(prefs, dict) else {} def g(key: str) -> str: v = prefs.get(key) return v.strip() if isinstance(v, str) else "" name = g("full_name") or getattr(user, "display_name", "") or getattr(user, "id", "?") street, postal, city = g("street"), g("postal_code"), g("city") city_line = " ".join(p for p in (postal, city) if p) address_short = ", ".join(p for p in (street, city_line) if p) return { "name": name, "street": street, "postal_code": postal, "city": city, "address_short": address_short, "phone_landline": g("phone_landline"), "phone_mobile": g("phone_mobile"), "birth_date": g("birth_date"), "medical_notes": g("medical_notes"), } def _geo_fields(geo) -> dict | None: """Normalisiert die Geoposition (pydantic-Model ODER dict) auf {lat,lon,accuracy,ts}.""" if geo is None: return None if isinstance(geo, dict): d = geo else: d = {k: getattr(geo, k, None) for k in ("lat", "lon", "accuracy", "ts")} if d.get("lat") is None or d.get("lon") is None: return None return {"lat": d["lat"], "lon": d["lon"], "accuracy": d.get("accuracy"), "ts": d.get("ts")} def _maps_link(lat, lon) -> str: return f"https://www.google.com/maps?q={lat},{lon}" def _person_phones(profile: dict) -> str: """Telefonnummer(n) der Person für die Meldung (Mobil / Festnetz).""" return " / ".join(t for t in (profile["phone_mobile"], profile["phone_landline"]) if t) def _device_line(device) -> str | None: """Beschriftet das auslösende Gerät (Mobilgerät/Festgerät + Plattform), falls bekannt.""" if not device: return None if isinstance(device, dict): mobile, platform = device.get("mobile"), device.get("platform") else: mobile, platform = getattr(device, "mobile", None), getattr(device, "platform", None) kind = "Mobilgerät" if mobile else ("Festgerät" if mobile is False else "Gerät") return f"Auslösegerät: {kind}" + (f" ({platform})" if platform else "") def _location_block(geo, geo_error: str | None = None) -> str: """Standort-Block für die E-Mail (Koordinaten + Karte, oder Grund der Nicht-Verfügbarkeit).""" g = _geo_fields(geo) if not g: reason = geo_error or "kein GPS / keine Freigabe" return f"Standort (Gerät): nicht verfügbar ({reason})." lines = [ f"Standort (Gerät): {g['lat']:.6f}, {g['lon']:.6f}", f"Karte: {_maps_link(g['lat'], g['lon'])}", ] if g.get("accuracy") is not None: lines.append(f"Genauigkeit: ca. {round(g['accuracy'])} m") if g.get("ts"): lines.append(f"Zeitpunkt des Fixes: {g['ts']}") return "\n".join(lines) def _emergency_texts(profile: dict, geo) -> dict[str, str]: """Fertige Texte für SMS und Sprachanruf (vom Webhook-Empfänger genutzt).""" name = profile["name"] addr = profile["address_short"] tel = _person_phones(profile) g = _geo_fields(geo) # SMS: kompakt — Name, Adresse, Telefon, Karten-Link (trägt den genauen Standort) sms = f"NOTRUF: {name} braucht Hilfe." if addr: sms += f" Adresse: {addr}." if tel: sms += f" Tel: {tel}." if g: sms += f" Standort: {_maps_link(g['lat'], g['lon'])}" # Anruf: gesprochen — Name + Wohnadresse, KEINE URL/Koordinaten, Verweis auf SMS/E-Mail call = f"Achtung. {name} hat einen Notruf ausgelöst und braucht dringend Hilfe." if addr: call += f" Wohnadresse: {addr}." call += " Genaue Standortdaten und weitere Details stehen in der SMS und der E-Mail." return {"sms": sms, "call": call} def _email_body(profile: dict, geo, geo_error: str | None = None, device=None) -> str: """Voll-Dossier für die E-Mail: Gerät + Angaben zur Person + Standort.""" lines = [ f"{profile['name']} hat in der Voice-Assistant-App einen Notruf ausgelöst.", "", "Bitte umgehend Kontakt aufnehmen.", "", ] dev = _device_line(device) if dev: lines += [dev, ""] lines.append("— Angaben zur Person —") fields = [ ("Adresse", profile["address_short"]), ("Mobil", profile["phone_mobile"]), ("Festnetz", profile["phone_landline"]), ("Geburtsdatum", profile["birth_date"]), ("Medizinische Hinweise", profile["medical_notes"]), ] info = [f"{label}: {value}" for label, value in fields if value] lines.extend(info or ["(keine Profilangaben hinterlegt)"]) lines += ["", "— Standort —", _location_block(geo, geo_error), "", "(Diese Meldung wurde automatisch von der Voice-Assistant-App gesendet.)"] return "\n".join(lines) # --- Versandkanäle --------------------------------------------------------- def _contacts(user, cfg) -> list[str]: """Kontakt-E-Mail(s): pro-Nutzer-Pref (CSV) falls vorhanden, sonst Default aus der Config.""" prefs = getattr(user, "prefs", None) or {} raw = (prefs.get("emergency_contacts") or "").strip() if isinstance(prefs, dict) else "" return _csv(raw) or _csv(cfg.emergency_contact_email) def _phones(user, cfg) -> list[str]: """Kontakt-Telefonnummer(n): pro-Nutzer-Pref (CSV) falls vorhanden, sonst Default aus der Config.""" prefs = getattr(user, "prefs", None) or {} raw = (prefs.get("emergency_phones") or "").strip() if isinstance(prefs, dict) else "" return _csv(raw) or _csv(cfg.emergency_contact_phone) def notify_readiness(user, cfg=settings) -> dict: """Ob für diesen Nutzer ein Benachrichtigungskanal *funktionsfähig* wäre. Wichtig: nicht „Kontakt hinterlegt", sondern „Kanal würde tatsächlich senden". E-Mail braucht einen SMTP-Host UND mindestens einen Empfänger; SMS/Anruf braucht eine Webhook-URL. Dient dem Frontend für eine Vorab-Warnung und der sauberen Status-Unterscheidung no_recipients vs. delivery_failed. """ email = bool(cfg.smtp_host) and bool(_contacts(user, cfg)) webhook = bool(cfg.emergency_webhook_url) return {"email": email, "webhook": webhook, "any": email or webhook} def _send_email_sync(to: list[str], subject: str, body: str, cfg) -> bool: """Versendet eine E-Mail per SMTP. Liefert True bei Erfolg, sonst False (best effort).""" if not cfg.smtp_host or not to: return False msg = EmailMessage() msg["From"] = cfg.smtp_from or cfg.smtp_user or "voice-assistant@localhost" msg["To"] = ", ".join(to) msg["Subject"] = subject msg.set_content(body) try: with smtplib.SMTP(cfg.smtp_host, int(cfg.smtp_port), timeout=10) as s: if cfg.smtp_starttls: s.starttls() if cfg.smtp_user: s.login(cfg.smtp_user, cfg.smtp_password) s.send_message(msg) return True except Exception: logger.exception("Notruf-E-Mail konnte nicht versendet werden") return False def _webhook_payload(user, language, phones, channels, profile, geo, geo_error=None, device=None) -> dict: """Provider-agnostisches Notruf-Payload für den Eskalations-Webhook.""" return { "event": "emergency", "category": "manual", "source": "voice-assistant", "timestamp": datetime.now(timezone.utc).isoformat(), "language": (language or "de"), "user": {"id": getattr(user, "id", None), "name": profile["name"]}, "channels": channels, "phones": phones, "geo": _geo_fields(geo), "geo_error": geo_error, "device": device if isinstance(device, dict) else None, "message": _emergency_texts(profile, geo), } async def _post_webhook(url: str, payload: dict, cfg) -> bool: """Sendet das Notruf-Payload per HTTP-POST an den Eskalations-Webhook (best effort).""" if not url: return False headers = {"Content-Type": "application/json"} if cfg.emergency_webhook_token: headers["Authorization"] = f"Bearer {cfg.emergency_webhook_token}" try: async with httpx.AsyncClient(timeout=10) as c: resp = await c.post(url, json=payload, headers=headers) return resp.is_success except Exception: logger.exception("Notruf-Webhook (SMS/Anruf) konnte nicht ausgelöst werden") return False def _resolve_cooldown_minutes(user, cfg) -> int: """Alarmpause in Minuten. Präzedenz: Nutzer-Pref > globale Einstellung > 5. Leerer/fehlender Pref-Wert erbt die globale Einstellung; "0" bedeutet aus. """ raw = (getattr(user, "prefs", None) or {}).get("emergency_cooldown_minutes") if raw is not None and str(raw).strip() != "": try: return max(0, int(raw)) except (TypeError, ValueError): pass try: return max(0, int(getattr(cfg, "emergency_cooldown_minutes", 5))) except (TypeError, ValueError): return 5 def _cooldown_seconds(user, cfg) -> int: return _resolve_cooldown_minutes(user, cfg) * 60 def emergency_cooldown_status(user, store, cfg=settings) -> dict: """Aktueller Sperr-Status: ist nach dem letzten manuellen Notruf noch eine Wiederauslöse-Sperre aktiv, wie lange ist er her, wie viel bleibt. Schützt verwirrte/verängstigte Senioren vor Mehrfach-Alarmen. Die Pause ist pro Nutzer einstellbar (sonst greift die globale Einstellung). """ total = _cooldown_seconds(user, cfg) last = store.last_manual_emergency_at(user.id) if total > 0 else None base = {"active": False, "cooldown_seconds": total, "remaining_seconds": 0, "minutes_ago": 0} if not last: return base try: last_dt = datetime.fromisoformat(last) except (ValueError, TypeError): return base elapsed = (datetime.now(timezone.utc) - last_dt).total_seconds() remaining = int(total - elapsed) return { "active": remaining > 0, "cooldown_seconds": total, "remaining_seconds": max(0, remaining), "minutes_ago": max(0, int(elapsed // 60)), } async def record_manual_emergency(user, store, language: str | None = None, geo=None, geo_error: str | None = None, device=None, cfg=settings) -> dict: """Protokolliert einen bestätigten Notruf, benachrichtigt Kontakte und liefert den Hinweis. Innerhalb der Wiederauslöse-Sperre (emergency_cooldown_minutes) wird KEIN neuer Alarm gesendet und kein Ereignis protokolliert — stattdessen liefert die Antwort status='cooldown' mit minutes_ago/remaining_seconds, damit das Frontend den beruhigenden Hinweis zeigen kann. """ cd = emergency_cooldown_status(user, store, cfg) if cd["active"]: metrics.inc("emergency_total", {"category": "manual", "source": "user_suppressed"}) return { "category": "manual", "status": "cooldown", "cooldown_seconds": cd["cooldown_seconds"], "remaining_seconds": cd["remaining_seconds"], "minutes_ago": cd["minutes_ago"], "notice": notify_notice(language, True), } # Bewusst minimaler Log-Eintrag — KEINE Adresse/Koordinaten (Admin-Log bleibt unkritisch). store.log_emergency(user.id, "manual", "Vom Nutzer ausgelöster Notruf") metrics.inc("emergency_total", {"category": "manual", "source": "user"}) profile = _profile(user) # Diagnose-Zeile im Server-Log: erklärt fehlenden Standort (geo_error) + Geräteart. logger.info("Notruf von %s: geo=%s geo_error=%r device=%r", profile["name"], bool(_geo_fields(geo)), geo_error, device) # Kanal 1: E-Mail (SMTP) — Voll-Dossier to = _contacts(user, cfg) subject = f"🆘 Notruf von {profile['name']}" body = _email_body(profile, geo, geo_error, device) email_sent = await asyncio.to_thread(_send_email_sync, to, subject, body, cfg) if email_sent: metrics.inc("emergency_notify_total", {"channel": "email"}) # Kanal 2: SMS/Anruf (provider-agnostischer Webhook) phones = _phones(user, cfg) channels = _csv(cfg.emergency_webhook_channels) payload = _webhook_payload(user, language, phones, channels, profile, geo, geo_error, device) webhook_sent = await _post_webhook(cfg.emergency_webhook_url, payload, cfg) if webhook_sent: metrics.inc("emergency_notify_total", {"channel": "webhook"}) notified = email_sent or webhook_sent ready = notify_readiness(user, cfg) # Drei ehrliche Endzustände: erreicht / kein Kanal eingerichtet / Zustellung gescheitert. if notified: status = "alerted" elif not ready["any"]: status = "no_recipients" else: status = "delivery_failed" if status != "alerted": logger.warning("Notruf von %s: status=%s, keine Benachrichtigung zugestellt " "(email_ready=%s webhook_ready=%s)", profile["name"], status, ready["email"], ready["webhook"]) return { "category": "manual", "status": status, "cooldown_seconds": cd["cooldown_seconds"], "notice": emergency_notice(language, status), "notified": notified, "notify_ready": ready, "email_sent": email_sent, "webhook_sent": webhook_sent, "channels": channels, }