from __future__ import annotations import csv import io import re import unicodedata import xml.etree.ElementTree as ET from dataclasses import dataclass from datetime import date, datetime from decimal import Decimal from pathlib import Path from typing import Any from uuid import uuid4 from ccma.domain.contributions import claim_balance from ccma.storage.repository import MemberRepository, RepositoryError, validate_iban PAIN_NAMESPACE = "urn:iso:std:iso:20022:tech:xsd:pain.008.001.08" SEPA_CHARACTERS = frozenset( "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" "':?,-(+.)/ ÄäÖöÜüß&*$%" ) ET.register_namespace("", PAIN_NAMESPACE) @dataclass(frozen=True, slots=True) class DirectDebit: member_id: str member_number: str account_holder: str iban: str bic: str mandate_reference: str mandate_signed_at: str amount: Decimal purpose: str claim_ids: tuple[str, ...] def _safe(value: Any, maximum: int) -> str: replacements = str.maketrans({"’": "'", "‘": "'", "´": "'", "`": "'", "–": "-", "—": "-"}) source = str(value).translate(replacements) normalized: list[str] = [] for character in source: if character in SEPA_CHARACTERS: normalized.append(character) continue decomposed = unicodedata.normalize("NFKD", character) replacement = "".join( part for part in decomposed if not unicodedata.combining(part) and part in SEPA_CHARACTERS ) normalized.append(replacement or " ") return re.sub(r"\s+", " ", "".join(normalized)).strip()[:maximum] def pending_direct_debits( repository: MemberRepository, *, due_until: date, due_from: date | None = None ) -> tuple[list[DirectDebit], list[str]]: if due_from and due_from > due_until: raise RepositoryError("„Fällig ab“ darf nicht nach „Fällig bis“ liegen.") debits: list[DirectDebit] = [] warnings: list[str] = [] for member in repository.list_members(): if not member.mandate_active: continue missing = [ label for value, label in ( (member.iban, "IBAN"), (member.mandate_reference, "Mandatsreferenz"), (member.mandate_signed_at, "Mandatsdatum"), ) if not str(value).strip() ] if missing: warnings.append(f"{member.member_number or member.display_name}: {', '.join(missing)} fehlt.") continue try: validate_iban(member.iban) date.fromisoformat(member.mandate_signed_at) except (RepositoryError, ValueError) as exc: warnings.append(f"{member.member_number or member.display_name}: {exc}") continue data = repository.get_contributions(member.member_id) selected = [] amount = Decimal("0") for claim in data.claims: if str(claim.get("status", "")) == "cancelled": continue try: claim_due = date.fromisoformat(str(claim.get("due_date", ""))) except ValueError: continue balance = claim_balance(data, claim) if (due_from is None or claim_due >= due_from) and claim_due <= due_until and balance > 0: selected.append(claim) amount += balance if not selected: continue titles = [_safe(claim.get("title") or "Forderung", 70) for claim in selected] purpose = _safe(f"{member.member_number} " + ", ".join(dict.fromkeys(titles)), 140) debits.append( DirectDebit( member_id=member.member_id, member_number=member.member_number, account_holder=member.account_holder.strip() or member.display_name, iban=member.iban.replace(" ", "").upper(), bic=member.bic.replace(" ", "").upper(), mandate_reference=_safe(member.mandate_reference, 35), mandate_signed_at=member.mandate_signed_at, amount=amount.quantize(Decimal("0.01")), purpose=purpose, claim_ids=tuple(str(claim.get("claim_id", "")) for claim in selected), ) ) return sorted(debits, key=lambda item: (item.member_number, item.account_holder)), warnings def csv_text(debits: list[DirectDebit], *, collection_date: date) -> str: target = io.StringIO(newline="") writer = csv.writer(target, delimiter=";", lineterminator="\r\n") writer.writerow( ["Einzugsdatum", "Mitgliedsnummer", "Kontoinhaber", "IBAN", "BIC", "Betrag", "Waehrung", "Mandatsreferenz", "Mandatsdatum", "Verwendungszweck"] ) for debit in debits: writer.writerow( [collection_date.isoformat(), debit.member_number, debit.account_holder, debit.iban, debit.bic, f"{debit.amount:.2f}".replace(".", ","), "EUR", debit.mandate_reference, debit.mandate_signed_at, debit.purpose] ) return target.getvalue() def write_csv(path: Path | str, debits: list[DirectDebit], *, collection_date: date) -> None: Path(path).write_text(csv_text(debits, collection_date=collection_date), encoding="utf-8-sig") def _el(parent: ET.Element, name: str, text: Any | None = None) -> ET.Element: child = ET.SubElement(parent, f"{{{PAIN_NAMESPACE}}}{name}") if text is not None: child.text = str(text) return child def pain008_bytes( debits: list[DirectDebit], *, collection_date: date, organization: dict[str, Any], message_id: str | None = None, created_at: datetime | None = None, ) -> bytes: if not debits: raise RepositoryError("Es gibt keine Lastschriften für den Export.") creditor_name = _safe(organization.get("name", ""), 70) creditor_iban = str(organization.get("iban", "")).replace(" ", "").upper() creditor_bic = str(organization.get("bic", "")).replace(" ", "").upper() creditor_id = _safe(organization.get("creditor_id", ""), 35) missing = [label for value, label in ( (creditor_name, "Vereinsname"), (creditor_iban, "Vereins-IBAN"), (creditor_bic, "Vereins-BIC"), (creditor_id, "Gläubiger-ID"), ) if not value] if missing: raise RepositoryError("Für den SEPA-XML-Export fehlen in den Optionen: " + ", ".join(missing) + ".") validate_iban(creditor_iban) timestamp = created_at or datetime.now().astimezone() msg_id = _safe(message_id or f"CCMA-{timestamp:%Y%m%d%H%M%S}-{uuid4().hex[:8]}", 35) total = sum((item.amount for item in debits), Decimal("0")) document = ET.Element(f"{{{PAIN_NAMESPACE}}}Document") initiation = _el(document, "CstmrDrctDbtInitn") header = _el(initiation, "GrpHdr") for name, value in (("MsgId", msg_id), ("CreDtTm", timestamp.isoformat(timespec="seconds")), ("NbOfTxs", len(debits)), ("CtrlSum", f"{total:.2f}")): _el(header, name, value) _el(_el(header, "InitgPty"), "Nm", creditor_name) payment = _el(initiation, "PmtInf") for name, value in (("PmtInfId", _safe(f"{msg_id}-1", 35)), ("PmtMtd", "DD"), ("BtchBookg", "true"), ("NbOfTxs", len(debits)), ("CtrlSum", f"{total:.2f}")): _el(payment, name, value) payment_type = _el(payment, "PmtTpInf") _el(_el(payment_type, "SvcLvl"), "Cd", "SEPA") _el(_el(payment_type, "LclInstrm"), "Cd", "CORE") _el(payment_type, "SeqTp", "RCUR") _el(payment, "ReqdColltnDt", collection_date.isoformat()) _el(_el(payment, "Cdtr"), "Nm", creditor_name) _el(_el(_el(payment, "CdtrAcct"), "Id"), "IBAN", creditor_iban) _el(_el(_el(payment, "CdtrAgt"), "FinInstnId"), "BICFI", creditor_bic) _el(payment, "ChrgBr", "SLEV") for index, debit in enumerate(debits, start=1): transaction = _el(payment, "DrctDbtTxInf") _el(_el(transaction, "PmtId"), "EndToEndId", _safe(f"{debit.member_number}-{index}", 35)) amount = _el(transaction, "InstdAmt", f"{debit.amount:.2f}") amount.set("Ccy", "EUR") direct_debit = _el(transaction, "DrctDbtTx") mandate = _el(direct_debit, "MndtRltdInf") _el(mandate, "MndtId", debit.mandate_reference) _el(mandate, "DtOfSgntr", debit.mandate_signed_at) other = _el(_el(_el(_el(direct_debit, "CdtrSchmeId"), "Id"), "PrvtId"), "Othr") _el(other, "Id", creditor_id) _el(_el(other, "SchmeNm"), "Prtry", "SEPA") financial = _el(_el(transaction, "DbtrAgt"), "FinInstnId") if debit.bic: _el(financial, "BICFI", debit.bic) else: _el(_el(financial, "Othr"), "Id", "NOTPROVIDED") _el(_el(transaction, "Dbtr"), "Nm", _safe(debit.account_holder, 70)) _el(_el(_el(transaction, "DbtrAcct"), "Id"), "IBAN", debit.iban) _el(_el(transaction, "RmtInf"), "Ustrd", debit.purpose) ET.indent(document, space=" ") return ET.tostring(document, encoding="utf-8", xml_declaration=True) def write_pain008( path: Path | str, debits: list[DirectDebit], *, collection_date: date, organization: dict[str, Any], ) -> None: Path(path).write_bytes( pain008_bytes(debits, collection_date=collection_date, organization=organization) )