7645f6a791
- Infinite Scroll ersetzt durch Zurück/Weiter + Seitenzahlen-Fenster (funktioniert ohne JavaScript) - Gemeinsames Makro seitennavigation.html (aria-current auf aktiver Seite, deaktivierte Buttons als span) - Neuheiten: Pagination bleibt beim HTMX-Filter-Swap innerhalb von #neuheiten-liste konsistent; Filter/Sortierung/Jahr bleiben in den Seiten-URLs erhalten - Audit-Log: gleiche Optik statt reiner Text-Links
303 lines
11 KiB
Python
303 lines
11 KiB
Python
"""Plugin „audit-log“ — Wer hat was wann verschoben, eingetragen oder geändert.
|
|
|
|
Voller Plugin-Vertrag: eigene Migration (Tabelle audit_eintraege), eigene
|
|
Routen/Templates (Admin-Ansicht mit Filtern und Paginierung) und eine
|
|
öffentliche API, die andere Plugins bei relevanten Ereignissen aufrufen.
|
|
|
|
Andere Plugins benutzen die API so:
|
|
|
|
audit = request.app.state.registry.get("audit-log")
|
|
if audit is not None:
|
|
await audit.log(user, "geaendert", "planungseintrag", eintrag.id,
|
|
details={"alt": alt, "neu": neu},
|
|
ip_adresse=request.client.host)
|
|
|
|
In synchronen Kontexten (FastAPI führt `def`-Routen im Threadpool aus) steht
|
|
`audit.log_sync(...)` mit denselben Argumenten bereit.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, time
|
|
from math import ceil
|
|
from urllib.parse import urlencode
|
|
|
|
from fastapi import Depends, Request
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from redaktionskern.auth.deps import get_db, require_roles, require_user
|
|
from redaktionskern.auth.models import Role, User
|
|
from redaktionskern.contracts import BasePlugin, DashboardKarte, Migration, NavEntry
|
|
|
|
from .models import AuditEintrag
|
|
|
|
SEITEN_GROESSE = 25
|
|
SYSTEM_NAME = "System"
|
|
|
|
# Kanonische Aktionstypen mit deutschen Anzeigenamen. Andere Plugins sollen
|
|
# diese Konstanten verwenden, damit die Filter-Auswahl konsistent bleibt.
|
|
AKTIONEN_ANZEIGE: dict[str, str] = {
|
|
"erstellt": "Eintrag erstellt",
|
|
"verschoben": "Eintrag verschoben",
|
|
"geaendert": "Eintrag geändert",
|
|
"geloescht": "Eintrag gelöscht",
|
|
"benachrichtigt": "Benachrichtigung versendet",
|
|
}
|
|
|
|
|
|
def _json_schoen(wert) -> str:
|
|
"""Jinja-Filter: Dict lesbar als JSON formatieren (leer → Leerstring)."""
|
|
if not wert:
|
|
return ""
|
|
return json.dumps(wert, ensure_ascii=False, indent=2, sort_keys=True)
|
|
|
|
|
|
def _aus_actor(actor: User | str | None) -> tuple[int | None, str]:
|
|
"""Normalisiert den Akteur auf (Benutzer-ID, Name im Log)."""
|
|
if actor is None:
|
|
return None, SYSTEM_NAME
|
|
if isinstance(actor, User):
|
|
return actor.id, actor.username
|
|
return None, str(actor)
|
|
|
|
|
|
def _tabelle_anlegen(conn) -> None:
|
|
"""Migration 0001: legt die Plugin-Tabelle an (portabel, idempotent)."""
|
|
AuditEintrag.__table__.create(conn, checkfirst=True)
|
|
|
|
|
|
def _parse_datum(roh: str) -> datetime | None:
|
|
"""Parst ein ISO-Datum (JJJJ-MM-TT); None bei leer, ValueError bei Unsinn."""
|
|
roh = (roh or "").strip()
|
|
if not roh:
|
|
return None
|
|
return datetime.strptime(roh, "%Y-%m-%d")
|
|
|
|
|
|
class AuditLogPlugin(BasePlugin):
|
|
name = "audit-log"
|
|
title = "Audit-Log"
|
|
description = "Wer hat was wann verschoben, eingetragen oder geändert."
|
|
|
|
# ---------- Plugin-Vertrag ----------
|
|
|
|
def migrations(self) -> list[Migration]:
|
|
return [Migration(version="0001_audit_eintraege", up=_tabelle_anlegen)]
|
|
|
|
def navigation(self) -> list[NavEntry]:
|
|
return [NavEntry(label=self.title, url=f"/{self.name}")]
|
|
|
|
def dashboard_karten(self, user: User) -> list[DashboardKarte]:
|
|
"""Dashboard: die letzten fünf Audit-Einträge (nur Admins)."""
|
|
if user.role != Role.ADMIN.value:
|
|
return []
|
|
with self.context.session_factory() as db:
|
|
letzte = db.scalars(
|
|
select(AuditEintrag).order_by(AuditEintrag.id.desc()).limit(5)
|
|
).all()
|
|
zeilen = tuple(
|
|
f"{e.erstellt_am.strftime('%d.%m. %H:%M') if e.erstellt_am else '—'}"
|
|
f" · {e.actor_name or 'System'}: {e.action}"
|
|
for e in letzte
|
|
)
|
|
return [
|
|
DashboardKarte(
|
|
titel="Letzte Audit-Log-Einträge",
|
|
zeilen=zeilen,
|
|
beschreibung="" if zeilen else "Noch keine Ereignisse protokolliert.",
|
|
link_url="/audit-log",
|
|
link_label="Zum Audit-Log",
|
|
)
|
|
]
|
|
|
|
def on_load(self, context) -> None:
|
|
super().on_load(context)
|
|
# Jinja-Filter für die JSON-Anzeige in der Admin-Ansicht.
|
|
context.templates.env.filters["json_schoen"] = _json_schoen
|
|
|
|
# ---------- Öffentliche API für andere Plugins ----------
|
|
|
|
async def log(
|
|
self,
|
|
actor: User | str | None,
|
|
action: str,
|
|
objekt_typ: str = "",
|
|
objekt_id: int | str | None = None,
|
|
details: dict | None = None,
|
|
*,
|
|
ip_adresse: str | None = None,
|
|
) -> None:
|
|
"""Protokolliert ein Ereignis (asynchrone Variante).
|
|
|
|
`details` ist ein freies Dict; die Schlüssel „alt“ und „neu“ werden
|
|
in die gleichnamigen JSON-Spalten übernommen, der Rest landet in
|
|
`details`.
|
|
"""
|
|
self._speichere(actor, action, objekt_typ, objekt_id, details, ip_adresse)
|
|
|
|
def log_sync(
|
|
self,
|
|
actor: User | str | None,
|
|
action: str,
|
|
objekt_typ: str = "",
|
|
objekt_id: int | str | None = None,
|
|
details: dict | None = None,
|
|
*,
|
|
ip_adresse: str | None = None,
|
|
) -> None:
|
|
"""Synchrone Variante von log() für `def`-Routen (Threadpool)."""
|
|
self._speichere(actor, action, objekt_typ, objekt_id, details, ip_adresse)
|
|
|
|
def _speichere(
|
|
self,
|
|
actor: User | str | None,
|
|
action: str,
|
|
objekt_typ: str,
|
|
objekt_id: int | str | None,
|
|
details: dict | None,
|
|
ip_adresse: str | None,
|
|
) -> None:
|
|
actor_id, actor_name = _aus_actor(actor)
|
|
alt = neu = rest = None
|
|
if details:
|
|
rest = dict(details)
|
|
alt = rest.pop("alt", None)
|
|
neu = rest.pop("neu", None)
|
|
if not rest:
|
|
rest = None
|
|
with self.context.session_factory() as db:
|
|
db.add(
|
|
AuditEintrag(
|
|
actor_id=actor_id,
|
|
actor_name=actor_name,
|
|
action=action,
|
|
objekt_typ=objekt_typ or "",
|
|
objekt_id="" if objekt_id is None else str(objekt_id),
|
|
alt=alt,
|
|
neu=neu,
|
|
details=rest,
|
|
ip_adresse=ip_adresse,
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
# ---------- Admin-Ansicht ----------
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
|
|
@self.router.get("/audit-log")
|
|
def ansicht(
|
|
request: Request,
|
|
db: Session = Depends(get_db),
|
|
benutzer: str = "",
|
|
aktion: str = "",
|
|
von: str = "",
|
|
bis: str = "",
|
|
seite: int = 1,
|
|
user: User = Depends(require_roles(Role.ADMIN.value)),
|
|
):
|
|
"""Admin-Ansicht: filterbar nach Benutzer, Aktion und Zeitraum."""
|
|
fehler: list[str] = []
|
|
|
|
filter_liste = []
|
|
if benutzer.strip():
|
|
filter_liste.append(AuditEintrag.actor_name == benutzer.strip())
|
|
if aktion.strip():
|
|
filter_liste.append(AuditEintrag.action == aktion.strip())
|
|
try:
|
|
von_dt = _parse_datum(von)
|
|
except ValueError:
|
|
von_dt = None
|
|
fehler.append("„Von“ ist kein gültiges Datum (JJJJ-MM-TT) — Filter ignoriert.")
|
|
try:
|
|
bis_dt = _parse_datum(bis)
|
|
except ValueError:
|
|
bis_dt = None
|
|
fehler.append("„Bis“ ist kein gültiges Datum (JJJJ-MM-TT) — Filter ignoriert.")
|
|
if von_dt is not None:
|
|
filter_liste.append(AuditEintrag.erstellt_am >= von_dt)
|
|
if bis_dt is not None:
|
|
filter_liste.append(
|
|
AuditEintrag.erstellt_am
|
|
<= datetime.combine(bis_dt.date(), time.max)
|
|
)
|
|
|
|
gesamt = db.scalar(
|
|
select(func.count()).select_from(AuditEintrag).where(*filter_liste)
|
|
) or 0
|
|
gesamt_seiten = max(1, ceil(gesamt / SEITEN_GROESSE))
|
|
seite = min(max(1, seite), gesamt_seiten)
|
|
offset = (seite - 1) * SEITEN_GROESSE
|
|
|
|
eintraege = (
|
|
db.scalars(
|
|
select(AuditEintrag)
|
|
.where(*filter_liste)
|
|
.order_by(AuditEintrag.erstellt_am.desc(), AuditEintrag.id.desc())
|
|
.offset(offset)
|
|
.limit(SEITEN_GROESSE)
|
|
)
|
|
.all()
|
|
)
|
|
|
|
# Auswahl für die Filter-Dropdowns (aus vorhandenen Einträgen).
|
|
benutzer_auswahl = [
|
|
name
|
|
for name in db.scalars(
|
|
select(AuditEintrag.actor_name).distinct().order_by(AuditEintrag.actor_name)
|
|
)
|
|
if name
|
|
]
|
|
aktionen_auswahl = db.scalars(
|
|
select(AuditEintrag.action).distinct().order_by(AuditEintrag.action)
|
|
).all()
|
|
|
|
# Paginierungs-Links mit erhaltenen Filtern.
|
|
parameter = {
|
|
schluessel: wert
|
|
for schluessel, wert in (
|
|
("benutzer", benutzer.strip()),
|
|
("aktion", aktion.strip()),
|
|
("von", von.strip()),
|
|
("bis", bis.strip()),
|
|
)
|
|
if wert
|
|
}
|
|
basis = urlencode(parameter)
|
|
|
|
def _seiten_url(nr: int) -> str:
|
|
return f"/audit-log?{basis}&seite={nr}" if basis else f"/audit-log?seite={nr}"
|
|
|
|
return self.context.templates.TemplateResponse(
|
|
request=request,
|
|
name="audit-log/index.html",
|
|
context={
|
|
"user": user,
|
|
"titel": self.title,
|
|
"eintraege": eintraege,
|
|
"gesamt": gesamt,
|
|
"seite": seite,
|
|
"gesamt_seiten":gesamt_seiten,
|
|
"seiten_groesse": SEITEN_GROESSE,
|
|
"start": offset + 1 if eintraege else 0,
|
|
"ende": offset + len(eintraege),
|
|
"url_zurueck": _seiten_url(seite - 1),
|
|
"url_weiter": _seiten_url(seite + 1),
|
|
"seiten_url": _seiten_url,
|
|
"url_ohne_seite": f"/audit-log?{basis}" if basis else "/audit-log",
|
|
"benutzer_auswahl": benutzer_auswahl,
|
|
"aktionen_auswahl": aktionen_auswahl,
|
|
"aktionen_anzeige": AKTIONEN_ANZEIGE,
|
|
"filter_benutzer": benutzer.strip(),
|
|
"filter_aktion": aktion.strip(),
|
|
"filter_von": von.strip(),
|
|
"filter_bis": bis.strip(),
|
|
"fehler": fehler,
|
|
},
|
|
)
|
|
|
|
|
|
plugin = AuditLogPlugin()
|