from __future__ import annotations from datetime import date from decimal import ROUND_HALF_UP, Decimal, InvalidOperation from typing import Any from ccma.domain.models import ContributionData CENT = Decimal("0.01") CLAIM_STATUS_LABELS = { "open": "OFFEN", "partially_paid": "TEILBEZAHLT", "paid": "BEZAHLT", "overpaid": "ÜBERZAHLT", "overdue": "ÜBERFÄLLIG", "credit": "GUTSCHRIFT", "cancelled": "STORNIERT", } DONATION_STATUS_LABELS = { "open": "OFFEN", "partially_allocated": "TEILWEISE ZUGEORDNET", "allocated": "ZUGEORDNET", "overallocated": "ÜBERZUGEORDNET", } def decimal_value(value: Any, field_name: str = "Betrag") -> Decimal: text = str(value).strip().replace(",", ".") try: return Decimal(text).quantize(CENT, rounding=ROUND_HALF_UP) except (InvalidOperation, ValueError) as exc: raise ValueError(f"{field_name} ist kein gültiger Geldbetrag.") from exc def money_text(value: Decimal | str) -> str: return f"{decimal_value(value):.2f}" def claim_items(claim: dict[str, Any]) -> list[dict[str, Any]]: items = claim.get("items") if isinstance(items, list) and items: return items amount = decimal_value(claim.get("amount", "0")) return [ { "item_id": "legacy-base", "type": "base", "description": str(claim.get("title") or "Forderung"), "quantity": "1.00", "unit_price": money_text(amount), "amount": money_text(amount), } ] def materialize_claim_items(claim: dict[str, Any]) -> list[dict[str, Any]]: if not isinstance(claim.get("items"), list) or not claim["items"]: claim["items"] = claim_items(claim) return claim["items"] def claim_total(claim: dict[str, Any]) -> Decimal: return sum((decimal_value(item.get("amount", "0")) for item in claim_items(claim)), Decimal("0")) def allocation_effect(data: ContributionData, allocation: dict[str, Any]) -> Decimal: amount = decimal_value(allocation.get("amount", "0")) if str(allocation.get("credit_id", "")): return -amount return amount def allocated_total(data: ContributionData, claim_id: str) -> Decimal: return sum( ( allocation_effect(data, allocation) for allocation in data.allocations if str(allocation.get("claim_id", "")) == claim_id ), Decimal("0"), ) def claim_settled_total(data: ContributionData, claim: dict[str, Any]) -> Decimal: allocated = allocated_total(data, str(claim.get("claim_id", ""))) if claim_total(claim) < 0: return abs(allocated).quantize(CENT) return allocated.quantize(CENT) def payment_allocated_total(data: ContributionData, payment_id: str) -> Decimal: return sum( ( decimal_value(allocation.get("amount", "0")) for allocation in data.allocations if str(allocation.get("payment_id", "")) == payment_id ), Decimal("0"), ) def credit_allocated_total(data: ContributionData, credit_id: str) -> Decimal: return sum( ( decimal_value(allocation.get("amount", "0")) for allocation in data.allocations if str(allocation.get("credit_id", "")) == credit_id ), Decimal("0"), ) def claim_balance(data: ContributionData, claim: dict[str, Any]) -> Decimal: return (claim_total(claim) - allocated_total(data, str(claim.get("claim_id", "")))).quantize(CENT) def donation_allocated_total(data: ContributionData, donation_id: str) -> Decimal: return sum( ( decimal_value(allocation.get("amount", "0")) for allocation in data.allocations if str(allocation.get("donation_id", "")) == donation_id ), Decimal("0"), ) def donation_amount(donation: dict[str, Any]) -> Decimal: return decimal_value(donation.get("amount", "0")) def donation_balance(data: ContributionData, donation: dict[str, Any]) -> Decimal: donation_id = str(donation.get("donation_id", "")) return (donation_amount(donation) - donation_allocated_total(data, donation_id)).quantize(CENT) def donation_status(data: ContributionData, donation: dict[str, Any]) -> str: balance = donation_balance(data, donation) allocated = donation_allocated_total(data, str(donation.get("donation_id", ""))) if balance < 0: return "overallocated" if balance == 0: return "allocated" if allocated > 0: return "partially_allocated" return "open" def claim_status(data: ContributionData, claim: dict[str, Any], *, today: date | None = None) -> str: if str(claim.get("status", "")) == "cancelled": return "cancelled" total = claim_total(claim) paid = allocated_total(data, str(claim.get("claim_id", ""))) balance = total - paid if total < 0: return "credit" if balance < 0: return "overpaid" if balance == 0: return "paid" if paid > 0: return "partially_paid" try: due = date.fromisoformat(str(claim.get("due_date", ""))) except ValueError: due = None if due and due < (today or date.today()): return "overdue" return "open"