mirror of
https://git.hiabuto.net/C3MA/CCMA.git
synced 2026-09-06 03:20:49 +02:00
feat: add SEPA direct debit exports and notifications
This commit is contained in:
Marcel Peterkau
@@ -0,0 +1,217 @@
|
||||
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)
|
||||
)
|
||||
@@ -0,0 +1,160 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from datetime import date, datetime
|
||||
from email.message import EmailMessage
|
||||
from email.policy import SMTP
|
||||
from email.utils import format_datetime, make_msgid
|
||||
from pathlib import Path
|
||||
|
||||
from ccma.services.sepa import DirectDebit
|
||||
from ccma.storage.repository import MemberRepository, RepositoryError
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class GeneratedDebitMail:
|
||||
member_id: str
|
||||
recipient: str
|
||||
export_path: Path
|
||||
archive_path: Path
|
||||
|
||||
|
||||
def _safe_filename(value: str) -> str:
|
||||
cleaned = re.sub(r"[^A-Za-z0-9._-]+", "-", value.strip()).strip("-._")
|
||||
return cleaned or "Mitglied"
|
||||
|
||||
|
||||
def _available_path(directory: Path, filename: str) -> Path:
|
||||
candidate = directory / filename
|
||||
if not candidate.exists():
|
||||
return candidate
|
||||
stem, suffix = Path(filename).stem, Path(filename).suffix
|
||||
index = 2
|
||||
while (directory / f"{stem}-{index}{suffix}").exists():
|
||||
index += 1
|
||||
return directory / f"{stem}-{index}{suffix}"
|
||||
|
||||
|
||||
def debit_mail_bytes(
|
||||
*,
|
||||
recipient: str,
|
||||
first_name: str,
|
||||
debit: DirectDebit,
|
||||
collection_date: date,
|
||||
creditor_id: str,
|
||||
sender_name: str,
|
||||
sender_email: str,
|
||||
signature: str,
|
||||
created_at: datetime | None = None,
|
||||
) -> bytes:
|
||||
if not recipient.strip():
|
||||
raise RepositoryError("Für das Mitglied ist keine E-Mail-Adresse hinterlegt.")
|
||||
if not sender_email.strip() or "@" not in sender_email:
|
||||
raise RepositoryError("Für die Info-Mails ist eine gültige Absenderadresse erforderlich.")
|
||||
timestamp = created_at or datetime.now().astimezone()
|
||||
message = EmailMessage(policy=SMTP)
|
||||
message["Message-ID"] = make_msgid(domain=sender_email.rsplit("@", 1)[-1])
|
||||
message["Date"] = format_datetime(timestamp)
|
||||
message["From"] = f"{sender_name.strip()} <{sender_email.strip()}>"
|
||||
message["To"] = recipient.strip()
|
||||
message["Subject"] = f"SEPA-Lastschrift C3MA – {collection_date.strftime('%d.%m.%Y')}"
|
||||
message["X-Mozilla-Draft-Info"] = (
|
||||
"internal/draft; vcard=0; receipt=0; DSN=0; uuencode=0; attachmentreminder=0"
|
||||
)
|
||||
greeting = first_name.strip() or "Mitglied"
|
||||
message.set_content(
|
||||
"\n".join(
|
||||
(
|
||||
f"Hallo {greeting},",
|
||||
"",
|
||||
"hiermit informieren wir dich, dass wir den folgenden Betrag per "
|
||||
"SEPA-Lastschrift einziehen werden:",
|
||||
"",
|
||||
f"Betrag: {debit.amount:.2f} Euro",
|
||||
f"Einzugsdatum: {collection_date.strftime('%d.%m.%Y')}",
|
||||
f"Verwendungszweck: {debit.purpose}",
|
||||
f"Mandatsreferenz: {debit.mandate_reference}",
|
||||
f"Gläubiger-Identifikationsnummer: {creditor_id}",
|
||||
"",
|
||||
"Sollten die Angaben nicht stimmen, melde dich bitte zeitnah bei uns.",
|
||||
"",
|
||||
"Viele Grüße",
|
||||
signature.strip(),
|
||||
)
|
||||
),
|
||||
charset="utf-8",
|
||||
)
|
||||
return message.as_bytes()
|
||||
|
||||
|
||||
def generate_debit_mails(
|
||||
repository: MemberRepository,
|
||||
debits: list[DirectDebit],
|
||||
*,
|
||||
collection_date: date,
|
||||
output_directory: Path | str,
|
||||
sender_name: str,
|
||||
sender_email: str,
|
||||
signature: str,
|
||||
) -> tuple[list[GeneratedDebitMail], list[str]]:
|
||||
output = Path(output_directory)
|
||||
output.mkdir(parents=True, exist_ok=True)
|
||||
organization = repository.get_configuration().get("organization") or {}
|
||||
creditor_id = str(organization.get("creditor_id", "")).strip()
|
||||
if not creditor_id:
|
||||
raise RepositoryError("In den Optionen fehlt die Gläubiger-ID.")
|
||||
generated: list[GeneratedDebitMail] = []
|
||||
warnings: list[str] = []
|
||||
for debit in debits:
|
||||
member = repository.get_member(debit.member_id)
|
||||
if not member.email.strip():
|
||||
warnings.append(f"{member.member_number or member.display_name}: E-Mail-Adresse fehlt.")
|
||||
continue
|
||||
content = debit_mail_bytes(
|
||||
recipient=member.email,
|
||||
first_name=member.first_name,
|
||||
debit=debit,
|
||||
collection_date=collection_date,
|
||||
creditor_id=creditor_id,
|
||||
sender_name=sender_name,
|
||||
sender_email=sender_email,
|
||||
signature=signature,
|
||||
)
|
||||
filename = (
|
||||
f"SEPA-Info-{collection_date.isoformat()}-"
|
||||
f"{_safe_filename(member.member_number or member.display_name)}.eml"
|
||||
)
|
||||
export_path = _available_path(output, filename)
|
||||
archive_dir = repository.members_root / member.member_id / "files" / "documents" / "SEPA"
|
||||
archive_dir.mkdir(parents=True, exist_ok=True)
|
||||
archive_path = _available_path(archive_dir, filename)
|
||||
export_path.write_bytes(content)
|
||||
try:
|
||||
archive_path.write_bytes(content)
|
||||
except OSError:
|
||||
export_path.unlink(missing_ok=True)
|
||||
raise
|
||||
relative_path = archive_path.relative_to(
|
||||
repository.members_root / member.member_id / "files"
|
||||
).as_posix()
|
||||
digest = hashlib.sha256(content).hexdigest()
|
||||
repository.append_event(
|
||||
member.member_id,
|
||||
event_type="sepa_notification_generated",
|
||||
summary=f"SEPA-Info-Mail erzeugt: {archive_path.name}",
|
||||
actor_type="user",
|
||||
actor_name="Vorstand",
|
||||
references={"document": relative_path},
|
||||
data={
|
||||
"sha256": digest,
|
||||
"recipient": member.email,
|
||||
"collection_date": collection_date.isoformat(),
|
||||
"amount": f"{debit.amount:.2f}",
|
||||
},
|
||||
)
|
||||
generated.append(
|
||||
GeneratedDebitMail(member.member_id, member.email, export_path, archive_path)
|
||||
)
|
||||
return generated, warnings
|
||||
Reference in New Issue
Block a user