"""Plugin „export“ — Listen als CSV und PDF herunterladen. Exportiert die drei Redaktionslisten **Neuheiten**, **Planung** und **Archiv**: - **CSV** (stdlib ``csv``): Semikolon-getrennt für Excel mit deutschem Gebietsschema, UTF-8 mit Byte-Order-Mark. - **PDF** (WeasyPrint): formatierte Tabellen mit deutscher Kopfzeile, Erstell-Datum und Seitenzahlen; die Kopfzeile wiederholt sich auf jeder Seite. WeasyPrint wird erst beim PDF-Download importiert — fehlen die System-Bibliotheken, bleiben App und CSV-Export voll nutzbar. Rollen: Admin/Redakteur dürfen alles exportieren, Rezensenten nur die Planungsliste. Die Prüfung passiert serverseitig am Download-Endpunkt; auf der Export-Seite werden nicht erlaubte Formate ausgegraut angezeigt. Jeder Download wird best effort ins Audit-Log geschrieben (Aktion „exportiert“). Eigene Routen/Templates, keine Logik im Kern — die Datenzugriffe laufen über die gemeinsame SQLAlchemy-Metadata, ohne Partner-Plugins zu importieren (siehe exporte.py). """ from __future__ import annotations import logging from datetime import datetime from fastapi import Depends, HTTPException, Request from fastapi.responses import RedirectResponse, StreamingResponse from sqlalchemy.orm import Session from redaktionskern.auth.deps import AccessDenied, get_db, require_user from redaktionskern.auth.models import Role, User from redaktionskern.contracts import BasePlugin, NavEntry from .exporte import LISTEN, csv_erzeugen, dateiname, pdf_erzeugen _logger = logging.getLogger("plugins.export") REDAKTION = (Role.ADMIN.value, Role.REDAKTEUR.value) #: Medien-Typen der Downloads. MEDIA_CSV = "text/csv; charset=utf-8" MEDIA_PDF = "application/pdf" def _als_stream(daten: bytes, blockgroesse: int = 64 * 1024): """Zerlegt ein fertiges Dokument in Blöcke für die StreamingResponse.""" for start in range(0, len(daten), blockgroesse): yield daten[start : start + blockgroesse] def _download(daten: bytes, medientyp: str, name: str) -> StreamingResponse: return StreamingResponse( _als_stream(daten), media_type=medientyp, headers={"Content-Disposition": f'attachment; filename="{name}"'}, ) class ExportPlugin(BasePlugin): name = "export" title = "Export" description = ( "Listen (Neuheiten, Planung, Archiv) als CSV für Excel oder als " "PDF zum Drucken herunterladen." ) def __init__(self) -> None: super().__init__() self._routen_registrieren() # ---------- Plugin-Vertrag ---------- def navigation(self) -> list[NavEntry]: return [NavEntry(label=self.title, url="/export")] # ---------- Hilfsfunktionen ---------- @staticmethod def _definition(liste: str): definition = LISTEN.get((liste or "").lower()) if definition is None: raise HTTPException(status_code=404, detail="Unbekannte Liste.") return definition @staticmethod def _pruefe_zugriff(user: User, definition) -> None: """Admin/Redakteur dürfen alles; Rezensenten nur die Planungsliste.""" if definition.nur_redaktion and user.role not in REDAKTION: raise AccessDenied( f"Die Liste „{definition.titel}“ darf nur von Admins und " "Redakteuren exportiert werden." ) def _lade_zeilen(self, db: Session, definition): """Holt die Zeilen über den Lader der Liste (None = Partner fehlt).""" return definition.lader(db) def _audit_export( self, request: Request, user: User, liste: str, format_name: str, anzahl: int, ) -> None: """Schreibt einen Audit-Eintrag „exportiert“ (best effort).""" audit = request.app.state.registry.get("audit-log") if audit is None: return try: audit.log_sync( user, "exportiert", f"liste-{liste}", None, details={"format": format_name, "anzahl": anzahl}, ip_adresse=request.client.host if request.client else None, ) except Exception: _logger.exception("Audit-Log-Eintrag für Export fehlgeschlagen.") # ---------- Routen ---------- def _routen_registrieren(self) -> None: @self.router.get("/export") def seite( request: Request, db: Session = Depends(get_db), fehler: str = "", user: User = Depends(require_user), ): """Export-Seite: Karten je Liste mit Format-Buttons.""" karten = [] for definition in LISTEN.values(): zeilen = self._lade_zeilen(db, definition) karten.append( { "key": definition.key, "titel": definition.titel, "beschreibung": definition.beschreibung, "erlaubt": not definition.nur_redaktion or user.role in REDAKTION, "verfuegbar": zeilen is not None, "anzahl": len(zeilen) if zeilen is not None else None, } ) return self.context.templates.TemplateResponse( request=request, name="export/index.html", context={ "user": user, "titel": self.title, "karten": karten, "ist_redaktion": user.role in REDAKTION, "fehler": fehler[:300], }, ) @self.router.get("/export/{liste}/csv") def csv_download( request: Request, liste: str, db: Session = Depends(get_db), user: User = Depends(require_user), ): """Liste als Semikolon-getrennte CSV (UTF-8 mit BOM) herunterladen.""" definition = self._definition(liste) self._pruefe_zugriff(user, definition) zeilen = self._lade_zeilen(db, definition) if zeilen is None: return RedirectResponse("/export?fehler=Liste+nicht+verfügbar", status_code=303) daten = csv_erzeugen(definition.spalten, zeilen) self._audit_export(request, user, definition.key, "csv", len(zeilen)) return _download( daten, MEDIA_CSV, dateiname(definition.key, "csv"), ) @self.router.get("/export/{liste}/pdf") def pdf_download( request: Request, liste: str, db: Session = Depends(get_db), user: User = Depends(require_user), ): """Liste als PDF (WeasyPrint) herunterladen.""" definition = self._definition(liste) self._pruefe_zugriff(user, definition) zeilen = self._lade_zeilen(db, definition) if zeilen is None: return RedirectResponse("/export?fehler=Liste+nicht+verfügbar", status_code=303) try: daten = pdf_erzeugen(definition.titel, definition.spalten, zeilen) except Exception: # Fehlende System-Bibliotheken (Pango & Co.) dürfen nur diesen # Download betreffen — App und CSV-Export bleiben nutzbar. _logger.exception("PDF-Erzeugung für Liste '%s' fehlgeschlagen.", definition.key) return RedirectResponse( "/export?fehler=PDF-Erzeugung+fehlgeschlagen+%28WeasyPrint%2FSystem-Bibliotheken+fehlen%3F%29", status_code=303, ) self._audit_export(request, user, definition.key, "pdf", len(zeilen)) return _download( daten, MEDIA_PDF, dateiname(definition.key, "pdf", datetime.now()), ) plugin = ExportPlugin()