moneyfyandClaude Opus 5
54c59c9f71
- 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
534 lines
18 KiB
Python
534 lines
18 KiB
Python
"""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)
|