scripte/MABEA
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Refactor: große Dateien in kleinere Module/Komponenten aufgeteilt
CI / backend-tests (push) Successful in 53s
Details
CI / frontend-build (push) Successful in 34s
Details

Browse Source 
Explizite Vorgabe: keine wachsenden Monolith-Dateien. AdminPage.tsx (349
Zeilen) in pages/admin/{Standort,Material,Objekt}Section.tsx aufgeteilt,
KontrollPage.tsx (277 Zeilen) in pages/kontrolle/{useKontrolle,
PositionCard,StatusBadge,types} aufgeteilt, services/kontrolle.py (332
Zeilen) in ein Package (lifecycle/erfassung/exceptions) aufgeteilt -
öffentliche API bleibt über __init__.py unverändert re-exportiert, kein
bestehender Import musste angepasst werden. Reines Refactoring, keine
Verhaltensänderung.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01L85hmKbvX7Cqkq47KnQhFt
This commit is contained in:
patrick
2026-09-04 00:45:18 +02:00
co-authored by Claude Sonnet 5
parent 08b10c7c37
commit f6084e2d5a
14 changed files with 1079 additions and 661 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/services/kontrolle/__init__.py
+30
View File
@@ -0,0 +1,30 @@
"""Kontroll-Domänenlogik (Prompt 02.3/09/16), aufgeteilt in Module:
- exceptions: ObjektGesperrtError/UnvollstaendigError
- lifecycle: Start/Übernahme/Abschluss/Abbruch, Mindermenge-Ablauf (U6)
- erfassung: Position erfassen + Fehlbestand-Synchronisierung
Öffentliche API bleibt hier re-exportiert, damit bestehender Code
(`from app.services.kontrolle import starte_kontrolle`) unverändert funktioniert.
"""
from .erfassung import erfasse_position
from .exceptions import ObjektGesperrtError, UnvollstaendigError
from .lifecycle import (
abbreche_kontrolle,
abschliesse_kontrolle,
aktive_kontrolle,
fehlende_material_ids,
starte_kontrolle,
)
__all__ = [
"ObjektGesperrtError",
"UnvollstaendigError",
"abbreche_kontrolle",
"abschliesse_kontrolle",
"aktive_kontrolle",
"erfasse_position",
"fehlende_material_ids",
"starte_kontrolle",
]
backend/app/services/kontrolle/erfassung.py
+153
View File
@@ -0,0 +1,153 @@
import uuid
from datetime import datetime, timezone
from decimal import Decimal
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.fehlbestand import Fehlbestand, FehlbestandStatus
from app.models.kontrolle import Kontrolle, Kontrollposition
from app.models.objekt import Objekt
from app.models.objektposition import Objektposition
from app.services import benachrichtigung as benachrichtigung_service
from app.services import historie as historie_service
from app.services.objekte import sollmenge_effektiv, sollmenge_map
async def erfasse_position(
db: AsyncSession,
*,
kontrolle: Kontrolle,
objekt: Objekt,
material_id: int,
istmenge: Decimal,
zustaendiger_server_id: int,
) -> tuple[Kontrollposition, uuid.UUID | None]:
"""Prompt 02.3/09 Vier-Kernbegriffe: Kontrolle erfasst nur, ändert nie die
persistente Objektposition.istmenge. Bei Ist < Soll entsteht automatisch ein
Fehlbestand (nie bei Überbestand, Prompt 02.4). Gibt zusätzlich die
Fehlbestand-ID zurück (falls einer aktiv ist), damit die UI direkt eine
Sofort-Nachfüllung (Karte 07) anbieten kann, ohne separat nachzufragen."""
vorlagen_sollmengen = await sollmenge_map(db, objekt.vorlage_id)
result = await db.execute(
select(Objektposition).where(
Objektposition.objekt_id == objekt.id, Objektposition.material_id == material_id
)
)
objektposition = result.scalar_one_or_none()
sollmenge = sollmenge_effektiv(objektposition, vorlagen_sollmengen) if objektposition else Decimal(0)
abweichung = istmenge != sollmenge
result = await db.execute(
select(Kontrollposition).where(
Kontrollposition.kontrolle_id == kontrolle.id, Kontrollposition.material_id == material_id
)
)
kontrollposition = result.scalar_one_or_none()
if kontrollposition is None:
kontrollposition = Kontrollposition(
kontrolle_id=kontrolle.id,
material_id=material_id,
sollmenge_snapshot=sollmenge,
istmenge_erfasst=istmenge,
abweichung=abweichung,
)
db.add(kontrollposition)
else:
kontrollposition.sollmenge_snapshot = sollmenge
kontrollposition.istmenge_erfasst = istmenge
kontrollposition.abweichung = abweichung
await db.flush()
await historie_service.log(
db,
zustaendiger_server_id=zustaendiger_server_id,
benutzer_id=kontrolle.benutzer_id,
ereignistyp="istmenge_erfasst",
entitaet_typ="kontrollposition",
entitaet_id=kontrollposition.id,
neuer_wert={"material_id": material_id, "sollmenge": str(sollmenge), "istmenge": str(istmenge)},
)
fehlbestand_id = await _synchronisiere_fehlbestand(
db,
kontrolle=kontrolle,
objekt=objekt,
material_id=material_id,
sollmenge=sollmenge,
istmenge=istmenge,
zustaendiger_server_id=zustaendiger_server_id,
)
return kontrollposition, fehlbestand_id
async def _synchronisiere_fehlbestand(
db: AsyncSession,
*,
kontrolle: Kontrolle,
objekt: Objekt,
material_id: int,
sollmenge: Decimal,
istmenge: Decimal,
zustaendiger_server_id: int,
) -> uuid.UUID | None:
result = await db.execute(
select(Fehlbestand).where(
Fehlbestand.kontrolle_id == kontrolle.id,
Fehlbestand.material_id == material_id,
Fehlbestand.status == FehlbestandStatus.offen,
)
)
bestehender = result.scalar_one_or_none()
unterdeckung = sollmenge - istmenge
if unterdeckung > 0:
# Ist < Soll (Prompt 02.3) - Überbestand (Ist > Soll) erzeugt laut Prompt 02.4
# ausdrücklich KEINEN Fehlbestand.
if bestehender is None:
bestehender = Fehlbestand(
erzeugt_von_server_id=zustaendiger_server_id,
objekt_id=objekt.id,
material_id=material_id,
standort_id=objekt.standort_id,
sollmenge=sollmenge,
istmenge=istmenge,
fehlmenge=unterdeckung,
entstanden_am=datetime.now(timezone.utc),
festgestellt_von=kontrolle.benutzer_id,
kontrolle_id=kontrolle.id,
status=FehlbestandStatus.offen,
)
db.add(bestehender)
await db.flush()
await historie_service.log(
db,
zustaendiger_server_id=zustaendiger_server_id,
benutzer_id=kontrolle.benutzer_id,
ereignistyp="fehlbestand_entstanden",
entitaet_typ="fehlbestand",
entitaet_id=bestehender.id,
neuer_wert={
"objekt_id": objekt.id,
"material_id": material_id,
"sollmenge": str(sollmenge),
"istmenge": str(istmenge),
"fehlmenge": str(unterdeckung),
},
)
await benachrichtigung_service.benachrichtige_neuer_fehlbestand(db, fehlbestand=bestehender)
else:
# Korrektur vor Abschluss (z. B. Tippfehler nochmal erfasst) - solange die
# Kontrolle noch läuft, ist der von IHR erzeugte Fehlbestand noch nicht
# final und darf berichtigt werden.
bestehender.sollmenge = sollmenge
bestehender.istmenge = istmenge
bestehender.fehlmenge = unterdeckung
elif bestehender is not None:
# Abweichung wurde vor Abschluss korrigiert - der von dieser Kontrolle
# erzeugte Fehlbestand hat sich erledigt, bevor er je "real" wurde.
await db.delete(bestehender)
bestehender = None
await db.flush()
return bestehender.id if bestehender is not None else None
backend/app/services/kontrolle/exceptions.py
+17
View File
@@ -0,0 +1,17 @@
from app.models.kontrolle import Kontrolle
class ObjektGesperrtError(Exception):
"""Prompt 02.8: Objekt wird bereits von einer anderen, laufenden Kontrolle belegt."""
def __init__(self, aktive_kontrolle: Kontrolle):
self.aktive_kontrolle = aktive_kontrolle
super().__init__("Objekt wird bereits kontrolliert")
class UnvollstaendigError(Exception):
"""Prompt 16.1: Abschluss mit unbestätigten Positionen."""
def __init__(self, fehlende_material_ids: list[int]):
self.fehlende_material_ids = fehlende_material_ids
super().__init__("Nicht alle Positionen wurden erfasst")
backend/app/services/kontrolle.py → backend/app/services/kontrolle/lifecycle.py
+2 -160
View File
@@ -1,34 +1,15 @@
import uuid
from datetime import datetime, timezone
from decimal import Decimal
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.fehlbestand import Fehlbestand, FehlbestandStatus
from app.models.fehlbestand import Fehlbestand
from app.models.kontrolle import Kontrolle, KontrollStatus, Kontrollposition
from app.models.mindermenge import MindermengeStatus, MindermengenGenehmigung
from app.models.objekt import Objekt
from app.models.objektposition import Objektposition, ObjektpositionStatus
from app.services import benachrichtigung as benachrichtigung_service
from app.services import historie as historie_service
from app.services.objekte import sollmenge_effektiv, sollmenge_map
class ObjektGesperrtError(Exception):
"""Prompt 02.8: Objekt wird bereits von einer anderen, laufenden Kontrolle belegt."""
def __init__(self, aktive_kontrolle: Kontrolle):
self.aktive_kontrolle = aktive_kontrolle
super().__init__("Objekt wird bereits kontrolliert")
class UnvollstaendigError(Exception):
"""Prompt 16.1: Abschluss mit unbestätigten Positionen."""
def __init__(self, fehlende_material_ids: list[int]):
self.fehlende_material_ids = fehlende_material_ids
super().__init__("Nicht alle Positionen wurden erfasst")
from .exceptions import ObjektGesperrtError, UnvollstaendigError
async def aktive_kontrolle(db: AsyncSession, objekt_id: int) -> Kontrolle | None:
@@ -136,145 +117,6 @@ async def _verwirf_kontrolldaten(db: AsyncSession, kontrolle_id) -> None:
await db.flush()
async def erfasse_position(
db: AsyncSession,
*,
kontrolle: Kontrolle,
objekt: Objekt,
material_id: int,
istmenge: Decimal,
zustaendiger_server_id: int,
) -> tuple[Kontrollposition, uuid.UUID | None]:
"""Prompt 02.3/09 Vier-Kernbegriffe: Kontrolle erfasst nur, ändert nie die
persistente Objektposition.istmenge. Bei Ist < Soll entsteht automatisch ein
Fehlbestand (nie bei Überbestand, Prompt 02.4). Gibt zusätzlich die
Fehlbestand-ID zurück (falls einer aktiv ist), damit die UI direkt eine
Sofort-Nachfüllung (Karte 07) anbieten kann, ohne separat nachzufragen."""
vorlagen_sollmengen = await sollmenge_map(db, objekt.vorlage_id)
result = await db.execute(
select(Objektposition).where(
Objektposition.objekt_id == objekt.id, Objektposition.material_id == material_id
)
)
objektposition = result.scalar_one_or_none()
sollmenge = sollmenge_effektiv(objektposition, vorlagen_sollmengen) if objektposition else Decimal(0)
abweichung = istmenge != sollmenge
result = await db.execute(
select(Kontrollposition).where(
Kontrollposition.kontrolle_id == kontrolle.id, Kontrollposition.material_id == material_id
)
)
kontrollposition = result.scalar_one_or_none()
if kontrollposition is None:
kontrollposition = Kontrollposition(
kontrolle_id=kontrolle.id,
material_id=material_id,
sollmenge_snapshot=sollmenge,
istmenge_erfasst=istmenge,
abweichung=abweichung,
)
db.add(kontrollposition)
else:
kontrollposition.sollmenge_snapshot = sollmenge
kontrollposition.istmenge_erfasst = istmenge
kontrollposition.abweichung = abweichung
await db.flush()
await historie_service.log(
db,
zustaendiger_server_id=zustaendiger_server_id,
benutzer_id=kontrolle.benutzer_id,
ereignistyp="istmenge_erfasst",
entitaet_typ="kontrollposition",
entitaet_id=kontrollposition.id,
neuer_wert={"material_id": material_id, "sollmenge": str(sollmenge), "istmenge": str(istmenge)},
)
fehlbestand_id = await _synchronisiere_fehlbestand(
db,
kontrolle=kontrolle,
objekt=objekt,
material_id=material_id,
sollmenge=sollmenge,
istmenge=istmenge,
zustaendiger_server_id=zustaendiger_server_id,
)
return kontrollposition, fehlbestand_id
async def _synchronisiere_fehlbestand(
db: AsyncSession,
*,
kontrolle: Kontrolle,
objekt: Objekt,
material_id: int,
sollmenge: Decimal,
istmenge: Decimal,
zustaendiger_server_id: int,
) -> uuid.UUID | None:
result = await db.execute(
select(Fehlbestand).where(
Fehlbestand.kontrolle_id == kontrolle.id,
Fehlbestand.material_id == material_id,
Fehlbestand.status == FehlbestandStatus.offen,
)
)
bestehender = result.scalar_one_or_none()
unterdeckung = sollmenge - istmenge
if unterdeckung > 0:
# Ist < Soll (Prompt 02.3) - Überbestand (Ist > Soll) erzeugt laut Prompt 02.4
# ausdrücklich KEINEN Fehlbestand.
if bestehender is None:
bestehender = Fehlbestand(
erzeugt_von_server_id=zustaendiger_server_id,
objekt_id=objekt.id,
material_id=material_id,
standort_id=objekt.standort_id,
sollmenge=sollmenge,
istmenge=istmenge,
fehlmenge=unterdeckung,
entstanden_am=datetime.now(timezone.utc),
festgestellt_von=kontrolle.benutzer_id,
kontrolle_id=kontrolle.id,
status=FehlbestandStatus.offen,
)
db.add(bestehender)
await db.flush()
await historie_service.log(
db,
zustaendiger_server_id=zustaendiger_server_id,
benutzer_id=kontrolle.benutzer_id,
ereignistyp="fehlbestand_entstanden",
entitaet_typ="fehlbestand",
entitaet_id=bestehender.id,
neuer_wert={
"objekt_id": objekt.id,
"material_id": material_id,
"sollmenge": str(sollmenge),
"istmenge": str(istmenge),
"fehlmenge": str(unterdeckung),
},
)
await benachrichtigung_service.benachrichtige_neuer_fehlbestand(db, fehlbestand=bestehender)
else:
# Korrektur vor Abschluss (z. B. Tippfehler nochmal erfasst) - solange die
# Kontrolle noch läuft, ist der von IHR erzeugte Fehlbestand noch nicht
# final und darf berichtigt werden.
bestehender.sollmenge = sollmenge
bestehender.istmenge = istmenge
bestehender.fehlmenge = unterdeckung
elif bestehender is not None:
# Abweichung wurde vor Abschluss korrigiert - der von dieser Kontrolle
# erzeugte Fehlbestand hat sich erledigt, bevor er je "real" wurde.
await db.delete(bestehender)
bestehender = None
await db.flush()
return bestehender.id if bestehender is not None else None
async def fehlende_material_ids(db: AsyncSession, kontrolle: Kontrolle) -> list[int]:
result = await db.execute(
select(Objektposition.material_id).where(