diff --git a/arbeitskacheln/01_foundation.md b/arbeitskacheln/01_foundation.md index 38c7a8f..297ddd9 100644 --- a/arbeitskacheln/01_foundation.md +++ b/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. --- diff --git a/backend/app/services/fristen.py b/backend/app/services/fristen.py new file mode 100644 index 0000000..44c03a3 --- /dev/null +++ b/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 diff --git a/backend/tests/test_fristen.py b/backend/tests/test_fristen.py new file mode 100644 index 0000000..cbadab2 --- /dev/null +++ b/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"