feat(notifications): täglicher Lauf über SMTP und Apprise

- Kanäle SMTP (HTML-Mail mit Logos als CID-Anhang) und Apprise, beide blockierenden Bibliotheken laufen in einem Thread - Vier Anlässe: Fälligkeiten im Vorlauf, Kündigungsfristen in drei Stufen, überschrittene Budgets je Monat, Vertragsverlängerungen im Folgemonat - APScheduler im Anwendungsprozess, täglich 07:00 Europe/Berlin, räumt zugleich abgelaufene Sitzungen auf - Duplikatsschutz über den Zieltag statt den Versandtag; fehlgeschlagener Versand wird beim nächsten Lauf erneut versucht - Endpunkte für Regeln, Protokoll, Testversand und sofortigen Lauf - Einstellungsseite mit Einrichtungsstand, Regelpflege und Protokoll - 25 neue Backend-Tests (260 gesamt), 9 neue Frontend-Tests (74 gesamt) Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014e7t8UpmoVNMtWivY5LiSH
2026-09-09 17:04:30 +02:00
parent 0adf154049
commit 54c59c9f71
17 changed files with 2500 additions and 1 deletions
@@ -0,0 +1,533 @@
+"""Benachrichtigungen: Ereignisse sammeln, Duplikate ausschließen, versenden.
+
+Der Duplikatsschutz hängt nicht am Versandtag, sondern am Zieltag des Ereignisses:
+Eine Fälligkeit am 15.03. wird genau einmal gemeldet, egal an welchem Tag des
+Vorlaufs der Job läuft. Ein zweiter Lauf am selben Tag erzeugt daher ebenso wenig
+eine zweite Nachricht wie ein Lauf am Folgetag.
+"""
+
+import logging
+from collections import defaultdict
+from dataclasses import dataclass, field
+from datetime import date, timedelta
+from decimal import Decimal
+from html import escape
+from pathlib import Path
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.clock import add_months, month_end, month_start, today, utcnow
+from app.core.config import settings
+from app.models import LogoAsset, Merchant, NotificationLog, NotificationRule
+from app.models.enums import (
+    EntryKind,
+    NotificationChannel,
+    NotificationStatus,
+    NotificationType,
+)
+from app.services.channels import Attachment, ChannelError, Notification, get_channel
+from app.services.occurrences import load_recurrences
+from app.services.recurrence import contract_term
+from app.services.reports import budget_status, flows
+
+logger = logging.getLogger(__name__)
+
+ZERO = Decimal("0.00")
+
+# Vorlaufstufen für Kündigungsfristen gemäß Fachspezifikation.
+NOTICE_STAGES = (30, 14, 7)
+
+REF_OCCURRENCE = "occurrence"
+REF_RECURRENCE = "recurrence"
+REF_BUDGET = "budget"
+
+TYPE_LABELS: dict[NotificationType, str] = {
+    NotificationType.DUE_SOON: "Bald fällig",
+    NotificationType.NOTICE_DEADLINE: "Kündigungsfrist läuft ab",
+    NotificationType.BUDGET_EXCEEDED: "Budget überschritten",
+    NotificationType.CONTRACT_RENEWAL: "Vertrag verlängert sich",
+}
+
+
+@dataclass(frozen=True, slots=True)
+class Event:
+    """Ein meldenswertes Ereignis."""
+
+    ref_type: str
+    ref_id: str
+    """Stabiler Schlüssel innerhalb des ref_type."""
+    dedupe_day: date
+    """Zieltag des Ereignisses – nicht der Versandtag."""
+    headline: str
+    detail: str
+    amount: Decimal | None = None
+    merchant_id: int | None = None
+    on: date | None = None
+
+
+@dataclass(slots=True)
+class RunResult:
+    """Ergebnis eines Laufs."""
+
+    checked: int = 0
+    sent: int = 0
+    skipped: int = 0
+    failed: int = 0
+    by_rule: dict[int, int] = field(default_factory=dict)
+
+
+# --- Ereignisse sammeln --------------------------------------------------------
+
+
+def _money(amount: Decimal | None) -> str:
+    if amount is None:
+        return ""
+    return f"{amount:,.2f} €".replace(",", "#").replace(".", ",").replace("#", ".")
+
+
+def _day(value: date) -> str:
+    return value.strftime("%d.%m.%Y")
+
+
+async def collect_due_soon(
+    session: AsyncSession, rule: NotificationRule, as_of: date
+) -> list[Event]:
+    """Fälligkeiten innerhalb des Vorlaufs."""
+    ende = as_of + timedelta(days=max(rule.lead_days, 0))
+
+    ereignisse: list[Event] = []
+    for eintrag in await flows(session, as_of, ende):
+        # Einmalige Buchungen sind bereits erfasst und brauchen keine Erinnerung.
+        if eintrag.source != "recurrence" or eintrag.recurrence_id is None:
+            continue
+        if eintrag.is_confirmed:
+            continue
+
+        richtung = "Einkunft" if eintrag.kind is EntryKind.INCOME else "Zahlung"
+        ereignisse.append(
+            Event(
+                ref_type=REF_OCCURRENCE,
+                ref_id=f"{eintrag.recurrence_id}:{eintrag.occurrence_date}",
+                dedupe_day=eintrag.on,
+                headline=eintrag.title,
+                detail=f"{richtung} am {_day(eintrag.on)} über {_money(eintrag.amount)}",
+                amount=eintrag.amount,
+                merchant_id=eintrag.merchant_id,
+                on=eintrag.on,
+            )
+        )
+    return ereignisse
+
+
+async def collect_notice_deadlines(
+    session: AsyncSession, rule: NotificationRule, as_of: date
+) -> list[Event]:
+    """Kündigungsfristen, die in 30, 14 oder 7 Tagen ablaufen.
+
+    Jede Stufe wird einmal gemeldet; der Schlüssel enthält deshalb die Stufe.
+    """
+    # Aufsteigend, damit die *engste* zutreffende Stufe gewinnt: bei 14 Resttagen
+    # ist es die 14er-Stufe, nicht erneut die 30er.
+    stufen = sorted({*NOTICE_STAGES, rule.lead_days} - {0})
+
+    ereignisse: list[Event] = []
+    for recurrence in await load_recurrences(session):
+        if recurrence.contract_cancelled_at is not None:
+            continue
+        laufzeit = contract_term(recurrence, as_of)
+        if laufzeit is None or laufzeit.notice_deadline is None:
+            continue
+
+        verbleibend = (laufzeit.notice_deadline - as_of).days
+        if verbleibend < 0:
+            continue
+
+        # Die erste Stufe, die der Termin gerade erreicht oder unterschritten hat.
+        stufe = next((wert for wert in stufen if verbleibend <= wert), None)
+        if stufe is None:
+            continue
+
+        ereignisse.append(
+            Event(
+                ref_type=REF_RECURRENCE,
+                ref_id=f"{recurrence.id}:{stufe}",
+                dedupe_day=laufzeit.notice_deadline,
+                headline=recurrence.title,
+                detail=(
+                    f"Kündigung bis {_day(laufzeit.notice_deadline)} möglich "
+                    f"(noch {verbleibend} Tage), Laufzeit endet am {_day(laufzeit.term_end)}"
+                ),
+                amount=recurrence.amount,
+                merchant_id=recurrence.merchant_id,
+                on=laufzeit.notice_deadline,
+            )
+        )
+    return ereignisse
+
+
+async def collect_budget_exceeded(
+    session: AsyncSession, rule: NotificationRule, as_of: date
+) -> list[Event]:
+    """Überschrittene Budgets – höchstens einmal je Monat und Kategorie."""
+    monat = month_start(as_of)
+
+    return [
+        Event(
+            ref_type=REF_BUDGET,
+            ref_id=str(eintrag.category_id),
+            dedupe_day=monat,
+            headline=eintrag.category_name,
+            detail=(
+                f"{_money(eintrag.spent)} von {_money(eintrag.available)} verbraucht "
+                f"({round(float(eintrag.ratio) * 100)} %), "
+                f"{_money(abs(eintrag.remaining))} zu viel"
+            ),
+            amount=eintrag.spent,
+            on=month_end(monat),
+        )
+        for eintrag in await budget_status(session, monat)
+        if eintrag.state == "exceeded"
+    ]
+
+
+async def collect_contract_renewals(
+    session: AsyncSession, rule: NotificationRule, as_of: date
+) -> list[Event]:
+    """Verträge, die sich im kommenden Monat automatisch verlängern."""
+    fenster_ende = month_end(add_months(as_of, 1))
+
+    ereignisse: list[Event] = []
+    for recurrence in await load_recurrences(session):
+        if recurrence.contract_cancelled_at is not None:
+            continue
+        laufzeit = contract_term(recurrence, as_of)
+        if laufzeit is None or laufzeit.renews_on is None:
+            continue
+        if not as_of <= laufzeit.renews_on <= fenster_ende:
+            continue
+
+        verlaengerung = recurrence.contract_auto_renew_months
+        ereignisse.append(
+            Event(
+                ref_type=REF_RECURRENCE,
+                ref_id=str(recurrence.id),
+                dedupe_day=laufzeit.renews_on,
+                headline=recurrence.title,
+                detail=(
+                    f"Verlängert sich am {_day(laufzeit.renews_on)}"
+                    + (f" um {verlaengerung} Monate" if verlaengerung else "")
+                ),
+                amount=recurrence.amount,
+                merchant_id=recurrence.merchant_id,
+                on=laufzeit.renews_on,
+            )
+        )
+    return ereignisse
+
+
+COLLECTORS = {
+    NotificationType.DUE_SOON: collect_due_soon,
+    NotificationType.NOTICE_DEADLINE: collect_notice_deadlines,
+    NotificationType.BUDGET_EXCEEDED: collect_budget_exceeded,
+    NotificationType.CONTRACT_RENEWAL: collect_contract_renewals,
+}
+
+
+# --- Nachricht bauen -----------------------------------------------------------
+
+
+async def _logo_attachments(session: AsyncSession, events: list[Event]) -> dict[int, Attachment]:
+    """Lädt die Logodateien der beteiligten Firmen für den Mailanhang."""
+    firmen_ids = {ereignis.merchant_id for ereignis in events if ereignis.merchant_id}
+    if not firmen_ids:
+        return {}
+
+    stmt = (
+        select(Merchant, LogoAsset)
+        .join(LogoAsset, LogoAsset.id == Merchant.logo_asset_id)
+        .where(Merchant.id.in_(firmen_ids))
+    )
+
+    anhaenge: dict[int, Attachment] = {}
+    for merchant, asset in (await session.execute(stmt)).all():
+        pfad = Path(settings.logo_storage_dir) / asset.file_path
+        if not pfad.exists():
+            continue
+        anhaenge[merchant.id] = Attachment(cid=f"logo-{merchant.id}", path=pfad, mime=asset.mime)
+    return anhaenge
+
+
+def build_notification(
+    rule_type: NotificationType,
+    events: list[Event],
+    attachments: dict[int, Attachment],
+) -> Notification:
+    """Formuliert aus den Ereignissen eine Nachricht in Text und HTML."""
+    ueberschrift = TYPE_LABELS[rule_type]
+    betreff = (
+        f"moneyfy: {ueberschrift}"
+        if len(events) == 1
+        else f"moneyfy: {ueberschrift} ({len(events)})"
+    )
+
+    text_zeilen = [ueberschrift, ""]
+    for ereignis in events:
+        text_zeilen.append(f"• {ereignis.headline}: {ereignis.detail}")
+    text_zeilen.extend(["", settings.public_base_url])
+
+    zeilen_html: list[str] = []
+    for ereignis in events:
+        anhang = attachments.get(ereignis.merchant_id or -1)
+        bild = (
+            f'<img src="cid:{anhang.cid}" alt="" width="28" height="28" '
+            'style="border-radius:6px;vertical-align:middle;margin-right:10px">'
+            if anhang
+            else ""
+        )
+        zeilen_html.append(
+            '<tr><td style="padding:10px 0;border-bottom:1px solid #e2e8f0">'
+            f"{bild}"
+            f'<strong style="color:#0f172a">{escape(ereignis.headline)}</strong><br>'
+            f'<span style="color:#475569;font-size:14px">{escape(ereignis.detail)}</span>'
+            "</td></tr>"
+        )
+
+    html = f"""<!doctype html>
+<html lang="de"><body style="margin:0;padding:24px;background:#f8fafc;
+  font-family:system-ui,-apple-system,'Segoe UI',Roboto,sans-serif">
+  <div style="max-width:560px;margin:0 auto;background:#ffffff;border-radius:14px;
+    border:1px solid #e2e8f0;padding:24px">
+    <h1 style="margin:0 0 4px;font-size:18px;color:#0f172a">{escape(ueberschrift)}</h1>
+    <p style="margin:0 0 16px;font-size:13px;color:#64748b">
+      {len(events)} {"Eintrag" if len(events) == 1 else "Einträge"}
+    </p>
+    <table style="width:100%;border-collapse:collapse">{"".join(zeilen_html)}</table>
+    <p style="margin:20px 0 0;font-size:13px">
+      <a href="{escape(settings.public_base_url)}" style="color:#16a34a">In moneyfy öffnen</a>
+    </p>
+  </div>
+</body></html>"""
+
+    beteiligte = {
+        ereignis.merchant_id for ereignis in events if ereignis.merchant_id in attachments
+    }
+    return Notification(
+        subject=betreff,
+        text="\n".join(text_zeilen),
+        html=html,
+        attachments=[attachments[firma_id] for firma_id in beteiligte if firma_id],
+    )
+
+
+# --- Versand und Protokoll -----------------------------------------------------
+
+
+async def _existing_log(
+    session: AsyncSession, rule_id: int, event: Event
+) -> NotificationLog | None:
+    stmt = select(NotificationLog).where(
+        NotificationLog.rule_id == rule_id,
+        NotificationLog.ref_type == event.ref_type,
+        NotificationLog.ref_id == event.ref_id,
+        NotificationLog.dedupe_day == event.dedupe_day,
+    )
+    return (await session.execute(stmt)).scalar_one_or_none()
+
+
+async def pending_events(
+    session: AsyncSession, rule: NotificationRule, as_of: date
+) -> tuple[list[Event], list[Event], dict[str, NotificationLog]]:
+    """Trennt neue Ereignisse von bereits gemeldeten.
+
+    Ein zuvor fehlgeschlagener Versand gilt als offen und wird erneut versucht;
+    der vorhandene Protokolleintrag wird dabei aktualisiert statt neu angelegt.
+    """
+    sammler = COLLECTORS[rule.type]
+    alle = await sammler(session, rule, as_of)
+
+    offen: list[Event] = []
+    erledigt: list[Event] = []
+    vorhandene: dict[str, NotificationLog] = {}
+
+    for ereignis in alle:
+        eintrag = await _existing_log(session, rule.id, ereignis)
+        if eintrag is None:
+            offen.append(ereignis)
+            continue
+        if eintrag.status is NotificationStatus.FAILED:
+            offen.append(ereignis)
+            vorhandene[f"{ereignis.ref_type}:{ereignis.ref_id}"] = eintrag
+            continue
+        erledigt.append(ereignis)
+
+    return offen, erledigt, vorhandene
+
+
+async def _record(
+    session: AsyncSession,
+    rule: NotificationRule,
+    events: list[Event],
+    existing: dict[str, NotificationLog],
+    status: NotificationStatus,
+    error: str | None,
+) -> None:
+    """Schreibt das Versandprotokoll für alle Ereignisse einer Nachricht."""
+    zeitpunkt = utcnow()
+    for ereignis in events:
+        schluessel = f"{ereignis.ref_type}:{ereignis.ref_id}"
+        eintrag = existing.get(schluessel)
+        if eintrag is not None:
+            eintrag.status = status
+            eintrag.error = error
+            eintrag.sent_at = zeitpunkt
+            continue
+
+        session.add(
+            NotificationLog(
+                rule_id=rule.id,
+                ref_type=ereignis.ref_type,
+                ref_id=ereignis.ref_id,
+                dedupe_day=ereignis.dedupe_day,
+                sent_at=zeitpunkt,
+                status=status,
+                error=error,
+            )
+        )
+    await session.flush()
+
+
+async def run_rule(
+    session: AsyncSession, rule: NotificationRule, as_of: date | None = None
+) -> tuple[int, int, int]:
+    """Führt eine Regel aus. Liefert (geprüft, versendet, übersprungen)."""
+    stichtag = as_of or today()
+    offen, erledigt, vorhandene = await pending_events(session, rule, stichtag)
+
+    if not offen:
+        return len(offen) + len(erledigt), 0, len(erledigt)
+
+    anhaenge = await _logo_attachments(session, offen)
+    nachricht = build_notification(rule.type, offen, anhaenge)
+
+    try:
+        await get_channel(rule.channel).send(nachricht, rule.target)
+    except ChannelError as fehler:
+        logger.warning("Regel %s konnte nicht zugestellt werden: %s", rule.id, fehler)
+        await _record(session, rule, offen, vorhandene, NotificationStatus.FAILED, str(fehler))
+        return len(offen) + len(erledigt), 0, len(erledigt)
+
+    await _record(session, rule, offen, vorhandene, NotificationStatus.SENT, None)
+    return len(offen) + len(erledigt), len(offen), len(erledigt)
+
+
+async def run_all(session: AsyncSession, as_of: date | None = None) -> RunResult:
+    """Führt alle aktiven Regeln aus – der tägliche Lauf des Schedulers."""
+    if not settings.notifications_enabled:
+        logger.info("Benachrichtigungen sind abgeschaltet.")
+        return RunResult()
+
+    stichtag = as_of or today()
+    stmt = select(NotificationRule).where(NotificationRule.is_active.is_(True))
+    regeln = list((await session.execute(stmt)).scalars())
+
+    ergebnis = RunResult()
+    for regel in regeln:
+        try:
+            geprueft, versendet, uebersprungen = await run_rule(session, regel, stichtag)
+        except Exception:
+            # Eine fehlerhafte Regel darf die übrigen nicht verhindern.
+            logger.exception("Regel %s ist fehlgeschlagen.", regel.id)
+            ergebnis.failed += 1
+            continue
+
+        ergebnis.checked += geprueft
+        ergebnis.sent += versendet
+        ergebnis.skipped += uebersprungen
+        if versendet:
+            ergebnis.by_rule[regel.id] = versendet
+
+    await session.commit()
+    logger.info(
+        "Benachrichtigungslauf: %d geprüft, %d versendet, %d übersprungen, %d fehlerhaft.",
+        ergebnis.checked,
+        ergebnis.sent,
+        ergebnis.skipped,
+        ergebnis.failed,
+    )
+    return ergebnis
+
+
+# --- Testversand ---------------------------------------------------------------
+
+
+def test_notification() -> Notification:
+    """Eine kurze Nachricht, mit der sich die Einrichtung prüfen lässt."""
+    zeitpunkt = utcnow().astimezone().strftime("%d.%m.%Y um %H:%M Uhr")
+    return Notification(
+        subject="moneyfy: Testnachricht",
+        text=(
+            "Das ist eine Testnachricht von moneyfy.\n"
+            f"Gesendet am {zeitpunkt}.\n\n"
+            "Wenn du sie liest, ist der Kanal richtig eingerichtet.\n\n"
+            f"{settings.public_base_url}"
+        ),
+        html=f"""<!doctype html>
+<html lang="de"><body style="margin:0;padding:24px;background:#f8fafc;
+  font-family:system-ui,-apple-system,'Segoe UI',Roboto,sans-serif">
+  <div style="max-width:560px;margin:0 auto;background:#ffffff;border-radius:14px;
+    border:1px solid #e2e8f0;padding:24px">
+    <h1 style="margin:0 0 8px;font-size:18px;color:#0f172a">Testnachricht</h1>
+    <p style="margin:0;font-size:14px;color:#475569">
+      Wenn du das liest, ist der Kanal richtig eingerichtet.<br>
+      Gesendet am {zeitpunkt}.
+    </p>
+    <p style="margin:20px 0 0;font-size:13px">
+      <a href="{escape(settings.public_base_url)}" style="color:#16a34a">moneyfy öffnen</a>
+    </p>
+  </div>
+</body></html>""",
+    )
+
+
+@dataclass(frozen=True, slots=True)
+class TestResult:
+    """Ergebnis eines Testversands je Kanal."""
+
+    channel: NotificationChannel
+    configured: bool
+    sent: bool
+    error: str | None = None
+
+
+async def send_test(target: str | None = None) -> list[TestResult]:
+    """Verschickt eine Testnachricht über beide Kanäle.
+
+    Nicht eingerichtete Kanäle werden gemeldet, aber nicht als Fehler gewertet.
+    """
+    nachricht = test_notification()
+
+    ergebnisse: list[TestResult] = []
+    for art in (NotificationChannel.SMTP, NotificationChannel.APPRISE):
+        kanal = get_channel(art)
+        if not kanal.is_configured():
+            ergebnisse.append(TestResult(channel=art, configured=False, sent=False))
+            continue
+        try:
+            await kanal.send(nachricht, target)
+        except ChannelError as fehler:
+            ergebnisse.append(
+                TestResult(channel=art, configured=True, sent=False, error=str(fehler))
+            )
+            continue
+        ergebnisse.append(TestResult(channel=art, configured=True, sent=True))
+
+    return ergebnisse
+
+
+def group_by_type(events: list[Event]) -> dict[str, list[Event]]:
+    """Hilfsfunktion für Vorschauen in der Oberfläche."""
+    gruppen: dict[str, list[Event]] = defaultdict(list)
+    for ereignis in events:
+        gruppen[ereignis.ref_type].append(ereignis)
+    return dict(gruppen)