"""Plugin „benachrichtigung“ — Kanäle E-Mail, Telegram und In-App. Architektur (Adapter-Muster): - `adapter.py` kapselt die Zustellkanäle. Fehlt die Konfiguration eines Kanals (z. B. kein SMTP-Host in der Entwicklung), protokolliert ein Dev-Log-Adapter die Nachricht, statt sie zu versenden. - Pro Benutzer speichern `models.Praeferenz` (gewählte Kanäle) und `models.Kontakt` (E-Mail-Adresse, Telegram-Chat-ID) die Einstellungen. - Andere Plugins rufen die öffentliche API auf: plugin = request.app.state.registry.get("benachrichtigung") bericht = await plugin.send_notification(user, "Titel", "Text", "kategorie") Der Kern bleibt unverändert; alle Logik liegt in diesem Plugin. """ from __future__ import annotations import logging from dataclasses import dataclass from fastapi import Depends, Form, Request from fastapi.responses import RedirectResponse from sqlalchemy import func, select from sqlalchemy.orm import Session from redaktionskern.auth.deps import get_db, require_user from redaktionskern.auth.models import User from redaktionskern.contracts import BasePlugin, DashboardKarte, Migration, NavEntry from .adapter import BenachrichtigungsAdapter, adapter_aus_konfiguration from .konfiguration import aus_umgebung from .models import KANAELLE, STANDARD_KANAELLE, Benachrichtigung, Kontakt, Praeferenz, gueltige_reihenfolge logger = logging.getLogger("redaktion.benachrichtigung") @dataclass(frozen=True) class Zustellbericht: """Ergebnis eines send_notification-Aufrufs.""" zugestellt: list[str] fehlgeschlagen: list[str] def _tabellen_erstellen(conn) -> None: """Eigene Migration des Plugins: legt alle drei Tabellen an.""" Benachrichtigung.__table__.create(conn, checkfirst=True) Praeferenz.__table__.create(conn, checkfirst=True) Kontakt.__table__.create(conn, checkfirst=True) class BenachrichtigungPlugin(BasePlugin): name = "benachrichtigung" title = "Benachrichtigungen" description = "E-Mail, Telegram und In-App — Kanal je Benutzer einstellbar." def __init__(self) -> None: super().__init__() self._adapters: dict[str, BenachrichtigungsAdapter] | None = None self._routen_registrieren() # ---------------- Lifecycle ---------------- def on_load(self, context) -> None: super().on_load(context) self._adapters = adapter_aus_konfiguration( aus_umgebung(), context.session_factory ) @property def adapters(self) -> dict[str, BenachrichtigungsAdapter]: """Kanal-Adapter der Installation (nach on_load verfügbar).""" if self._adapters is None: raise RuntimeError("Plugin ist nicht geladen (on_load fehlt).") return self._adapters def migrations(self) -> list[Migration]: return [Migration(version="0001_tabellen", up=_tabellen_erstellen)] def navigation(self) -> list[NavEntry]: return [NavEntry(label=self.title, url="/benachrichtigung")] def dashboard_karten(self, user: User) -> list[DashboardKarte]: """Dashboard: Anzahl ungelesener In-App-Benachrichtigungen.""" with self.context.session_factory() as db: ungelesen = db.scalar( select(func.count()) .select_from(Benachrichtigung) .where( Benachrichtigung.user_id == user.id, Benachrichtigung.gelesen.is_(False), ) ) or 0 return [ DashboardKarte( titel="Ungelesene Benachrichtigungen", wert=str(ungelesen), beschreibung="Nachrichten in deinem Posteingang", link_url="/benachrichtigung", link_label="Zum Posteingang", ton="warnung" if ungelesen else "ok", ) ] # ---------------- Öffentliche Plugin-API ---------------- async def send_notification( self, user: User, titel: str, text: str, kategorie: str = "allgemein", ) -> Zustellbericht: """Stellt eine Benachrichtigung über alle gewählten Kanäle zu. Andere Plugins holen sich dieses Plugin über die Registry und rufen diese Methode. Die In-App-Nachricht wird persistiert; E-Mail/Telegram gehen an die im Profil hinterlegten Kontaktdaten. Fehlt einem Kanal die Empfängeradresse oder schlägt die Zustellung fehl, wird das protokolliert und im Bericht gemeldet — send_notification wirft nicht. """ assert self.context is not None with self.context.session_factory() as db: praeferenz = db.get(Praeferenz, user.id) kontakt = db.get(Kontakt, user.id) kanaele = ( praeferenz.kanaele_als_liste() if praeferenz is not None else [STANDARD_KANAELLE] ) zugestellt: list[str] = [] fehlgeschlagen: list[str] = [] for schluessel in kanaele: adapter = self.adapters.get(schluessel) if adapter is None: continue empfaenger: str | None = None if schluessel == "email" and kontakt is not None: empfaenger = kontakt.email elif schluessel == "telegram" and kontakt is not None: empfaenger = kontakt.telegram_chat_id try: if schluessel != "inapp" and not empfaenger: raise ValueError( "Keine Empfängeradresse für diesen Kanal hinterlegt." ) await adapter.senden(user.id, empfaenger, titel, text, kategorie) if adapter.dev_fallback: logger.info( "Kanal '%s' nur protokolliert (nicht konfiguriert): '%s'", schluessel, titel, ) zugestellt.append(schluessel) except Exception as exc: logger.exception( "Zustellung über Kanal '%s' fehlgeschlagen: %s", schluessel, exc ) fehlgeschlagen.append(schluessel) return Zustellbericht(zugestellt=zugestellt, fehlgeschlagen=fehlgeschlagen) # ---------------- Routen ---------------- def _routen_registrieren(self) -> None: @self.router.get("/benachrichtigung") def posteingang( request: Request, db: Session = Depends(get_db), user: User = Depends(require_user), ): mitteilungen = db.scalars( select(Benachrichtigung) .where(Benachrichtigung.user_id == user.id) .order_by(Benachrichtigung.created_at.desc(), Benachrichtigung.id.desc()) .limit(200) ).all() ungelesen = db.scalar( select(func.count()) .select_from(Benachrichtigung) .where( Benachrichtigung.user_id == user.id, Benachrichtigung.gelesen.is_(False), ) ) praef = db.get(Praeferenz, user.id) gewaehlte = ( praeferenz.kanaele_als_liste() if praef is not None else [STANDARD_KANAELLE] ) return self.context.templates.TemplateResponse( request=request, name="benachrichtigung/index.html", context={ "user": user, "titel": self.title, "mitteilungen": mitteilungen, "ungelesen": ungelesen or 0, "gewaehlte_kanaelle": [ KANAELLE[k] for k in gueltige_reihenfolge(gewaehlte) ], }, ) @self.router.post("/benachrichtigung/alle-gelesen") def alle_gelesen( db: Session = Depends(get_db), user: User = Depends(require_user), ): db.query(Benachrichtigung).filter( Benachrichtigung.user_id == user.id, Benachrichtigung.gelesen.is_(False), ).update({"gelesen": True}, synchronize_session=False) db.commit() return RedirectResponse("/benachrichtigung", status_code=303) @self.router.post("/benachrichtigung/{id}/gelesen") def einzel_gelesen( id: int, db: Session = Depends(get_db), user: User = Depends(require_user), ): nachricht = db.scalar( select(Benachrichtigung).where( Benachrichtigung.id == id, Benachrichtigung.user_id == user.id, ) ) if nachricht is not None and not nachricht.gelesen: nachricht.gelesen = True db.commit() return RedirectResponse("/benachrichtigung", status_code=303) @self.router.get("/benachrichtigung/einstellungen") def einstellungen_formular( request: Request, gespeichert: bool = False, db: Session = Depends(get_db), user: User = Depends(require_user), ): praef = db.get(Praeferenz, user.id) kontakt = db.get(Kontakt, user.id) gewaehlte = ( praef.kanaele_als_liste() if praef is not None else [STANDARD_KANAELLE] ) kanal_status = { schluessel: { "label": label, "aktiv": not self.adapters[schluessel].dev_fallback, } for schluessel, label in KANAELLE.items() } return self.context.templates.TemplateResponse( request=request, name="benachrichtigung/einstellungen.html", context={ "user": user, "titel": f"{self.title} — Einstellungen", "kanal_status": kanal_status, "gewaehlte": gewaehlte, "emailadresse": kontakt.email if kontakt else "", "telegram_chat_id": kontakt.telegram_chat_id if kontakt else "", "gespeichert": gespeichert, "fehler": request.session.pop("benachrichtigung_fehler", None), }, ) @self.router.post("/benachrichtigung/einstellungen") def einstellungen_speichern( request: Request, db: Session = Depends(get_db), user: User = Depends(require_user), kanaele: list[str] = Form(default=[]), emailadresse: str = Form(""), telegram_chat_id: str = Form(""), ): auswahl = gueltige_reihenfolge( [k for k in kanaele if k in KANAELLE] ) emailadresse = emailadresse.strip() telegram_chat_id = telegram_chat_id.strip() fehler: str | None = None if emailadresse and ("@" not in emailadresse or " " in emailadresse): fehler = "Die E-Mail-Adresse sieht nicht gültig aus." if telegram_chat_id.lstrip("-").isdigit() is False and telegram_chat_id: fehler = "Die Telegram-Chat-ID darf nur Ziffern enthalten (Gruppen mit vorangestelltem Minus)." if fehler is not None: request.session["benachrichtigung_fehler"] = fehler return RedirectResponse( "/benachrichtigung/einstellungen", status_code=303 ) kontakt = db.get(Kontakt, user.id) if kontakt is None: kontakt = Kontakt(user_id=user.id) db.add(kontakt) kontakt.email = emailadresse or None kontakt.telegram_chat_id = telegram_chat_id or None praef = db.get(Praeferenz, user.id) if praef is None: praef = Praeferenz(user_id=user.id) db.add(praef) praef.kanaele = ",".join(auswahl) if auswahl else "" db.commit() return RedirectResponse( "/benachrichtigung/einstellungen?gespeichert=true", status_code=303 ) plugin = BenachrichtigungPlugin()