Karte 14 Schritt 2 (Admin-Teil) + Frontend für Anhänger-Kopplung
CI / backend-tests (push) Failing after 1m21s
CI / frontend-build (push) Successful in 22s

Backend: geraet_instanz-CRUD (Liste/Anlegen/Status ändern inkl. optionalem
Fehlbestand pro Gerät/Löschen), pruefintervall_monate in Objektposition-
Schemas verdrahtet (Prüfdatum -> naechste_pruefung wird automatisch
berechnet, Monatsrechnung ohne neue Dependency).

Frontend: GeraeteInstanzenListe ersetzt das Einzel-SN-Feld im Admin-Portal
(ObjektPositionenPanel) durch eine Instanz-Liste mit Status/Prüfdatum/
Bemerkung je Exemplar - löst das Karte-14-Kernproblem (mehrere Geräte pro
Position). ObjektSection: Anhänger-Kopplung bei Anlage und nachträglich
änderbar.

Kontroll-Erfassung (PositionCard/KontrollPage) bleibt unverändert - die
komplett neue Erfassungs-UI für geraet_sn (Umsetzungsschritt 4, Karte 14)
ist bewusst nicht Teil dieses Commits.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CVgbozhYmuEhiEJHffRXCV
This commit is contained in:
2026-09-04 20:55:09 +02:00
co-authored by Claude Sonnet 5
parent eff9fe4956
commit 6934276e8a
14 changed files with 685 additions and 7 deletions
+130
View File
@@ -0,0 +1,130 @@
import calendar
import uuid
from datetime import date, 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.geraet_instanz import GeraetInstanz, GeraetStatus
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
def _plus_monate(stichtag: date, monate: int) -> date:
"""Kein python-dateutil als Dependency für eine einzelne Monatsaddition -
Tag wird auf das Monatsende gekappt, falls der Zielmonat kürzer ist
(z.B. 31. Januar + 1 Monat -> 28./29. Februar)."""
monats_index = stichtag.month - 1 + monate
jahr = stichtag.year + monats_index // 12
monat = monats_index % 12 + 1
letzter_tag = calendar.monthrange(jahr, monat)[1]
return date(jahr, monat, min(stichtag.day, letzter_tag))
async def liste_fuer_position(db: AsyncSession, objektposition_id) -> list[GeraetInstanz]:
result = await db.execute(
select(GeraetInstanz).where(GeraetInstanz.objektposition_id == objektposition_id)
)
return list(result.scalars().all())
async def erstelle(db: AsyncSession, *, objektposition_id, seriennummer: str) -> GeraetInstanz:
instanz = GeraetInstanz(objektposition_id=objektposition_id, seriennummer=seriennummer)
db.add(instanz)
await db.flush()
return instanz
async def loesche(db: AsyncSession, instanz: GeraetInstanz) -> None:
await db.delete(instanz)
async def aendere_status(
db: AsyncSession,
*,
instanz: GeraetInstanz,
status: GeraetStatus,
pruefdatum: date | None,
bemerkung: str | None,
erzeuge_fehlbestand: bool,
benutzer_id: int,
zustaendiger_server_id: int,
) -> tuple[GeraetInstanz, uuid.UUID | None]:
"""Karte 14: Status ändern (einsatzbereit/defekt/in_reparatur), optional
Prüfung erfassen (naechste_pruefung wird neu berechnet, sofern die
Objektposition ein pruefintervall_monate hat), optional einen Fehlbestand
erzeugen (pro Gerät entscheidbar, nutzt bestehende Fehlbestand-
Infrastruktur statt eigenem Parallel-Konzept)."""
alter_status = instanz.status
instanz.status = status
instanz.bemerkung = bemerkung
if pruefdatum is not None:
instanz.pruefdatum = pruefdatum
objektposition = await db.get(Objektposition, instanz.objektposition_id)
if objektposition is not None and objektposition.pruefintervall_monate:
instanz.naechste_pruefung = _plus_monate(pruefdatum, objektposition.pruefintervall_monate)
await db.flush()
await historie_service.log(
db,
zustaendiger_server_id=zustaendiger_server_id,
benutzer_id=benutzer_id,
ereignistyp="geraet_status_geaendert",
entitaet_typ="geraet_instanz",
entitaet_id=instanz.id,
alter_wert={"status": alter_status.value},
neuer_wert={"status": status.value, "seriennummer": instanz.seriennummer},
begruendung=bemerkung,
)
fehlbestand_id = None
if erzeuge_fehlbestand and status != GeraetStatus.einsatzbereit:
fehlbestand_id = await _erzeuge_fehlbestand(
db,
instanz=instanz,
benutzer_id=benutzer_id,
zustaendiger_server_id=zustaendiger_server_id,
)
return instanz, fehlbestand_id
async def _erzeuge_fehlbestand(
db: AsyncSession, *, instanz: GeraetInstanz, benutzer_id: int, zustaendiger_server_id: int
):
objektposition = await db.get(Objektposition, instanz.objektposition_id)
objekt = await db.get(Objekt, objektposition.objekt_id)
fehlbestand = Fehlbestand(
erzeugt_von_server_id=zustaendiger_server_id,
objekt_id=objektposition.objekt_id,
material_id=objektposition.material_id,
standort_id=objekt.standort_id,
sollmenge=Decimal(1),
istmenge=Decimal(0),
fehlmenge=Decimal(1),
entstanden_am=datetime.now(timezone.utc),
festgestellt_von=benutzer_id,
ursache=f"Gerät {instanz.seriennummer} ({instanz.status.value})",
status=FehlbestandStatus.offen,
)
db.add(fehlbestand)
await db.flush()
await historie_service.log(
db,
zustaendiger_server_id=zustaendiger_server_id,
benutzer_id=benutzer_id,
ereignistyp="fehlbestand_entstanden",
entitaet_typ="fehlbestand",
entitaet_id=fehlbestand.id,
neuer_wert={"ursache": fehlbestand.ursache, "geraet_instanz_id": str(instanz.id)},
)
await benachrichtigung_service.benachrichtige_neuer_fehlbestand(db, fehlbestand=fehlbestand)
return fehlbestand.id