MABEA/backend/scripts/import_excel.py

"""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

            try:
                menge_zahl = float(menge)
            except ValueError:
                # Titel-/Deckblattzeilen (z. B. "Rucksack PAX SEG.xlsx" hat den
                # Objekttyp-Namen in Spalte C statt einer Menge) sind keine Artikel-
                # zeilen - überspringen statt den Import abzubrechen.
                print(f"Übersprungen (Menge nicht numerisch): {artikel!r} / {menge!r}")
                continue

            ergebnis.zeilen.append(
                ImportZeile(
                    fach=aktuelles_fach,
                    artikel=artikel,
                    menge=menge_zahl,
                    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()

    positionen_je_material: dict[int, Vorlagenposition] = {}
    for zeile in datei.zeilen:
        material = await _hole_oder_erstelle_material(db, zeile)
        bestehende_position = positionen_je_material.get(material.id)
        if bestehende_position is not None:
            # Gleiches Material in mehreren Fächern derselben Liste (z. B. Verbandsmaterial
            # in mehreren Taschen) - Schema erlaubt nur eine Zeile pro Material+Vorlage,
            # daher Mengen zusammenführen statt zweite Zeile einzufügen (führt sonst zu
            # UniqueViolationError auf vorlage_id+material_id).
            bestehende_position.sollmenge += zeile.menge
            if zeile.fach and zeile.fach not in (bestehende_position.fach or ""):
                bestehende_position.fach = f"{bestehende_position.fach}, {zeile.fach}" if bestehende_position.fach else zeile.fach
            continue

        position = Vorlagenposition(
            vorlage_id=vorlage.id,
            material_id=material.id,
            fach=zeile.fach,
            sollmenge=zeile.menge,
        )
        db.add(position)
        positionen_je_material[material.id] = position
    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()