from __future__ import annotations import hashlib import re from contextlib import nullcontext 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.mail_delivery import append_to_imap_drafts, imap_session, send_via_smtp, smtp_session 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 # None for "send"/"drafts" delivery -- see reminder_mail.GeneratedReminderMail. export_path: Path | None 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, delivery_mode: str, output_directory: Path | str | None = None, sender_name: str, sender_email: str, signature: str, ) -> tuple[list[GeneratedDebitMail], list[str]]: output: Path | None = None email_settings: dict | None = None if delivery_mode == "local": if not output_directory: raise RepositoryError("Kein Zielordner für die lokale Ablage angegeben.") output = Path(output_directory) output.mkdir(parents=True, exist_ok=True) session = nullcontext(None) elif delivery_mode == "send": email_settings = repository.get_email_settings() session = smtp_session(email_settings) elif delivery_mode == "drafts": email_settings = repository.get_email_settings() session = imap_session(email_settings) else: raise RepositoryError(f"Unbekannter Versandmodus: {delivery_mode}") 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] = [] # One connection is opened for the whole batch (a run can cover dozens of # members) instead of reconnecting/re-authenticating per recipient. with session as client: 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" ) 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: Path | None = None if delivery_mode == "local": export_path = _available_path(output, filename) export_path.write_bytes(content) elif delivery_mode == "send": send_via_smtp(client, content) else: append_to_imap_drafts(client, content, folder=email_settings["imap_drafts_folder"]) try: archive_path.write_bytes(content) except OSError: if export_path is not None: 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}", "delivery_mode": delivery_mode, }, ) generated.append( GeneratedDebitMail(member.member_id, member.email, export_path, archive_path) ) return generated, warnings