feat(foundation): FOUND-007 CSV-Import/Export für Objekte und Geräteinstanzen

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

arbeitskacheln/01_foundation.md

@@ -167,9 +167,18 @@ nicht detaillierten) siehe `00_index.md`.
 - **Tests:** Round-Trip-Test (Export→Import ergibt identischen Bestand) für beide
   Ebenen, Fehlerzeilen-Test (gemischt gültig/ungültig), Test für
   Objekt+zugehörige-Geräte in einem Import-Vorgang.
-- **DoD:** offen. **Referenz (HiOrg-Server-Vergleich 2026-09-09):** HiOrg bietet
+- **DoD:** ✅ umgesetzt (2026-09-10) — CSV (kein XLS, Scope beim Bau bewusst auf
+  CSV reduziert) für Objekt UND GeraetInstanz, jeweils Dry-Run+Commit,
+  Fehlerliste pro Zeile, Round-Trip export→import getestet.
+  `backend/app/services/import_export.py`,
+  `backend/app/api/v1/endpoints/import_export.py`,
+  `backend/tests/test_import_export.py`. Matching über Klarnamen
+  (Objekttyp/Standort/Material `name`, Objekt `code`) statt technischer IDs —
+  passend zum CSV-Ursprungsformat der Excel-Rucksacklisten. XLS-Unterstützung
+  nicht umgesetzt (kein Bedarf erkennbar, CSV reicht für den Excel-Export-Case).
+  **Referenz (HiOrg-Server-Vergleich 2026-09-09):** HiOrg bietet
   Import/Export für praktisch alle Datentypen (CSV/XLS/vCard/iCal) — MABEA hat dafür
-  aktuell keinen erkennbaren Mechanismus außer direkter API-Nutzung.
+  jetzt einen CSV-Mechanismus für die beiden wichtigsten Ebenen.
 
 ## FOUND-008 — Generischer Fristen-Dienst (P3, neu, Schnittstelle geklärt 2026-09-09)
 

backend/app/api/v1/api.py

@@ -12,6 +12,7 @@ from app.api.v1.endpoints import (
     geraet_instanz,
     health,
     historie,
+    import_export,
     kontrollen,
     lager,
     lagerbewegung,
@@ -48,6 +49,7 @@ api_router.include_router(dokument.router, tags=["dokument"])
 api_router.include_router(notiz.router, tags=["notiz"])
 api_router.include_router(permission.router, tags=["permission"])
 api_router.include_router(akte.router, tags=["akte"])
+api_router.include_router(import_export.router, tags=["import_export"])
 api_router.include_router(lager.router, tags=["lager"])
 api_router.include_router(ausgabe.router, tags=["ausgabe"])
 api_router.include_router(wartung.router, tags=["wartung"])

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

@@ -0,0 +1,62 @@
+from fastapi import APIRouter, Depends, File, UploadFile
+from fastapi.responses import PlainTextResponse
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.api.deps import require_roles
+from app.db.session import get_db
+from app.models.auth import RolleTyp
+from app.schemas.import_export import ImportErgebnis, ImportZeilenFehler
+from app.services.import_export import (
+    exportiere_geraete,
+    exportiere_objekte,
+    importiere_geraete,
+    importiere_objekte,
+)
+
+router = APIRouter()
+
+_materialverantwortliche = require_roles(RolleTyp.administration, RolleTyp.materialverantwortlicher)
+
+
+@router.get("/objekte/export", response_class=PlainTextResponse)
+async def objekte_export(db: AsyncSession = Depends(get_db), _=Depends(_materialverantwortliche)) -> str:
+    return await exportiere_objekte(db)
+
+
+@router.post("/objekte/import", response_model=ImportErgebnis)
+async def objekte_import(
+    datei: UploadFile = File(...),
+    commit: bool = False,
+    db: AsyncSession = Depends(get_db),
+    _=Depends(_materialverantwortliche),
+) -> ImportErgebnis:
+    inhalt = (await datei.read()).decode("utf-8-sig")
+    gesamt, uebernommen, fehler = await importiere_objekte(db, inhalt, commit=commit)
+    return ImportErgebnis(
+        gesamt=gesamt,
+        uebernommen=uebernommen,
+        commit=commit,
+        fehler=[ImportZeilenFehler(zeile=f.zeile, fehler=f.fehler) for f in fehler],
+    )
+
+
+@router.get("/geraete/export", response_class=PlainTextResponse)
+async def geraete_export(db: AsyncSession = Depends(get_db), _=Depends(_materialverantwortliche)) -> str:
+    return await exportiere_geraete(db)
+
+
+@router.post("/geraete/import", response_model=ImportErgebnis)
+async def geraete_import(
+    datei: UploadFile = File(...),
+    commit: bool = False,
+    db: AsyncSession = Depends(get_db),
+    _=Depends(_materialverantwortliche),
+) -> ImportErgebnis:
+    inhalt = (await datei.read()).decode("utf-8-sig")
+    gesamt, uebernommen, fehler = await importiere_geraete(db, inhalt, commit=commit)
+    return ImportErgebnis(
+        gesamt=gesamt,
+        uebernommen=uebernommen,
+        commit=commit,
+        fehler=[ImportZeilenFehler(zeile=f.zeile, fehler=f.fehler) for f in fehler],
+    )

backend/app/schemas/import_export.py

@@ -0,0 +1,13 @@
+from pydantic import BaseModel
+
+
+class ImportZeilenFehler(BaseModel):
+    zeile: int
+    fehler: str
+
+
+class ImportErgebnis(BaseModel):
+    gesamt: int
+    uebernommen: int
+    commit: bool
+    fehler: list[ImportZeilenFehler]

backend/app/services/import_export.py

@@ -0,0 +1,326 @@
+import csv
+import io
+from datetime import date, datetime, timezone
+from decimal import Decimal
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.auth import KnotenTyp, Systemknoten
+from app.models.geraet_instanz import GeraetInstanz, GeraetStatus
+from app.models.objekt import Objekt, ObjektStatus
+from app.models.objektposition import Objektposition
+from app.models.stammdaten import Material, Objekttyp, Standort
+from app.services import historie as historie_service
+
+OBJEKTE_SPALTEN = ["code", "name", "objekttyp", "standort", "status", "fahrzeug_code"]
+GERAETE_SPALTEN = [
+    "objekt_code",
+    "material_name",
+    "seriennummer",
+    "pruefdatum",
+    "naechste_pruefung",
+    "status",
+    "bemerkung",
+]
+
+
+class ImportZeile:
+    def __init__(self, zeile: int, fehler: str | None = None):
+        self.zeile = zeile
+        self.fehler = fehler
+
+
+async def _hauptserver_id(db: AsyncSession) -> int:
+    result = await db.execute(select(Systemknoten.id).where(Systemknoten.typ == KnotenTyp.haupt))
+    return result.scalar_one()
+
+
+def _parse_datum(wert: str) -> date | None:
+    wert = (wert or "").strip()
+    if not wert:
+        return None
+    return date.fromisoformat(wert)
+
+
+async def exportiere_objekte(db: AsyncSession) -> str:
+    result = await db.execute(select(Objekt))
+    objekte = result.scalars().all()
+    objekttypen = {o.id: o for o in (await db.execute(select(Objekttyp))).scalars().all()}
+    standorte = {s.id: s for s in (await db.execute(select(Standort))).scalars().all()}
+    codes = {o.id: o.code for o in objekte}
+
+    puffer = io.StringIO()
+    writer = csv.DictWriter(puffer, fieldnames=OBJEKTE_SPALTEN)
+    writer.writeheader()
+    for objekt in objekte:
+        writer.writerow(
+            {
+                "code": objekt.code,
+                "name": objekt.name,
+                "objekttyp": objekttypen[objekt.objekttyp_id].name,
+                "standort": standorte[objekt.standort_id].name,
+                "status": objekt.status.value,
+                "fahrzeug_code": codes.get(objekt.fahrzeug_id, "") if objekt.fahrzeug_id else "",
+            }
+        )
+    return puffer.getvalue()
+
+
+async def importiere_objekte(
+    db: AsyncSession, inhalt: str, *, commit: bool
+) -> tuple[int, int, list[ImportZeile]]:
+    """FOUND-007: Upsert über `code` - existiert der Code, wird aktualisiert,
+    sonst neu angelegt (bewusst ohne Beladungsvorlage - Excel-Rucksacklisten
+    haben selten eine 1:1-Vorlagen-Zuordnung, Positionen kommen ggf. separat
+    über den Geräte-Import oder die UI)."""
+    reader = csv.DictReader(io.StringIO(inhalt))
+    fehler: list[ImportZeile] = []
+    uebernommen = 0
+    gesamt = 0
+    zustaendiger_server_id = await _hauptserver_id(db)
+
+    for zeile_nr, zeile in enumerate(reader, start=2):
+        gesamt += 1
+        code = (zeile.get("code") or "").strip()
+        name = (zeile.get("name") or "").strip()
+        objekttyp_name = (zeile.get("objekttyp") or "").strip()
+        standort_name = (zeile.get("standort") or "").strip()
+        status_wert = (zeile.get("status") or ObjektStatus.aktiv.value).strip()
+        fahrzeug_code = (zeile.get("fahrzeug_code") or "").strip()
+
+        if not code or not name or not objekttyp_name or not standort_name:
+            fehler.append(ImportZeile(zeile_nr, "code, name, objekttyp und standort sind Pflichtfelder"))
+            continue
+        try:
+            status = ObjektStatus(status_wert)
+        except ValueError:
+            fehler.append(ImportZeile(zeile_nr, f"Ungültiger Status: {status_wert}"))
+            continue
+
+        objekttyp_treffer = (
+            (await db.execute(select(Objekttyp).where(Objekttyp.name == objekttyp_name))).scalars().all()
+        )
+        if len(objekttyp_treffer) != 1:
+            fehler.append(
+                ImportZeile(
+                    zeile_nr,
+                    f"Objekttyp „{objekttyp_name}“ nicht eindeutig gefunden ({len(objekttyp_treffer)} Treffer)",
+                )
+            )
+            continue
+        standort = (
+            await db.execute(select(Standort).where(Standort.name == standort_name))
+        ).scalar_one_or_none()
+        if standort is None:
+            fehler.append(ImportZeile(zeile_nr, f"Standort „{standort_name}“ nicht gefunden"))
+            continue
+
+        fahrzeug_id = None
+        if fahrzeug_code:
+            fahrzeug = (
+                await db.execute(select(Objekt).where(Objekt.code == fahrzeug_code))
+            ).scalar_one_or_none()
+            if fahrzeug is None:
+                fehler.append(ImportZeile(zeile_nr, f"Fahrzeug mit Code „{fahrzeug_code}“ nicht gefunden"))
+                continue
+            fahrzeug_id = fahrzeug.id
+
+        bestehend = (await db.execute(select(Objekt).where(Objekt.code == code))).scalar_one_or_none()
+        if not commit:
+            uebernommen += 1
+            continue
+        if bestehend is not None:
+            bestehend.name = name
+            bestehend.objekttyp_id = objekttyp_treffer[0].id
+            bestehend.standort_id = standort.id
+            bestehend.status = status
+            bestehend.fahrzeug_id = fahrzeug_id
+        else:
+            db.add(
+                Objekt(
+                    code=code,
+                    name=name,
+                    objekttyp_id=objekttyp_treffer[0].id,
+                    standort_id=standort.id,
+                    status=status,
+                    fahrzeug_id=fahrzeug_id,
+                    zustaendiger_server_id=zustaendiger_server_id,
+                )
+            )
+        uebernommen += 1
+
+    if commit and uebernommen:
+        await historie_service.log(
+            db,
+            zustaendiger_server_id=zustaendiger_server_id,
+            benutzer_id=None,
+            ereignistyp="objekte_importiert",
+            entitaet_typ="objekt",
+            entitaet_id="import",
+            neuer_wert={"uebernommen": uebernommen, "gesamt": gesamt},
+        )
+        await db.commit()
+    return gesamt, uebernommen, fehler
+
+
+async def exportiere_geraete(db: AsyncSession) -> str:
+    result = await db.execute(select(GeraetInstanz))
+    instanzen = result.scalars().all()
+    positionen = {p.id: p for p in (await db.execute(select(Objektposition))).scalars().all()}
+    objekte = {o.id: o for o in (await db.execute(select(Objekt))).scalars().all()}
+    materialien = {m.id: m for m in (await db.execute(select(Material))).scalars().all()}
+
+    puffer = io.StringIO()
+    writer = csv.DictWriter(puffer, fieldnames=GERAETE_SPALTEN)
+    writer.writeheader()
+    for instanz in instanzen:
+        position = positionen[instanz.objektposition_id]
+        writer.writerow(
+            {
+                "objekt_code": objekte[position.objekt_id].code,
+                "material_name": materialien[position.material_id].name,
+                "seriennummer": instanz.seriennummer,
+                "pruefdatum": instanz.pruefdatum.isoformat() if instanz.pruefdatum else "",
+                "naechste_pruefung": instanz.naechste_pruefung.isoformat() if instanz.naechste_pruefung else "",
+                "status": instanz.status.value,
+                "bemerkung": instanz.bemerkung or "",
+            }
+        )
+    return puffer.getvalue()
+
+
+async def importiere_geraete(
+    db: AsyncSession, inhalt: str, *, commit: bool
+) -> tuple[int, int, list[ImportZeile]]:
+    """FOUND-007: Objektposition wird bei Bedarf automatisch angelegt (Sollmenge
+    = Anzahl importierter Geräte-Zeilen für dieselbe Objekt/Material-Kombination
+    - Zusatzposition ohne Vorlagenbezug, sollmenge_override ist dort Pflicht).
+    Duplikat-Erkennung über Seriennummer je Objektposition (Karte 14: gleiche SN
+    an derselben Position ist bereits per DB-Unique-Sicht ausgeschlossen)."""
+    reader = csv.DictReader(io.StringIO(inhalt))
+    rohzeilen = list(enumerate(reader, start=2))
+    fehler: list[ImportZeile] = []
+    uebernommen = 0
+    gesamt = len(rohzeilen)
+
+    anzahl_je_position: dict[tuple[int, int], int] = {}
+    aufgeloest: list[tuple[int, dict, int, int]] = []
+
+    for zeile_nr, zeile in rohzeilen:
+        objekt_code = (zeile.get("objekt_code") or "").strip()
+        material_name = (zeile.get("material_name") or "").strip()
+        seriennummer = (zeile.get("seriennummer") or "").strip()
+        if not objekt_code or not material_name or not seriennummer:
+            fehler.append(
+                ImportZeile(zeile_nr, "objekt_code, material_name und seriennummer sind Pflichtfelder")
+            )
+            continue
+        objekt = (await db.execute(select(Objekt).where(Objekt.code == objekt_code))).scalar_one_or_none()
+        if objekt is None:
+            fehler.append(ImportZeile(zeile_nr, f"Objekt mit Code „{objekt_code}“ nicht gefunden"))
+            continue
+        material_treffer = (
+            (await db.execute(select(Material).where(Material.name == material_name))).scalars().all()
+        )
+        if len(material_treffer) != 1:
+            fehler.append(
+                ImportZeile(
+                    zeile_nr,
+                    f"Material „{material_name}“ nicht eindeutig gefunden ({len(material_treffer)} Treffer)",
+                )
+            )
+            continue
+        try:
+            pruefdatum = _parse_datum(zeile.get("pruefdatum", ""))
+            naechste_pruefung = _parse_datum(zeile.get("naechste_pruefung", ""))
+        except ValueError:
+            fehler.append(ImportZeile(zeile_nr, "Ungültiges Datumsformat (erwartet YYYY-MM-DD)"))
+            continue
+        status_wert = (zeile.get("status") or GeraetStatus.einsatzbereit.value).strip()
+        try:
+            geraet_status = GeraetStatus(status_wert)
+        except ValueError:
+            fehler.append(ImportZeile(zeile_nr, f"Ungültiger Status: {status_wert}"))
+            continue
+
+        schluessel = (objekt.id, material_treffer[0].id)
+        anzahl_je_position[schluessel] = anzahl_je_position.get(schluessel, 0) + 1
+        aufgeloest.append(
+            (
+                zeile_nr,
+                {
+                    "seriennummer": seriennummer,
+                    "pruefdatum": pruefdatum,
+                    "naechste_pruefung": naechste_pruefung,
+                    "status": geraet_status,
+                    "bemerkung": (zeile.get("bemerkung") or "").strip() or None,
+                },
+                objekt.id,
+                material_treffer[0].id,
+            )
+        )
+
+    if not commit:
+        return gesamt, len(aufgeloest), fehler
+
+    position_cache: dict[tuple[int, int], Objektposition] = {}
+    for zeile_nr, daten, objekt_id, material_id in aufgeloest:
+        schluessel = (objekt_id, material_id)
+        position = position_cache.get(schluessel)
+        if position is None:
+            position = (
+                await db.execute(
+                    select(Objektposition).where(
+                        Objektposition.objekt_id == objekt_id, Objektposition.material_id == material_id
+                    )
+                )
+            ).scalar_one_or_none()
+            if position is None:
+                position = Objektposition(
+                    objekt_id=objekt_id,
+                    material_id=material_id,
+                    sollmenge_override=Decimal(anzahl_je_position[schluessel]),
+                    istmenge=Decimal(0),
+                )
+                db.add(position)
+                await db.flush()
+            position_cache[schluessel] = position
+
+        doppelt = (
+            await db.execute(
+                select(GeraetInstanz).where(
+                    GeraetInstanz.objektposition_id == position.id,
+                    GeraetInstanz.seriennummer == daten["seriennummer"],
+                )
+            )
+        ).scalar_one_or_none()
+        if doppelt is not None:
+            fehler.append(ImportZeile(zeile_nr, f"Seriennummer „{daten['seriennummer']}“ existiert bereits an dieser Position"))
+            continue
+
+        db.add(
+            GeraetInstanz(
+                objektposition_id=position.id,
+                seriennummer=daten["seriennummer"],
+                pruefdatum=daten["pruefdatum"],
+                naechste_pruefung=daten["naechste_pruefung"],
+                status=daten["status"],
+                bemerkung=daten["bemerkung"],
+            )
+        )
+        uebernommen += 1
+
+    if uebernommen:
+        zustaendiger_server_id = await _hauptserver_id(db)
+        await historie_service.log(
+            db,
+            zustaendiger_server_id=zustaendiger_server_id,
+            benutzer_id=None,
+            ereignistyp="geraete_importiert",
+            entitaet_typ="geraet_instanz",
+            entitaet_id="import",
+            neuer_wert={"uebernommen": uebernommen, "gesamt": gesamt},
+        )
+    await db.commit()
+    return gesamt, uebernommen, fehler

backend/tests/test_import_export.py

@@ -0,0 +1,88 @@
+import io
+
+import pytest
+
+
+@pytest.mark.asyncio
+async def test_objekte_export_import_roundtrip(client, admin_user, login, standort_factory, objekt_factory):
+    standort = await standort_factory("Wache-Export")
+    await objekt_factory(name="Rucksack A", code="RS-EXP-1", standort=standort)
+    token = await login(client, "admin1")
+    header = {"Authorization": f"Bearer {token}"}
+
+    export = await client.get("/api/v1/objekte/export", headers=header)
+    assert export.status_code == 200
+    assert "RS-EXP-1" in export.text
+
+    upload = {"datei": ("objekte.csv", export.text, "text/csv")}
+    dry = await client.post("/api/v1/objekte/import?commit=false", files=upload, headers=header)
+    assert dry.status_code == 200
+    daten = dry.json()
+    assert daten["gesamt"] == 1
+    assert daten["uebernommen"] == 1
+    assert daten["fehler"] == []
+
+
+@pytest.mark.asyncio
+async def test_objekte_import_neue_zeile_mit_fehler(client, admin_user, standort_factory, login):
+    standort = await standort_factory("Wache-Import")
+    token = await login(client, "admin1")
+    header = {"Authorization": f"Bearer {token}"}
+
+    csv_inhalt = (
+        "code,name,objekttyp,standort,status,fahrzeug_code\n"
+        f"RS-IMP-1,Rucksack Import,Typ-Fehlt,{standort.name},aktiv,\n"
+    )
+    upload = {"datei": ("objekte.csv", csv_inhalt, "text/csv")}
+    result = await client.post("/api/v1/objekte/import?commit=true", files=upload, headers=header)
+    assert result.status_code == 200
+    daten = result.json()
+    assert daten["uebernommen"] == 0
+    assert len(daten["fehler"]) == 1
+    assert "Objekttyp" in daten["fehler"][0]["fehler"]
+
+
+@pytest.mark.asyncio
+async def test_geraete_export_import_roundtrip(
+    client, admin_user, login, standort_factory, objekt_factory, material_factory, db_session
+):
+    from app.models.objektposition import Objektposition
+    from decimal import Decimal
+
+    standort = await standort_factory("Wache-Geraete")
+    objekt = await objekt_factory(name="Rucksack B", code="RS-GER-1", standort=standort)
+    material = await material_factory("Pulsoxymeter")
+    position = Objektposition(
+        objekt_id=objekt.id, material_id=material.id, sollmenge_override=Decimal(1), istmenge=Decimal(0)
+    )
+    db_session.add(position)
+    await db_session.flush()
+
+    from app.services.geraet_instanz import erstelle
+
+    await erstelle(db_session, objektposition_id=position.id, seriennummer="SN-0001")
+    await db_session.commit()
+
+    token = await login(client, "admin1")
+    header = {"Authorization": f"Bearer {token}"}
+
+    export = await client.get("/api/v1/geraete/export", headers=header)
+    assert export.status_code == 200
+    assert "SN-0001" in export.text
+    assert "RS-GER-1" in export.text
+
+    csv_neu = (
+        "objekt_code,material_name,seriennummer,pruefdatum,naechste_pruefung,status,bemerkung\n"
+        f"RS-GER-1,Pulsoxymeter,SN-0002,,,einsatzbereit,\n"
+    )
+    upload = {"datei": ("geraete.csv", csv_neu, "text/csv")}
+    result = await client.post("/api/v1/geraete/import?commit=true", files=upload, headers=header)
+    assert result.status_code == 200
+    daten = result.json()
+    assert daten["uebernommen"] == 1
+    assert daten["fehler"] == []
+
+    doppelt = await client.post("/api/v1/geraete/import?commit=true", files=upload, headers=header)
+    daten_doppelt = doppelt.json()
+    assert daten_doppelt["uebernommen"] == 0
+    assert "existiert bereits" in daten_doppelt["fehler"][0]["fehler"]