"""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'' if anhang else "" ) zeilen_html.append( '' f"{bild}" f'{escape(ereignis.headline)}
' f'{escape(ereignis.detail)}' "" ) html = f"""

{escape(ueberschrift)}

{len(events)} {"Eintrag" if len(events) == 1 else "Einträge"}

{"".join(zeilen_html)}

In moneyfy öffnen

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

Testnachricht

Wenn du das liest, ist der Kanal richtig eingerichtet.
Gesendet am {zeitpunkt}.

moneyfy öffnen

""", ) @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)