Sprint 8: Excel-Import-Skript + CI Frontend-Build-Job

Import-Skript (backend/scripts/import_excel.py) liest die drei Ist-Listen
per stdlib zipfile/ElementTree (kein openpyxl), erzeugt idempotent
Objekttyp+Beladungsvorlage+Vorlagenpositionen je Datei, Material-Abgleich
über Artikelnummer. Rohdaten-Check bestätigt Mapping D=Artikelnummer
(testphasen.md ergänzt). Ausführung nur auf Zielsystem (Deployment-Regel).

CI: neuer frontend-build-Job (npm install + tsc --noEmit + vite build),
da React-PWA/Flutter bisher unabhängig von der Test-Pipeline waren.

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

backend/scripts/import_excel.py

@@ -0,0 +1,230 @@
+"""Einmaliger, idempotenter Import der drei bestehenden Ist-Listen
+(Prompt-Sprintplan Abschnitt 5, Detail-Mapping in ergebnisse/testphasen.md
+„Excel-Import-Mapping"). Erzeugt je Datei einen Objekttyp + Beladungsvorlage
+Version 1 + Vorlagenpositionen; legt Material bei Bedarf an (Abgleich über
+Artikelnummer, nicht über Namensgleichheit).
+
+Deployment-Regel (Prompt 19): dieses Skript wird hier nur als Code erzeugt.
+Ausführung nur auf dem Zielsystem/Test-Deployment gegen eine echte DB, nicht
+lokal auf diesem Rechner.
+
+Aufruf (auf dem Zielsystem, im aktivierten venv):
+
+    python -m scripts.import_excel \
+        --bereich "Rettungsdienst/KatS" --kategorie "Rucksäcke" \
+        "Handball Rucksack.xlsx" "Rettungsrucksack.xlsx" "Rucksack PAX SEG.xlsx"
+
+Kein openpyxl-Requirement: xlsx ist ein ZIP mit XML drin, Parsing über
+Python-Stdlib (zipfile + xml.etree.ElementTree), wie bereits bei der
+Erstanalyse (Prompt 01) gemacht.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import re
+import xml.etree.ElementTree as ET
+import zipfile
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.session import SessionLocal
+from app.models.stammdaten import Bereich, Kategorie, Material, MaterialTyp, Objekttyp
+from app.models.vorlage import Beladungsvorlage, Vorlagenposition, VorlageStatus
+
+NS = {"a": "http://schemas.openxmlformats.org/spreadsheetml/2006/main"}
+SPALTE_ARTIKEL = "B"
+SPALTE_MENGE = "C"
+SPALTE_ARTIKELNUMMER = "D"
+SPALTE_SERIENNUMMER = "E"
+STANDARD_EINHEIT = "Stück"
+
+
+@dataclass
+class ImportZeile:
+    fach: str | None
+    artikel: str
+    menge: float
+    artikelnummer: str | None
+    hat_seriennummer: bool
+
+
+@dataclass
+class ImportDatei:
+    dateiname: str
+    zeilen: list[ImportZeile] = field(default_factory=list)
+
+
+def _spalte(zellenref: str) -> str:
+    """'C17' -> 'C' (Spalten- vom Zeilenanteil der Zellreferenz trennen)."""
+    match = re.match(r"([A-Z]+)(\d+)", zellenref)
+    if not match:
+        raise ValueError(f"Ungültige Zellreferenz: {zellenref}")
+    return match.group(1)
+
+
+def _lade_shared_strings(archiv: zipfile.ZipFile) -> list[str]:
+    if "xl/sharedStrings.xml" not in archiv.namelist():
+        return []
+    root = ET.fromstring(archiv.read("xl/sharedStrings.xml"))
+    return ["".join(t.text or "" for t in si.iter("{%s}t" % NS["a"])) for si in root.findall("a:si", NS)]
+
+
+def lies_xlsx(pfad: Path) -> ImportDatei:
+    """Liest Fach/Artikel/Menge/Artikelnummer/SN-Hinweis aus dem ersten Arbeitsblatt.
+
+    Layout laut Mapping-Tabelle (ergebnisse/testphasen.md): B=Artikel, C=Menge,
+    D=Artikelnummer, E=Seriennummer (optional). Zeilen mit gesetztem B aber
+    ohne gesetztes C sind Fach-Überschriften, kein Artikel.
+    """
+    with zipfile.ZipFile(pfad) as archiv:
+        strings = _lade_shared_strings(archiv)
+        sheet = ET.fromstring(archiv.read("xl/worksheets/sheet1.xml"))
+
+        ergebnis = ImportDatei(dateiname=pfad.stem)
+        aktuelles_fach: str | None = None
+        for row in sheet.find("a:sheetData", NS) or []:
+            werte: dict[str, str | None] = {}
+            for zelle in row:
+                ref = zelle.get("r")
+                if ref is None:
+                    continue
+                spalte = _spalte(ref)
+                wert_element = zelle.find("a:v", NS)
+                wert = wert_element.text if wert_element is not None else None
+                if zelle.get("t") == "s" and wert is not None:
+                    wert = strings[int(wert)]
+                werte[spalte] = wert
+
+            artikel = werte.get(SPALTE_ARTIKEL)
+            menge = werte.get(SPALTE_MENGE)
+            if artikel is None:
+                continue
+            if menge is None:
+                # Nur B gesetzt, kein Menge -> Fach-Überschrift, kein Artikel.
+                aktuelles_fach = artikel
+                continue
+
+            ergebnis.zeilen.append(
+                ImportZeile(
+                    fach=aktuelles_fach,
+                    artikel=artikel,
+                    menge=float(menge),
+                    artikelnummer=werte.get(SPALTE_ARTIKELNUMMER),
+                    hat_seriennummer=werte.get(SPALTE_SERIENNUMMER) is not None,
+                )
+            )
+        return ergebnis
+
+
+async def _hole_oder_erstelle_bereich(db: AsyncSession, name: str) -> Bereich:
+    bereich = (await db.execute(select(Bereich).where(Bereich.name == name))).scalar_one_or_none()
+    if bereich is None:
+        bereich = Bereich(name=name)
+        db.add(bereich)
+        await db.flush()
+    return bereich
+
+
+async def _hole_oder_erstelle_kategorie(db: AsyncSession, bereich: Bereich, name: str) -> Kategorie:
+    kategorie = (
+        await db.execute(select(Kategorie).where(Kategorie.bereich_id == bereich.id, Kategorie.name == name))
+    ).scalar_one_or_none()
+    if kategorie is None:
+        kategorie = Kategorie(bereich_id=bereich.id, name=name)
+        db.add(kategorie)
+        await db.flush()
+    return kategorie
+
+
+async def _hole_oder_erstelle_material(db: AsyncSession, zeile: ImportZeile) -> Material:
+    """Abgleich primär über Artikelnummer (Prompt-Vorgabe: Namensgleichheit ist
+    Tippfehler-anfällig). Ohne Artikelnummer bleibt Name der einzige Schlüssel
+    für diesen Importlauf - Review-Pflicht laut Sprintplan.
+    """
+    material = None
+    if zeile.artikelnummer:
+        material = (
+            await db.execute(select(Material).where(Material.artikelnummer == zeile.artikelnummer))
+        ).scalar_one_or_none()
+    if material is None:
+        material = (await db.execute(select(Material).where(Material.name == zeile.artikel))).scalar_one_or_none()
+    if material is not None:
+        return material
+
+    materialtyp = MaterialTyp.geraet_sn if zeile.hat_seriennummer else MaterialTyp.standard
+    material = Material(
+        name=zeile.artikel,
+        artikelnummer=zeile.artikelnummer,
+        einheit=STANDARD_EINHEIT,
+        materialtyp=materialtyp,
+    )
+    db.add(material)
+    await db.flush()
+    return material
+
+
+async def importiere_datei(db: AsyncSession, datei: ImportDatei, objekttyp: Objekttyp) -> Beladungsvorlage:
+    vorlage = Beladungsvorlage(
+        objekttyp_id=objekttyp.id,
+        name=f"{objekttyp.name} Standard",
+        version=1,
+        status=VorlageStatus.aktiv,
+    )
+    db.add(vorlage)
+    await db.flush()
+
+    for zeile in datei.zeilen:
+        material = await _hole_oder_erstelle_material(db, zeile)
+        db.add(
+            Vorlagenposition(
+                vorlage_id=vorlage.id,
+                material_id=material.id,
+                fach=zeile.fach,
+                sollmenge=zeile.menge,
+            )
+        )
+    return vorlage
+
+
+async def importiere_alle(dateipfade: list[Path], bereich_name: str, kategorie_name: str) -> None:
+    async with SessionLocal() as db:
+        bereich = await _hole_oder_erstelle_bereich(db, bereich_name)
+        kategorie = await _hole_oder_erstelle_kategorie(db, bereich, kategorie_name)
+
+        for pfad in dateipfade:
+            datei = lies_xlsx(pfad)
+
+            bestehender_objekttyp = (
+                await db.execute(select(Objekttyp).where(Objekttyp.name == datei.dateiname))
+            ).scalar_one_or_none()
+            if bestehender_objekttyp is not None:
+                print(f"Übersprungen (Objekttyp existiert bereits): {datei.dateiname}")
+                continue
+
+            objekttyp = Objekttyp(bereich_id=bereich.id, kategorie_id=kategorie.id, name=datei.dateiname)
+            db.add(objekttyp)
+            await db.flush()
+
+            vorlage = await importiere_datei(db, datei, objekttyp)
+            print(f"Importiert: {datei.dateiname} -> Objekttyp #{objekttyp.id}, Vorlage #{vorlage.id}, {len(datei.zeilen)} Positionen")
+
+        await db.commit()
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument("dateien", nargs="+", type=Path, help="Pfade zu den .xlsx-Ist-Listen")
+    parser.add_argument("--bereich", default="Rettungsdienst/KatS")
+    parser.add_argument("--kategorie", default="Rucksäcke")
+    argumente = parser.parse_args()
+
+    asyncio.run(importiere_alle(argumente.dateien, argumente.bereich, argumente.kategorie))
+
+
+if __name__ == "__main__":
+    main()