spiele-redaktion/plugins/export/exporte.py

"""Export-Logik des Plugins „export“ — Listen als CSV und PDF.

Diese Datei enthält die komplette Fachlogik des Export-Plugins:

- **CSV** über das Standardmodul ``csv``, Semikolon-getrennt für Excel mit
  deutschem Gebietsschema, kodiert als UTF-8 mit Byte-Order-Mark (BOM).
- **PDF** über WeasyPrint: sauber formatierte Tabellen mit deutscher
  Kopfzeile (auf jeder Seite wiederholt), Erstell-Datum und Seitenzahlen.
  WeasyPrint wird erst beim PDF-Aufruf importiert (lazy) — ohne die
  System-Bibliotheken (Pango usw.) bleiben CSV-Exporte und die App nutzbar.

Die Datenzugriffe lesen die Tabellen der Partner-Plugins über die gemeinsame
SQLAlchemy-Metadata (``Base.metadata.tables``) — genau wie das planung-Plugin
es vorlebt. Fehlt ein Partner-Plugin, meldet der jeweilige Lader ``None``;
die Liste erscheint dann im Export als nicht verfügbar.
"""
from __future__ import annotations

import csv
import html
import io
from dataclasses import dataclass, field
from datetime import datetime

from sqlalchemy import select
from sqlalchemy.orm import Session

from redaktionskern.auth.models import User
from redaktionskern.db import Base

# ---------------- Status-Anzeigenamen (lokale Kopien, keine Plugin-Importe) ----------------

NEUHEITEN_STATUS_ANZEIGE: dict[str, str] = {
    "neuheit": "Neuheit",
    "planung": "In Planung",
    "archiv": "Archiviert",
}

PLANUNG_STATUS_ANZEIGE: dict[str, str] = {
    "offen": "Offen",
    "in_bearbeitung": "In Bearbeitung",
    "abgeschlossen": "Abgeschlossen",
}

#: Titel, die das archiv-Plugin später automatisch setzt (12-Monats-Regel).
STATUS_ARCHIV = "archiv"


def _datum_de(wert: datetime | None) -> str:
    """Deutsches Kurzformat (TT.MM.JJJJ HH:MM); leer bei None."""
    return wert.strftime("%d.%m.%Y %H:%M") if wert else ""


def _oder_leer(wert) -> str:
    """None/Leer → Leerstring, alles andere als Text."""
    return "" if wert is None else str(wert)


# ---------------- Listen-Definitionen ----------------


@dataclass(frozen=True)
class ListenDefinition:
    """Eine exportierbare Liste: Anzeige-Texte, Spalten und Daten-Lader."""

    key: str
    titel: str
    beschreibung: str
    nur_redaktion: bool
    spalten: list[str] = field(default_factory=list)
    lader: object = None  # Callable[[Session], list[list[str]] | None]


def _lade_neuheiten(db: Session, *, nur_archiv: bool = False) -> list[list[str]] | None:
    """Zeilen der Neuheiten-Tabelle (Archiv = Filter auf Status „archiv“).

    Gibt ``None`` zurück, wenn das neuheiten-Plugin nicht geladen ist.
    """
    tabelle = Base.metadata.tables.get("neuheiten")
    if tabelle is None:
        return None
    abfrage = select(tabelle)
    if nur_archiv:
        abfrage = abfrage.where(tabelle.c.status == STATUS_ARCHIV)
    zeilen: list[list[str]] = []
    # Core-Select auf die Tabelle: jede Zeile ist ein Row mit Attributzugriff.
    for zeile in db.execute(abfrage.order_by(tabelle.c.titel.asc(), tabelle.c.id.asc())):
        zeilen.append(
            [
                zeile.titel,
                _oder_leer(zeile.verlag),
                _oder_leer(zeile.autor),
                _oder_leer(zeile.erscheinungsjahr),
                NEUHEITEN_STATUS_ANZEIGE.get(zeile.status, zeile.status),
                _datum_de(zeile.aktualisiert_am),
            ]
        )
    return zeilen


def _lade_planung(db: Session) -> list[list[str]] | None:
    """Zeilen der Planungsliste inklusive Rezensenten-Namen.

    Gibt ``None`` zurück, wenn das planung-Plugin nicht geladen ist.
    """
    tabelle = Base.metadata.tables.get("planungsliste")
    if tabelle is None:
        return None
    namen = {
        benutzer.id: (benutzer.display_name or benutzer.username)
        for benutzer in db.scalars(select(User)).all()
    }
    zeilen: list[list[str]] = []
    abfrage = select(tabelle).order_by(tabelle.c.titel.asc(), tabelle.c.id.asc())
    for zeile in db.execute(abfrage):
        zeilen.append(
            [
                zeile.titel,
                _oder_leer(zeile.verlag),
                _oder_leer(zeile.autor),
                _oder_leer(zeile.ausgabe),
                namen.get(zeile.rezensent_id, ""),
                PLANUNG_STATUS_ANZEIGE.get(zeile.status, zeile.status),
                _oder_leer(zeile.notizen),
                _datum_de(zeile.aktualisiert_am),
            ]
        )
    return zeilen


NEUHEITEN_SPALTEN = ["Titel", "Verlag", "Autor", "Erscheinungsjahr", "Status", "Aktualisiert am"]
PLANUNG_SPALTEN = ["Titel", "Verlag", "Autor", "Ausgabe", "Rezensent", "Status", "Notizen", "Aktualisiert am"]

#: Alle exportierbaren Listen (Reihenfolge = Reihenfolge auf der Export-Seite).
LISTEN: dict[str, ListenDefinition] = {
    definition.key: definition
    for definition in (
        ListenDefinition(
            key="neuheiten",
            titel="Neuheitenliste",
            beschreibung=(
                "Aktuelle Neuheiten aus der BoardGameGeek-Synchronisation — "
                "inklusive Status (Neuheit, in Planung, archiviert)."
            ),
            nur_redaktion=True,
            spalten=NEUHEITEN_SPALTEN,
            lader=lambda db: _lade_neuheiten(db),
        ),
        ListenDefinition(
            key="planung",
            titel="Planungsliste",
            beschreibung=(
                "Geplante Rezensionen mit Zuordnung an Rezensentin/Rezensent, "
                "Ausgabe und Bearbeitungsstatus."
            ),
            nur_redaktion=False,
            spalten=PLANUNG_SPALTEN,
            lader=_lade_planung,
        ),
        ListenDefinition(
            key="archiv",
            titel="Archiv",
            beschreibung=(
                "Archivierte Titel (älter als 12 Monate, Status „archiviert“) — "
                "gefüllt durch den Autopilot des archiv-Plugins."
            ),
            nur_redaktion=True,
            spalten=list(NEUHEITEN_SPALTEN),
            lader=lambda db: _lade_neuheiten(db, nur_archiv=True),
        ),
    )
}


# ---------------- Formaterzeugung ----------------


def csv_erzeugen(kopfzeilen: list[str], zeilen: list[list[str]]) -> bytes:
    """CSV-Dokument als Bytes: Semikolon-getrennt, UTF-8 mit BOM (Excel-DE)."""
    puffer = io.StringIO(newline="")
    writer = csv.writer(puffer, delimiter=";")
    writer.writerow(kopfzeilen)
    writer.writerows(zeilen)
    # utf-8-sig schreibt das Byte-Order-Mark (EF BB BF) — Excel erkennt
    # daran die UTF-8-Kodierung auch ohne Datei-Import-Assistenten.
    return puffer.getvalue().encode("utf-8-sig")


def _zelle(wert: str) -> str:
    return html.escape(_oder_leer(wert))


def _html_tabelle(listen_titel: str, kopfzeilen: list[str], zeilen: list[list[str]], erstellt_am: datetime) -> str:
    """Baut das HTML-Gerüst für den PDF-Druck (deutsche Kopfzeile, Datum, Seitenzahlen)."""
    kopf = "".join(f"<th>{_zelle(spalte)}</th>" for spalte in kopfzeilen)
    koerper = "".join(
        "<tr>" + "".join(f"<td>{_zelle(zelle)}</td>" for zelle in zeile) + "</tr>"
        for zeile in zeilen
    )
    leer = "" if zeilen else '<p class="hinweis">Diese Liste enthält zurzeit keine Einträge.</p>'
    datum_text = erstellt_am.strftime("%d.%m.%Y, %H:%M")
    return f"""<!doctype html>
<html lang="de">
<head><meta charset="utf-8"><title>{html.escape(listen_titel)}</title>
<style>
@page {{
    size: A4;
    margin: 20mm 15mm 18mm 15mm;
    @bottom-center {{
        content: "Seite " counter(page) " von " counter(pages);
        font-size: 9pt;
        color: #64748b;
    }}
}}
body {{ font-family: sans-serif; color: #0f172a; font-size: 10pt; }}
h1 {{ font-size: 17pt; margin: 0 0 2mm 0; }}
.meta {{ font-size: 9pt; color: #475569; margin: 0 0 6mm 0; }}
table {{ width: 100%; border-collapse: collapse; }}
thead {{ display: table-header-group; }} /* Kopfzeile auf jeder Seite */
th {{
    background-color: #ecfdf5;
    border-bottom: 1pt solid #047857;
    text-align: left;
    padding: 2mm 2.5mm;
    font-size: 9.5pt;
}}
td {{
    border-bottom: 0.4pt solid #cbd5e1;
    padding: 1.8mm 2.5mm;
    vertical-align: top;
    word-wrap: break-word;
}}
tr {{ page-break-inside: avoid; }}
.hinweis {{ color: #475569; font-style: italic; }}
</style></head>
<body>
<h1>{html.escape(listen_titel)}</h1>
<p class="meta">Spiele-Redaktion &middot; Erstellt am {datum_text} Uhr &middot;
{len(zeilen)} Eintr&auml;ge</p>
{leer}
<table>
<thead><tr>{kopf}</tr></thead>
<tbody>{koerper}</tbody>
</table>
</body></html>"""


def pdf_erzeugen(
    listen_titel: str,
    kopfzeilen: list[str],
    zeilen: list[list[str]],
    erstellt_am: datetime | None = None,
) -> bytes:
    """PDF-Dokument über WeasyPrint (Import lazy, damit CSV ohne System-Bibliotheken bleibt)."""
    from weasyprint import HTML  # noqa: PLC0415 — bewusst spät

    dokument = _html_tabelle(
        listen_titel, kopfzeilen, zeilen, erstellt_am or datetime.now()
    )
    return HTML(string=dokument).write_pdf()


def dateiname(liste: str, endung: str, jetzt: datetime | None = None) -> str:
    """ASCII-sicherer Download-Name, z. B. ``spiele-planung-export-20250821-1430.csv``."""
    stempel = (jetzt or datetime.now()).strftime("%Y%m%d-%H%M")
    return f"spiele-{liste}-export-{stempel}.{endung}"