feat(foundation): FOUND-008 gemeinsame Fristen-Normalisierung (Dringlichkeits-Ampel)

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV

arbeitskacheln/01_foundation.md

@@ -228,11 +228,21 @@ nicht detaillierten) siehe `00_index.md`.
   Duplikation der Kernlogik hinzufügbar.
 - **Tests:** Regressionstests der drei bestehenden Fristen-Berechnungen gegen die
   neue gemeinsame Implementierung.
-- **DoD:** offen, **P3 — reines Refactoring, kein MVP-Bestandteil**. Schnittstelle
-  geklärt (2026-09-09), bereit für Implementierung. **Referenz
-  (HiOrg-Server-Vergleich 2026-09-09):** HiOrg berücksichtigt Qualifikations-Ablauf
-  generisch bei der Diensteinteilung — technische Entsprechung wäre bei MABEA ein
-  gemeinsamer Fristen-Mechanismus statt drei getrennter Implementierungen.
+- **DoD:** ✅ Normalisierungsschicht umgesetzt (2026-09-10),
+  `app/services/fristen.py` — `fristen_fuer_chargen`/`fristen_fuer_qualifikationen`/
+  `fristen_fuer_wartung` rufen die drei bestehenden Rohfunktionen unverändert auf
+  und normalisieren nur das Ergebnis auf `FristTreffer` (Ampel + `rest_anzeige`).
+  **Bewusst NICHT umgesetzt:** Umstellung der bestehenden Aufrufer (Dashboard-/
+  Akte-/Personal-Endpunkte) auf die neue Schicht — schrittweise Migration wie
+  in der Schnittstellen-Entscheidung vorgesehen, noch nicht angegangen, aktuell
+  reine Bibliotheksfunktion ohne eigenen Endpunkt. Wartungs-Ampel für km/
+  Betriebsstunden bleibt vereinfacht auf ok/überfällig (Rohfunktion liefert kein
+  Zwischenmaß, siehe Docstring in `fristen.py`) — echte bald/kritisch-Abstufung
+  bräuchte einen Umbau von `faellige_wartungen_fuer_objekt`, aus Scope-Gründen
+  zurückgestellt. `backend/tests/test_fristen.py` als Regressionstest gegen die
+  drei Rohfunktionen. **Referenz (HiOrg-Server-Vergleich 2026-09-09):** HiOrg
+  berücksichtigt Qualifikations-Ablauf generisch bei der Diensteinteilung —
+  technische Entsprechung ist jetzt der gemeinsame Fristen-Mechanismus.
 
 ---
 

backend/app/services/fristen.py

@@ -0,0 +1,141 @@
+import enum
+
+from pydantic import BaseModel
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.objekt import Objekt
+from app.services.dashboard import bevorstehende_ablaufdaten, bevorstehende_qualifikationsablaeufe
+from app.services.wartung import faellige_wartungen_fuer_objekt
+
+
+class FristTyp(str, enum.Enum):
+    charge_ablauf = "charge_ablauf"
+    qualifikation_ablauf = "qualifikation_ablauf"
+    wartung_faellig = "wartung_faellig"
+
+
+class Dringlichkeit(str, enum.Enum):
+    ok = "ok"
+    bald = "bald"
+    kritisch = "kritisch"
+    ueberfaellig = "ueberfaellig"
+
+
+class FristTreffer(BaseModel):
+    objekt_id: int | None
+    frist_typ: FristTyp
+    bezeichnung: str
+    dringlichkeit: Dringlichkeit
+    rest_anzeige: str
+
+
+def ampel_aus_tagen(verbleibende_tage: int, warnzeitraum_tage: int) -> Dringlichkeit:
+    """FOUND-008: gemeinsamer Schwellenwert für alle datumsbasierten Fristen
+    (Charge-Ablauf, Qualifikations-Ablauf, zeitbasierte Wartungsintervalle) -
+    "kritisch" ab einem Viertel des jeweiligen Warnzeitraums, nicht fix in Tagen
+    (60 Tage Vorlauf bei Qualifikationen brauchen einen anderen Kritisch-Punkt
+    als 30 Tage bei Chargen)."""
+    if verbleibende_tage < 0:
+        return Dringlichkeit.ueberfaellig
+    if verbleibende_tage <= warnzeitraum_tage * 0.25:
+        return Dringlichkeit.kritisch
+    if verbleibende_tage <= warnzeitraum_tage:
+        return Dringlichkeit.bald
+    return Dringlichkeit.ok
+
+
+def ampel_aus_rest(rest, intervall_wert) -> Dringlichkeit:
+    """km-/Betriebsstunden-Fristen haben keine Zeitachse (Kachel-Entscheidung
+    2026-09-10: keine erfundene Tage-Umrechnung) - Ampel relativ zur
+    Intervallgröße statt zu Tagen: < 0 Rest = überfällig, < 10% kritisch,
+    < 25% bald."""
+    if rest < 0:
+        return Dringlichkeit.ueberfaellig
+    if intervall_wert and rest <= intervall_wert * 0.10:
+        return Dringlichkeit.kritisch
+    if intervall_wert and rest <= intervall_wert * 0.25:
+        return Dringlichkeit.bald
+    return Dringlichkeit.ok
+
+
+async def fristen_fuer_chargen(db: AsyncSession) -> list[FristTreffer]:
+    """Normalisiert INV-005 (app/services/dashboard.py:bevorstehende_ablaufdaten)
+    auf FristTreffer - Rohfunktion bleibt unverändert (schrittweise Migration,
+    kein Big-Bang)."""
+    treffer = []
+    for eintrag in await bevorstehende_ablaufdaten(db):
+        # Rohfunktion liefert nur abgelaufen/bald_ablaufend (kein eigenes
+        # warnzeitraum_tage im Rückgabewert, um "kritisch" davon abzuleiten -
+        # daher hier grob: <= 7 Tage Rest gilt als kritisch, unabhängig vom
+        # materialspezifischen Warnzeitraum).
+        if eintrag["status"] == "abgelaufen":
+            dringlichkeit = Dringlichkeit.ueberfaellig
+        elif eintrag["verbleibende_tage"] <= 7:
+            dringlichkeit = Dringlichkeit.kritisch
+        else:
+            dringlichkeit = Dringlichkeit.bald
+        treffer.append(
+            FristTreffer(
+                objekt_id=eintrag["objekt_id"],
+                frist_typ=FristTyp.charge_ablauf,
+                bezeichnung=f"{eintrag['material_name']} ({eintrag['objekt_name']})",
+                dringlichkeit=dringlichkeit,
+                rest_anzeige=(
+                    f"seit {abs(eintrag['verbleibende_tage'])} Tagen abgelaufen"
+                    if eintrag["verbleibende_tage"] < 0
+                    else f"in {eintrag['verbleibende_tage']} Tagen"
+                ),
+            )
+        )
+    return treffer
+
+
+async def fristen_fuer_qualifikationen(db: AsyncSession, warnzeitraum_tage: int = 60) -> list[FristTreffer]:
+    """Normalisiert PERS-007 (app/services/dashboard.py:bevorstehende_qualifikationsablaeufe)."""
+    treffer = []
+    for eintrag in await bevorstehende_qualifikationsablaeufe(db, warnzeitraum_tage):
+        dringlichkeit = ampel_aus_tagen(eintrag["verbleibende_tage"], warnzeitraum_tage)
+        treffer.append(
+            FristTreffer(
+                objekt_id=None,
+                frist_typ=FristTyp.qualifikation_ablauf,
+                bezeichnung=f"{eintrag['qualifikationstyp_name']} ({eintrag['benutzer_name']})",
+                dringlichkeit=dringlichkeit,
+                rest_anzeige=(
+                    f"seit {abs(eintrag['verbleibende_tage'])} Tagen abgelaufen"
+                    if eintrag["verbleibende_tage"] < 0
+                    else f"in {eintrag['verbleibende_tage']} Tagen"
+                ),
+            )
+        )
+    return treffer
+
+
+async def fristen_fuer_wartung(db: AsyncSession, objekt: Objekt) -> list[FristTreffer]:
+    """Normalisiert MAINT-002 (app/services/wartung.py:faellige_wartungen_fuer_objekt) -
+    die Rohfunktion liefert nur ein binäres `faellig`, hier wird daraus die
+    Ampel (kein Zwischenzustand ableitbar ohne Umbau der Rohfunktion - "bald"/
+    "kritisch" sind für km/Betriebsstunden daher konservativ nicht unterscheidbar,
+    nur ok/überfällig; künftiger Ausbauschritt könnte die Rohfunktion um
+    Rest-Distanz erweitern)."""
+    treffer = []
+    for eintrag in await faellige_wartungen_fuer_objekt(db, objekt):
+        dringlichkeit = Dringlichkeit.ueberfaellig if eintrag["faellig"] else Dringlichkeit.ok
+        if eintrag["faellig_am"] is not None:
+            rest_anzeige = eintrag["faellig_am"].isoformat()
+        elif eintrag["faellig_bei_km"] is not None:
+            rest_anzeige = f"bei {eintrag['faellig_bei_km']} km"
+        elif eintrag["faellig_bei_betriebsstunden"] is not None:
+            rest_anzeige = f"bei {eintrag['faellig_bei_betriebsstunden']} Bh"
+        else:
+            rest_anzeige = "noch nie durchgeführt" if eintrag["faellig"] else "ok"
+        treffer.append(
+            FristTreffer(
+                objekt_id=eintrag["objekt_id"],
+                frist_typ=FristTyp.wartung_faellig,
+                bezeichnung=eintrag["wartungsart"],
+                dringlichkeit=dringlichkeit,
+                rest_anzeige=rest_anzeige,
+            )
+        )
+    return treffer

backend/tests/test_fristen.py

@@ -0,0 +1,96 @@
+from datetime import date, timedelta
+
+import pytest
+
+from app.models.objekt import Objekt
+from app.models.objektposition import Objektposition
+from app.models.personal import BenutzerQualifikation, Qualifikationskategorie, Qualifikationstyp
+from app.models.stammdaten import MaterialTyp
+from app.services.fristen import Dringlichkeit, FristTyp, fristen_fuer_chargen, fristen_fuer_qualifikationen, fristen_fuer_wartung
+from app.services.wartung import erstelle_wartungsplan
+from app.schemas.wartung import WartungsplanPositionCreate
+from tests.conftest import login
+
+
+@pytest.mark.asyncio
+async def test_fristen_fuer_chargen_normalisiert_bevorstehende_ablaufdaten(
+    db_session, hauptserver_id, standort_factory, objekt_factory, material_factory
+):
+    from decimal import Decimal
+
+    standort = await standort_factory("Wache-Frist")
+    objekt = await objekt_factory(name="Rucksack Frist", code="RS-FRIST-1", standort=standort)
+    material = await material_factory("Verbandmaterial")
+    material.materialtyp = MaterialTyp.ablauf_charge
+
+    abgelaufen = Objektposition(
+        objekt_id=objekt.id,
+        material_id=material.id,
+        sollmenge_override=Decimal(1),
+        istmenge=Decimal(1),
+        ablaufdatum=date.today() - timedelta(days=5),
+    )
+    db_session.add(abgelaufen)
+    await db_session.commit()
+
+    treffer = await fristen_fuer_chargen(db_session)
+    passende = [t for t in treffer if t.objekt_id == objekt.id]
+    assert len(passende) == 1
+    assert passende[0].frist_typ == FristTyp.charge_ablauf
+    assert passende[0].dringlichkeit == Dringlichkeit.ueberfaellig
+    assert "abgelaufen" in passende[0].rest_anzeige
+
+
+@pytest.mark.asyncio
+async def test_fristen_fuer_qualifikationen_normalisiert(db_session, admin_user):
+    qualifikationstyp = Qualifikationstyp(
+        name="Führerschein-Frist", kategorie=Qualifikationskategorie.fuehrerschein
+    )
+    db_session.add(qualifikationstyp)
+    await db_session.flush()
+    zuordnung = BenutzerQualifikation(
+        benutzer_id=admin_user.id,
+        qualifikationstyp_id=qualifikationstyp.id,
+        erworben_am=date.today() - timedelta(days=300),
+        gueltig_bis=date.today() + timedelta(days=10),
+    )
+    db_session.add(zuordnung)
+    await db_session.commit()
+
+    treffer = await fristen_fuer_qualifikationen(db_session, warnzeitraum_tage=60)
+    passende = [t for t in treffer if t.bezeichnung.startswith("Führerschein-Frist")]
+    assert len(passende) == 1
+    assert passende[0].frist_typ == FristTyp.qualifikation_ablauf
+    # 10 Tage Rest bei 60 Tagen Warnzeitraum -> <= 25% (15 Tage) => kritisch
+    assert passende[0].dringlichkeit == Dringlichkeit.kritisch
+
+
+@pytest.mark.asyncio
+async def test_fristen_fuer_wartung_normalisiert_faellig(
+    client, db_session, hauptserver_id, admin_user, objekttyp_factory, standort_factory
+):
+    objekttyp = await objekttyp_factory("RTW-Frist")
+    standort = await standort_factory("Standort-Frist")
+    objekt = Objekt(
+        code="RTW-FRIST-1",
+        name="RTW Frist",
+        objekttyp_id=objekttyp.id,
+        standort_id=standort.id,
+        zustaendiger_server_id=hauptserver_id,
+    )
+    db_session.add(objekt)
+    await db_session.flush()
+
+    await erstelle_wartungsplan(
+        db_session,
+        objekttyp_id=objekttyp.id,
+        name="Plan-Frist",
+        positionen=[WartungsplanPositionCreate(wartungsart="Ölwechsel", intervalle=[])],
+    )
+    await db_session.commit()
+
+    treffer = await fristen_fuer_wartung(db_session, objekt)
+    assert len(treffer) == 1
+    assert treffer[0].frist_typ == FristTyp.wartung_faellig
+    assert treffer[0].dringlichkeit == Dringlichkeit.ueberfaellig
+    assert treffer[0].rest_anzeige == "noch nie durchgeführt"