moneyfy/backend/app/services/notifications.py

"""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, 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, current_period, flows
from app.services.settings import month_start_day

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 Abrechnungsmonat und Kategorie."""
    monatsbeginn = await month_start_day(session)
    zeitraum = current_period(monatsbeginn, as_of)

    return [
        Event(
            ref_type=REF_BUDGET,
            ref_id=str(eintrag.category_id),
            dedupe_day=zeitraum.key,
            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=zeitraum.end,
        )
        for eintrag in await budget_status(session, zeitraum.key, monatsbeginn)
        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)