feat(recurrence): Expansions-Engine für wiederkehrende Posten
Datenbankfreie Kernberechnung auf Basis schlanker Protokolle, sodass sowohl ORM-Objekte als auch Testdatenklassen verarbeitet werden. - expand() mit voller RFC-5545-Unterstützung, Fensterbegrenzung und Overlays - expand_by_due_date() für Sichten, die nach dem tatsächlichen Zahltag gruppieren - Werktagsverschiebung über den NRW-Feiertagskalender, nominales Datum als Schlüssel - Preishistorie, Ratenzahlungen, Vertragsfristen und Rücklagenberechnung - validate_rrule() und next_dates() für Eingabeprüfung und Terminvorschau - 81 Unit-Tests, davon alle in Abschnitt 2 geforderten Fälle Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014e7t8UpmoVNMtWivY5LiSH
This commit is contained in:
moneyfy
@@ -0,0 +1,577 @@
|
||||
"""Recurrence-Engine: virtuelle Expansion wiederkehrender Posten.
|
||||
|
||||
Das Modul ist bewusst frei von Datenbankzugriffen. Alle Funktionen arbeiten auf
|
||||
schlanken Protokollen, sodass sowohl ORM-Objekte als auch einfache Datenklassen
|
||||
übergeben werden können. Preishistorie und materialisierte Occurrences werden als
|
||||
Sequenzen hereingereicht – das Laden ist Aufgabe der aufrufenden Schicht.
|
||||
|
||||
Zentrale Regel: Schlüssel einer Fälligkeit ist immer das **nominale** Datum, also
|
||||
das von der RRULE gelieferte Datum vor einer Werktagsverschiebung. Nur so bleibt
|
||||
die Zuordnung zu `occurrence`-Zeilen stabil, wenn sich Feiertage oder die
|
||||
Verschiebungsregel ändern.
|
||||
"""
|
||||
|
||||
from collections.abc import Iterable, Sequence
|
||||
from dataclasses import dataclass
|
||||
from datetime import date, datetime, timedelta
|
||||
from decimal import ROUND_HALF_UP, Decimal
|
||||
from functools import lru_cache
|
||||
from typing import Protocol, runtime_checkable
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
import holidays
|
||||
from dateutil.relativedelta import relativedelta
|
||||
from dateutil.rrule import rrulestr
|
||||
|
||||
from app.models.enums import BusinessDayShift, EntryKind, OccurrenceStatus
|
||||
|
||||
ZERO = Decimal("0.00")
|
||||
CENT = Decimal("0.01")
|
||||
|
||||
# Obergrenze für die Werktagssuche – schützt vor Endlosschleifen bei absurden Kalendern.
|
||||
MAX_SHIFT_DAYS = 30
|
||||
|
||||
# Puffer, um den das nominale Fenster erweitert wird, wenn nach Ist-Fälligkeit
|
||||
# gefiltert wird. Eine Verschiebung überschreitet realistisch nie wenige Tage.
|
||||
DUE_DATE_WINDOW_PADDING_DAYS = 21
|
||||
|
||||
|
||||
class RecurrenceError(ValueError):
|
||||
"""Fachlicher Fehler bei der Auswertung einer Wiederholungsregel."""
|
||||
|
||||
|
||||
class InvalidRRuleError(RecurrenceError):
|
||||
"""Die RRULE konnte nicht geparst werden."""
|
||||
|
||||
|
||||
# --- Protokolle ---------------------------------------------------------------
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class AmountVersionLike(Protocol):
|
||||
"""Eine Preisversion; gültig ab `valid_from` (einschließlich)."""
|
||||
|
||||
amount: Decimal
|
||||
valid_from: date
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class OccurrenceLike(Protocol):
|
||||
"""Eine materialisierte Fälligkeit, die das virtuelle Ergebnis überlagert."""
|
||||
|
||||
id: int | None
|
||||
occurrence_date: date
|
||||
status: OccurrenceStatus
|
||||
actual_amount: Decimal | None
|
||||
actual_date: date | None
|
||||
account_id: int | None
|
||||
note: str | None
|
||||
|
||||
|
||||
@runtime_checkable
|
||||
class RecurrenceLike(Protocol):
|
||||
"""Die von der Engine benötigten Felder einer Recurrence."""
|
||||
|
||||
id: int | None
|
||||
kind: EntryKind
|
||||
amount: Decimal
|
||||
is_variable: bool
|
||||
rrule: str
|
||||
dtstart: date
|
||||
until: date | None
|
||||
business_day_shift: BusinessDayShift
|
||||
holiday_region: str
|
||||
installments_total: int | None
|
||||
principal_amount: Decimal | None
|
||||
contract_start: date | None
|
||||
contract_min_term_months: int | None
|
||||
contract_notice_period_days: int | None
|
||||
contract_auto_renew_months: int | None
|
||||
contract_cancelled_at: date | None
|
||||
reserve_enabled: bool
|
||||
account_id: int | None
|
||||
|
||||
|
||||
# --- Ergebnistypen ------------------------------------------------------------
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class PlannedOccurrence:
|
||||
"""Eine einzelne Fälligkeit, virtuell berechnet und ggf. durch ein Overlay ergänzt."""
|
||||
|
||||
recurrence_id: int | None
|
||||
kind: EntryKind
|
||||
nominal_date: date
|
||||
"""Von der RRULE geliefertes Datum – stabiler Schlüssel, auch nach Verschiebung."""
|
||||
due_date: date
|
||||
"""Tatsächlicher Zahltag nach Wochenend-/Feiertagsverschiebung."""
|
||||
amount: Decimal
|
||||
"""Sollbetrag laut Preishistorie zum `nominal_date`."""
|
||||
status: OccurrenceStatus
|
||||
is_variable: bool
|
||||
occurrence_id: int | None = None
|
||||
actual_amount: Decimal | None = None
|
||||
actual_date: date | None = None
|
||||
account_id: int | None = None
|
||||
note: str | None = None
|
||||
installment_number: int | None = None
|
||||
installments_total: int | None = None
|
||||
|
||||
@property
|
||||
def is_materialised(self) -> bool:
|
||||
"""True, wenn zu dieser Fälligkeit bereits eine `occurrence`-Zeile existiert."""
|
||||
return self.occurrence_id is not None
|
||||
|
||||
@property
|
||||
def is_skipped(self) -> bool:
|
||||
return self.status is OccurrenceStatus.SKIPPED
|
||||
|
||||
@property
|
||||
def effective_amount(self) -> Decimal:
|
||||
"""Ist-Betrag, sofern bestätigt, sonst der Sollbetrag. Ausgelassene zählen nicht."""
|
||||
if self.status is OccurrenceStatus.SKIPPED:
|
||||
return ZERO
|
||||
if self.actual_amount is not None:
|
||||
return self.actual_amount
|
||||
return self.amount
|
||||
|
||||
@property
|
||||
def effective_date(self) -> date:
|
||||
"""Ist-Datum, sofern erfasst, sonst der geplante Zahltag."""
|
||||
return self.actual_date or self.due_date
|
||||
|
||||
@property
|
||||
def signed_amount(self) -> Decimal:
|
||||
"""Betrag mit Vorzeichen: Ausgaben negativ, Einkünfte positiv."""
|
||||
amount = self.effective_amount
|
||||
return -amount if self.kind is EntryKind.EXPENSE else amount
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class InstallmentStatus:
|
||||
"""Stand einer Ratenzahlung zu einem Stichtag."""
|
||||
|
||||
total: int
|
||||
paid: int
|
||||
remaining: int
|
||||
paid_amount: Decimal
|
||||
remaining_amount: Decimal
|
||||
"""Restschuld: `principal_amount` abzüglich geleisteter Raten, sonst Summe der Restraten."""
|
||||
final_due_date: date | None
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ContractTerm:
|
||||
"""Aktuelle Vertragsperiode samt Kündigungstermin."""
|
||||
|
||||
term_start: date
|
||||
term_end: date
|
||||
"""Letzter Tag der laufenden Periode."""
|
||||
notice_deadline: date | None
|
||||
"""Spätester Kündigungstermin; None, wenn keine Frist hinterlegt ist."""
|
||||
renews_on: date | None
|
||||
"""Beginn der Folgeperiode bei automatischer Verlängerung, sonst None."""
|
||||
is_cancelled: bool
|
||||
|
||||
|
||||
# --- Feiertage und Werktagsverschiebung ---------------------------------------
|
||||
|
||||
|
||||
@lru_cache(maxsize=16)
|
||||
def _holiday_calendar(region: str) -> holidays.HolidayBase:
|
||||
"""Feiertagskalender für eine Region wie `DE-NW`. Jahre werden bei Bedarf nachgeladen."""
|
||||
country, _, subdiv = region.partition("-")
|
||||
try:
|
||||
return holidays.country_holidays(country.upper(), subdiv=subdiv.upper() or None)
|
||||
except NotImplementedError as exc: # pragma: no cover - nur bei Fehlkonfiguration
|
||||
raise RecurrenceError(f"Unbekannte Feiertagsregion: {region}") from exc
|
||||
|
||||
|
||||
def is_business_day(day: date, region: str = "DE-NW") -> bool:
|
||||
"""Ein Werktag ist Montag bis Freitag und kein gesetzlicher Feiertag der Region."""
|
||||
if day.weekday() >= 5:
|
||||
return False
|
||||
return day not in _holiday_calendar(region)
|
||||
|
||||
|
||||
def shift_to_business_day(
|
||||
day: date,
|
||||
shift: BusinessDayShift = BusinessDayShift.NEXT,
|
||||
region: str = "DE-NW",
|
||||
) -> date:
|
||||
"""Verschiebt einen Termin auf den nächsten bzw. vorherigen Werktag."""
|
||||
if shift is BusinessDayShift.NONE:
|
||||
return day
|
||||
|
||||
step = timedelta(days=1 if shift is BusinessDayShift.NEXT else -1)
|
||||
candidate = day
|
||||
for _ in range(MAX_SHIFT_DAYS):
|
||||
if is_business_day(candidate, region):
|
||||
return candidate
|
||||
candidate += step
|
||||
# Unerreichbar bei realen Kalendern; lieber das Originaldatum als eine Endlosschleife.
|
||||
return day
|
||||
|
||||
|
||||
# --- RRULE --------------------------------------------------------------------
|
||||
|
||||
|
||||
def build_rule(rrule: str, dtstart: date):
|
||||
"""Parst eine RRULE ohne DTSTART und bindet sie an den Startzeitpunkt."""
|
||||
text = (rrule or "").strip()
|
||||
if not text:
|
||||
raise InvalidRRuleError("Die Wiederholungsregel darf nicht leer sein.")
|
||||
if "DTSTART" in text.upper():
|
||||
raise InvalidRRuleError("Die Wiederholungsregel darf kein DTSTART enthalten.")
|
||||
try:
|
||||
return rrulestr(text, dtstart=_as_datetime(dtstart))
|
||||
except Exception as exc:
|
||||
raise InvalidRRuleError(f"Ungültige Wiederholungsregel: {exc}") from exc
|
||||
|
||||
|
||||
def validate_rrule(rrule: str, dtstart: date) -> str:
|
||||
"""Prüft eine RRULE und liefert sie normalisiert zurück. Wirft `InvalidRRuleError`."""
|
||||
normalised = (rrule or "").strip()
|
||||
rule = build_rule(normalised, dtstart)
|
||||
# Eine Regel, die niemals feuert, ist mit Sicherheit ein Eingabefehler.
|
||||
horizon = _as_datetime(dtstart + relativedelta(years=25))
|
||||
if not rule.between(_as_datetime(dtstart), horizon, inc=True):
|
||||
raise InvalidRRuleError("Die Wiederholungsregel ergibt keine Termine.")
|
||||
return normalised
|
||||
|
||||
|
||||
def next_dates(
|
||||
recurrence: RecurrenceLike,
|
||||
*,
|
||||
count: int = 5,
|
||||
after: date | None = None,
|
||||
) -> list[date]:
|
||||
"""Die nächsten `count` nominalen Termine nach `after` (einschließlich)."""
|
||||
start = after or _today()
|
||||
end = start
|
||||
results: list[date] = []
|
||||
# Fenster schrittweise vergrößern, bis genug Termine gefunden sind.
|
||||
for years in (1, 5, 25):
|
||||
end = start + relativedelta(years=years)
|
||||
results = [item.nominal_date for item in expand(recurrence, start, end)]
|
||||
if len(results) >= count:
|
||||
break
|
||||
return results[:count]
|
||||
|
||||
|
||||
# --- Kernfunktion -------------------------------------------------------------
|
||||
|
||||
|
||||
def expand(
|
||||
recurrence: RecurrenceLike,
|
||||
window_start: date,
|
||||
window_end: date,
|
||||
*,
|
||||
amount_versions: Sequence[AmountVersionLike] | None = None,
|
||||
occurrences: Sequence[OccurrenceLike] | None = None,
|
||||
) -> list[PlannedOccurrence]:
|
||||
"""Expandiert eine Recurrence über ein Zeitfenster.
|
||||
|
||||
Das Fenster bezieht sich auf das **nominale** Datum. Eine Verschiebung auf den
|
||||
nächsten Werktag kann `due_date` daher über `window_end` hinausschieben; wer nach
|
||||
dem tatsächlichen Zahltag filtern will, nutzt `expand_by_due_date`.
|
||||
|
||||
Args:
|
||||
recurrence: Die auszuwertende Regel.
|
||||
window_start: Erster Tag des Fensters (einschließlich).
|
||||
window_end: Letzter Tag des Fensters (einschließlich).
|
||||
amount_versions: Preishistorie. `None` übernimmt `recurrence.amount_versions`,
|
||||
falls vorhanden.
|
||||
occurrences: Materialisierte Fälligkeiten, die das Ergebnis überlagern.
|
||||
"""
|
||||
if window_end < window_start:
|
||||
return []
|
||||
|
||||
versions = _resolve_versions(recurrence, amount_versions)
|
||||
overlays = _index_overlays(occurrences)
|
||||
|
||||
rule = build_rule(recurrence.rrule, recurrence.dtstart)
|
||||
|
||||
# Ratenzahlung: harte Obergrenze an Vorkommen, unabhängig von COUNT in der RRULE.
|
||||
installment_numbers, installment_cutoff = _installment_plan(rule, recurrence)
|
||||
|
||||
series_end = _series_end(recurrence, installment_cutoff)
|
||||
effective_end = window_end if series_end is None else min(window_end, series_end)
|
||||
if effective_end < window_start:
|
||||
return []
|
||||
|
||||
nominal_dates = rule.between(_as_datetime(window_start), _as_datetime(effective_end), inc=True)
|
||||
|
||||
results: list[PlannedOccurrence] = []
|
||||
for moment in nominal_dates:
|
||||
nominal = moment.date()
|
||||
overlay = overlays.get(nominal)
|
||||
results.append(
|
||||
PlannedOccurrence(
|
||||
recurrence_id=getattr(recurrence, "id", None),
|
||||
kind=recurrence.kind,
|
||||
nominal_date=nominal,
|
||||
due_date=shift_to_business_day(
|
||||
nominal, recurrence.business_day_shift, recurrence.holiday_region
|
||||
),
|
||||
amount=resolve_amount(recurrence, nominal, versions),
|
||||
status=overlay.status if overlay else OccurrenceStatus.PLANNED,
|
||||
is_variable=recurrence.is_variable,
|
||||
occurrence_id=getattr(overlay, "id", None) if overlay else None,
|
||||
actual_amount=overlay.actual_amount if overlay else None,
|
||||
actual_date=overlay.actual_date if overlay else None,
|
||||
account_id=(
|
||||
overlay.account_id
|
||||
if overlay and overlay.account_id is not None
|
||||
else getattr(recurrence, "account_id", None)
|
||||
),
|
||||
note=overlay.note if overlay else None,
|
||||
installment_number=installment_numbers.get(nominal),
|
||||
installments_total=recurrence.installments_total,
|
||||
)
|
||||
)
|
||||
return results
|
||||
|
||||
|
||||
def expand_by_due_date(
|
||||
recurrence: RecurrenceLike,
|
||||
window_start: date,
|
||||
window_end: date,
|
||||
*,
|
||||
amount_versions: Sequence[AmountVersionLike] | None = None,
|
||||
occurrences: Sequence[OccurrenceLike] | None = None,
|
||||
) -> list[PlannedOccurrence]:
|
||||
"""Wie `expand`, filtert aber nach dem tatsächlichen Zahltag (`effective_date`).
|
||||
|
||||
Für Kalender und Monatsauswertungen ist das die richtige Sicht: eine Rate vom
|
||||
31.05., die auf den 02.06. rutscht, gehört in den Juni.
|
||||
"""
|
||||
padding = timedelta(days=DUE_DATE_WINDOW_PADDING_DAYS)
|
||||
candidates = expand(
|
||||
recurrence,
|
||||
window_start - padding,
|
||||
window_end + padding,
|
||||
amount_versions=amount_versions,
|
||||
occurrences=occurrences,
|
||||
)
|
||||
return [item for item in candidates if window_start <= item.effective_date <= window_end]
|
||||
|
||||
|
||||
def resolve_amount(
|
||||
recurrence: RecurrenceLike,
|
||||
on: date,
|
||||
amount_versions: Sequence[AmountVersionLike] | None = None,
|
||||
) -> Decimal:
|
||||
"""Der zum Stichtag gültige Sollbetrag: größtes `valid_from <= on`."""
|
||||
versions = _resolve_versions(recurrence, amount_versions)
|
||||
applicable = [version for version in versions if version.valid_from <= on]
|
||||
if not applicable:
|
||||
return recurrence.amount
|
||||
return max(applicable, key=lambda version: version.valid_from).amount
|
||||
|
||||
|
||||
# --- Raten --------------------------------------------------------------------
|
||||
|
||||
|
||||
def installments_remaining(
|
||||
recurrence: RecurrenceLike,
|
||||
as_of: date | None = None,
|
||||
*,
|
||||
amount_versions: Sequence[AmountVersionLike] | None = None,
|
||||
occurrences: Sequence[OccurrenceLike] | None = None,
|
||||
) -> InstallmentStatus | None:
|
||||
"""Restraten und Restschuld einer Ratenzahlung. `None`, wenn keine Raten definiert sind.
|
||||
|
||||
Als geleistet gilt jede Rate, deren Zahltag am Stichtag bereits erreicht ist.
|
||||
Die Restschuld folgt `principal_amount` abzüglich der geleisteten Beträge; ohne
|
||||
hinterlegte Darlehenssumme wird die Summe der verbleibenden Raten gebildet – das
|
||||
berücksichtigt Betragsänderungen mitten in der Serie korrekt.
|
||||
"""
|
||||
total = recurrence.installments_total
|
||||
if not total:
|
||||
return None
|
||||
|
||||
reference = as_of or _today()
|
||||
schedule = expand(
|
||||
recurrence,
|
||||
recurrence.dtstart,
|
||||
# Weit genug in die Zukunft, um garantiert alle Raten zu erfassen.
|
||||
recurrence.dtstart + relativedelta(years=100),
|
||||
amount_versions=amount_versions,
|
||||
occurrences=occurrences,
|
||||
)
|
||||
|
||||
paid = [item for item in schedule if item.effective_date <= reference and not item.is_skipped]
|
||||
outstanding = [
|
||||
item for item in schedule if item.effective_date > reference and not item.is_skipped
|
||||
]
|
||||
|
||||
paid_amount = _sum(item.effective_amount for item in paid)
|
||||
if recurrence.principal_amount is not None:
|
||||
remaining_amount = recurrence.principal_amount - paid_amount
|
||||
else:
|
||||
remaining_amount = _sum(item.amount for item in outstanding)
|
||||
|
||||
return InstallmentStatus(
|
||||
total=total,
|
||||
paid=len(paid),
|
||||
remaining=len(outstanding),
|
||||
paid_amount=paid_amount,
|
||||
remaining_amount=max(remaining_amount, ZERO),
|
||||
final_due_date=schedule[-1].due_date if schedule else None,
|
||||
)
|
||||
|
||||
|
||||
# --- Verträge und Kündigungsfristen -------------------------------------------
|
||||
|
||||
|
||||
def contract_term(recurrence: RecurrenceLike, as_of: date | None = None) -> ContractTerm | None:
|
||||
"""Die zum Stichtag laufende Vertragsperiode.
|
||||
|
||||
Die Erstlaufzeit beginnt mit `contract_start` (ersatzweise `dtstart`) und endet
|
||||
am Tag vor Ablauf von `contract_min_term_months`. Ist eine automatische
|
||||
Verlängerung hinterlegt, wird so lange verlängert, bis die Periode den Stichtag
|
||||
einschließt.
|
||||
"""
|
||||
if not recurrence.contract_min_term_months:
|
||||
return None
|
||||
|
||||
start = recurrence.contract_start or recurrence.dtstart
|
||||
reference = as_of or _today()
|
||||
|
||||
term_start = start
|
||||
term_end = start + relativedelta(months=recurrence.contract_min_term_months) - timedelta(days=1)
|
||||
renew_months = recurrence.contract_auto_renew_months
|
||||
|
||||
if renew_months:
|
||||
# Höchstens 200 Verlängerungen – schützt vor Endlosschleifen bei Fehldaten.
|
||||
for _ in range(200):
|
||||
if term_end >= reference:
|
||||
break
|
||||
term_start = term_end + timedelta(days=1)
|
||||
term_end = term_start + relativedelta(months=renew_months) - timedelta(days=1)
|
||||
|
||||
deadline: date | None = None
|
||||
if recurrence.contract_notice_period_days is not None:
|
||||
deadline = term_end - timedelta(days=recurrence.contract_notice_period_days)
|
||||
|
||||
cancelled = recurrence.contract_cancelled_at is not None
|
||||
renews_on = None if cancelled or not renew_months else term_end + timedelta(days=1)
|
||||
|
||||
return ContractTerm(
|
||||
term_start=term_start,
|
||||
term_end=term_end,
|
||||
notice_deadline=deadline,
|
||||
renews_on=renews_on,
|
||||
is_cancelled=cancelled,
|
||||
)
|
||||
|
||||
|
||||
def notice_deadline(recurrence: RecurrenceLike, as_of: date | None = None) -> date | None:
|
||||
"""Letzter Kündigungstermin: Vertragsende minus Kündigungsfrist.
|
||||
|
||||
`None`, wenn kein Vertrag hinterlegt, keine Frist gesetzt oder bereits gekündigt ist.
|
||||
"""
|
||||
term = contract_term(recurrence, as_of)
|
||||
if term is None or term.is_cancelled:
|
||||
return None
|
||||
return term.notice_deadline
|
||||
|
||||
|
||||
# --- Rücklagen ----------------------------------------------------------------
|
||||
|
||||
|
||||
def annual_burden(
|
||||
recurrence: RecurrenceLike,
|
||||
as_of: date | None = None,
|
||||
*,
|
||||
amount_versions: Sequence[AmountVersionLike] | None = None,
|
||||
) -> Decimal:
|
||||
"""Belastung der kommenden zwölf Monate aus der tatsächlichen Expansion.
|
||||
|
||||
Bewusst kein fester Intervallfaktor: eine halbjährliche Zahlung, die im Fenster
|
||||
nur einmal fällt, ergibt auch nur eine Belastung.
|
||||
"""
|
||||
start = as_of or _today()
|
||||
end = start + relativedelta(years=1) - timedelta(days=1)
|
||||
schedule = expand(recurrence, start, end, amount_versions=amount_versions)
|
||||
return _sum(item.amount for item in schedule)
|
||||
|
||||
|
||||
def monthly_reserve(
|
||||
recurrence: RecurrenceLike,
|
||||
as_of: date | None = None,
|
||||
*,
|
||||
amount_versions: Sequence[AmountVersionLike] | None = None,
|
||||
) -> Decimal:
|
||||
"""Monatlich zurückzulegender Betrag für nicht-monatliche Posten.
|
||||
|
||||
Jahresbelastung geteilt durch zwölf, kaufmännisch auf Cent gerundet. Ohne
|
||||
aktivierte Rücklagenbildung ist das Ergebnis 0,00.
|
||||
"""
|
||||
if not recurrence.reserve_enabled:
|
||||
return ZERO
|
||||
burden = annual_burden(recurrence, as_of, amount_versions=amount_versions)
|
||||
return (burden / 12).quantize(CENT, rounding=ROUND_HALF_UP)
|
||||
|
||||
|
||||
# --- Interne Helfer -----------------------------------------------------------
|
||||
|
||||
|
||||
def _today() -> date:
|
||||
"""Heutiges Datum in der fachlichen Zeitzone."""
|
||||
return datetime.now(ZoneInfo("Europe/Berlin")).date()
|
||||
|
||||
|
||||
def _as_datetime(day: date) -> datetime:
|
||||
"""dateutil rechnet intern mit datetime; die Uhrzeit ist fachlich bedeutungslos."""
|
||||
return datetime(day.year, day.month, day.day)
|
||||
|
||||
|
||||
def _sum(values: Iterable[Decimal]) -> Decimal:
|
||||
return sum(values, ZERO)
|
||||
|
||||
|
||||
def _resolve_versions(
|
||||
recurrence: RecurrenceLike, explicit: Sequence[AmountVersionLike] | None
|
||||
) -> Sequence[AmountVersionLike]:
|
||||
"""Explizit übergebene Preisversionen haben Vorrang vor denen am Objekt."""
|
||||
if explicit is not None:
|
||||
return explicit
|
||||
return getattr(recurrence, "amount_versions", None) or ()
|
||||
|
||||
|
||||
def _index_overlays(
|
||||
occurrences: Sequence[OccurrenceLike] | None,
|
||||
) -> dict[date, OccurrenceLike]:
|
||||
"""Overlays nach nominalem Datum indizieren."""
|
||||
if not occurrences:
|
||||
return {}
|
||||
return {item.occurrence_date: item for item in occurrences}
|
||||
|
||||
|
||||
def _series_end(recurrence: RecurrenceLike, installment_cutoff: date | None) -> date | None:
|
||||
"""Frühestes Serienende aus `until`, Kündigung und Ratenzahl."""
|
||||
candidates = [
|
||||
value
|
||||
for value in (recurrence.until, recurrence.contract_cancelled_at, installment_cutoff)
|
||||
if value is not None
|
||||
]
|
||||
return min(candidates) if candidates else None
|
||||
|
||||
|
||||
def _installment_plan(rule, recurrence: RecurrenceLike) -> tuple[dict[date, int], date | None]:
|
||||
"""Nummeriert die Raten ab `dtstart` und liefert das Datum der letzten Rate."""
|
||||
total = recurrence.installments_total
|
||||
if not total:
|
||||
return {}, None
|
||||
|
||||
numbers: dict[date, int] = {}
|
||||
last: date | None = None
|
||||
for index, moment in enumerate(rule, start=1):
|
||||
if index > total:
|
||||
break
|
||||
day = moment.date()
|
||||
numbers[day] = index
|
||||
last = day
|
||||
return numbers, last
|
||||
Reference in New Issue
Block a user