ScarriffleandClaude Opus 5
fa5a2bc4fa
Eine Teilmenge auszulagern ging rechnerisch schon, aber es blieb keine Spur davon: Eine Charge mit 900 von urspruenglich 1000 ml war von einer frisch eingelagerten 900-ml-Charge nicht zu unterscheiden, und eine geoeffnete Milch galt weiter bis zum aufgedruckten MHD. Anbrechen spaltet jetzt eine Packung ab. Aus 4 x 1 l wird eine offene Charge mit 1 l und eine ungeoeffnete mit 3 l - sonst truegen die restlichen 3 l das kurze Verbrauchsdatum mit, obwohl sie noch zu sind. Der Gesamtbestand aendert sich dabei nicht, es ist eine Umbuchung wie das Aufteilen und schreibt wie dieses keine Bewegung. Verbrauchen geht in einem Schritt: Wer "100 ml verbraucht" eintraegt, bricht die Charge dabei an (abschaltbar ueber auto_open). Gerechnet wird vor dem Anbrechen - scheitert die Einheit, soll keine Packung grundlos offen stehen. MHD und "zu verbrauchen bis" stehen nebeneinander und bleiben beide sichtbar; gewarnt wird nach dem FRUEHEREN (Lot.faellig_am). Dafuer musste in /expiring die Vorbedingung "best_before IS NOT NULL" weg, sonst fehlte ausgerechnet die angebrochene Packung ohne aufgedrucktes MHD. ExpiringItem traegt beide Daten plus warn_date und warn_source, und best_before ist dort jetzt optional. Die Frist kommt aus drei Ebenen: Artikel schlaegt Kategorie, Kategorie schlaegt ueber ihre Oberkategorien hinweg den globalen Wert. Der Helfer dafuer (services/vererbung.py) liest die Kategoriekette von services/fields.py rueckwaerts - dort werden Felder ALLER Vorfahren gesammelt, hier gewinnt der naechste. Dieselbe Mechanik traegt die Hausmasse. Hausmasse sind der zweite Teil: "2 EL Mehl" liess sich bisher nicht buchen, weil eine globale Einheit "Essloeffel" bei Mehl und Zucker zwangslaeufig gleich schwer waere. Der Name ist deshalb global, die Menge haengt an Kategorie und Artikel. to_base loest sie nur mit hausmasse=True auf - beim Auslagern und Verbrauchen ja, beim Einlagern und bei Mindestbestaenden ausdruecklich nicht, denn ein Loeffel ist ein Schaetzwert. Verwaltete Einheiten werden zuerst geprueft, damit ein Mass namens "Liter" die echte Einheit nicht verdeckt. Zwei Stellen waeren sonst still kaputtgegangen: - Die Konsolidierung haette die offene Packung beim naechsten Einlagern am selben Ort einkassiert und ihr Verbrauchsdatum verschluckt. Angebrochene Chargen werden deshalb nie zusammengefasst - in consolidate_lot_group und im Start-Aufraeumer, beide bisher ohne jeden Test. Das Netz dafuer ist jetzt da (test_konsolidierung.py). - FEFO sortiert nach dem wirksamen Datum statt nur nach dem MHD. Eine offene Packung kommt damit meist von selbst zuerst, aber eine ungeoeffnete, die morgen ablaeuft, geht vor - Verderb schlaegt Bequemlichkeit. Erst bei gleichem Datum wird das Angebrochene vorgezogen. Dashboard bekommt "angebrochen" als fuenften Zustand, mit Vorrang abgelaufen > bald > angebrochen > in Ordnung: Was offen ist und morgen ablaeuft, gehoert in die Warnung, nicht in den ruhigeren Zustand. Oberflaechen in Web und App folgen in eigenen Commits. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1095 lines
42 KiB
Python
1095 lines
42 KiB
Python
"""Export und Import von Beständen (CSV für Tabellen, JSON für Backups).
|
||
|
||
Der Import arbeitet ausschließlich additiv: Unbekanntes wird angelegt, Chargen
|
||
werden ergänzt – es wird nie etwas gelöscht oder überschrieben.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import csv
|
||
import io
|
||
import json
|
||
from datetime import date, datetime, timezone
|
||
|
||
from fastapi import APIRouter, Body, Depends, File, HTTPException, UploadFile, status
|
||
from fastapi.responses import Response
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..database import get_db
|
||
from ..deps import get_current_user, require_admin
|
||
from ..services.master_data import export_master_data, import_master_data
|
||
from ..services.dates import MONTH, clean_precision, normalize_best_before
|
||
from ..services import gruppen as gruppen_graph
|
||
from ..services.group_codes import sync as sync_group_code
|
||
from ..services.min_stock import lies_ueberall, schreibe_ueberall
|
||
from ..services.etiketten import FELD_PRAEFIX, konfiguration_lesen
|
||
from ..models import (
|
||
BaseUnit,
|
||
Category,
|
||
CategoryTracking,
|
||
DatePrecision,
|
||
EtikettenVorlage,
|
||
FieldDefinition,
|
||
Group,
|
||
GroupLocationMinStock,
|
||
Item,
|
||
Location,
|
||
Lot,
|
||
Movement,
|
||
MovementType,
|
||
Product,
|
||
ProductLocationMinStock,
|
||
Shop,
|
||
Unit,
|
||
UnitKind,
|
||
User,
|
||
)
|
||
from ..services.conversion import BASE_OF_KIND, display_unit_info, find_unit
|
||
from ..services.fields import (
|
||
FieldError,
|
||
apply_field_values,
|
||
effective_field_definitions,
|
||
options_list,
|
||
slugify,
|
||
)
|
||
|
||
router = APIRouter(tags=["transfer"])
|
||
|
||
CSV_FIELDS = [
|
||
"barcode", "name", "marke", "einheit", "packungsgroesse", "gebinde",
|
||
"gruppe", "kategorie", "art", "mindestbestand", "menge", "menge_einheit", "mhd", "lagerort",
|
||
]
|
||
|
||
|
||
def _tracking_from_art(text) -> str:
|
||
"""Spalte „art" → Verwaltungsart einer (neu angelegten) Kategorie.
|
||
|
||
Leer/unbekannt ⇒ Lebensmittel (bewahrt das Verhalten für zurückgespielte
|
||
Lebensmittel-Exporte).
|
||
"""
|
||
t = (str(text) if text is not None else "").strip().lower()
|
||
if t in ("gegenstand", "gegenstände", "object", "objekt", "non-food", "nonfood"):
|
||
return CategoryTracking.object.value
|
||
return CategoryTracking.food.value
|
||
|
||
|
||
def _art_label(product: Product) -> str:
|
||
if product.category and product.category.tracking == CategoryTracking.object.value:
|
||
return "Gegenstand"
|
||
return "Lebensmittel"
|
||
PACKAGE_TOKENS = {"packung", "package", "pkg", "pack"}
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Export
|
||
# --------------------------------------------------------------------------
|
||
def _zeitstempel() -> str:
|
||
"""Ortszeit als YYYY-MM-DD_HHMM für Dateinamen.
|
||
|
||
Ohne Zeitstempel heißen mehrere Ausleitungen alle gleich und der Browser
|
||
haengt (1), (2) an – dann weiß niemand mehr, welche die aktuelle ist.
|
||
"""
|
||
return datetime.now().strftime("%Y-%m-%d_%H%M")
|
||
|
||
|
||
def _article_unit(product: Product) -> tuple[float, str]:
|
||
"""Faktor und Bezeichnung der Artikeleinheit (Gebinde, sonst Produkteinheit)."""
|
||
unit_name, unit_factor = display_unit_info(product)
|
||
if product.package_size and product.package_size > 0:
|
||
return product.package_size, (product.package_label or "Packung")
|
||
return unit_factor, unit_name
|
||
|
||
|
||
@router.get("/export/stock.csv")
|
||
def export_stock_csv(
|
||
db: Session = Depends(get_db), _: User = Depends(get_current_user)
|
||
) -> Response:
|
||
"""Eine Zeile je Charge – direkt in Excel/LibreOffice bearbeitbar."""
|
||
buf = io.StringIO()
|
||
writer = csv.writer(buf, delimiter=";", lineterminator="\r\n")
|
||
writer.writerow(CSV_FIELDS)
|
||
|
||
for product in db.query(Product).order_by(Product.name).all():
|
||
unit_name, _ = display_unit_info(product)
|
||
factor, amount_label = _article_unit(product)
|
||
base = [
|
||
product.barcode or "",
|
||
product.name,
|
||
product.brand or "",
|
||
unit_name,
|
||
product.package_size if product.package_size else "",
|
||
product.package_label or "",
|
||
product.group.name if product.group else "",
|
||
_category_path(db, product.category),
|
||
_art_label(product),
|
||
# „Ueberall"-Bedarf (frueher der Gesamt-Mindestbestand am Artikel).
|
||
# Die CSV kennt nur diese eine Spalte; Ort-Bedarfe stehen im
|
||
# JSON-Backup.
|
||
_ueberall_csv(product),
|
||
]
|
||
lots = (
|
||
db.query(Lot)
|
||
.filter(Lot.product_id == product.id)
|
||
.order_by(Lot.best_before.is_(None), Lot.best_before)
|
||
.all()
|
||
)
|
||
if not lots:
|
||
# Produkt ohne Bestand: Stammdaten trotzdem mitnehmen.
|
||
writer.writerow(base + ["", "", "", ""])
|
||
continue
|
||
for lot in lots:
|
||
location = db.get(Location, lot.location_id) if lot.location_id else None
|
||
writer.writerow(
|
||
base
|
||
+ [
|
||
round(lot.quantity / factor, 6),
|
||
amount_label,
|
||
_format_best_before(lot),
|
||
location.name if location else "",
|
||
]
|
||
)
|
||
|
||
# BOM, damit Excel die Umlaute korrekt erkennt.
|
||
return Response(
|
||
content="" + buf.getvalue(),
|
||
media_type="text/csv; charset=utf-8",
|
||
headers={
|
||
"Content-Disposition": f'attachment; filename="bestand_{_zeitstempel()}.csv"'
|
||
},
|
||
)
|
||
|
||
|
||
@router.get("/export/backup.json")
|
||
def export_backup_json(
|
||
db: Session = Depends(get_db), _: User = Depends(get_current_user)
|
||
) -> Response:
|
||
"""Vollständiges Backup: Einheiten, Gruppen, Lagerorte, Produkte, Chargen."""
|
||
locations = db.query(Location).order_by(Location.id).all()
|
||
loc_name = {loc.id: loc.name for loc in locations}
|
||
|
||
data = {
|
||
# v4: Etiketten-Vorlagen samt hinterlegter P-touch-Datei.
|
||
# v3: Gruppen mit Obergruppen und Gebinde, Mindestbestaende je Ort
|
||
# (Ort null = „Ueberall") in Basiseinheiten. Aeltere Sicherungen bleiben
|
||
# lesbar – ihnen fehlen diese Listen einfach.
|
||
"version": 4,
|
||
"exported_at": datetime.now(timezone.utc).isoformat(),
|
||
"exported_at_local": datetime.now().isoformat(timespec="seconds"),
|
||
"units": [
|
||
{"name": u.name, "kind": u.kind.value, "factor": u.factor}
|
||
for u in db.query(Unit).order_by(Unit.id).all()
|
||
],
|
||
"groups": [
|
||
{
|
||
"name": g.name,
|
||
"min_stock_unit": g.min_stock_unit.name if g.min_stock_unit else None,
|
||
"package_size": g.package_size,
|
||
"package_label": g.package_label,
|
||
"min_stock_in_packages": g.min_stock_in_packages,
|
||
# Namen statt IDs: die sind zwischen zwei Instanzen nicht gleich.
|
||
"parents": sorted(p.name for p in g.parents),
|
||
"min_stocks": _min_stock_liste(g, loc_name),
|
||
}
|
||
for g in db.query(Group).order_by(Group.id).all()
|
||
],
|
||
"locations": [
|
||
{"name": loc.name, "parent": loc_name.get(loc.parent_id)}
|
||
for loc in locations
|
||
],
|
||
# Kategorien mit Verwaltungsart, damit Lebensmittel/Gegenstände beim
|
||
# Wiederherstellen erhalten bleiben (auch leere Kategorien).
|
||
"categories": [
|
||
{"path": _category_path(db, c), "tracking": c.tracking}
|
||
for c in db.query(Category).order_by(Category.id).all()
|
||
],
|
||
# Bezugsquellen (nur für Gegenstände).
|
||
"shops": [
|
||
{"name": s.name, "website": s.website}
|
||
for s in db.query(Shop).order_by(Shop.id).all()
|
||
],
|
||
# Selbst definierte Felder je Kategorie.
|
||
"field_definitions": [
|
||
{
|
||
"category": _category_path(db, db.get(Category, fd.category_id)),
|
||
"label": fd.label,
|
||
"field_type": fd.field_type,
|
||
"unit": fd.unit,
|
||
"options": options_list(fd),
|
||
"required": fd.required,
|
||
"position": fd.position,
|
||
}
|
||
for fd in db.query(FieldDefinition)
|
||
.order_by(FieldDefinition.category_id, FieldDefinition.position, FieldDefinition.id)
|
||
.all()
|
||
],
|
||
"products": [],
|
||
}
|
||
|
||
for p in db.query(Product).order_by(Product.id).all():
|
||
unit_name, _ = display_unit_info(p)
|
||
lots = db.query(Lot).filter(Lot.product_id == p.id).all()
|
||
data["products"].append(
|
||
{
|
||
"barcode": p.barcode,
|
||
"name": p.name,
|
||
"brand": p.brand,
|
||
"image_url": p.image_url,
|
||
"unit": unit_name,
|
||
"package_size": p.package_size,
|
||
"package_label": p.package_label,
|
||
"date_precision": p.date_precision,
|
||
"group": p.group.name if p.group else None,
|
||
"category": _category_path(db, p.category),
|
||
"min_stock_unit": p.min_stock_unit.name if p.min_stock_unit else None,
|
||
"min_stock_in_packages": bool(p.min_stock_in_packages),
|
||
# Zweiteinheit-Brücke („3 Stück ≙ 250 g"), in Basiseinheiten.
|
||
"secondary_base": p.secondary_base,
|
||
"secondary_count": p.secondary_count,
|
||
"secondary_amount": p.secondary_amount,
|
||
"min_stocks": _min_stock_liste(p, loc_name),
|
||
# Gegenstands-Felder:
|
||
"shop": p.shop.name if p.shop else None,
|
||
"product_url": p.product_url,
|
||
"field_values": {
|
||
pfv.field_definition.label: pfv.value for pfv in p.field_values
|
||
},
|
||
"lots": [
|
||
{
|
||
"quantity": lot.quantity,
|
||
"best_before": lot.best_before.isoformat() if lot.best_before else None,
|
||
"best_before_precision": lot.best_before_precision,
|
||
"location": loc_name.get(lot.location_id),
|
||
# Angebrochene Packungen sollen eine Sicherung ueberleben –
|
||
# sonst waere nach dem Zurueckspielen alles wieder zu.
|
||
"opened_on": lot.opened_on.isoformat() if lot.opened_on else None,
|
||
"consume_by": lot.consume_by.isoformat() if lot.consume_by else None,
|
||
"opened_quantity": lot.opened_quantity,
|
||
}
|
||
for lot in lots
|
||
],
|
||
}
|
||
)
|
||
|
||
data["etiketten_vorlagen"] = _vorlagen_exportieren(db)
|
||
|
||
return Response(
|
||
content=json.dumps(data, ensure_ascii=False, indent=2),
|
||
media_type="application/json; charset=utf-8",
|
||
headers={
|
||
"Content-Disposition":
|
||
f'attachment; filename="vorrania-backup_{_zeitstempel()}.json"'
|
||
},
|
||
)
|
||
|
||
|
||
@router.get("/export/master-data")
|
||
def export_masterdata(
|
||
db: Session = Depends(get_db), _: User = Depends(require_admin)
|
||
) -> dict:
|
||
"""Stammdaten (Kategorien, Felder, Lagerorte, Einheiten, Gebinde) als JSON –
|
||
mit IDs, für Sicherung/Umzug."""
|
||
return export_master_data(db)
|
||
|
||
|
||
@router.post("/import/master-data")
|
||
def import_masterdata(
|
||
payload: dict = Body(...),
|
||
mode: str = "skip",
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(require_admin),
|
||
) -> dict:
|
||
"""Stammdaten aus JSON einspielen. mode=skip (Vorhandenes lassen) oder
|
||
overwrite (per ID überschreiben). IDs werden aus der Datei übernommen."""
|
||
try:
|
||
return import_master_data(db, payload, mode=mode)
|
||
except ValueError as e:
|
||
raise HTTPException(status.HTTP_400_BAD_REQUEST, str(e))
|
||
|
||
|
||
@router.get("/export/labels")
|
||
def export_labels(
|
||
category_ids: str | None = None,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(get_current_user),
|
||
) -> list[dict]:
|
||
"""Zeilen für den QR-Etiketten-Export (P-touch & Co.) – je Einzelstück eine.
|
||
|
||
``category_ids`` (z.B. "1,3") grenzt auf Kategorien samt Unterkategorien ein;
|
||
leer = alle Einzelstücke. Den QR-Inhalt (…/i/<UID>) setzt die Weboberfläche
|
||
dazu, weil nur sie die öffentliche Adresse kennt.
|
||
"""
|
||
from .categories import descendant_ids
|
||
|
||
ids: set[int] | None = None
|
||
if category_ids:
|
||
ids = set()
|
||
for part in category_ids.split(","):
|
||
teil = part.strip()
|
||
if teil.isdigit():
|
||
ids |= descendant_ids(db, int(teil))
|
||
|
||
query = db.query(Item)
|
||
if ids is not None:
|
||
query = (
|
||
query.join(Product, Item.product_id == Product.id)
|
||
.filter(Product.category_id.in_(ids))
|
||
)
|
||
rows: list[dict] = []
|
||
for item in query.order_by(Item.id).all():
|
||
product = item.product
|
||
rows.append({
|
||
"uid": item.uid,
|
||
"product": product.name if product else "",
|
||
"brand": (product.brand if product else "") or "",
|
||
"category": _category_path(db, product.category) if product else "",
|
||
"location": item.location.name if item.location else "",
|
||
})
|
||
return rows
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Import (nur additiv)
|
||
# --------------------------------------------------------------------------
|
||
def _parse_tag(value) -> "date | None":
|
||
"""ISO-Datum aus einer Sicherung lesen; Muell wird still zu None.
|
||
|
||
Der Import ist bewusst additiv und soll nicht an einer einzelnen Zeile
|
||
scheitern – dieselbe Haltung wie bei ``_parse_best_before``.
|
||
"""
|
||
if not value:
|
||
return None
|
||
try:
|
||
return date.fromisoformat(str(value)[:10])
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def _zahl_oder_none(value) -> float | None:
|
||
try:
|
||
return float(value) if value is not None else None
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
|
||
def _num(value) -> float | None:
|
||
if value is None:
|
||
return None
|
||
text = str(value).strip().replace(",", ".")
|
||
if text == "":
|
||
return None
|
||
return float(text)
|
||
|
||
|
||
def _parse_date(value) -> date | None:
|
||
text = (str(value) if value is not None else "").strip()
|
||
if not text:
|
||
return None
|
||
for fmt in ("%Y-%m-%d", "%d.%m.%Y", "%d.%m.%y"):
|
||
try:
|
||
return datetime.strptime(text, fmt).date()
|
||
except ValueError:
|
||
continue
|
||
raise ValueError(f"Datum nicht lesbar: {text}")
|
||
|
||
|
||
def _parse_best_before(value, precision=None) -> tuple[date | None, str]:
|
||
"""Liest ein MHD und erkennt dabei, ob nur Monat/Jahr angegeben war.
|
||
|
||
Steht die Genauigkeit ausdrücklich in der Datei (JSON-Backup), gilt sie.
|
||
Sonst wird sie aus der Schreibweise abgeleitet – "09/2026" ist eine
|
||
Monatsangabe, "2026-09-07" eine tagesgenaue.
|
||
"""
|
||
text = (str(value) if value is not None else "").strip()
|
||
if not text:
|
||
return None, DatePrecision.day.value
|
||
|
||
if precision == DatePrecision.month.value:
|
||
return normalize_best_before(_parse_date(text), MONTH), MONTH
|
||
|
||
for fmt in ("%m/%Y", "%Y-%m", "%m.%Y"):
|
||
try:
|
||
parsed = datetime.strptime(text, fmt).date()
|
||
except ValueError:
|
||
continue
|
||
return normalize_best_before(parsed, MONTH), MONTH
|
||
|
||
return _parse_date(text), DatePrecision.day.value
|
||
|
||
|
||
def _format_best_before(lot: Lot) -> str:
|
||
"""Monatsangaben als "09/2026" schreiben, damit die Datei lesbar bleibt."""
|
||
if not lot.best_before:
|
||
return ""
|
||
if lot.best_before_precision == MONTH:
|
||
return lot.best_before.strftime("%m/%Y")
|
||
return lot.best_before.isoformat()
|
||
|
||
|
||
IMPORT_MODES = {"add", "replace_listed", "replace_all"}
|
||
|
||
|
||
def _clear_lots(db: Session, product: Product, user: User) -> None:
|
||
"""Entfernt alle Chargen eines Produkts und protokolliert das als Korrektur."""
|
||
lots = db.query(Lot).filter(Lot.product_id == product.id).all()
|
||
total = float(sum(lot.quantity for lot in lots))
|
||
for lot in lots:
|
||
db.delete(lot)
|
||
if total > 0:
|
||
db.add(
|
||
Movement(
|
||
product_id=product.id,
|
||
lot_id=None,
|
||
user_id=user.id,
|
||
type=MovementType.adjust,
|
||
quantity=-total,
|
||
unit_used="base",
|
||
note="Import: Bestand ersetzt",
|
||
)
|
||
)
|
||
db.flush()
|
||
|
||
|
||
def _clear_all_lots(db: Session, user: User) -> None:
|
||
for product in db.query(Product).all():
|
||
_clear_lots(db, product, user)
|
||
|
||
|
||
def _get_or_create_location(db: Session, name: str | None) -> Location | None:
|
||
name = (name or "").strip()
|
||
if not name:
|
||
return None
|
||
loc = db.query(Location).filter(Location.name == name).first()
|
||
if loc is None:
|
||
loc = Location(name=name)
|
||
db.add(loc)
|
||
db.flush()
|
||
return loc
|
||
|
||
|
||
def _vorlagen_exportieren(db: Session) -> list[dict]:
|
||
"""Etiketten-Vorlagen fuer die Sicherung.
|
||
|
||
Kategorien als PFAD und eigene Felder als LABEL – IDs sind zwischen zwei
|
||
Instanzen nicht dieselben, genau wie bei Gruppen und Artikeln.
|
||
|
||
Die hinterlegte .lbx kommt als Base64 mit. Das ist das erste Binaerformat in
|
||
dieser Sicherung; vertretbar, weil eine .lbx wenige Kilobyte gross ist und
|
||
eine Vorlage ohne ihre Datei nur eine halbe Vorlage waere.
|
||
"""
|
||
zeilen: list[dict] = []
|
||
for v in db.query(EtikettenVorlage).order_by(EtikettenVorlage.position).all():
|
||
konf = konfiguration_lesen(v)
|
||
spalten: list[str] = []
|
||
for schluessel in konf["spalten"]:
|
||
if schluessel.startswith(FELD_PRAEFIX):
|
||
rest = schluessel[len(FELD_PRAEFIX):]
|
||
fd = db.get(FieldDefinition, int(rest)) if rest.isdigit() else None
|
||
# Kategoriepfad mit dazu: Zwei Kategorien duerfen dasselbe Label
|
||
# fuehren, erst beides zusammen ist eindeutig.
|
||
spalten.append(
|
||
f"{FELD_PRAEFIX}{_category_path(db, db.get(Category, fd.category_id))}"
|
||
f"::{fd.label}" if fd else schluessel
|
||
)
|
||
else:
|
||
spalten.append(schluessel)
|
||
zeilen.append({
|
||
"name": v.name,
|
||
"art": v.art,
|
||
"position": v.position,
|
||
"spalten": spalten,
|
||
"kategorien": [
|
||
_category_path(db, db.get(Category, kid)) for kid in konf["kategorie_ids"]
|
||
],
|
||
"trennzeichen": konf["trennzeichen"],
|
||
"datei_name": v.datei_name,
|
||
"datei_base64": base64.b64encode(v.datei).decode() if v.datei else None,
|
||
})
|
||
return zeilen
|
||
|
||
|
||
def _vorlagen_importieren(db: Session, zeilen: list, errors: list[str]) -> int:
|
||
"""Vorlagen additiv einspielen – Bestehendes bleibt unberuehrt."""
|
||
angelegt = 0
|
||
pfad_zu_id: dict[str, int] = {}
|
||
for cat in db.query(Category).all():
|
||
pfad_zu_id[_category_path(db, cat)] = cat.id
|
||
|
||
for entry in zeilen:
|
||
try:
|
||
name = (entry.get("name") or "").strip()
|
||
if not name or db.query(EtikettenVorlage).filter(
|
||
EtikettenVorlage.name == name
|
||
).first():
|
||
continue
|
||
|
||
spalten: list[str] = []
|
||
for roh in entry.get("spalten", []):
|
||
if not str(roh).startswith(FELD_PRAEFIX):
|
||
spalten.append(str(roh))
|
||
continue
|
||
rest = str(roh)[len(FELD_PRAEFIX):]
|
||
pfad, _, label = rest.rpartition("::")
|
||
kat_id = pfad_zu_id.get(pfad)
|
||
fd = (
|
||
db.query(FieldDefinition)
|
||
.filter(
|
||
FieldDefinition.category_id == kat_id,
|
||
FieldDefinition.label == label,
|
||
).first()
|
||
if kat_id else None
|
||
)
|
||
# Gibt es das Feld hier nicht, faellt die Spalte weg statt die
|
||
# ganze Vorlage scheitern zu lassen.
|
||
if fd is not None:
|
||
spalten.append(f"{FELD_PRAEFIX}{fd.id}")
|
||
|
||
kategorie_ids = [
|
||
pfad_zu_id[p] for p in entry.get("kategorien", []) if p in pfad_zu_id
|
||
]
|
||
roh_datei = entry.get("datei_base64")
|
||
db.add(EtikettenVorlage(
|
||
name=name,
|
||
art=entry.get("art") or "items",
|
||
position=int(entry.get("position") or 0),
|
||
konfiguration=json.dumps({
|
||
"spalten": spalten,
|
||
"kategorie_ids": kategorie_ids,
|
||
"trennzeichen": entry.get("trennzeichen") or ",",
|
||
}, ensure_ascii=False),
|
||
datei_name=entry.get("datei_name"),
|
||
datei=base64.b64decode(roh_datei) if roh_datei else None,
|
||
))
|
||
angelegt += 1
|
||
except Exception as exc:
|
||
errors.append(f"Etiketten-Vorlage {entry.get('name')!r}: {exc}")
|
||
db.commit()
|
||
return angelegt
|
||
|
||
|
||
def _category_path(db: Session, category: Category | None) -> str:
|
||
"""Unterkategorien als "Süßwaren & Snacks > Schokolade" in EINE Spalte.
|
||
|
||
So bleibt die CSV in Excel bedienbar, statt für jede Ebene eine eigene
|
||
Spalte zu brauchen.
|
||
"""
|
||
if category is None:
|
||
return ""
|
||
teile = [category.name]
|
||
aktuell, tiefe = category, 0
|
||
while aktuell.parent_id is not None and tiefe < 10: # Schutz vor einem Ring
|
||
aktuell = db.get(Category, aktuell.parent_id)
|
||
if aktuell is None:
|
||
break
|
||
teile.append(aktuell.name)
|
||
tiefe += 1
|
||
return " > ".join(reversed(teile))
|
||
|
||
|
||
def _get_or_create_category(
|
||
db: Session, path: str | None, tracking: str = CategoryTracking.food.value
|
||
) -> Category | None:
|
||
"""Legt den ganzen Pfad an, falls Teile davon fehlen.
|
||
|
||
Importierte Kategorien sind standardmäßig „food": Backups stammen aus der
|
||
Lebensmittel-Ausgabe, und so verhalten sich wiederhergestellte Artikel wie
|
||
zuvor. Der Modus aus einem neueren Backup (Liste ``categories``) überschreibt
|
||
das anschließend.
|
||
"""
|
||
path = (path or "").strip()
|
||
if not path:
|
||
return None
|
||
parent: Category | None = None
|
||
for name in [t.strip() for t in path.split(">") if t.strip()]:
|
||
query = db.query(Category).filter(Category.name == name)
|
||
query = query.filter(
|
||
Category.parent_id == (parent.id if parent else None)
|
||
)
|
||
node = query.first()
|
||
if node is None:
|
||
node = Category(
|
||
name=name,
|
||
parent_id=parent.id if parent else None,
|
||
tracking=tracking,
|
||
)
|
||
db.add(node)
|
||
db.flush()
|
||
parent = node
|
||
return parent
|
||
|
||
|
||
def _get_or_create_shop(db: Session, name: str | None, website: str | None = None) -> Shop | None:
|
||
name = (name or "").strip()
|
||
if not name:
|
||
return None
|
||
shop = db.query(Shop).filter(Shop.name == name).first()
|
||
if shop is None:
|
||
shop = Shop(name=name, website=(website or None))
|
||
db.add(shop)
|
||
db.flush()
|
||
return shop
|
||
|
||
|
||
def _ueberall_csv(besitzer) -> float | str:
|
||
"""Der „Ueberall"-Mindestbestand fuer die CSV-Spalte (leer, wenn keiner)."""
|
||
wert = lies_ueberall(besitzer.location_min_stocks)
|
||
return "" if wert is None else wert
|
||
|
||
|
||
def _min_stock_liste(besitzer, loc_name: dict[str, str]) -> list[dict]:
|
||
"""Mindestbestaende eines Artikels/einer Gruppe je Ort, fuer die Sicherung.
|
||
|
||
Mengen in Basiseinheiten, Orte als Name; ``location = null`` ist „Ueberall".
|
||
"""
|
||
return [
|
||
{
|
||
"location": None if e.location_id is None else loc_name.get(e.location_id),
|
||
"min_stock": e.min_stock,
|
||
}
|
||
for e in besitzer.location_min_stocks
|
||
]
|
||
|
||
|
||
def _min_stocks_einspielen(db: Session, besitzer, eintraege: list[dict] | None) -> None:
|
||
"""Mindestbestaende aus der Sicherung setzen – nur, wenn noch keine da sind.
|
||
|
||
Wie der ganze Import additiv: Vorhandenes wird nie ueberschrieben.
|
||
"""
|
||
if not eintraege or besitzer.location_min_stocks:
|
||
return
|
||
for eintrag in eintraege:
|
||
menge = eintrag.get("min_stock")
|
||
if menge is None or float(menge) <= 0:
|
||
continue
|
||
ort_name = eintrag.get("location")
|
||
ort = None
|
||
if ort_name:
|
||
ort = db.query(Location).filter(Location.name == ort_name).first()
|
||
if ort is None:
|
||
continue # Ort fehlt in dieser Instanz – Eintrag entfaellt
|
||
# Beziehung mitsetzen, nicht nur die ID: sonst liefert ``.location`` bis
|
||
# zum naechsten Commit None, obwohl der Ort feststeht.
|
||
if isinstance(besitzer, Product):
|
||
besitzer.location_min_stocks.append(ProductLocationMinStock(
|
||
product_id=besitzer.id, location=ort,
|
||
location_id=ort.id if ort else None, min_stock=float(menge)))
|
||
else:
|
||
besitzer.location_min_stocks.append(GroupLocationMinStock(
|
||
group_id=besitzer.id, location=ort,
|
||
location_id=ort.id if ort else None, min_stock=float(menge)))
|
||
|
||
|
||
def _get_or_create_group(db: Session, name: str | None) -> Group | None:
|
||
name = (name or "").strip()
|
||
if not name:
|
||
return None
|
||
group = db.query(Group).filter(Group.name == name).first()
|
||
if group is None:
|
||
group = Group(name=name)
|
||
db.add(group)
|
||
db.flush()
|
||
return group
|
||
|
||
|
||
def _bildadresse_ergaenzen(product: Product, row: dict) -> None:
|
||
"""Fehlende Bildadresse aus dem Backup nachtragen – vorhandene bleibt.
|
||
|
||
Nur ergaenzend: Ein Import soll Daten hinzufuegen, nicht stillschweigend
|
||
etwas ueberschreiben, das jemand von Hand gesetzt hat.
|
||
"""
|
||
adresse = (row.get("image_url") or "").strip()
|
||
if adresse and not product.image_url:
|
||
product.image_url = adresse
|
||
|
||
|
||
def _get_or_create_product(db: Session, row: dict, created: list[str]) -> Product:
|
||
"""Sucht per Barcode, sonst per Name; legt das Produkt sonst an."""
|
||
barcode = (row.get("barcode") or "").strip() or None
|
||
name = (row.get("name") or "").strip()
|
||
if barcode:
|
||
product = db.query(Product).filter(Product.barcode == barcode).first()
|
||
if product:
|
||
_bildadresse_ergaenzen(product, row)
|
||
return product
|
||
if name:
|
||
product = db.query(Product).filter(Product.name == name).first()
|
||
if product:
|
||
_bildadresse_ergaenzen(product, row)
|
||
return product
|
||
if not name:
|
||
raise ValueError("Produktname fehlt")
|
||
|
||
unit = find_unit(db, (row.get("einheit") or "Stück"))
|
||
if unit is None:
|
||
raise ValueError(f"Unbekannte Einheit: {row.get('einheit')}")
|
||
group = _get_or_create_group(db, row.get("gruppe"))
|
||
category = _get_or_create_category(
|
||
db, row.get("kategorie"), tracking=_tracking_from_art(row.get("art"))
|
||
)
|
||
product = Product(
|
||
barcode=barcode,
|
||
name=name,
|
||
brand=(row.get("marke") or "").strip() or None,
|
||
# Die Bildadresse stand zwar immer im JSON-Backup, wurde beim Einlesen
|
||
# aber verworfen - eingelesene Artikel blieben deshalb dauerhaft ohne
|
||
# Bild. Die lokale Kopie holt sich der Bildabruf beim ersten Ansehen.
|
||
image_url=(row.get("image_url") or "").strip() or None,
|
||
base_unit=BASE_OF_KIND[unit.kind],
|
||
display_unit_id=unit.id,
|
||
package_size=_num(row.get("packungsgroesse")),
|
||
package_label=(row.get("gebinde") or "").strip() or None,
|
||
# Nur im JSON-Backup enthalten; aus der CSV kommt hier nichts.
|
||
date_precision=clean_precision((row.get("mhd_genauigkeit") or "").strip() or None),
|
||
group_id=group.id if group else None,
|
||
category_id=category.id if category else None,
|
||
source="import",
|
||
)
|
||
db.add(product)
|
||
db.flush()
|
||
# Die CSV-Spalte „mindestbestand" (und das v2-Backup) meinen den Bedarf
|
||
# ohne Ortsangabe – das ist jetzt die „Ueberall"-Zeile. Basiseinheiten.
|
||
schreibe_ueberall(db, product, _num(row.get("mindestbestand")))
|
||
# Damit ein eingelesenes Backup denselben Stand erzeugt wie das Anlegen
|
||
# ueber die Oberflaeche.
|
||
sync_group_code(db, product)
|
||
created.append(product.name)
|
||
return product
|
||
|
||
|
||
def _quantity_to_base(db: Session, product: Product, amount: float, token: str) -> float:
|
||
"""Rechnet eine Importmenge in Basiseinheiten um (kennt auch das Gebinde)."""
|
||
text = (token or "").strip().lower()
|
||
package_label = (product.package_label or "Packung").strip().lower()
|
||
if text in PACKAGE_TOKENS or (text and text == package_label):
|
||
if not product.package_size:
|
||
raise ValueError(f"'{token}': keine Packungsgröße für {product.name} hinterlegt")
|
||
return amount * product.package_size
|
||
if not text:
|
||
return amount # ohne Angabe: Basiseinheiten
|
||
unit = find_unit(db, text)
|
||
if unit is None:
|
||
raise ValueError(f"Unbekannte Einheit: {token}")
|
||
return amount * unit.factor
|
||
|
||
|
||
def _import_csv(db: Session, content: bytes, user: User, mode: str) -> dict:
|
||
text = content.decode("utf-8-sig", errors="replace")
|
||
sample = text[:2048]
|
||
delimiter = ";" if sample.count(";") >= sample.count(",") else ","
|
||
reader = csv.DictReader(io.StringIO(text), delimiter=delimiter)
|
||
|
||
created_products: list[str] = []
|
||
lots_added = 0
|
||
errors: list[str] = []
|
||
cleared: set[int] = set()
|
||
|
||
if mode == "replace_all":
|
||
_clear_all_lots(db, user)
|
||
|
||
for index, raw in enumerate(reader, start=2): # Zeile 1 = Kopfzeile
|
||
row = {(k or "").strip().lower(): (v if v is not None else "") for k, v in raw.items()}
|
||
if not any(str(v).strip() for v in row.values()):
|
||
continue
|
||
try:
|
||
product = _get_or_create_product(db, row, created_products)
|
||
# Beim Ersetzen: Bestand des Produkts einmalig leeren, bevor die
|
||
# Zeilen dieser Datei dazukommen.
|
||
if mode == "replace_listed" and product.id not in cleared:
|
||
_clear_lots(db, product, user)
|
||
cleared.add(product.id)
|
||
amount = _num(row.get("menge"))
|
||
if amount is None or amount <= 0:
|
||
continue # Zeile ohne Bestand: nur Stammdaten
|
||
quantity = _quantity_to_base(db, product, amount, row.get("menge_einheit", ""))
|
||
location = _get_or_create_location(db, row.get("lagerort"))
|
||
best_before, precision = _parse_best_before(row.get("mhd"))
|
||
lot = Lot(
|
||
product_id=product.id,
|
||
quantity=quantity,
|
||
best_before=best_before,
|
||
best_before_precision=precision,
|
||
location_id=location.id if location else None,
|
||
)
|
||
db.add(lot)
|
||
db.flush()
|
||
db.add(
|
||
Movement(
|
||
product_id=product.id,
|
||
lot_id=lot.id,
|
||
user_id=user.id,
|
||
type=MovementType.in_,
|
||
quantity=quantity,
|
||
unit_used=row.get("menge_einheit") or "base",
|
||
note="Import",
|
||
)
|
||
)
|
||
lots_added += 1
|
||
except Exception as exc: # eine fehlerhafte Zeile darf den Rest nicht stoppen
|
||
errors.append(f"Zeile {index}: {exc}")
|
||
|
||
return {
|
||
"format": "csv",
|
||
"mode": mode,
|
||
"products_created": len(created_products),
|
||
"products_cleared": len(cleared),
|
||
"lots_added": lots_added,
|
||
"errors": errors,
|
||
}
|
||
|
||
|
||
def _import_json(db: Session, content: bytes, user: User, mode: str) -> dict:
|
||
data = json.loads(content.decode("utf-8-sig", errors="replace"))
|
||
created_products: list[str] = []
|
||
lots_added = 0
|
||
units_created = 0
|
||
errors: list[str] = []
|
||
cleared: set[int] = set()
|
||
|
||
if mode == "replace_all":
|
||
_clear_all_lots(db, user)
|
||
|
||
for entry in data.get("units", []):
|
||
try:
|
||
if db.query(Unit).filter(Unit.name == entry["name"]).first():
|
||
continue
|
||
db.add(
|
||
Unit(
|
||
name=entry["name"],
|
||
kind=UnitKind(entry.get("kind", "count")),
|
||
factor=float(entry.get("factor", 1)),
|
||
is_builtin=False,
|
||
)
|
||
)
|
||
units_created += 1
|
||
except Exception as exc:
|
||
errors.append(f"Einheit {entry!r}: {exc}")
|
||
db.flush()
|
||
|
||
for entry in data.get("groups", []):
|
||
gruppe = _get_or_create_group(db, entry.get("name"))
|
||
if gruppe is None:
|
||
continue
|
||
# Gebinde und Einheit nur nachtragen, wenn die Gruppe noch nackt ist –
|
||
# der Import ueberschreibt grundsaetzlich nichts Vorhandenes.
|
||
if gruppe.package_size is None and entry.get("package_size"):
|
||
gruppe.package_size = float(entry["package_size"])
|
||
gruppe.package_label = entry.get("package_label")
|
||
gruppe.min_stock_in_packages = bool(entry.get("min_stock_in_packages"))
|
||
if gruppe.min_stock_unit_id is None and entry.get("min_stock_unit"):
|
||
einheit = db.query(Unit).filter(Unit.name == entry["min_stock_unit"]).first()
|
||
if einheit is not None:
|
||
gruppe.min_stock_unit_id = einheit.id
|
||
for entry in data.get("locations", []):
|
||
_get_or_create_location(db, entry.get("name"))
|
||
db.flush()
|
||
# Übergeordnete Lagerorte nachziehen
|
||
for entry in data.get("locations", []):
|
||
parent_name = entry.get("parent")
|
||
if not parent_name:
|
||
continue
|
||
child = db.query(Location).filter(Location.name == entry.get("name")).first()
|
||
parent = db.query(Location).filter(Location.name == parent_name).first()
|
||
if child and parent and child.parent_id is None and child.id != parent.id:
|
||
child.parent_id = parent.id
|
||
|
||
# Zweiter Durchgang für die Gruppen: Ober-/Untergruppen und Mindestbestände
|
||
# lassen sich erst setzen, wenn alle Gruppen und Lagerorte existieren.
|
||
for entry in data.get("groups", []):
|
||
gruppe = db.query(Group).filter(Group.name == entry.get("name")).first()
|
||
if gruppe is None:
|
||
continue
|
||
if not gruppe.parents:
|
||
for eltern_name in entry.get("parents") or []:
|
||
eltern = db.query(Group).filter(Group.name == eltern_name).first()
|
||
# Selbstkante und Ringe abweisen: eine beschädigte oder
|
||
# manipulierte Datei darf keinen einschleusen, der danach jede
|
||
# Auswertung im Kreis laufen liesse.
|
||
if eltern is None or eltern.id == gruppe.id:
|
||
continue
|
||
if eltern.id in gruppen_graph.nachfahren_ids(gruppe):
|
||
errors.append(f"Gruppe {gruppe.name!r}: {eltern_name!r} wäre ein Ring")
|
||
continue
|
||
gruppe.parents.append(eltern)
|
||
_min_stocks_einspielen(db, gruppe, entry.get("min_stocks"))
|
||
db.flush()
|
||
|
||
# Kategorien samt Verwaltungsart (Backup v2). Ältere Backups ohne diese Liste
|
||
# legen ihre Kategorien weiter über die Produktpfade an (Standard: food).
|
||
for entry in data.get("categories", []):
|
||
cat = _get_or_create_category(
|
||
db,
|
||
entry.get("path"),
|
||
tracking=entry.get("tracking") or CategoryTracking.food.value,
|
||
)
|
||
if cat is not None and entry.get("tracking"):
|
||
cat.tracking = entry["tracking"]
|
||
for entry in data.get("shops", []):
|
||
_get_or_create_shop(db, entry.get("name"), entry.get("website"))
|
||
db.flush()
|
||
for entry in data.get("field_definitions", []):
|
||
cat = _get_or_create_category(db, entry.get("category"))
|
||
label = (entry.get("label") or "").strip()
|
||
if cat is None or not label:
|
||
continue
|
||
if db.query(FieldDefinition).filter_by(category_id=cat.id, label=label).first():
|
||
continue
|
||
ftype = entry.get("field_type") or "text"
|
||
optionen = entry.get("options") or []
|
||
db.add(
|
||
FieldDefinition(
|
||
category_id=cat.id,
|
||
label=label,
|
||
key=slugify(label),
|
||
field_type=ftype,
|
||
unit=(entry.get("unit") or None),
|
||
options=json.dumps(optionen) if ftype == "select" and optionen else None,
|
||
required=bool(entry.get("required")),
|
||
position=int(entry.get("position") or 0),
|
||
)
|
||
)
|
||
db.flush()
|
||
|
||
for entry in data.get("products", []):
|
||
try:
|
||
row = {
|
||
"barcode": entry.get("barcode") or "",
|
||
"name": entry.get("name") or "",
|
||
"marke": entry.get("brand") or "",
|
||
# Ohne diesen Eintrag laeuft die Bildadresse ins Leere: Das
|
||
# Backup enthaelt sie, aber _get_or_create_product sieht nur
|
||
# diese Zeile - nicht den urspruenglichen JSON-Eintrag.
|
||
"image_url": entry.get("image_url") or "",
|
||
"einheit": entry.get("unit") or "Stück",
|
||
"packungsgroesse": entry.get("package_size") or "",
|
||
"gebinde": entry.get("package_label") or "",
|
||
"mhd_genauigkeit": entry.get("date_precision") or "",
|
||
"kategorie": entry.get("category") or "",
|
||
"gruppe": entry.get("group") or "",
|
||
"mindestbestand": entry.get("min_stock") if entry.get("min_stock") is not None else "",
|
||
}
|
||
product = _get_or_create_product(db, row, created_products)
|
||
# Mindestbestände je Ort (v3). Ältere Sicherungen haben nur den
|
||
# Gesamtwert, der oben schon über „mindestbestand" gesetzt wurde.
|
||
_min_stocks_einspielen(db, product, entry.get("min_stocks"))
|
||
# Zweiteinheit nur nachtragen, wenn der Artikel noch keine hat –
|
||
# der Import ueberschreibt grundsaetzlich nichts Vorhandenes.
|
||
if not product.secondary_base and entry.get("secondary_base"):
|
||
product.secondary_base = entry["secondary_base"]
|
||
product.secondary_count = entry.get("secondary_count")
|
||
product.secondary_amount = entry.get("secondary_amount")
|
||
# Gegenstands-Zusatzfelder – nur ergänzend, nichts überschreiben.
|
||
shop_name = (entry.get("shop") or "").strip()
|
||
if shop_name and product.shop_id is None:
|
||
shop = _get_or_create_shop(db, shop_name)
|
||
product.shop_id = shop.id if shop else None
|
||
if entry.get("product_url") and not product.product_url:
|
||
product.product_url = entry["product_url"]
|
||
feldwerte = entry.get("field_values") or {}
|
||
if feldwerte and product.category_id:
|
||
nach_label = {
|
||
fd.label: fd
|
||
for fd, _ in effective_field_definitions(db, product.category_id)
|
||
}
|
||
for label, value in feldwerte.items():
|
||
fd = nach_label.get(label)
|
||
if fd is None:
|
||
continue
|
||
try:
|
||
apply_field_values(db, product, {fd.id: value})
|
||
except FieldError:
|
||
pass
|
||
if mode == "replace_listed" and product.id not in cleared:
|
||
_clear_lots(db, product, user)
|
||
cleared.add(product.id)
|
||
for lot_entry in entry.get("lots", []):
|
||
quantity = float(lot_entry.get("quantity") or 0)
|
||
if quantity <= 0:
|
||
continue
|
||
location = _get_or_create_location(db, lot_entry.get("location"))
|
||
best_before, precision = _parse_best_before(
|
||
lot_entry.get("best_before"), lot_entry.get("best_before_precision")
|
||
)
|
||
lot = Lot(
|
||
product_id=product.id,
|
||
quantity=quantity,
|
||
best_before=best_before,
|
||
best_before_precision=precision,
|
||
location_id=location.id if location else None,
|
||
opened_on=_parse_tag(lot_entry.get("opened_on")),
|
||
consume_by=_parse_tag(lot_entry.get("consume_by")),
|
||
opened_quantity=_zahl_oder_none(lot_entry.get("opened_quantity")),
|
||
)
|
||
db.add(lot)
|
||
db.flush()
|
||
db.add(
|
||
Movement(
|
||
product_id=product.id,
|
||
lot_id=lot.id,
|
||
user_id=user.id,
|
||
type=MovementType.in_,
|
||
quantity=quantity,
|
||
unit_used="base",
|
||
note="Import",
|
||
)
|
||
)
|
||
lots_added += 1
|
||
except Exception as exc:
|
||
errors.append(f"Produkt {entry.get('name')!r}: {exc}")
|
||
|
||
vorlagen_created = _vorlagen_importieren(
|
||
db, data.get("etiketten_vorlagen", []), errors
|
||
)
|
||
|
||
return {
|
||
"format": "json",
|
||
"mode": mode,
|
||
"etiketten_vorlagen_created": vorlagen_created,
|
||
"units_created": units_created,
|
||
"products_created": len(created_products),
|
||
"products_cleared": len(cleared),
|
||
"lots_added": lots_added,
|
||
"errors": errors,
|
||
}
|
||
|
||
|
||
@router.post("/import/stock")
|
||
def import_stock(
|
||
mode: str = "add",
|
||
file: UploadFile = File(...),
|
||
db: Session = Depends(get_db),
|
||
user: User = Depends(require_admin),
|
||
) -> dict:
|
||
"""Importiert CSV oder JSON.
|
||
|
||
mode:
|
||
add – nur ergänzen (Standard, löscht nie etwas)
|
||
replace_listed – Bestand der in der Datei genannten Produkte ersetzen
|
||
replace_all – alle Bestände vorher leeren (vollständige Wiederherstellung)
|
||
"""
|
||
if mode not in IMPORT_MODES:
|
||
raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Unbekannter Modus: {mode}")
|
||
|
||
content = file.file.read()
|
||
if not content:
|
||
raise HTTPException(status.HTTP_400_BAD_REQUEST, "Datei ist leer")
|
||
|
||
filename = (file.filename or "").lower()
|
||
stripped = content.lstrip()[:1]
|
||
is_json = filename.endswith(".json") or stripped in (b"{", b"[")
|
||
|
||
try:
|
||
result = (
|
||
_import_json(db, content, user, mode)
|
||
if is_json
|
||
else _import_csv(db, content, user, mode)
|
||
)
|
||
except json.JSONDecodeError as exc:
|
||
db.rollback()
|
||
raise HTTPException(status.HTTP_400_BAD_REQUEST, f"JSON nicht lesbar: {exc}") from exc
|
||
except Exception as exc:
|
||
db.rollback()
|
||
raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Import fehlgeschlagen: {exc}") from exc
|
||
|
||
db.commit()
|
||
return result
|