044f63446f
- Web: Kategorien/Unterkategorien und eigene Felder direkt im Artikelformular
anlegen (ohne Umweg über die Kategorien-Seite).
- Artikelfoto per Kamera oder Galerie hochladen – Backend-Endpunkt
(PUT/DELETE /products/{id}/image), Web (Foto machen / Galerie) und iOS
(Kamera + PhotosPicker, inline in der Produktansicht angezeigt).
- CSV-Import: neue Spalte "art" (Gegenstand/Lebensmittel) steuert den Modus
automatisch angelegter Kategorien; Export schreibt sie mit.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
776 lines
28 KiB
Python
776 lines
28 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 csv
|
||
import io
|
||
import json
|
||
from datetime import date, datetime, timezone
|
||
|
||
from fastapi import APIRouter, 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.dates import MONTH, clean_precision, normalize_best_before
|
||
from ..services.group_codes import sync as sync_group_code
|
||
from ..models import (
|
||
BaseUnit,
|
||
Category,
|
||
CategoryTracking,
|
||
DatePrecision,
|
||
FieldDefinition,
|
||
Group,
|
||
Location,
|
||
Lot,
|
||
Movement,
|
||
MovementType,
|
||
Product,
|
||
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),
|
||
product.min_stock if product.min_stock is not None else "",
|
||
]
|
||
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 = {
|
||
"version": 2,
|
||
"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": g.min_stock,
|
||
"min_stock_unit": g.min_stock_unit.name if g.min_stock_unit else None,
|
||
}
|
||
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": p.min_stock,
|
||
"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),
|
||
# 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),
|
||
}
|
||
for lot in lots
|
||
],
|
||
}
|
||
)
|
||
|
||
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"'
|
||
},
|
||
)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# Import (nur additiv)
|
||
# --------------------------------------------------------------------------
|
||
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 _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 _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,
|
||
min_stock=_num(row.get("mindestbestand")),
|
||
source="import",
|
||
)
|
||
db.add(product)
|
||
db.flush()
|
||
# 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", []):
|
||
_get_or_create_group(db, entry.get("name"))
|
||
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
|
||
|
||
# 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)
|
||
# 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,
|
||
)
|
||
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}")
|
||
|
||
return {
|
||
"format": "json",
|
||
"mode": mode,
|
||
"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
|