"""SEPA-XML-Erstellung: Lastschrift (SepaDD) und Gutschrift (SepaCT).""" import logging import math import os import random from datetime import datetime, date, timedelta from SEPA import SepaCT, SepaDD from core.models import _AV_KONSTANTEN, DEFAULT_DATE, Mitglied from core.utils import mask_name_for_log logger = logging.getLogger(__name__) def create_sepa_xml_lastschrift(system) -> None: """Erstellt die SEPA-Lastschrift-XML für Mitglieder mit positiven Kosten. system: Abrechnungssystem-Instanz. """ try: _validate_sepa_folder(system.folder_sepa) lastschrift_mitglieder = [m for m in system.mitglieder if m.Kosten > 0] if not lastschrift_mitglieder: logger.info("Keine Mitglieder für SEPA-Lastschrift gefunden.") return _validate_sepa_config(system) _iban, _bic, _creditor = _sepa_ids(system) sepa_config = { "name": system.config["Vereinsname"], "IBAN": _iban, "BIC": _bic, "creditor_id": _creditor, } random_suffix = _random_suffix() zvr = str(system.config.get("ZVR", "")).strip() if not zvr: logger.warning("ZVR-Nummer fehlt – SEPA-Lastschrift-XML wird ohne ZVR erstellt.") sepa = SepaDD(sepa_config) sepa.add_group_header( message_id=getattr(system, "_ls_msgid", None) or f"{system.config['Vereinsname']}-LS-{system.periode}-{random_suffix}", creation_datetime=datetime.now(), initiating_party_name=system.config["Vereinsname"], ZVR=zvr, ) sepa.add_payment_info( payment_id=f"Lastschrift-{system.periode}-{_random_suffix()}", requested_execution_date=system.datum_abbuchung, creditor_name=system.config["Vereinsname"], creditor_iban=system.config["SEPA_IBAN"], creditor_bic=system.config["SEPA_BIC"], creditor_id=system.config["SEPA_Creditor_ID"], ) for idx, mitglied in enumerate(lastschrift_mitglieder, start=1): try: # Mandatsreferenz = Zählpunktnummer (Bezug), global eindeutig in AT # Fallback auf Mitglieds-ID falls ZP_Consumption fehlt _zp_raw = str(getattr(mitglied, "ZP_Consumption", "") or "").strip().replace(" ", "") _mandate_ref = _zp_raw if _zp_raw and _zp_raw.lower() not in ("nan", "none", "") else str(mitglied.ID) # Mandatsdatum = Eintrittsdatum (= Vertragsunterzeichnung); Fallback: heute _eintritt = getattr(mitglied, "Eintrittsdatum", DEFAULT_DATE) _mandate_date = ( _eintritt.date() if (_eintritt and _eintritt != DEFAULT_DATE) else date.today() ) sepa.add_transaction( amount=mitglied.Kosten, debtor_name=mitglied.Name, debtor_iban=mitglied.IBAN, debtor_bic=mitglied.BIC, mandate_id=_mandate_ref, mandate_date=_mandate_date, reference=f"{system.config['Vereinsname']}-{system.periode}-{mitglied.Rechnungsnummer}", id=idx, ) except Exception as e: logger.error(f"Fehler bei SEPA-Transaktion für {mask_name_for_log(mitglied.Name)}: {e}") raise output_filename = ( f"{system.config['Vereinsname']}-SEPA_Lastschrift_" f"{datetime.now().strftime('%Y-%m-%d')}.xml" ) output_path = os.path.join(system.folder_sepa, output_filename) sepa.export(output_path) logger.info(f"SEPA-Lastschrift-XML erstellt: {output_path}") except (FileNotFoundError, PermissionError, ValueError) as e: logger.error(f"SEPA-Lastschrift Fehler: {e} – ABBRUCH") raise except Exception as e: logger.error(f"Unbekannter Fehler bei SEPA-Lastschrift: {e} – ABBRUCH") raise def create_sepa_xml_gutschrift(system) -> None: """Erstellt die SEPA-Gutschrift-XML (Credit Transfer) für Mitglieder mit negativen Kosten. system: Abrechnungssystem-Instanz. """ try: _validate_sepa_folder(system.folder_sepa) gutschrift_mitglieder = [m for m in system.mitglieder if m.Kosten < 0] if not gutschrift_mitglieder: logger.info("Keine Mitglieder für SEPA-Gutschrift gefunden.") return _validate_sepa_config(system) sepa_ct = _create_sepa_ct_object(system) _add_gutschrift_transactions(sepa_ct, gutschrift_mitglieder, system) output_path = os.path.join( system.folder_sepa, f"{system.config['Vereinsname']}-SEPA_Gutschrift_" f"{datetime.now().strftime('%Y-%m-%d')}.xml", ) _export_sepa_ct(sepa_ct, output_path, system) logger.info(f"SEPA-Gutschrift-XML erstellt: {output_path}") except (FileNotFoundError, PermissionError, ValueError) as e: logger.error(f"SEPA-Gutschrift Fehler: {e} – ABBRUCH") raise except Exception as e: logger.error(f"Unbekannter Fehler bei SEPA-Gutschrift: {e} – ABBRUCH") raise # ── Hilfsfunktionen ─────────────────────────────────────────────────────────── def _random_suffix(k: int = 6) -> str: """Erzeugt einen zufälligen alphanumerischen Suffix der Länge k.""" return "".join(random.choices( "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789", k=k )) def _sepa_ids(system) -> tuple: """Gibt (IBAN, BIC, Creditor_ID) zurück – im AV-Modus aus den AV-Konstanten, sonst aus Config.""" if _is_av_mode(system): return ( _AV_KONSTANTEN["iban"], _AV_KONSTANTEN["bic"], _AV_KONSTANTEN["creditor_id"], ) return ( system.config.get("SEPA_IBAN", ""), system.config.get("SEPA_BIC", ""), system.config.get("SEPA_Creditor_ID", ""), ) def _is_av_mode(system) -> bool: """True wenn Abbuchungsmodus = abrechnungsverein.""" return system.config.get( "Abbuchung_Modus", system.config.get("abbuchung_modus", "ueberweisung") ).strip().lower() == "abrechnungsverein" def _validate_sepa_folder(sepa_folder: str) -> None: """Prüft ob der SEPA-Ordner existiert und schreibbar ist.""" if not os.path.exists(sepa_folder): raise FileNotFoundError(f"SEPA-Ordner existiert nicht: {sepa_folder}") if not os.access(sepa_folder, os.W_OK): raise PermissionError(f"Keine Schreibrechte für SEPA-Ordner: {sepa_folder}") def _validate_sepa_config(system) -> None: """Prüft ob alle SEPA-Konfigurationswerte gültig sind. Im AV-Modus werden IBAN/BIC/Creditor_ID nicht geprüft (kommen aus _AV_KONSTANTEN). """ iban, bic, creditor = _sepa_ids(system) abbuchung_modus = system.config.get( "Abbuchung_Modus", system.config.get("abbuchung_modus", "ueberweisung") ).strip().lower() required_fields = { "Vereinsname": (system.config.get("Vereinsname", ""), lambda x: bool(x.strip())), "Vereinsstrasse": (system.config.get("Vereinsstrasse", ""), lambda x: bool(x.strip())), "Vereinsort": (system.config.get("Vereinsort", ""), lambda x: bool(x.strip())), } if abbuchung_modus == "eigenes_konto": required_fields["SEPA_IBAN"] = (iban, lambda x: len(x.replace(" ", "")) >= 15 and x.replace(" ", "").isalnum()) required_fields["SEPA_BIC"] = (bic, lambda x: len(x) in [8, 11] and x.isalnum() and x.isupper()) required_fields["SEPA_Creditor_ID"] = (creditor, lambda x: x.isalnum() and len(x) <= 35) for field, (value, validator) in required_fields.items(): if not value: raise ValueError(f"Fehlende SEPA-Konfiguration: {field}") if not validator(value): raise ValueError(f"Ungültige SEPA-Konfiguration: {field} = {value}") def _create_sepa_ct_object(system) -> SepaCT: """Erstellt und konfiguriert ein SEPA-CT-Objekt für Gutschriften.""" sepa_ct = SepaCT(config={}) sepa_ct.add_group_header( message_id=getattr(system, "_gs_msgid", None) or f"{system.config['Vereinsname']}-GS-{system.periode}-{_random_suffix()}", creation_datetime=datetime.now(), ) plz_ort = system.config["Vereinsort"] plz, *ort_teile = plz_ort.split(" ", 1) ort = " ".join(ort_teile) sepa_ct.add_payment_info( payment_id=f"Gutschrift-{system.periode}-{_random_suffix()}", requested_execution_date=system.datum_abbuchung + timedelta(days=1), debtor_name=system.config["Vereinsname"], debtor_address={ "Vereinsadresse": system.config["Vereinsstrasse"], "Vereinsort": ort, "PLZ": plz, }, debtor_iban=_sepa_ids(system)[0], debtor_bic=_sepa_ids(system)[1], ) return sepa_ct def _add_gutschrift_transactions(sepa_ct: SepaCT, gutschrift_mitglieder: list, system) -> None: """Fügt Gutschrift-Transaktionen zum SEPA-CT-Objekt hinzu.""" for mitglied in gutschrift_mitglieder: try: amount = abs(float(mitglied.Kosten)) if math.isnan(amount): logger.error(f"Ungültiger Betrag (NaN) für {mask_name_for_log(mitglied.Name)} – überspringe.") continue if not mitglied.IBAN or not mitglied.BIC: logger.error(f"{mask_name_for_log(mitglied.Name)} hat keine gültige IBAN/BIC – überspringe.") continue sepa_ct.add_transaction( amount=amount, creditor_name=mitglied.Name4Rechnung, creditor_address={ "Strasse": mitglied.Strasse, "PLZ": mitglied.PLZ, "Ort": mitglied.Ort, }, creditor_iban=mitglied.IBAN, creditor_bic=mitglied.BIC, reference=( f"{system.config['Vereinsname']}-{mitglied.ID}-" f"{system.periode}-{mitglied.Rechnungsnummer}" ), id=mitglied.ID, ) except Exception as e: logger.error(f"Fehler bei Transaktion für {mask_name_for_log(mitglied.Name)}: {e} – überspringe.") continue def _export_sepa_ct(sepa_ct: SepaCT, output_path: str, system) -> None: """Exportiert das SEPA-CT-Objekt als XML-Datei.""" try: sepa_ct.export( output_path, initiating_party_name=system.config["Vereinsname"], initiating_party_address={ "Vereinsadresse": system.config["Vereinsstrasse"], "Vereinsort": system.config["Vereinsort"], }, ) except Exception as e: raise IOError(f"Fehler beim Exportieren der SEPA-Gutschrift: {e}")