P1: Rollenbasiertes Dashboard mit Plugin-Karten

- Plugin-Vertrag erweitert: optionaler Hook dashboard_karten(user) +
  DashboardKarte-Dataclass (Kern bleibt schlank, Fachlogik im Plugin)
- Karten: Neuheiten+System-Status (neuheiten), Planungs-Stats bzw.
  eigene Einträge (planung), offene Dedup-Konflikte (dedup), nächste
  Redaktionsschlüsse (erinnerung), ungelesene Benachrichtigungen
  (benachrichtigung), letzte Audit-Einträge (audit-log, nur Admin),
  Benutzer-Anzahl (Kern, nur Admin)
- 'Geladene Plugins' als aufklappbarer Admin-Technikbereich
This commit is contained in:
Flo Hartmann
2026-08-26 08:48:30 +00:00
parent 742c265df8
commit 9bd3b700c4
9 changed files with 326 additions and 17 deletions

View File

@@ -28,7 +28,7 @@ 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, Migration, NavEntry
from redaktionskern.contracts import BasePlugin, DashboardKarte, Migration, NavEntry
from .models import AuditEintrag
@@ -88,6 +88,29 @@ class AuditLogPlugin(BasePlugin):
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.

View File

@@ -25,7 +25,7 @@ from sqlalchemy.orm import Session
from redaktionskern.auth.deps import get_db, require_user
from redaktionskern.auth.models import User
from redaktionskern.contracts import BasePlugin, Migration, NavEntry
from redaktionskern.contracts import BasePlugin, DashboardKarte, Migration, NavEntry
from .adapter import BenachrichtigungsAdapter, adapter_aus_konfiguration
from .konfiguration import aus_umgebung
@@ -80,6 +80,28 @@ class BenachrichtigungPlugin(BasePlugin):
def navigation(self) -> list[NavEntry]:
return [NavEntry(label=self.title, url="/benachrichtigung")]
def dashboard_karten(self, user: User) -> list[DashboardKarte]:
"""Dashboard: Anzahl ungelesener In-App-Benachrichtigungen."""
with self.context.session_factory() as db:
ungelesen = db.scalar(
select(func.count())
.select_from(Benachrichtigung)
.where(
Benachrichtigung.user_id == user.id,
Benachrichtigung.gelesen.is_(False),
)
) or 0
return [
DashboardKarte(
titel="Ungelesene Benachrichtigungen",
wert=str(ungelesen),
beschreibung="Nachrichten in deinem Posteingang",
link_url="/benachrichtigung",
link_label="Zum Posteingang",
ton="warnung" if ungelesen else "ok",
)
]
# ---------------- Öffentliche Plugin-API ----------------
async def send_notification(

View File

@@ -33,12 +33,12 @@ import logging
import os
from fastapi import Depends, Form, Request
from sqlalchemy import select
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from redaktionskern.auth.deps import get_db, require_user
from redaktionskern.auth.models import User
from redaktionskern.contracts import BasePlugin, Migration, NavEntry
from redaktionskern.contracts import BasePlugin, DashboardKarte, Migration, NavEntry
from .bgg import BggPruefClient
from .ki_pruefung import KiPruefClient, lade_konfiguration
@@ -87,6 +87,36 @@ class DedupPlugin(BasePlugin):
def navigation(self) -> list[NavEntry]:
return [NavEntry(label=self.title, url="/dedup")]
def dashboard_karten(self, user: User) -> list[DashboardKarte]:
"""Dashboard: offene Dedup-Konflikte (nur Redaktion)."""
from redaktionskern.auth.models import Role
from redaktionskern.db import Base
if user.role not in (Role.ADMIN.value, Role.REDAKTEUR.value):
return []
tabelle = Base.metadata.tables.get("planungsliste")
if tabelle is None:
return []
with self.context.session_factory() as db:
offene = db.scalar(
select(func.count())
.select_from(tabelle)
.where(
tabelle.c.pruefung.is_not(None),
tabelle.c.status != "abgeschlossen",
)
) or 0
return [
DashboardKarte(
titel="Offene Dedup-Konflikte",
wert=str(offene),
beschreibung="Planungseinträge mit Prüfhinweis",
link_url="/planung",
link_label="Zur Planung",
ton="warnung" if offene else "ok",
)
]
def on_load(self, context) -> None:
super().on_load(context)
self._bgg_aktiv = os.environ.get("SPIELE_DEDUP_BGG_AKTIV", "1").strip() != "0"

View File

@@ -36,7 +36,13 @@ 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, Migration, NavEntry, PluginContext
from redaktionskern.contracts import (
BasePlugin,
DashboardKarte,
Migration,
NavEntry,
PluginContext,
)
# Bewusst `from .dienst import …` (wie `from .models import …` überall sonst):
# `from . import dienst` funktioniert unter dem dateibasierten Plugin-Loader
@@ -116,6 +122,39 @@ class ErinnerungPlugin(BasePlugin):
def navigation(self) -> list[NavEntry]:
return [NavEntry(label=self.title, url="/erinnerung")]
def dashboard_karten(self, user: User) -> list[DashboardKarte]:
"""Dashboard: die nächsten Redaktionsschlüsse (für alle Rollen)."""
from .dienst import heute
with self.context.session_factory() as db:
ausgaben = db.scalars(
select(Ausgabe)
.where(Ausgabe.redaktionsschluss >= heute())
.order_by(Ausgabe.redaktionsschluss)
.limit(3)
).all()
if not ausgaben:
return [
DashboardKarte(
titel="Nächste Redaktionsschlüsse",
beschreibung="Aktuell ist kein Redaktionsschluss geplant.",
link_url="/erinnerung",
)
]
zeilen = tuple(
f"{a.redaktionsschluss.strftime('%d.%m.%Y')}{a.name}"
for a in ausgaben
)
return [
DashboardKarte(
titel="Nächste Redaktionsschlüsse",
wert=str(len(ausgaben)),
zeilen=zeilen,
link_url="/erinnerung",
link_label="Alle Ausgaben",
)
]
def on_load(self, context: PluginContext) -> None:
super().on_load(context)
import os

View File

@@ -35,7 +35,13 @@ from sqlalchemy import func, or_, select
from redaktionskern.auth.deps import require_roles, require_user
from redaktionskern.auth.models import Role, User
from redaktionskern.contracts import BasePlugin, Migration, NavEntry, PluginContext
from redaktionskern.contracts import (
BasePlugin,
DashboardKarte,
Migration,
NavEntry,
PluginContext,
)
from .bgg import BggClient, BggFehler
from .models import Neuheit, QuellenStatus
@@ -394,6 +400,43 @@ class NeuheitenPlugin(BasePlugin):
def navigation(self) -> list[NavEntry]:
return [NavEntry(label=self.title, url="/neuheiten")]
def dashboard_karten(self, user: User) -> list[DashboardKarte]:
"""Dashboard: Neuheiten-Übersicht (Redaktion) + System-Status (Admin)."""
if user.role not in (Role.ADMIN.value, Role.REDAKTEUR.value):
return []
with self.context.session_factory() as db:
neu = db.scalar(
select(func.count())
.select_from(Neuheit)
.where(Neuheit.status == "neuheit")
) or 0
karten = [
DashboardKarte(
titel="Neuheiten",
wert=str(neu),
beschreibung="Offene Einträge in der Neuheitenliste",
link_url="/neuheiten",
link_label="Zu den Neuheiten",
ton="ok" if neu else "neutral",
)
]
if user.role == Role.ADMIN.value:
token = self._aktives_bgg_token()
sync_aktiv = self._scheduler is not None and self._scheduler.running
karten.append(
DashboardKarte(
titel="System-Status",
zeilen=(
f"BGG-API-Token: {'konfiguriert' if token else 'fehlt'}",
f"Hintergrund-Sync: {'aktiv' if sync_aktiv else 'aus'}",
),
ton="ok" if token else "fehler",
link_url="/neuheiten",
link_label="Neuheiten synchronisieren",
)
)
return karten
def on_load(self, context: PluginContext) -> None:
super().on_load(context)
import os

View File

@@ -33,7 +33,7 @@ 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, Migration, NavEntry
from redaktionskern.contracts import BasePlugin, DashboardKarte, Migration, NavEntry
from redaktionskern.db import Base
from .models import (
@@ -134,6 +134,55 @@ class PlanungPlugin(BasePlugin):
def navigation(self) -> list[NavEntry]:
return [NavEntry(label=self.title, url="/planung")]
def dashboard_karten(self, user: User) -> list[DashboardKarte]:
"""Dashboard: Planungs-Stats — Rezensenten ihre eigenen, die
Redaktion die Gesamtzahlen je Status."""
with self.context.session_factory() as db:
zaehler = dict(
db.execute(
select(Planungseintrag.status, func.count())
.group_by(Planungseintrag.status)
).all()
)
if user.role in REDAKTION:
return [
DashboardKarte(
titel="Planung",
wert=str(zaehler.get("offen", 0)),
beschreibung="offene Einträge",
zeilen=tuple(
f"{STATUS_ANZEIGE[s]}: {zaehler.get(s, 0)}"
for s in GUELTIGE_STATUS
),
link_url="/planung",
link_label="Zur Planung",
ton="neutral",
)
]
eigene = dict(
db.execute(
select(Planungseintrag.status, func.count())
.where(Planungseintrag.rezensent_id == user.id)
.group_by(Planungseintrag.status)
).all()
)
return [
DashboardKarte(
titel="Meine offenen Planungseinträge",
wert=str(eigene.get("offen", 0)),
link_url="/planung?status=offen",
link_label="Anzeigen",
ton="neutral",
),
DashboardKarte(
titel="Meine Einträge in Bearbeitung",
wert=str(eigene.get("in_bearbeitung", 0)),
link_url="/planung?status=in_bearbeitung",
link_label="Anzeigen",
ton="neutral",
),
]
# ---------- Hilfsfunktionen ----------
@staticmethod