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
den Namen - die Listen haben keine echte Artikelnummer-Spalte, siehe unten).

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_ABLAUFHINWEIS = "D"
SPALTE_SERIENNUMMER = "E"
STANDARD_EINHEIT = "Stück"

# Excel-Datumsserial (Tage seit 1899-12-30) - Spalte D ist als Datum formatiert
# und enthält bei den meisten Zeilen ein echtes Ablaufdatum (z. B. 47209 ->
# 2029-04-01, geprüft anhand der Reihe "Kompresse 10x10 / 5 / Apr 29" aus der
# Quelldatei). Ursprüngliche Annahme "D=Artikelnummer" war falsch - es gibt in
# den drei Listen gar keine echte Artikelnummer-Spalte. Plausibilitätsfenster
# grenzt Platzhalter-/Fehlwerte aus (z. B. 12754 -> 1934-12-01, mehrfach mit
# identischem Wert bei völlig unterschiedlichen Artikeln - kein echtes Datum).
_EXCEL_EPOCH_TAGE_MIN = 42000  # ca. 2015-01-01
_EXCEL_EPOCH_TAGE_MAX = 51000  # ca. 2039-08-01


def _ist_plausibles_ablaufdatum(roh_wert: str | None) -> bool:
    if roh_wert is None:
        return False
    try:
        tage = int(roh_wert)
    except ValueError:
        # Freitext wie "Jahr 2023" oder "MTK 23" kommt in den Quelldaten vor -
        # kein auswertbares Datum, aber auch kein Grund den Import abzubrechen.
        return False
    return _EXCEL_EPOCH_TAGE_MIN <= tage <= _EXCEL_EPOCH_TAGE_MAX


@dataclass
class ImportZeile:
    fach: str | None
    artikel: str
    menge: float
    hat_ablaufhinweis: bool
    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/Ablaufhinweis/SN-Hinweis aus dem ersten Arbeitsblatt.

    Layout laut Mapping-Tabelle (ergebnisse/testphasen.md): B=Artikel, C=Menge,
    D=Ablaufdatum (Excel-Datumsserial, oft plausibel, gelegentlich Platzhalter/
    Freitext), 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,
                    hat_ablaufhinweis=_ist_plausibles_ablaufdatum(werte.get(SPALTE_ABLAUFHINWEIS)),
                    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 über den Namen - die drei Quelllisten haben keine echte
    Artikelnummer-Spalte (Review-Fund, siehe Modul-Docstring). Tippfehler-Risiko
    bleibt daher wie im Sprintplan vermerkt eine manuelle Review-Pflicht.
    """
    material = (await db.execute(select(Material).where(Material.name == zeile.artikel))).scalar_one_or_none()
    if material is not None:
        return material

    if zeile.hat_seriennummer:
        materialtyp = MaterialTyp.geraet_sn
    elif zeile.hat_ablaufhinweis:
        materialtyp = MaterialTyp.ablauf_charge
    else:
        materialtyp = MaterialTyp.standard

    material = Material(
        name=zeile.artikel,
        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()