from __future__ import annotations import calendar from dataclasses import dataclass, field from datetime import date, timedelta from ccma.domain.dates import ( DateValidationError, parse_iso_date, validate_birth_date, validate_member_dates, ) from ccma.domain.models import HousekeeperFinding from ccma.services.intervals import AnniversaryInterval, parse_anniversary_intervals from ccma.storage.repository import MemberRepository @dataclass(frozen=True, slots=True) class HousekeeperSettings: birthday_days_before: int = 7 birthday_days_after: int = 2 anniversary_days_before: int = 14 anniversary_days_after: int = 7 anniversary_intervals: tuple[AnniversaryInterval, ...] = field( default_factory=lambda: tuple(parse_anniversary_intervals("1Y;5Y;10Y;25Y;50Y")) ) @classmethod def from_values( cls, *, birthday_days_before: int, birthday_days_after: int, anniversary_days_before: int, anniversary_days_after: int, anniversary_intervals: str, ) -> HousekeeperSettings: return cls( birthday_days_before=min(365, max(0, birthday_days_before)), birthday_days_after=min(365, max(0, birthday_days_after)), anniversary_days_before=min(365, max(0, anniversary_days_before)), anniversary_days_after=min(365, max(0, anniversary_days_after)), anniversary_intervals=tuple(parse_anniversary_intervals(anniversary_intervals)), ) class Housekeeper: def __init__(self, repository: MemberRepository, settings: HousekeeperSettings | None = None): self.repository = repository self.settings = settings or HousekeeperSettings() def run(self, today: date | None = None) -> list[HousekeeperFinding]: current_date = today or date.today() findings: list[HousekeeperFinding] = [] for member in self.repository.list_members(): try: validate_member_dates( birth_date=member.birth_date, accepted_at=member.accepted_at, membership_started_at=member.membership_started_at, today=current_date, ) except DateValidationError as exc: findings.append( HousekeeperFinding( severity="error", member_id=member.member_id, code="invalid_member_dates", title=f"{member.display_name}: Ungültige Datumsangabe", detail=str(exc), ) ) if member.status in { "active", "suspended_contribution", "resigned_end_of_year", "honorary", }: birthday = self._birthday_finding( member.member_id, member.display_name, member.birth_date, current_date ) if birthday: findings.append(birthday) findings.extend( self._anniversary_findings( member.member_id, member.display_name, member.membership_started_at, current_date, ) ) if not member.contribution_rule_id and not member.honorary: findings.append( HousekeeperFinding( severity="error", member_id=member.member_id, code="missing_contribution_rule", title=f"{member.display_name}: Beitragsregel fehlt", detail="Dem Mitglied ist keine Beitragsregel zugeordnet.", ) ) if member.status == "accepted_pending_payment" and member.accepted_at: accepted = _parse_date(member.accepted_at) if accepted: deadline = accepted + timedelta(days=28) days = (deadline - current_date).days if days < 0: findings.append( HousekeeperFinding( severity="error", member_id=member.member_id, code="initial_payment_overdue", title=f"{member.display_name}: Erstzahlung überfällig", detail=f"Die Vierwochenfrist ist seit {-days} Tagen überschritten.", due_date=deadline, ) ) elif days <= 7: findings.append( HousekeeperFinding( severity="warning", member_id=member.member_id, code="initial_payment_due_soon", title=f"{member.display_name}: Erstzahlung bald fällig", detail=f"Die Vierwochenfrist endet in {days} Tagen.", due_date=deadline, ) ) contributions = self.repository.get_contributions(member.member_id) for claim in contributions.claims: if str(claim.get("status", "open")) not in {"open", "partially_paid"}: continue due = _parse_date(str(claim.get("due_date", ""))) if not due: continue days = (due - current_date).days title = str(claim.get("title") or "Beitragsforderung") if days < 0: findings.append( HousekeeperFinding( severity="error", member_id=member.member_id, code="claim_overdue", title=f"{member.display_name}: {title} überfällig", detail=f"Fälligkeit war vor {-days} Tagen.", due_date=due, ) ) elif days <= 14: findings.append( HousekeeperFinding( severity="info", member_id=member.member_id, code="claim_due_soon", title=f"{member.display_name}: {title} bald fällig", detail=f"Fälligkeit in {days} Tagen.", due_date=due, ) ) severity_order = {"error": 0, "warning": 1, "info": 2} return sorted( findings, key=lambda item: (severity_order.get(item.severity, 9), item.due_date or date.max) ) def _birthday_finding( self, member_id: str, name: str, birth_date_value: str, today: date, ) -> HousekeeperFinding | None: try: birth_date = validate_birth_date(birth_date_value, today=today) except DateValidationError: return None if not birth_date: return None occurrences = [_birthday_in_year(birth_date, year) for year in range(today.year - 1, today.year + 2)] occurrence = min(occurrences, key=lambda value: abs((value - today).days)) delta = (occurrence - today).days if delta > self.settings.birthday_days_before or delta < -self.settings.birthday_days_after: return None age = occurrence.year - birth_date.year title = _relative_title(name, delta, "Geburtstag") detail = f"Wird {age} Jahre alt." if delta >= 0 else f"Ist {age} Jahre alt geworden." return HousekeeperFinding( severity="info", member_id=member_id, code="birthday", title=title, detail=detail, due_date=occurrence, ) def _anniversary_findings( self, member_id: str, name: str, started_at_value: str, today: date, ) -> list[HousekeeperFinding]: try: started_at = parse_iso_date(started_at_value, "Mitglied seit") except DateValidationError: return [] if not started_at: return [] findings: list[HousekeeperFinding] = [] for interval in self.settings.anniversary_intervals: try: target = interval.target_date(started_at) except (OverflowError, ValueError): continue delta = (target - today).days if delta > self.settings.anniversary_days_before or delta < -self.settings.anniversary_days_after: continue occasion = _anniversary_name(interval) findings.append( HousekeeperFinding( severity="info", member_id=member_id, code="membership_anniversary", title=_relative_title(name, delta, occasion), detail=f"Mitglied seit {started_at:%d.%m.%Y}.", due_date=target, ) ) return findings def _parse_date(value: str) -> date | None: try: return date.fromisoformat(value[:10]) except (TypeError, ValueError): return None def _birthday_in_year(birth_date: date, year: int) -> date: day = min(birth_date.day, calendar.monthrange(year, birth_date.month)[1]) return date(year, birth_date.month, day) def _relative_title(name: str, delta: int, occasion: str) -> str: if delta == 0: return f"{name} hat heute {occasion}" days = "Tag" if abs(delta) == 1 else "Tagen" if delta > 0: return f"{name} hat in {delta} {days} {occasion}" return f"{name} hatte vor {-delta} {days} {occasion}" def _anniversary_name(interval: AnniversaryInterval) -> str: if interval.unit == "D": return f"{interval.value}-Tage-Mitgliedsjubiläum" if interval.unit == "M": return f"{interval.value}-Monats-Mitgliedsjubiläum" return f"{interval.value}-jähriges Mitgliedsjubiläum"