my_voice_assistant_v3_jamulix/app/safety/emergency.py

"""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).

Testphase: Bei einem Notruf wird eine E-Mail an die Kontaktperson(en) des
Nutzers geschickt (Default: settings.emergency_contact_email). SMS/Anruf-
Bridges folgen später. Ohne konfigurierten SMTP-Host wird nichts versendet
und der Nutzer erhält einen klaren Hinweis darauf.
"""

import asyncio
import logging
import smtplib
from email.message import EmailMessage

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 die E-Mail rausging.
_SENT_NOTICE: dict[str, str] = {
    "de": "Hilfe wurde verständigt. Eine Kontaktperson wurde per E-Mail benachrichtigt.",
    "en": "Help has been alerted. A contact person was notified by email.",
    "fr": "Les secours ont été alertés. Un proche a été prévenu par e-mail.",
    "es": "Se ha avisado a ayuda. Se notificó a un contacto por correo electrónico.",
    "it": "L'aiuto è stato allertato. Una persona di contatto è stata avvisata via e-mail.",
    "pt": "A ajuda foi avisada. Uma pessoa de contato foi notificada por e-mail.",
    "pl": "Pomoc została powiadomiona. Osoba kontaktowa otrzymała e-mail.",
    "ar": "تم تنبيه المساعدة. تم إخطار شخص الاتصال عبر البريد الإلكتروني.",
    "ru": "Помощь оповещена. Контактное лицо уведомлено по электронной почте.",
    "zh": "已通知求助。已通过电子邮件通知联系人。",
}


def notify_notice(language: str | None, sent: bool) -> str:
    """Hinweis in Nutzersprache (Fallback Deutsch), je nachdem ob eine Mail rausging."""
    table = _SENT_NOTICE if sent else _NO_NOTIFY_NOTICE
    return table.get((language or "de").lower(), table["de"])


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 ""
    if not raw:
        raw = cfg.emergency_contact_email or ""
    return [a.strip() for a in raw.replace(";", ",").split(",") if a.strip()]


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


async def record_manual_emergency(user, store, language: str | None = None, cfg=settings) -> dict:
    """Protokolliert einen bestätigten Notruf, benachrichtigt Kontakte und liefert den Hinweis."""
    store.log_emergency(user.id, "manual", "Vom Nutzer ausgelöster Notruf")
    metrics.inc("emergency_total", {"category": "manual", "source": "user"})

    to = _contacts(user, cfg)
    name = getattr(user, "display_name", None) or getattr(user, "id", "?")
    subject = f"🆘 Notruf von {name}"
    body = (
        f"{name} hat in der Voice-Assistant-App einen Notruf ausgelöst.\n\n"
        f"Bitte umgehend Kontakt aufnehmen.\n\n"
        f"(Testphase: automatische Eskalation per SMS/Anruf ist noch nicht aktiv.)"
    )
    sent = await asyncio.to_thread(_send_email_sync, to, subject, body, cfg)
    if sent:
        metrics.inc("emergency_notify_total", {"channel": "email"})
    return {"category": "manual", "notice": notify_notice(language, sent), "email_sent": sent}