feat(material): Material-Dubletten zusammenführen

Verwendung je Material einsehbar (Objektposition/Vorlagenposition/
Bestand/Kontrollposition/Fehlbestand/Ausgabe/Nachfüllung/
Materialbewegung), POST /materialien/{id}/ersetzen hängt alle
Referenzen auf ein Ziel-Material um und löscht die Quelle. Bricht bei
Konflikten ab (z.B. dasselbe Objekt hätte danach zwei Positionen für
dasselbe Material) statt Daten stillschweigend zu verlieren. Admin-UI:
"Zusammenführen" je Material zeigt Verwendung + Ziel-Auswahl.

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

backend/app/api/v1/endpoints/stammdaten.py

@@ -8,6 +8,7 @@ from app.models.auth import RolleTyp
 from app.models.objekt import Objekt
 from app.models.stammdaten import Bereich, Fach, Kategorie, Material, Objekttyp, Standort
 from app.models.vorlage import Beladungsvorlage, Vorlagenposition
+from app.services.material_merge import finde_konflikte, fuehre_zusammen, hole_verwendung
 from app.schemas.stammdaten import (
     BereichCreate,
     BereichRead,
@@ -23,6 +24,8 @@ from app.schemas.stammdaten import (
     MaterialCreate,
     MaterialRead,
     MaterialUpdate,
+    MaterialZusammenfuehren,
+    MaterialZusammenfuehrenErgebnis,
     ObjekttypCreate,
     ObjekttypRead,
     ObjekttypUpdate,
@@ -365,3 +368,44 @@ async def aendere_material(
         setattr(material, feld, wert)
     await db.flush()
     return material
+
+
+@router.get("/materialien/{material_id}/verwendung", response_model=dict[str, int])
+async def hole_material_verwendung(
+    material_id: int,
+    db: AsyncSession = Depends(get_db),
+    _=Depends(require_roles(RolleTyp.administration)),
+) -> dict[str, int]:
+    if await db.get(Material, material_id) is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Material nicht gefunden")
+    return await hole_verwendung(db, material_id)
+
+
+@router.post("/materialien/{material_id}/ersetzen", response_model=MaterialZusammenfuehrenErgebnis)
+async def ersetze_material(
+    material_id: int,
+    payload: MaterialZusammenfuehren,
+    db: AsyncSession = Depends(get_db),
+    _=Depends(require_roles(RolleTyp.administration)),
+) -> MaterialZusammenfuehrenErgebnis:
+    """Führt zwei Material-Datensätze zusammen (z. B. Import-Dubletten):
+    quelle_id (aus der URL) wird gelöscht, alle Referenzen zeigen danach auf
+    ziel_material_id. Bricht bei Konflikten ab (z. B. dasselbe Objekt hätte
+    danach zwei Positionen für dasselbe Material) statt Daten zu verlieren."""
+    quelle = await db.get(Material, material_id)
+    if quelle is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Material nicht gefunden")
+    if payload.ziel_material_id == material_id:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Quelle und Ziel sind identisch")
+    ziel = await db.get(Material, payload.ziel_material_id)
+    if ziel is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Ziel-Material nicht gefunden")
+
+    konflikte = await finde_konflikte(db, quelle_id=material_id, ziel_id=payload.ziel_material_id)
+    if konflikte:
+        return MaterialZusammenfuehrenErgebnis(erfolgreich=False, konflikte=konflikte)
+
+    await fuehre_zusammen(db, quelle_id=material_id, ziel_id=payload.ziel_material_id)
+    await db.delete(quelle)
+    await db.flush()
+    return MaterialZusammenfuehrenErgebnis(erfolgreich=True, konflikte=[])

backend/app/schemas/stammdaten.py

@@ -124,3 +124,12 @@ class MaterialUpdate(BaseModel):
 class MaterialRead(MaterialCreate):
     model_config = ConfigDict(from_attributes=True)
     id: int
+
+
+class MaterialZusammenfuehren(BaseModel):
+    ziel_material_id: int
+
+
+class MaterialZusammenfuehrenErgebnis(BaseModel):
+    erfolgreich: bool
+    konflikte: list[str]

backend/app/services/material_merge.py

@@ -0,0 +1,63 @@
+from sqlalchemy import func, select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.ausgabe import Ausgabe
+from app.models.fehlbestand import Fehlbestand
+from app.models.kontrolle import Kontrollposition
+from app.models.lager import Bestand, Materialbewegung
+from app.models.nachfuellung import Nachfuellung
+from app.models.objektposition import Objektposition
+from app.models.vorlage import Vorlagenposition
+
+# (Modell, Gruppierungs-Spalte für den Konflikt-Check, Anzeige-Name)
+_KONFLIKT_TABELLEN = [
+    (Objektposition, Objektposition.objekt_id, "Objektposition"),
+    (Vorlagenposition, Vorlagenposition.vorlage_id, "Vorlagenposition"),
+    (Bestand, Bestand.lagerplatz_id, "Bestand"),
+    (Kontrollposition, Kontrollposition.kontrolle_id, "Kontrollposition"),
+]
+# Reine Protokoll-/Bewegungstabellen ohne Eindeutigkeits-Constraint auf
+# material_id - können immer gefahrlos umgehängt werden.
+_SICHERE_TABELLEN = [Fehlbestand, Ausgabe, Nachfuellung, Materialbewegung]
+
+
+async def hole_verwendung(db: AsyncSession, material_id: int) -> dict[str, int]:
+    verwendung = {}
+    for model, _, name in _KONFLIKT_TABELLEN:
+        result = await db.execute(select(func.count()).select_from(model).where(model.material_id == material_id))
+        verwendung[name] = result.scalar_one()
+    for model in _SICHERE_TABELLEN:
+        result = await db.execute(select(func.count()).select_from(model).where(model.material_id == material_id))
+        verwendung[model.__tablename__] = result.scalar_one()
+    return verwendung
+
+
+async def finde_konflikte(db: AsyncSession, *, quelle_id: int, ziel_id: int) -> list[str]:
+    """Für Tabellen mit UNIQUE(gruppen_spalte, material_id) darf nach dem
+    Umhängen nicht dieselbe Kombination doppelt entstehen - z. B. wenn ein
+    Objekt schon eine Position mit dem Ziel-Material hat UND eine mit dem
+    Quell-Material."""
+    konflikte: list[str] = []
+    for model, gruppen_spalte, name in _KONFLIKT_TABELLEN:
+        quelle_gruppen = (
+            (await db.execute(select(gruppen_spalte).where(model.material_id == quelle_id))).scalars().all()
+        )
+        if not quelle_gruppen:
+            continue
+        ziel_gruppen = set(
+            (await db.execute(select(gruppen_spalte).where(model.material_id == ziel_id))).scalars().all()
+        )
+        ueberschneidung = set(quelle_gruppen) & ziel_gruppen
+        if ueberschneidung:
+            konflikte.append(f"{name}: {len(ueberschneidung)} Konflikt(e) - beide Materialien dort gleichzeitig vorhanden")
+    return konflikte
+
+
+async def fuehre_zusammen(db: AsyncSession, *, quelle_id: int, ziel_id: int) -> None:
+    """Voraussetzung: finde_konflikte() wurde vorher aufgerufen und war leer -
+    der Aufrufer (Endpunkt) verweigert den Merge sonst."""
+    for model, _, _ in _KONFLIKT_TABELLEN:
+        await db.execute(update(model).where(model.material_id == quelle_id).values(material_id=ziel_id))
+    for model in _SICHERE_TABELLEN:
+        await db.execute(update(model).where(model.material_id == quelle_id).values(material_id=ziel_id))
+    await db.flush()