from __future__ import annotations import copy import hashlib import os import re import shutil import subprocess import tempfile import zipfile from dataclasses import dataclass from datetime import date, datetime from pathlib import Path from xml.etree import ElementTree from ccma.domain.contributions import ( CLAIM_STATUS_LABELS, claim_balance, claim_items, claim_settled_total, claim_status, claim_total, money_text, reminder_items_total, ) from ccma.domain.dates import format_date_for_display from ccma.domain.models import MEMBERSHIP_STATUS_LABELS, Member from ccma.storage.repository import MemberRepository SUPPORTED_TEMPLATE_SUFFIXES = {".fodt", ".odt", ".ott"} PLACEHOLDER_PATTERN = re.compile(r"\{\{\s*([a-z][a-z0-9_.]*)\s*\}\}", re.IGNORECASE) REPEAT_PATTERN = re.compile(r"\{\{\s*([#/])\s*([a-z][a-z0-9_.]*)\s*\}\}", re.IGNORECASE) class DocumentError(RuntimeError): pass @dataclass(frozen=True, slots=True) class DocumentTemplate: name: str path: Path relative_path: str @dataclass(frozen=True, slots=True) class GeneratedDocument: path: Path relative_path: str sha256: str class DocumentService: def __init__(self, repository: MemberRepository): self.repository = repository self.templates_root = repository.root / "templates" def list_templates(self) -> list[DocumentTemplate]: templates: list[DocumentTemplate] = [] if not self.templates_root.is_dir(): return templates for path in sorted(self.templates_root.rglob("*")): if path.is_file() and path.suffix.casefold() in SUPPORTED_TEMPLATE_SUFFIXES: relative = path.relative_to(self.templates_root).as_posix() templates.append(DocumentTemplate(path.stem.replace("_", " "), path, relative)) return templates def compatible_templates( self, *, has_claim: bool, has_reminder: bool, ) -> list[DocumentTemplate]: compatible: list[DocumentTemplate] = [] for template in self.list_templates(): fields = _template_fields(template.path) if not has_claim and any(field.startswith("claim.") for field in fields): continue if not has_reminder and any(field.startswith("reminder.") for field in fields): continue compatible.append(template) return compatible def generate( self, template: DocumentTemplate, member_id: str, *, output_name: str, claim_id: str | None = None, reminder_id: str | None = None, ) -> GeneratedDocument: template_path = template.path.resolve() try: template_path.relative_to(self.templates_root.resolve()) except ValueError as exc: raise DocumentError("Das Template liegt nicht im Template-Verzeichnis.") from exc if not template_path.is_file() or template_path.suffix.casefold() not in SUPPORTED_TEMPLATE_SUFFIXES: raise DocumentError("Das ausgewählte OpenDocument-Template ist nicht verfügbar.") member = self.repository.get_member(member_id) data = None claim = None reminder = None if claim_id: data, claim = self.repository.get_claim(member_id, claim_id) if reminder_id: if not claim_id or data is None: raise DocumentError("Eine Mahnung benötigt den Kontext ihrer Forderung.") reminder = next( ( item for item in data.reminders if str(item.get("claim_id", "")) == claim_id and str(item.get("reminder_id", "")) == reminder_id ), None, ) if reminder is None: raise DocumentError("Die ausgewählte Mahnung wurde nicht gefunden.") organization = self.repository.get_configuration().get("organization") or {} if not isinstance(organization, dict): organization = {} values, repeats = _template_values( member, data=data, claim=claim, reminder=reminder, organization=organization, ) destination_dir = self.repository.members_root / member_id / "files" / "documents" destination_dir.mkdir(parents=True, exist_ok=True) destination = _available_path(destination_dir, _safe_pdf_name(output_name)) with tempfile.TemporaryDirectory(prefix="ccma-document-") as temporary_name: temporary = Path(temporary_name) rendered = temporary / f"rendered{template_path.suffix.casefold()}" _render_template(template_path, rendered, values, repeats) converted = _convert_to_pdf(rendered, temporary) temporary_destination = destination.with_name(f".{destination.name}.tmp") try: shutil.copyfile(converted, temporary_destination) os.replace(temporary_destination, destination) finally: temporary_destination.unlink(missing_ok=True) relative_path = destination.relative_to( self.repository.members_root / member_id / "files" ).as_posix() digest = hashlib.sha256(destination.read_bytes()).hexdigest() if reminder_id and claim_id: self.repository.register_reminder_document( member_id, claim_id, reminder_id, relative_path=relative_path, sha256=digest, template=template.relative_path, ) else: references = {"document": relative_path} if claim_id: references["claim_id"] = claim_id self.repository.append_event( member_id, event_type="document_generated", summary=f"Dokument erzeugt: {destination.name}", actor_type="user", actor_name="Vorstand", references=references, data={"template": template.relative_path, "sha256": digest}, ) return GeneratedDocument(destination, relative_path, digest) def _template_values( member: Member, *, data=None, claim: dict | None = None, reminder: dict | None = None, organization: dict | None = None, ) -> tuple[dict[str, str], dict[str, list[dict[str, str]]]]: organization = organization or {} created_at = datetime.now().astimezone() created_date = format_date_for_display(created_at.date().isoformat()) created_timestamp = created_at.strftime("%d.%m.%Y %H:%M") organization_address = " ".join( part for part in ( str(organization.get("street", "")), str(organization.get("postal_code", "")), str(organization.get("city", "")), ) if part ) values = { "document.created_date": created_date, "document.created_at": created_timestamp, "current_date": created_date, "current_datetime": created_timestamp, "member.id": member.member_id, "member.number": member.member_number, "member.first_name": member.first_name, "member.last_name": member.last_name, "member.nickname": member.nickname, "member.full_name": member.display_name, "member.email": member.email, "member.phone": member.phone, "member.birth_date": format_date_for_display(member.birth_date), "member.status": MEMBERSHIP_STATUS_LABELS.get(member.status, member.status), "member.accepted_at": format_date_for_display(member.accepted_at), "member.started_at": format_date_for_display(member.membership_started_at), "member.ended_at": format_date_for_display(member.membership_ended_at), "member.street": member.street, "member.address_addition": member.address_addition, "member.postal_code": member.postal_code, "member.city": member.city, "member.country": member.country, "member.address_line": " ".join( part for part in (member.street, member.postal_code, member.city) if part ), "member.account_holder": member.account_holder, "member.iban": member.iban, "member.bic": member.bic, "member.mandate_reference": member.mandate_reference, "member.mandate_signed_at": format_date_for_display(member.mandate_signed_at), "member.mandate_revoked_at": format_date_for_display(member.mandate_revoked_at), "member.mandate_active": "Ja" if member.mandate_active else "Nein", "organization.name": str(organization.get("name", "")), "organization.street": str(organization.get("street", "")), "organization.postal_code": str(organization.get("postal_code", "")), "organization.city": str(organization.get("city", "")), "organization.country": str(organization.get("country", "")), "organization.address_line": organization_address, "organization.email": str(organization.get("email", "")), "organization.phone": str(organization.get("phone", "")), "organization.website": str(organization.get("website", "")), "organization.iban": str(organization.get("iban", "")), "organization.bic": str(organization.get("bic", "")), "organization.creditor_id": str(organization.get("creditor_id", "")), } repeats: dict[str, list[dict[str, str]]] = {} if claim is not None and data is not None: claim_id = str(claim.get("claim_id", "")) item_lines = [ f"{item.get('description', '')}: {item.get('amount', '0.00')} EUR" for item in claim_items(claim) ] status = claim_status(data, claim) values.update( { "claim.id": claim_id, "claim.title": str(claim.get("title", "")), "claim.due_date": format_date_for_display(str(claim.get("due_date", ""))), "claim.created_date": _display_date_from_timestamp( str(claim.get("created_at", "")) ), "claim.created_at": _display_timestamp(str(claim.get("created_at", ""))), "claim.total": f"{money_text(claim_total(claim))} EUR", "claim.paid": f"{money_text(claim_settled_total(data, claim))} EUR", "claim.balance": f"{money_text(claim_balance(data, claim))} EUR", "claim.status": CLAIM_STATUS_LABELS.get(status, status), "claim.items": "; ".join(item_lines), } ) item_type_labels = { "base": "Grundposition", "product": "Produkt", "service": "Dienstleistung", "fee": "Gebühr", "discount": "Rabatt", "credit": "Gutschrift", "correction": "Korrektur", } repeats["claim.items"] = [ { "item.type": item_type_labels.get(str(item.get("type", "")), str(item.get("type", ""))), "item.description": str(item.get("description", "")), "item.quantity": str(item.get("quantity", "1")), "item.unit_price": str(item.get("unit_price", item.get("amount", ""))), "item.amount": str(item.get("amount", "")), } for item in claim_items(claim) ] if reminder is not None: reminder_status_labels = { "draft": "Entwurf", "generated": "Dokument erzeugt", "sent": "Versandt", "cancelled": "Verworfen", } channel_labels = {"email": "E-Mail", "letter": "Brief", "personal": "Persönlich"} status = str(reminder.get("status", "draft")) channel = str(reminder.get("channel", "")) reminder_items = reminder.get("items") or [] item_lines = [ f"{item.get('description', '')}: {item.get('amount', '0.00')} EUR" for item in reminder_items ] values.update( { "reminder.id": str(reminder.get("reminder_id", "")), "reminder.level": str(reminder.get("level", "")), "reminder.name": str(reminder.get("name", "")), "reminder.status": reminder_status_labels.get(status, status), "reminder.created_at": _display_timestamp(str(reminder.get("created_at", ""))), "reminder.sent_at": _display_timestamp(str(reminder.get("sent_at") or "")), "reminder.payment_deadline": format_date_for_display( str(reminder.get("payment_deadline") or "") ), "reminder.payment_deadline_days": str(reminder.get("payment_deadline_days", "")), "reminder.fee": f"{money_text(reminder_items_total(reminder))} EUR", "reminder.items": "; ".join(item_lines), "reminder.detail": str(reminder.get("detail", "")), "reminder.channel": channel_labels.get(channel, channel), } ) repeats["reminder.items"] = [ { "item.description": str(item.get("description", "")), "item.amount": str(item.get("amount", "")), } for item in reminder_items ] return values, repeats def _render_template( source: Path, destination: Path, values: dict[str, str], repeats: dict[str, list[dict[str, str]]], ) -> None: if source.suffix.casefold() == ".fodt": try: content = source.read_bytes() except OSError as exc: raise DocumentError(f"Template konnte nicht gelesen werden: {exc}") from exc destination.write_bytes(_replace_xml_placeholders(content, values, repeats)) return try: with zipfile.ZipFile(source, "r") as archive, zipfile.ZipFile(destination, "w") as output: for info in archive.infolist(): content = archive.read(info.filename) if info.filename in {"content.xml", "styles.xml"}: content = _replace_xml_placeholders(content, values, repeats) output.writestr(info, content) except (OSError, zipfile.BadZipFile) as exc: raise DocumentError(f"OpenDocument-Template ist beschädigt: {exc}") from exc def _template_fields(source: Path) -> set[str]: try: if source.suffix.casefold() == ".fodt": contents = [source.read_bytes()] else: with zipfile.ZipFile(source, "r") as archive: contents = [ archive.read(name) for name in ("content.xml", "styles.xml") if name in archive.namelist() ] except (OSError, zipfile.BadZipFile): return set() fields: set[str] = set() for content in contents: try: root = ElementTree.fromstring(content) except ElementTree.ParseError: continue for paragraph in root.iter(): if _local_name(paragraph.tag) in {"p", "h"}: combined = "".join(value for _node, _attribute, value in _text_slots(paragraph)) fields.update(match.group(1) for match in PLACEHOLDER_PATTERN.finditer(combined)) fields.update(match.group(2) for match in REPEAT_PATTERN.finditer(combined)) return fields def _replace_xml_placeholders( content: bytes, values: dict[str, str], repeats: dict[str, list[dict[str, str]]] | None = None, ) -> bytes: try: root = ElementTree.fromstring(content) except ElementTree.ParseError as exc: raise DocumentError(f"Template-XML ist beschädigt: {exc}") from exc _expand_repeating_rows(root, repeats or {}) unknown: set[str] = set() for paragraph in root.iter(): if _local_name(paragraph.tag) not in {"p", "h"}: continue slots = _text_slots(paragraph) combined = "".join(value for _element, _attribute, value in slots) matches = list(PLACEHOLDER_PATTERN.finditer(combined)) unknown.update(match.group(1) for match in matches if match.group(1) not in values) _replace_matches(slots, matches, values) if unknown: names = ", ".join(sorted(unknown)) raise DocumentError(f"Unbekannte oder im Kontext nicht verfügbare Platzhalter: {names}") return ElementTree.tostring(root, encoding="utf-8", xml_declaration=True) def _expand_repeating_rows(root, repeats: dict[str, list[dict[str, str]]]) -> None: for parent in root.iter(): for index, row in reversed(list(enumerate(list(parent)))): if _local_name(row.tag) != "table-row": continue slots = _text_slots(row) combined = "".join(value for _node, _attribute, value in slots) markers = list(REPEAT_PATTERN.finditer(combined)) starts = [match for match in markers if match.group(1) == "#"] if not starts: continue if len(starts) != 1: raise DocumentError("Eine Tabellenzeile darf genau eine Wiederholung öffnen.") section = starts[0].group(2) if not any(match.group(1) == "/" and match.group(2) == section for match in markers): raise DocumentError(f"Wiederholung {section} wird in derselben Tabellenzeile nicht beendet.") if section not in repeats: raise DocumentError(f"Wiederholung ist im aktuellen Kontext nicht verfügbar: {section}") parent.remove(row) for offset, item_values in enumerate(repeats[section]): clone = copy.deepcopy(row) clone_slots = _text_slots(clone) clone_text = "".join(value for _node, _attribute, value in clone_slots) clone_markers = list(REPEAT_PATTERN.finditer(clone_text)) _replace_matches(clone_slots, clone_markers, {section: ""}, key_group=2) clone_slots = _text_slots(clone) clone_text = "".join(value for _node, _attribute, value in clone_slots) item_matches = list(PLACEHOLDER_PATTERN.finditer(clone_text)) _replace_matches(clone_slots, item_matches, item_values) parent.insert(index + offset, clone) def _text_slots(element) -> list[tuple[object, str, str]]: slots: list[tuple[object, str, str]] = [] def collect(node) -> None: if node.text: slots.append((node, "text", node.text)) for child in node: collect(child) if child.tail: slots.append((child, "tail", child.tail)) collect(element) return slots def _replace_matches(slots, matches, values: dict[str, str], *, key_group: int = 1) -> None: boundaries: list[tuple[int, int]] = [] offset = 0 for _element, _attribute, text in slots: boundaries.append((offset, offset + len(text))) offset += len(text) for match in reversed(matches): key = match.group(key_group) if key not in values: continue start_index, start_offset = _slot_at(boundaries, match.start()) end_index, end_offset = _slot_at(boundaries, match.end() - 1) start_element, start_attribute, start_text = slots[start_index] end_element, end_attribute, end_text = slots[end_index] end_offset += 1 if start_index == end_index: replacement = start_text[:start_offset] + values[key] + start_text[end_offset:] setattr(start_element, start_attribute, replacement) slots[start_index] = (start_element, start_attribute, replacement) continue replacement = start_text[:start_offset] + values[key] setattr(start_element, start_attribute, replacement) slots[start_index] = (start_element, start_attribute, replacement) for index in range(start_index + 1, end_index): current_element, current_attribute, _current_text = slots[index] setattr(current_element, current_attribute, "") slots[index] = (current_element, current_attribute, "") suffix = end_text[end_offset:] setattr(end_element, end_attribute, suffix) slots[end_index] = (end_element, end_attribute, suffix) def _slot_at(boundaries: list[tuple[int, int]], position: int) -> tuple[int, int]: for index, (start, end) in enumerate(boundaries): if start <= position < end: return index, position - start raise DocumentError("Platzhalter konnte im Template nicht zugeordnet werden.") def _convert_to_pdf(source: Path, output_directory: Path) -> Path: executable = _office_executable() if not executable: raise DocumentError("LibreOffice/OpenOffice wurde nicht gefunden; PDF-Erzeugung ist nicht möglich.") profile = output_directory / "libreoffice-profile" command = [ executable, f"-env:UserInstallation={profile.resolve().as_uri()}", "--headless", "--convert-to", "pdf", "--outdir", str(output_directory), str(source), ] try: result = subprocess.run(command, capture_output=True, text=True, timeout=60, check=False) except (OSError, subprocess.TimeoutExpired) as exc: raise DocumentError(f"PDF-Konvertierung konnte nicht ausgeführt werden: {exc}") from exc converted = output_directory / f"{source.stem}.pdf" if result.returncode != 0 or not converted.is_file(): detail = (result.stderr or result.stdout).strip() or "keine PDF-Datei erzeugt" raise DocumentError(f"LibreOffice konnte das Dokument nicht konvertieren: {detail}") return converted def _office_executable() -> str | None: executable = shutil.which("soffice") or shutil.which("libreoffice") if executable: return executable if os.name != "nt": return None windows_roots: list[str] = [] for variable in ("PROGRAMFILES", "ProgramW6432", "PROGRAMFILES(X86)"): value = os.environ.get(variable) if value and value not in windows_roots: windows_roots.append(value) for root in windows_roots: candidate = os.path.join(root, "LibreOffice", "program", "soffice.exe") if os.path.isfile(candidate): return candidate return None def _safe_pdf_name(value: str) -> str: stem = Path(value.strip()).stem stem = re.sub(r"[^A-Za-z0-9ÄÖÜäöüß._ -]+", "-", stem).strip(" .-") if not stem: stem = f"Dokument-{date.today().isoformat()}" return f"{stem}.pdf" def _available_path(directory: Path, filename: str) -> Path: candidate = directory / filename counter = 2 while candidate.exists(): candidate = directory / f"{Path(filename).stem}-{counter}.pdf" counter += 1 return candidate def _display_timestamp(value: str) -> str: if not value: return "" try: return datetime.fromisoformat(value).strftime("%d.%m.%Y %H:%M") except ValueError: return value[:16] def _display_date_from_timestamp(value: str) -> str: if not value: return "" try: return format_date_for_display(datetime.fromisoformat(value).date().isoformat()) except ValueError: return value[:10] def _local_name(tag: str) -> str: return tag.rsplit("}", 1)[-1]