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