Etiketten-Vorlagen: gespeicherte Exporte mit frei waehlbaren Spalten
Der Etikettenexport konnte genau eine Sache, und zwar unveraenderlich: sechs
feste Spalten ueber Einzelstuecke. Was fehlte, war genau das, was ein Etikett
brauchbar macht - die eigenen Felder. Eine Kabellaenge oder die Kapazitaet einer
Powerbank steht laengst am Artikel, kam aber in keiner CSV vor. Und jede
Zusammenstellung war beim naechsten Mal wieder weg: Kategorien im React-State,
Trennzeichen im localStorage, Spalten hartkodiert.
Jetzt gibt es benannte Vorlagen mit eigener Spaltenwahl und Reihenfolge, fuer
Einzelstuecke, Artikel und Lagerorte. Eine Tabelle nach dem Vorbild von
DashboardLayout: benannt, Nutzlast als JSON-Text. Ohne user_id - eine Vorlage
gehoert allen, nicht einem Konto.
Spaltenschluessel fuer eigene Felder ist die ID, nicht der key: FieldDefinition.key
ist nur JE KATEGORIE eindeutig, zwei Kategorien duerfen beide "kapazitaet" fuehren.
Die CSV entsteht jetzt im Server statt im Browser. Mit waehlbaren Spalten,
eigenen Feldern und drei Bezugsarten gehoert das an eine Stelle - und zwar
dorthin, wo es sich pruefen laesst. Formel-Injektionsschutz, BOM und CRLF sind
aus der Browser-Fassung mitgenommen. Der Grund, aus dem der QR-Inhalt bisher im
Browser entstand ("nur die Oberflaeche kennt die oeffentliche Adresse"), gilt
allein fuer Lagerorte; dafuer reicht sie ihre Herkunft als Parameter durch.
Die beiden alten Abschnitte weichen den Vorlagen. Beim ersten Start entstehen
zwei mitgelieferte, die genau die bisherigen Spalten fuehren - wer nichts
einstellt, bekommt dasselbe Ergebnis wie vorher.
Dazu laesst sich eine P-touch-Datei hinterlegen. Geprueft wird ueber Endung und
ZIP-Inhalt (label.xml), nicht ueber den Content-Type: Fuer .lbx meldet der
Browser meist application/octet-stream. Eine .lbx merkt sich den Pfad zu ihrer
Datenquelle - auf einem fremden Rechner muss die CSV in P-touch einmal neu
verknuepft werden. Das laesst sich von hier aus nicht loesen; die Datei
dabeizuhaben erspart trotzdem das Suchen.
Sicherung auf Version 4: Vorlagen kommen mit, Kategorien als Pfad und eigene
Felder als Label, weil IDs zwischen zwei Instanzen nicht dieselben sind. Die
.lbx als Base64 - das erste Binaerformat in dieser Sicherung, vertretbar bei
wenigen Kilobyte und einer Vorlage, die ohne ihre Datei nur eine halbe waere.
23 neue Tests (343 gesamt), dazu im Browser durchgespielt: Kabel-Kategorie mit
eigenem Feld, Spalte hinzugefuegt, CSV mit "Länge (m)" und dem Wert, .lbx
hoch- und wieder heruntergeladen (Pruefsumme identisch), PNG und Nicht-ZIP
sauber abgelehnt.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Scarriffle
co-authored by
Claude Opus 5
Claude Opus 5
parent
7372590171
commit
c741df5963
@@ -12,6 +12,7 @@ from .routers import (
|
||||
branding,
|
||||
categories,
|
||||
dashboard,
|
||||
etiketten,
|
||||
field_definitions,
|
||||
groups,
|
||||
items,
|
||||
@@ -31,6 +32,7 @@ from .routers import (
|
||||
)
|
||||
from .seed import (
|
||||
ensure_builtin_categories,
|
||||
ensure_builtin_etiketten_vorlagen,
|
||||
ensure_builtin_package_types,
|
||||
ensure_builtin_units,
|
||||
ensure_example_object_categories,
|
||||
@@ -255,6 +257,7 @@ async def lifespan(app: FastAPI):
|
||||
try:
|
||||
ensure_builtin_units(db)
|
||||
ensure_builtin_package_types(db)
|
||||
ensure_builtin_etiketten_vorlagen(db)
|
||||
ensure_builtin_categories(db)
|
||||
ensure_example_object_categories(db)
|
||||
ensure_first_admin(db)
|
||||
@@ -291,6 +294,7 @@ def health() -> dict:
|
||||
|
||||
|
||||
app.include_router(auth.router)
|
||||
app.include_router(etiketten.router)
|
||||
app.include_router(oidc.router)
|
||||
app.include_router(users.router)
|
||||
app.include_router(products.router)
|
||||
|
||||
@@ -637,6 +637,43 @@ class DashboardLayout(Base):
|
||||
)
|
||||
|
||||
|
||||
class EtikettenVorlage(Base):
|
||||
"""Eine gespeicherte Zusammenstellung fuer den Etikettenexport.
|
||||
|
||||
Benannt, damit man sie wieder aufrufen kann („Kabelexport"), und mit frei
|
||||
gewaehlten Spalten - erst dadurch kommen eigene Felder wie eine Kabellaenge
|
||||
oder die Kapazitaet einer Powerbank ueberhaupt aufs Etikett.
|
||||
|
||||
Vorbild ist :class:`DashboardLayout`: eigene Tabelle statt ``Setting``
|
||||
(dessen ``value`` fasst 255 Zeichen und haelt genau einen Wert je Schluessel),
|
||||
Nutzlast als JSON-Text. Anders als dort ohne ``user_id`` - eine Vorlage
|
||||
gehoert allen, nicht einem Konto.
|
||||
"""
|
||||
|
||||
__tablename__ = "etiketten_vorlagen"
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True)
|
||||
name: Mapped[str] = mapped_column(String(80), unique=True, nullable=False)
|
||||
# Worauf sich die Vorlage bezieht: items | products | locations.
|
||||
# Als kurzer String und nicht als DB-Enum, wie DatePrecision und
|
||||
# CategoryTracking - so laesst sich eine weitere Art spaeter ohne neuen
|
||||
# Postgres-Typ nachziehen.
|
||||
art: Mapped[str] = mapped_column(String(16), nullable=False, default="items")
|
||||
position: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
|
||||
# Spaltenwahl, Reihenfolge, Filter und Trennzeichen als JSON-Text.
|
||||
konfiguration: Mapped[str] = mapped_column(Text, nullable=False, default="{}")
|
||||
# Hinterlegte P-touch-Vorlage (.lbx). Hoechstens eine je Eintrag, deshalb
|
||||
# Spalten derselben Zeile statt einer zweiten Tabelle.
|
||||
datei_name: Mapped[str | None] = mapped_column(String(255), nullable=True)
|
||||
# deferred: Sonst zoege jede Liste der Vorlagen die Binaerdaten mit - genau
|
||||
# das vermeidet ItemDocument.data aus demselben Grund.
|
||||
datei: Mapped[bytes | None] = mapped_column(LargeBinary, nullable=True, deferred=True)
|
||||
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
|
||||
updated_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), default=_now, onupdate=_now
|
||||
)
|
||||
|
||||
|
||||
class Setting(Base):
|
||||
__tablename__ = "settings"
|
||||
|
||||
|
||||
@@ -0,0 +1,300 @@
|
||||
"""Etiketten-Vorlagen: gespeicherte Zusammenstellungen fuer den Export.
|
||||
|
||||
Eine Vorlage haelt fest, worauf sie sich bezieht (Einzelstuecke, Artikel oder
|
||||
Lagerorte), welche Spalten in welcher Reihenfolge exportiert werden und wie
|
||||
gefiltert wird. Dazu kann eine P-touch-Datei (.lbx) hinterlegt werden, damit
|
||||
man sie auf einem fremden Rechner nicht suchen muss.
|
||||
|
||||
Vorlagen gehoeren allen: Anlegen und Aendern bleibt Administratoren
|
||||
vorbehalten, Exportieren darf jeder Angemeldete – es ist ein Lesevorgang.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import json
|
||||
import re
|
||||
import zipfile
|
||||
|
||||
from fastapi import APIRouter, Depends, File, HTTPException, Response, UploadFile, status
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from ..database import get_db
|
||||
from ..deps import get_current_user, require_admin
|
||||
from ..models import EtikettenVorlage, User
|
||||
from ..schemas import (
|
||||
EtikettenVorlageCreate,
|
||||
EtikettenVorlageOut,
|
||||
EtikettenVorlageUpdate,
|
||||
SpalteOut,
|
||||
)
|
||||
from ..services.etiketten import (
|
||||
ARTEN,
|
||||
csv_erzeugen,
|
||||
konfiguration_lesen,
|
||||
verfuegbare_spalten,
|
||||
)
|
||||
|
||||
router = APIRouter(prefix="/etiketten", tags=["etiketten"])
|
||||
|
||||
# Eine .lbx ist ein ZIP von wenigen Kilobyte. Die Grenze ist grosszuegig und
|
||||
# bleibt weit unter der nginx-Grenze (client_max_body_size 12m).
|
||||
LBX_MAX_BYTES = 2 * 1024 * 1024
|
||||
|
||||
|
||||
def _vorlage_oder_404(db: Session, vorlage_id: int) -> EtikettenVorlage:
|
||||
vorlage = db.get(EtikettenVorlage, vorlage_id)
|
||||
if vorlage is None:
|
||||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Vorlage nicht gefunden")
|
||||
return vorlage
|
||||
|
||||
|
||||
def _safe_filename(name: str | None) -> str:
|
||||
"""Steuerzeichen und Anfuehrungszeichen raus – sonst Header-Injektion."""
|
||||
cleaned = re.sub(r'[\r\n"]', "", (name or "").strip())
|
||||
return cleaned[:255] or "vorlage.lbx"
|
||||
|
||||
|
||||
def _pruefe_lbx(dateiname: str, daten: bytes) -> None:
|
||||
"""Ist das plausibel eine P-touch-Vorlage?
|
||||
|
||||
Eine .lbx ist ein ZIP mit label.xml (und prop.xml) darin. Geprueft wird
|
||||
genau das – nicht der vom Browser gemeldete Content-Type, denn fuer .lbx
|
||||
meldet der meist application/octet-stream oder gar nichts.
|
||||
"""
|
||||
if not daten:
|
||||
raise HTTPException(status.HTTP_400_BAD_REQUEST, "Die Datei ist leer.")
|
||||
if len(daten) > LBX_MAX_BYTES:
|
||||
raise HTTPException(
|
||||
status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
|
||||
f"Datei zu groß (max. {LBX_MAX_BYTES // (1024 * 1024)} MB).",
|
||||
)
|
||||
if not dateiname.lower().endswith(".lbx"):
|
||||
raise HTTPException(
|
||||
status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
|
||||
"Nur .lbx-Dateien aus dem P-touch Editor.",
|
||||
)
|
||||
try:
|
||||
with zipfile.ZipFile(io.BytesIO(daten)) as archiv:
|
||||
namen = {n.lower() for n in archiv.namelist()}
|
||||
except zipfile.BadZipFile:
|
||||
raise HTTPException(
|
||||
status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
|
||||
"Das ist keine P-touch-Vorlage (die Datei ist kein ZIP-Archiv).",
|
||||
) from None
|
||||
if "label.xml" not in namen:
|
||||
raise HTTPException(
|
||||
status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
|
||||
"Das ist keine P-touch-Vorlage (label.xml fehlt).",
|
||||
)
|
||||
|
||||
|
||||
def _als_out(vorlage: EtikettenVorlage) -> EtikettenVorlageOut:
|
||||
konf = konfiguration_lesen(vorlage)
|
||||
return EtikettenVorlageOut(
|
||||
id=vorlage.id,
|
||||
name=vorlage.name,
|
||||
art=vorlage.art,
|
||||
position=vorlage.position,
|
||||
spalten=konf["spalten"],
|
||||
kategorie_ids=konf["kategorie_ids"],
|
||||
trennzeichen=konf["trennzeichen"],
|
||||
datei_name=vorlage.datei_name,
|
||||
updated_at=vorlage.updated_at,
|
||||
)
|
||||
|
||||
|
||||
def _konfiguration(spalten: list[str], kategorie_ids: list[int], trennzeichen: str) -> str:
|
||||
return json.dumps(
|
||||
{
|
||||
"spalten": spalten,
|
||||
"kategorie_ids": kategorie_ids,
|
||||
"trennzeichen": trennzeichen if trennzeichen in (",", ";", "\t") else ",",
|
||||
},
|
||||
ensure_ascii=False,
|
||||
)
|
||||
|
||||
|
||||
@router.get("/spalten", response_model=list[SpalteOut])
|
||||
def spalten(
|
||||
art: str = "items",
|
||||
kategorie_ids: str | None = None,
|
||||
db: Session = Depends(get_db),
|
||||
_: User = Depends(get_current_user),
|
||||
) -> list[SpalteOut]:
|
||||
"""Welche Spalten fuer diese Bezugsart und Kategorien zur Wahl stehen.
|
||||
|
||||
Die eigenen Felder haengen an den Kategorien – ohne diesen Endpunkt
|
||||
wuesste die Oberflaeche nicht, dass es „Kapazität (mAh)" ueberhaupt gibt.
|
||||
"""
|
||||
if art not in ARTEN:
|
||||
raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Unbekannte Bezugsart „{art}“")
|
||||
ids = [int(t) for t in (kategorie_ids or "").split(",") if t.strip().isdigit()]
|
||||
return [SpalteOut(**s) for s in verfuegbare_spalten(db, art, ids)]
|
||||
|
||||
|
||||
@router.get("/vorlagen", response_model=list[EtikettenVorlageOut])
|
||||
def vorlagen_auflisten(
|
||||
db: Session = Depends(get_db),
|
||||
_: User = Depends(get_current_user),
|
||||
) -> list[EtikettenVorlageOut]:
|
||||
rows = (
|
||||
db.query(EtikettenVorlage)
|
||||
.order_by(EtikettenVorlage.position, EtikettenVorlage.id)
|
||||
.all()
|
||||
)
|
||||
return [_als_out(v) for v in rows]
|
||||
|
||||
|
||||
@router.post(
|
||||
"/vorlagen", response_model=EtikettenVorlageOut, status_code=status.HTTP_201_CREATED
|
||||
)
|
||||
def vorlage_anlegen(
|
||||
payload: EtikettenVorlageCreate,
|
||||
db: Session = Depends(get_db),
|
||||
_: User = Depends(require_admin),
|
||||
) -> EtikettenVorlageOut:
|
||||
name = payload.name.strip()
|
||||
if db.query(EtikettenVorlage).filter(EtikettenVorlage.name == name).first():
|
||||
raise HTTPException(status.HTTP_409_CONFLICT, "Es gibt schon eine Vorlage mit diesem Namen")
|
||||
hoechste = (
|
||||
db.query(EtikettenVorlage).order_by(EtikettenVorlage.position.desc()).first()
|
||||
)
|
||||
vorlage = EtikettenVorlage(
|
||||
name=name,
|
||||
art=payload.art,
|
||||
position=(hoechste.position + 1) if hoechste else 0,
|
||||
konfiguration=_konfiguration(
|
||||
payload.spalten, payload.kategorie_ids, payload.trennzeichen
|
||||
),
|
||||
)
|
||||
db.add(vorlage)
|
||||
db.commit()
|
||||
db.refresh(vorlage)
|
||||
return _als_out(vorlage)
|
||||
|
||||
|
||||
@router.patch("/vorlagen/{vorlage_id}", response_model=EtikettenVorlageOut)
|
||||
def vorlage_aendern(
|
||||
vorlage_id: int,
|
||||
payload: EtikettenVorlageUpdate,
|
||||
db: Session = Depends(get_db),
|
||||
_: User = Depends(require_admin),
|
||||
) -> EtikettenVorlageOut:
|
||||
vorlage = _vorlage_oder_404(db, vorlage_id)
|
||||
data = payload.model_dump(exclude_unset=True)
|
||||
|
||||
if "name" in data:
|
||||
neuer = data["name"].strip()
|
||||
doppelt = (
|
||||
db.query(EtikettenVorlage)
|
||||
.filter(EtikettenVorlage.name == neuer, EtikettenVorlage.id != vorlage.id)
|
||||
.first()
|
||||
)
|
||||
if doppelt:
|
||||
raise HTTPException(
|
||||
status.HTTP_409_CONFLICT, "Es gibt schon eine Vorlage mit diesem Namen"
|
||||
)
|
||||
vorlage.name = neuer
|
||||
if "art" in data:
|
||||
vorlage.art = data["art"]
|
||||
if "position" in data:
|
||||
vorlage.position = data["position"]
|
||||
|
||||
# Die drei Konfigurationswerte liegen zusammen in einem JSON-Text; wer nur
|
||||
# einen schickt, darf die anderen nicht verlieren.
|
||||
if {"spalten", "kategorie_ids", "trennzeichen"} & data.keys():
|
||||
alt = konfiguration_lesen(vorlage)
|
||||
vorlage.konfiguration = _konfiguration(
|
||||
data.get("spalten", alt["spalten"]),
|
||||
data.get("kategorie_ids", alt["kategorie_ids"]),
|
||||
data.get("trennzeichen", alt["trennzeichen"]),
|
||||
)
|
||||
|
||||
db.commit()
|
||||
db.refresh(vorlage)
|
||||
return _als_out(vorlage)
|
||||
|
||||
|
||||
@router.delete("/vorlagen/{vorlage_id}")
|
||||
def vorlage_loeschen(
|
||||
vorlage_id: int,
|
||||
db: Session = Depends(get_db),
|
||||
_: User = Depends(require_admin),
|
||||
) -> Response:
|
||||
db.delete(_vorlage_oder_404(db, vorlage_id))
|
||||
db.commit()
|
||||
return Response(status_code=status.HTTP_204_NO_CONTENT)
|
||||
|
||||
|
||||
@router.get("/vorlagen/{vorlage_id}/csv")
|
||||
def vorlage_csv(
|
||||
vorlage_id: int,
|
||||
basis_url: str = "",
|
||||
db: Session = Depends(get_db),
|
||||
_: User = Depends(get_current_user),
|
||||
) -> Response:
|
||||
"""Die CSV zu einer Vorlage.
|
||||
|
||||
``basis_url`` braucht nur der Lagerort-QR (…/l/<id>); Einzelstuecke tragen
|
||||
vorrania://i/<UID> ohne jede Adresse. Die Oberflaeche reicht ihre eigene
|
||||
Herkunft durch, weil nur sie weiss, unter welcher Adresse sie erreichbar ist.
|
||||
"""
|
||||
vorlage = _vorlage_oder_404(db, vorlage_id)
|
||||
inhalt = csv_erzeugen(db, vorlage, basis_url=basis_url)
|
||||
dateiname = _safe_filename(f"{vorlage.name}.csv".replace("/", "-"))
|
||||
return Response(
|
||||
content=inhalt.encode("utf-8"),
|
||||
media_type="text/csv; charset=utf-8",
|
||||
headers={"Content-Disposition": f'attachment; filename="{dateiname}"'},
|
||||
)
|
||||
|
||||
|
||||
@router.post("/vorlagen/{vorlage_id}/datei", response_model=EtikettenVorlageOut)
|
||||
async def datei_hochladen(
|
||||
vorlage_id: int,
|
||||
file: UploadFile = File(...),
|
||||
db: Session = Depends(get_db),
|
||||
_: User = Depends(require_admin),
|
||||
) -> EtikettenVorlageOut:
|
||||
"""Die P-touch-Vorlage (.lbx) hinterlegen."""
|
||||
vorlage = _vorlage_oder_404(db, vorlage_id)
|
||||
daten = await file.read()
|
||||
name = _safe_filename(file.filename)
|
||||
_pruefe_lbx(name, daten)
|
||||
vorlage.datei_name = name
|
||||
vorlage.datei = daten
|
||||
db.commit()
|
||||
db.refresh(vorlage)
|
||||
return _als_out(vorlage)
|
||||
|
||||
|
||||
@router.get("/vorlagen/{vorlage_id}/datei")
|
||||
def datei_herunterladen(
|
||||
vorlage_id: int,
|
||||
db: Session = Depends(get_db),
|
||||
_: User = Depends(get_current_user),
|
||||
) -> Response:
|
||||
vorlage = _vorlage_oder_404(db, vorlage_id)
|
||||
if vorlage.datei is None:
|
||||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Zu dieser Vorlage ist keine Datei hinterlegt")
|
||||
return Response(
|
||||
content=vorlage.datei,
|
||||
# attachment und nicht inline: Die Datei soll gespeichert werden, der
|
||||
# Browser kann mit ihr ohnehin nichts anfangen.
|
||||
media_type="application/octet-stream",
|
||||
headers={"Content-Disposition": f'attachment; filename="{vorlage.datei_name}"'},
|
||||
)
|
||||
|
||||
|
||||
@router.delete("/vorlagen/{vorlage_id}/datei")
|
||||
def datei_entfernen(
|
||||
vorlage_id: int,
|
||||
db: Session = Depends(get_db),
|
||||
_: User = Depends(require_admin),
|
||||
) -> Response:
|
||||
vorlage = _vorlage_oder_404(db, vorlage_id)
|
||||
vorlage.datei = None
|
||||
vorlage.datei_name = None
|
||||
db.commit()
|
||||
return Response(status_code=status.HTTP_204_NO_CONTENT)
|
||||
@@ -6,6 +6,7 @@ werden ergänzt – es wird nie etwas gelöscht oder überschrieben.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import csv
|
||||
import io
|
||||
import json
|
||||
@@ -22,11 +23,13 @@ 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,
|
||||
@@ -166,10 +169,11 @@ def export_backup_json(
|
||||
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. v2-Sicherungen bleiben
|
||||
# (Ort null = „Ueberall") in Basiseinheiten. Aeltere Sicherungen bleiben
|
||||
# lesbar – ihnen fehlen diese Listen einfach.
|
||||
"version": 3,
|
||||
"version": 4,
|
||||
"exported_at": datetime.now(timezone.utc).isoformat(),
|
||||
"exported_at_local": datetime.now().isoformat(timespec="seconds"),
|
||||
"units": [
|
||||
@@ -262,6 +266,8 @@ def export_backup_json(
|
||||
}
|
||||
)
|
||||
|
||||
data["etiketten_vorlagen"] = _vorlagen_exportieren(db)
|
||||
|
||||
return Response(
|
||||
content=json.dumps(data, ensure_ascii=False, indent=2),
|
||||
media_type="application/json; charset=utf-8",
|
||||
@@ -435,6 +441,106 @@ def _get_or_create_location(db: Session, name: str | None) -> Location | None:
|
||||
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.
|
||||
|
||||
@@ -901,9 +1007,14 @@ def _import_json(db: Session, content: bytes, user: User, mode: str) -> dict:
|
||||
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),
|
||||
|
||||
@@ -878,6 +878,49 @@ class ItemOut(BaseModel):
|
||||
gehoert_zu: ItemKurz | None = None
|
||||
|
||||
|
||||
# ---- Etiketten-Vorlagen ----
|
||||
class SpalteOut(BaseModel):
|
||||
"""Eine waehlbare Spalte – fest oder ein eigenes Feld."""
|
||||
|
||||
schluessel: str
|
||||
titel: str
|
||||
|
||||
|
||||
class EtikettenVorlageOut(BaseModel):
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
id: int
|
||||
name: str
|
||||
art: str
|
||||
position: int
|
||||
# Aus dem JSON-Text der Konfiguration geholt, damit die Oberflaeche nicht
|
||||
# selbst parsen muss.
|
||||
spalten: list[str] = []
|
||||
kategorie_ids: list[int] = []
|
||||
trennzeichen: str = ","
|
||||
# Nur der Name; die Binaerdaten gehen ueber einen eigenen Endpunkt.
|
||||
datei_name: str | None = None
|
||||
updated_at: datetime
|
||||
|
||||
|
||||
class EtikettenVorlageCreate(BaseModel):
|
||||
name: str = Field(min_length=1, max_length=80)
|
||||
art: str = Field(default="items", pattern="^(items|products|locations)$")
|
||||
spalten: list[str] = []
|
||||
kategorie_ids: list[int] = []
|
||||
trennzeichen: str = ","
|
||||
|
||||
|
||||
class EtikettenVorlageUpdate(BaseModel):
|
||||
"""Nur mitgeschickte Felder zaehlen."""
|
||||
|
||||
name: str | None = Field(default=None, min_length=1, max_length=80)
|
||||
art: str | None = Field(default=None, pattern="^(items|products|locations)$")
|
||||
spalten: list[str] | None = None
|
||||
kategorie_ids: list[int] | None = None
|
||||
trennzeichen: str | None = None
|
||||
position: int | None = None
|
||||
|
||||
|
||||
# ---- Views ----
|
||||
class ShoppingNeed(BaseModel):
|
||||
"""Fehlmenge als Gebinde-Leitangabe – bei zaehlbaren Packungen (Glas/Dose)
|
||||
|
||||
+41
-1
@@ -5,7 +5,17 @@ import json
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from .config import get_settings
|
||||
from .models import Category, CategoryTracking, FieldDefinition, PackageType, Role, Unit, UnitKind, User
|
||||
from .models import (
|
||||
Category,
|
||||
CategoryTracking,
|
||||
EtikettenVorlage,
|
||||
FieldDefinition,
|
||||
PackageType,
|
||||
Role,
|
||||
Unit,
|
||||
UnitKind,
|
||||
User,
|
||||
)
|
||||
from .security import hash_password
|
||||
from .services.fields import slugify
|
||||
|
||||
@@ -197,3 +207,33 @@ def ensure_first_admin(db: Session) -> None:
|
||||
)
|
||||
db.add(admin)
|
||||
db.commit()
|
||||
|
||||
|
||||
# Die zwei Vorlagen, die den frueheren festen Etikettenexport nachbilden: Wer
|
||||
# nichts einstellt, bekommt genau dieselben Spalten wie bisher.
|
||||
BUILTIN_ETIKETTEN_VORLAGEN = [
|
||||
("Einzelstücke", "items",
|
||||
["qr", "uid", "produkt", "marke", "kategorie", "lagerort"]),
|
||||
("Lagerorte", "locations", ["qr", "id", "name", "pfad"]),
|
||||
]
|
||||
|
||||
|
||||
def ensure_builtin_etiketten_vorlagen(db: Session) -> None:
|
||||
"""Legt die mitgelieferten Etiketten-Vorlagen an – einmalig.
|
||||
|
||||
Nur wenn noch GAR KEINE Vorlage existiert: Wer seine eigenen gebaut und die
|
||||
mitgelieferten geloescht hat, soll sie nicht bei jedem Start zurueckbekommen.
|
||||
"""
|
||||
if db.query(EtikettenVorlage).first() is not None:
|
||||
return
|
||||
for position, (name, art, spalten) in enumerate(BUILTIN_ETIKETTEN_VORLAGEN):
|
||||
db.add(EtikettenVorlage(
|
||||
name=name,
|
||||
art=art,
|
||||
position=position,
|
||||
konfiguration=json.dumps(
|
||||
{"spalten": spalten, "kategorie_ids": [], "trennzeichen": ","},
|
||||
ensure_ascii=False,
|
||||
),
|
||||
))
|
||||
db.commit()
|
||||
|
||||
@@ -0,0 +1,351 @@
|
||||
"""Etikettenexport: welche Spalten es gibt und wie daraus eine CSV wird.
|
||||
|
||||
Frueher entstand die CSV in der Weboberflaeche, mit sechs festen Spalten. Mit
|
||||
frei waehlbaren Spalten, eigenen Feldern und drei Bezugsarten gehoert das an
|
||||
eine Stelle – und zwar hierher, wo es sich pruefen laesst: Fuer die Oberflaeche
|
||||
gibt es keine Tests, fuer das Backend ueber dreihundert.
|
||||
|
||||
Alles hier ist bewusst frei von HTTP: Eine Vorlage, eine Datenbank-Sitzung und
|
||||
eine Basis-Adresse hinein, fertige CSV heraus.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import io
|
||||
import json
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Callable
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from ..models import (
|
||||
Category,
|
||||
EtikettenVorlage,
|
||||
FieldDefinition,
|
||||
Item,
|
||||
Location,
|
||||
Product,
|
||||
ProductFieldValue,
|
||||
)
|
||||
|
||||
# Bezugsarten einer Vorlage.
|
||||
ARTEN = ("items", "products", "locations")
|
||||
|
||||
# Zellen, die damit beginnen, deutet Excel als Formel. Ein vorangestelltes
|
||||
# Apostroph macht sie zu reinem Text. Uebernommen aus der fruaeheren
|
||||
# Browser-Fassung (web/src/pages/Transfer.jsx).
|
||||
_FORMEL_START = re.compile(r"^[=+\-@\t\r]")
|
||||
|
||||
# Praefix eines selbst definierten Feldes. Die ID und nicht der key: FieldDefinition.key
|
||||
# ist nur JE KATEGORIE eindeutig, zwei Kategorien duerfen beide "kapazitaet" fuehren.
|
||||
FELD_PRAEFIX = "feld:"
|
||||
|
||||
|
||||
@dataclass
|
||||
class Kontext:
|
||||
"""Was die Spaltenfunktionen ausser der Zeile selbst brauchen."""
|
||||
|
||||
db: Session
|
||||
# Oeffentliche Adresse der Oberflaeche – nur fuer Lagerort-QR noetig
|
||||
# (…/l/<id>). Einzelstuecke tragen vorrania://i/<UID> ohne jede Adresse.
|
||||
basis_url: str = ""
|
||||
# Werte der eigenen Felder, vorab gesammelt: {product_id: {fd_id: wert}}.
|
||||
feldwerte: dict[int, dict[int, str | None]] | None = None
|
||||
|
||||
|
||||
def _kategorie_pfad(db: Session, category: Category | None) -> str:
|
||||
"""„Elektronik > Powerbanks" in einer Spalte."""
|
||||
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 _lagerort_pfad(db: Session, ort: Location | None) -> str:
|
||||
if ort is None:
|
||||
return ""
|
||||
teile = [ort.name]
|
||||
aktuell, tiefe = ort, 0
|
||||
while aktuell.parent_id is not None and tiefe < 10:
|
||||
aktuell = db.get(Location, aktuell.parent_id)
|
||||
if aktuell is None:
|
||||
break
|
||||
teile.append(aktuell.name)
|
||||
tiefe += 1
|
||||
return " > ".join(reversed(teile))
|
||||
|
||||
|
||||
# Signatur wie alle Spaltenfunktionen (Zeile, Kontext) – auch wenn der
|
||||
# Kontext hier nicht gebraucht wird.
|
||||
def _preis(item: Item, _kontext: "Kontext") -> str:
|
||||
if item.price_cents is None:
|
||||
return ""
|
||||
return f"{item.price_cents / 100:.2f}"
|
||||
|
||||
|
||||
def _offener_verleih(item: Item):
|
||||
for loan in item.loans:
|
||||
if loan.returned_on is None:
|
||||
return loan
|
||||
return None
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Spalten-Registry je Bezugsart: Schluessel -> (Titel, Funktion)
|
||||
# --------------------------------------------------------------------------
|
||||
SPALTEN: dict[str, dict[str, tuple[str, Callable[[Any, Kontext], str]]]] = {
|
||||
"items": {
|
||||
"qr": ("QR-Inhalt", lambda i, k: f"vorrania://i/{i.uid}"),
|
||||
"uid": ("UID", lambda i, k: i.uid),
|
||||
"produkt": ("Produkt", lambda i, k: i.product.name if i.product else ""),
|
||||
"marke": ("Marke", lambda i, k: (i.product.brand if i.product else "") or ""),
|
||||
"kategorie": (
|
||||
"Kategorie",
|
||||
lambda i, k: _kategorie_pfad(k.db, i.product.category if i.product else None),
|
||||
),
|
||||
"lagerort": ("Lagerort", lambda i, k: i.location.name if i.location else ""),
|
||||
"lagerort_pfad": ("Lagerort (Pfad)", lambda i, k: _lagerort_pfad(k.db, i.location)),
|
||||
"gekauft_am": ("Gekauft am", lambda i, k: str(i.acquired_on or "")),
|
||||
"garantie_bis": ("Garantie bis", lambda i, k: str(i.warranty_until or "")),
|
||||
"gekauft_bei": ("Gekauft bei", lambda i, k: i.shop.name if i.shop else ""),
|
||||
"kaufpreis": ("Kaufpreis", _preis),
|
||||
"waehrung": ("Währung", lambda i, k: i.currency or ""),
|
||||
"notiz": ("Notiz", lambda i, k: i.note or ""),
|
||||
"verliehen_an": (
|
||||
"Verliehen an",
|
||||
lambda i, k: (_offener_verleih(i).borrower if _offener_verleih(i) else ""),
|
||||
),
|
||||
"gehoert_zu": (
|
||||
"Gehört zu",
|
||||
lambda i, k: (
|
||||
f"{i.gehoert_zu.product.name if i.gehoert_zu.product else ''} "
|
||||
f"({i.gehoert_zu.uid})".strip()
|
||||
if i.gehoert_zu else ""
|
||||
),
|
||||
),
|
||||
},
|
||||
"products": {
|
||||
"name": ("Name", lambda p, k: p.name),
|
||||
"marke": ("Marke", lambda p, k: p.brand or ""),
|
||||
"kategorie": ("Kategorie", lambda p, k: _kategorie_pfad(k.db, p.category)),
|
||||
"barcode": ("Barcode", lambda p, k: p.barcode or ""),
|
||||
"gebinde": ("Gebinde", lambda p, k: p.package_label or ""),
|
||||
"gebindegroesse": (
|
||||
"Gebindegröße",
|
||||
lambda p, k: "" if p.package_size is None else _zahl(p.package_size),
|
||||
),
|
||||
"mindestbestand": (
|
||||
"Mindestbestand",
|
||||
lambda p, k: "" if p.min_stock is None else _zahl(p.min_stock),
|
||||
),
|
||||
"gekauft_bei": ("Gekauft bei", lambda p, k: p.shop.name if p.shop else ""),
|
||||
"produktlink": ("Produktlink", lambda p, k: p.product_url or ""),
|
||||
},
|
||||
"locations": {
|
||||
"qr": ("QR-Inhalt", lambda o, k: f"{k.basis_url.rstrip('/')}/l/{o.id}"),
|
||||
"id": ("ID", lambda o, k: o.id),
|
||||
"name": ("Lagerort", lambda o, k: o.name),
|
||||
"pfad": ("Pfad", lambda o, k: _lagerort_pfad(k.db, o)),
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _zahl(wert: float) -> str:
|
||||
"""Ganze Zahlen ohne „.0" – auf einem Etikett stoert das."""
|
||||
return str(int(wert)) if float(wert).is_integer() else str(wert)
|
||||
|
||||
|
||||
def feld_spalten(db: Session, art: str, kategorie_ids: list[int]) -> list[dict]:
|
||||
"""Die eigenen Felder, die als Spalten zur Wahl stehen.
|
||||
|
||||
Eigene Felder haengen an Kategorien und damit am ARTIKEL – nicht am
|
||||
Einzelstueck. Alle Stuecke eines Artikels teilen sich denselben Wert; fuer
|
||||
ein Etikett ist genau das gewollt.
|
||||
"""
|
||||
if art == "locations":
|
||||
return [] # Lagerorte haben keine eigenen Felder
|
||||
frage = db.query(FieldDefinition)
|
||||
if kategorie_ids:
|
||||
# Ein Feld gilt auch fuer Unterkategorien; umgekehrt heisst das: alle
|
||||
# Definitionen entlang der Vorfahren der gewaehlten Kategorien.
|
||||
from ..routers.categories import descendant_ids
|
||||
|
||||
erlaubt: set[int] = set()
|
||||
for kid in kategorie_ids:
|
||||
erlaubt |= descendant_ids(db, kid)
|
||||
erlaubt |= {c.id for c in _vorfahren(db, kid)}
|
||||
frage = frage.filter(FieldDefinition.category_id.in_(erlaubt or {-1}))
|
||||
gesehen: set[int] = set()
|
||||
spalten: list[dict] = []
|
||||
for fd in frage.order_by(FieldDefinition.position, FieldDefinition.id).all():
|
||||
if fd.id in gesehen:
|
||||
continue
|
||||
gesehen.add(fd.id)
|
||||
titel = f"{fd.label} ({fd.unit})" if fd.unit else fd.label
|
||||
spalten.append({"schluessel": f"{FELD_PRAEFIX}{fd.id}", "titel": titel})
|
||||
return spalten
|
||||
|
||||
|
||||
def _vorfahren(db: Session, category_id: int) -> list[Category]:
|
||||
kette: list[Category] = []
|
||||
aktuell = db.get(Category, category_id)
|
||||
gesehen: set[int] = set()
|
||||
while aktuell is not None and aktuell.id not in gesehen:
|
||||
gesehen.add(aktuell.id)
|
||||
kette.append(aktuell)
|
||||
aktuell = db.get(Category, aktuell.parent_id) if aktuell.parent_id else None
|
||||
return kette
|
||||
|
||||
|
||||
def verfuegbare_spalten(db: Session, art: str, kategorie_ids: list[int]) -> list[dict]:
|
||||
"""Feste Spalten der Bezugsart plus die passenden eigenen Felder."""
|
||||
fest = [
|
||||
{"schluessel": s, "titel": titel}
|
||||
for s, (titel, _) in SPALTEN.get(art, {}).items()
|
||||
]
|
||||
return fest + feld_spalten(db, art, kategorie_ids)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Konfiguration
|
||||
# --------------------------------------------------------------------------
|
||||
def konfiguration_lesen(vorlage: EtikettenVorlage) -> dict:
|
||||
"""JSON-Text zu einem Woerterbuch – kaputter Inhalt ergibt Vorgaben."""
|
||||
try:
|
||||
roh = json.loads(vorlage.konfiguration or "{}")
|
||||
except (ValueError, TypeError):
|
||||
roh = {}
|
||||
if not isinstance(roh, dict):
|
||||
roh = {}
|
||||
spalten = [str(s) for s in roh.get("spalten", []) if isinstance(s, (str, int))]
|
||||
kategorie_ids = [int(k) for k in roh.get("kategorie_ids", []) if str(k).lstrip("-").isdigit()]
|
||||
trenn = roh.get("trennzeichen") or ","
|
||||
if trenn not in (",", ";", "\t"):
|
||||
trenn = ","
|
||||
return {"spalten": spalten, "kategorie_ids": kategorie_ids, "trennzeichen": trenn}
|
||||
|
||||
|
||||
def _zeilen(db: Session, art: str, kategorie_ids: list[int]) -> list:
|
||||
if art == "locations":
|
||||
return db.query(Location).order_by(Location.name).all()
|
||||
|
||||
from ..routers.categories import descendant_ids
|
||||
|
||||
ids: set[int] | None = None
|
||||
if kategorie_ids:
|
||||
ids = set()
|
||||
for kid in kategorie_ids:
|
||||
ids |= descendant_ids(db, kid)
|
||||
|
||||
if art == "products":
|
||||
frage = db.query(Product)
|
||||
if ids is not None:
|
||||
frage = frage.filter(Product.category_id.in_(ids or {-1}))
|
||||
return frage.order_by(Product.name).all()
|
||||
|
||||
frage = db.query(Item)
|
||||
if ids is not None:
|
||||
frage = frage.join(Product, Item.product_id == Product.id).filter(
|
||||
Product.category_id.in_(ids or {-1})
|
||||
)
|
||||
return frage.order_by(Item.id).all()
|
||||
|
||||
|
||||
def _feldwerte(db: Session, art: str, zeilen: list) -> dict[int, dict[int, str | None]]:
|
||||
"""Alle Feldwerte der beteiligten Artikel in EINER Abfrage.
|
||||
|
||||
Ohne das fragt jede Zeile einzeln nach – bei ein paar hundert Etiketten
|
||||
ist das der Unterschied zwischen einer Abfrage und Hunderten.
|
||||
"""
|
||||
if art == "locations":
|
||||
return {}
|
||||
produkt_ids = {
|
||||
(z.product_id if art == "items" else z.id) for z in zeilen
|
||||
}
|
||||
if not produkt_ids:
|
||||
return {}
|
||||
werte: dict[int, dict[int, str | None]] = {}
|
||||
for pfv in (
|
||||
db.query(ProductFieldValue)
|
||||
.filter(ProductFieldValue.product_id.in_(produkt_ids))
|
||||
.all()
|
||||
):
|
||||
werte.setdefault(pfv.product_id, {})[pfv.field_definition_id] = pfv.value
|
||||
return werte
|
||||
|
||||
|
||||
def _zelle(schluessel: str, zeile, art: str, kontext: Kontext) -> str | None:
|
||||
"""Wert einer Spalte – None, wenn es die Spalte nicht (mehr) gibt."""
|
||||
if schluessel.startswith(FELD_PRAEFIX):
|
||||
rest = schluessel[len(FELD_PRAEFIX):]
|
||||
if not rest.isdigit():
|
||||
return None
|
||||
fd_id = int(rest)
|
||||
if kontext.db.get(FieldDefinition, fd_id) is None:
|
||||
return None # Feld wurde geloescht – Spalte faellt still weg
|
||||
produkt_id = zeile.product_id if art == "items" else getattr(zeile, "id", None)
|
||||
return (kontext.feldwerte or {}).get(produkt_id, {}).get(fd_id) or ""
|
||||
eintrag = SPALTEN.get(art, {}).get(schluessel)
|
||||
if eintrag is None:
|
||||
return None
|
||||
return eintrag[1](zeile, kontext)
|
||||
|
||||
|
||||
def _titel(schluessel: str, art: str, db: Session) -> str | None:
|
||||
if schluessel.startswith(FELD_PRAEFIX):
|
||||
rest = schluessel[len(FELD_PRAEFIX):]
|
||||
if not rest.isdigit():
|
||||
return None
|
||||
fd = db.get(FieldDefinition, int(rest))
|
||||
if fd is None:
|
||||
return None
|
||||
return f"{fd.label} ({fd.unit})" if fd.unit else fd.label
|
||||
eintrag = SPALTEN.get(art, {}).get(schluessel)
|
||||
return eintrag[0] if eintrag else None
|
||||
|
||||
|
||||
def _schuetzen(wert: str) -> str:
|
||||
"""Formel-Injektion verhindern (=, +, -, @ am Zellenanfang)."""
|
||||
text = "" if wert is None else str(wert)
|
||||
return f"'{text}" if _FORMEL_START.match(text) else text
|
||||
|
||||
|
||||
def csv_erzeugen(db: Session, vorlage: EtikettenVorlage, basis_url: str = "") -> str:
|
||||
"""Die fertige CSV zu einer Vorlage.
|
||||
|
||||
Mit BOM und CRLF, damit Excel und P-touch die Umlaute richtig lesen.
|
||||
Unbekannte Spalten (geloeschtes Feld, geaenderte Bezugsart) fallen still
|
||||
weg, statt den Export zu zerreissen – so wie die Tabellen in der
|
||||
Oberflaeche verschwundene Spalten aussortieren.
|
||||
"""
|
||||
art = vorlage.art if vorlage.art in ARTEN else "items"
|
||||
konf = konfiguration_lesen(vorlage)
|
||||
|
||||
gueltig = [s for s in konf["spalten"] if _titel(s, art, db) is not None]
|
||||
if not gueltig:
|
||||
# Lieber die festen Spalten der Bezugsart als eine leere Datei.
|
||||
gueltig = list(SPALTEN.get(art, {}).keys())
|
||||
|
||||
zeilen = _zeilen(db, art, konf["kategorie_ids"])
|
||||
kontext = Kontext(db=db, basis_url=basis_url, feldwerte=_feldwerte(db, art, zeilen))
|
||||
|
||||
puffer = io.StringIO()
|
||||
schreiber = csv.writer(
|
||||
puffer, delimiter=konf["trennzeichen"], lineterminator="\r\n",
|
||||
quoting=csv.QUOTE_MINIMAL,
|
||||
)
|
||||
schreiber.writerow([_schuetzen(_titel(s, art, db) or s) for s in gueltig])
|
||||
for zeile in zeilen:
|
||||
schreiber.writerow(
|
||||
[_schuetzen(_zelle(s, zeile, art, kontext) or "") for s in gueltig]
|
||||
)
|
||||
return "" + puffer.getvalue()
|
||||
@@ -60,7 +60,8 @@ def test_backup_nimmt_obergruppen_und_bedarfe_mit(db, user, monkeypatch):
|
||||
db.commit()
|
||||
|
||||
daten = json.loads(_sicherung(db, user))
|
||||
assert daten["version"] == 3
|
||||
# v4 seit den Etiketten-Vorlagen; aeltere Sicherungen bleiben lesbar.
|
||||
assert daten["version"] == 4
|
||||
nach_name = {g["name"]: g for g in daten["groups"]}
|
||||
assert sorted(nach_name["Grillwurst"]["parents"]) == ["Grillgut", "Wurst"]
|
||||
assert {"location": None, "min_stock": 5000} in nach_name["Wurst"]["min_stocks"]
|
||||
|
||||
@@ -0,0 +1,353 @@
|
||||
"""Etiketten-Vorlagen: Spaltenwahl, eigene Felder, CSV-Form und .lbx-Prüfung.
|
||||
|
||||
Der Export ist eine reine Funktion über einer Datenbank-Sitzung – geprüft wird
|
||||
deshalb ohne HTTP, direkt am Ergebnis.
|
||||
"""
|
||||
|
||||
import io
|
||||
import json
|
||||
import zipfile
|
||||
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
|
||||
from app.models import (
|
||||
Category,
|
||||
CategoryTracking,
|
||||
EtikettenVorlage,
|
||||
FieldDefinition,
|
||||
Item,
|
||||
Location,
|
||||
Product,
|
||||
ProductFieldValue,
|
||||
Role,
|
||||
User,
|
||||
)
|
||||
from app.routers.etiketten import _pruefe_lbx, spalten
|
||||
from app.services.etiketten import csv_erzeugen, konfiguration_lesen
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def admin(db):
|
||||
person = User(username="admin", password_hash="x", role=Role.admin)
|
||||
db.add(person)
|
||||
db.commit()
|
||||
db.refresh(person)
|
||||
return person
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def kabel(db):
|
||||
"""Kategorie „Kabel" mit eigenem Feld „Länge" in Metern."""
|
||||
cat = Category(name="Kabel", tracking=CategoryTracking.object.value)
|
||||
db.add(cat)
|
||||
db.flush()
|
||||
fd = FieldDefinition(
|
||||
category_id=cat.id, label="Länge", key="laenge", field_type="number", unit="m"
|
||||
)
|
||||
db.add(fd)
|
||||
db.commit()
|
||||
db.refresh(cat)
|
||||
db.refresh(fd)
|
||||
return cat, fd
|
||||
|
||||
|
||||
def _vorlage(db, spalten, art="items", kategorie_ids=None, trenn=","):
|
||||
v = EtikettenVorlage(
|
||||
name=f"Test {len(db.query(EtikettenVorlage).all())}",
|
||||
art=art,
|
||||
konfiguration=json.dumps({
|
||||
"spalten": spalten,
|
||||
"kategorie_ids": kategorie_ids or [],
|
||||
"trennzeichen": trenn,
|
||||
}),
|
||||
)
|
||||
db.add(v)
|
||||
db.commit()
|
||||
db.refresh(v)
|
||||
return v
|
||||
|
||||
|
||||
def _stueck(db, cat, name="HDMI-Kabel", laenge=None, fd=None, ort=None):
|
||||
product = Product(name=name, category_id=cat.id, individual=True, package_size=1)
|
||||
db.add(product)
|
||||
db.flush()
|
||||
if laenge is not None and fd is not None:
|
||||
db.add(ProductFieldValue(
|
||||
product_id=product.id, field_definition_id=fd.id, value=laenge
|
||||
))
|
||||
item = Item(uid=f"UID{product.id:05d}", product_id=product.id,
|
||||
location_id=(ort.id if ort else None))
|
||||
db.add(item)
|
||||
db.commit()
|
||||
db.refresh(item)
|
||||
return item
|
||||
|
||||
|
||||
def _zeilen(csv_text):
|
||||
"""Ohne BOM, als Liste von Zeilen."""
|
||||
assert csv_text.startswith(""), "BOM fehlt – Excel zeigt sonst kaputte Umlaute"
|
||||
return csv_text[1:].rstrip("\r\n").split("\r\n")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Spalten und Reihenfolge
|
||||
# --------------------------------------------------------------------------
|
||||
def test_spalten_kommen_in_gespeicherter_reihenfolge(db, kabel):
|
||||
cat, _ = kabel
|
||||
_stueck(db, cat)
|
||||
v = _vorlage(db, ["produkt", "uid", "qr"])
|
||||
|
||||
kopf = _zeilen(csv_erzeugen(db, v))[0]
|
||||
assert kopf == "Produkt,UID,QR-Inhalt"
|
||||
|
||||
|
||||
def test_eigenes_feld_landet_in_der_csv(db, kabel):
|
||||
cat, fd = kabel
|
||||
_stueck(db, cat, laenge="3", fd=fd)
|
||||
v = _vorlage(db, ["uid", f"feld:{fd.id}"])
|
||||
|
||||
kopf, zeile = _zeilen(csv_erzeugen(db, v))
|
||||
assert kopf == "UID,Länge (m)"
|
||||
assert zeile.endswith(",3")
|
||||
|
||||
|
||||
def test_fehlender_feldwert_gibt_eine_leere_zelle(db, kabel):
|
||||
"""Nicht etwa eine Luecke – sonst verrutschen alle folgenden Spalten."""
|
||||
cat, fd = kabel
|
||||
_stueck(db, cat) # ohne Wert
|
||||
v = _vorlage(db, ["uid", f"feld:{fd.id}", "produkt"])
|
||||
|
||||
zeile = _zeilen(csv_erzeugen(db, v))[1]
|
||||
assert zeile.count(",") == 2
|
||||
assert ",," in zeile
|
||||
|
||||
|
||||
def test_geloeschtes_feld_zerreisst_den_export_nicht(db, kabel):
|
||||
cat, fd = kabel
|
||||
_stueck(db, cat)
|
||||
v = _vorlage(db, ["uid", f"feld:{fd.id}", "produkt"])
|
||||
db.delete(fd)
|
||||
db.commit()
|
||||
|
||||
kopf = _zeilen(csv_erzeugen(db, v))[0]
|
||||
assert kopf == "UID,Produkt" # die tote Spalte faellt still weg
|
||||
|
||||
|
||||
def test_ohne_gueltige_spalte_greifen_die_festen(db, kabel):
|
||||
cat, _ = kabel
|
||||
_stueck(db, cat)
|
||||
v = _vorlage(db, ["gibtsnicht"])
|
||||
|
||||
kopf = _zeilen(csv_erzeugen(db, v))[0]
|
||||
assert kopf.startswith("QR-Inhalt,UID,Produkt")
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# CSV-Form
|
||||
# --------------------------------------------------------------------------
|
||||
def test_trennzeichen_semikolon_und_tab(db, kabel):
|
||||
cat, _ = kabel
|
||||
_stueck(db, cat)
|
||||
assert _zeilen(csv_erzeugen(db, _vorlage(db, ["uid", "produkt"], trenn=";")))[0] \
|
||||
== "UID;Produkt"
|
||||
assert _zeilen(csv_erzeugen(db, _vorlage(db, ["uid", "produkt"], trenn="\t")))[0] \
|
||||
== "UID\tProdukt"
|
||||
|
||||
|
||||
def test_zelle_mit_trennzeichen_wird_eingefasst(db, kabel):
|
||||
cat, _ = kabel
|
||||
_stueck(db, cat, name="Kabel, lang")
|
||||
v = _vorlage(db, ["produkt"])
|
||||
|
||||
assert _zeilen(csv_erzeugen(db, v))[1] == '"Kabel, lang"'
|
||||
|
||||
|
||||
def test_formel_wird_entschaerft(db, kabel):
|
||||
"""Ohne das fuehrt Excel den Zellinhalt als Formel aus."""
|
||||
cat, _ = kabel
|
||||
_stueck(db, cat, name="=1+1")
|
||||
v = _vorlage(db, ["produkt"])
|
||||
|
||||
assert _zeilen(csv_erzeugen(db, v))[1] == "'=1+1"
|
||||
|
||||
|
||||
def test_zeilenende_ist_crlf(db, kabel):
|
||||
cat, _ = kabel
|
||||
_stueck(db, cat)
|
||||
assert "\r\n" in csv_erzeugen(db, _vorlage(db, ["uid"]))
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Bezugsarten und Filter
|
||||
# --------------------------------------------------------------------------
|
||||
def test_kategoriefilter_greift_samt_unterkategorien(db, kabel):
|
||||
cat, _ = kabel
|
||||
unter = Category(name="HDMI", tracking=CategoryTracking.object.value, parent_id=cat.id)
|
||||
andere = Category(name="Werkzeug", tracking=CategoryTracking.object.value)
|
||||
db.add_all([unter, andere])
|
||||
db.commit()
|
||||
_stueck(db, unter, name="HDMI 2.1")
|
||||
_stueck(db, andere, name="Hammer")
|
||||
|
||||
v = _vorlage(db, ["produkt"], kategorie_ids=[cat.id])
|
||||
zeilen = _zeilen(csv_erzeugen(db, v))
|
||||
assert zeilen[1:] == ["HDMI 2.1"]
|
||||
|
||||
|
||||
def test_lagerorte_brauchen_die_basis_adresse_fuer_den_qr(db):
|
||||
ort = Location(name="Keller")
|
||||
db.add(ort)
|
||||
db.commit()
|
||||
v = _vorlage(db, ["qr", "name"], art="locations")
|
||||
|
||||
zeile = _zeilen(csv_erzeugen(db, v, basis_url="https://haus.example/"))[1]
|
||||
assert zeile.startswith(f"https://haus.example/l/{ort.id},")
|
||||
|
||||
|
||||
def test_artikel_als_bezugsart(db, kabel):
|
||||
cat, fd = kabel
|
||||
_stueck(db, cat, name="USB-C", laenge="1", fd=fd)
|
||||
v = _vorlage(db, ["name", f"feld:{fd.id}"], art="products")
|
||||
|
||||
kopf, zeile = _zeilen(csv_erzeugen(db, v))
|
||||
assert kopf == "Name,Länge (m)"
|
||||
assert zeile == "USB-C,1"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Konfiguration lesen
|
||||
# --------------------------------------------------------------------------
|
||||
def test_kaputte_konfiguration_ergibt_vorgaben(db):
|
||||
v = EtikettenVorlage(name="Kaputt", art="items", konfiguration="{nicht json")
|
||||
db.add(v)
|
||||
db.commit()
|
||||
|
||||
konf = konfiguration_lesen(v)
|
||||
assert konf == {"spalten": [], "kategorie_ids": [], "trennzeichen": ","}
|
||||
|
||||
|
||||
def test_unerlaubtes_trennzeichen_faellt_auf_komma_zurueck(db):
|
||||
v = _vorlage(db, ["uid"], trenn="|")
|
||||
assert konfiguration_lesen(v)["trennzeichen"] == ","
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Spaltenliste für die Oberfläche
|
||||
# --------------------------------------------------------------------------
|
||||
def test_spaltenliste_nennt_eigene_felder_der_kategorie(db, admin, kabel):
|
||||
cat, fd = kabel
|
||||
liste = spalten(art="items", kategorie_ids=str(cat.id), db=db, _=admin)
|
||||
schluessel = {s.schluessel for s in liste}
|
||||
|
||||
assert "qr" in schluessel and "uid" in schluessel
|
||||
assert f"feld:{fd.id}" in schluessel
|
||||
assert next(s.titel for s in liste if s.schluessel == f"feld:{fd.id}") == "Länge (m)"
|
||||
|
||||
|
||||
def test_lagerorte_haben_keine_eigenen_felder(db, admin, kabel):
|
||||
liste = spalten(art="locations", kategorie_ids=None, db=db, _=admin)
|
||||
assert not any(s.schluessel.startswith("feld:") for s in liste)
|
||||
|
||||
|
||||
def test_unbekannte_bezugsart_wird_abgelehnt(db, admin):
|
||||
with pytest.raises(HTTPException) as fehler:
|
||||
spalten(art="quatsch", kategorie_ids=None, db=db, _=admin)
|
||||
assert fehler.value.status_code == 400
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# .lbx-Prüfung
|
||||
# --------------------------------------------------------------------------
|
||||
def _zip(namen):
|
||||
puffer = io.BytesIO()
|
||||
with zipfile.ZipFile(puffer, "w") as z:
|
||||
for n in namen:
|
||||
z.writestr(n, "<xml/>")
|
||||
return puffer.getvalue()
|
||||
|
||||
|
||||
def test_echte_lbx_wird_angenommen(db):
|
||||
_pruefe_lbx("etikett.lbx", _zip(["label.xml", "prop.xml"])) # wirft nicht
|
||||
|
||||
|
||||
def test_zip_ohne_label_xml_wird_abgelehnt(db):
|
||||
with pytest.raises(HTTPException) as fehler:
|
||||
_pruefe_lbx("etikett.lbx", _zip(["irgendwas.txt"]))
|
||||
assert fehler.value.status_code == 415
|
||||
|
||||
|
||||
def test_nicht_zip_wird_abgelehnt(db):
|
||||
with pytest.raises(HTTPException) as fehler:
|
||||
_pruefe_lbx("etikett.lbx", b"\x89PNG\r\n\x1a\n und noch was")
|
||||
assert fehler.value.status_code == 415
|
||||
|
||||
|
||||
def test_falsche_endung_wird_abgelehnt(db):
|
||||
with pytest.raises(HTTPException) as fehler:
|
||||
_pruefe_lbx("bild.png", _zip(["label.xml"]))
|
||||
assert fehler.value.status_code == 415
|
||||
|
||||
|
||||
def test_leere_datei_wird_abgelehnt(db):
|
||||
with pytest.raises(HTTPException) as fehler:
|
||||
_pruefe_lbx("etikett.lbx", b"")
|
||||
assert fehler.value.status_code == 400
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------
|
||||
# Backup-Umlauf
|
||||
# --------------------------------------------------------------------------
|
||||
def test_vorlage_uebersteht_die_sicherung(db, admin, kabel):
|
||||
"""Kategorien als Pfad, eigene Felder als Label – IDs sind zwischen zwei
|
||||
Instanzen nicht dieselben."""
|
||||
import json as _json
|
||||
|
||||
from app.routers.transfer import _import_json, export_backup_json
|
||||
from app.seed import ensure_builtin_units
|
||||
|
||||
cat, fd = kabel
|
||||
_stueck(db, cat, laenge="5", fd=fd)
|
||||
v = _vorlage(db, ["uid", f"feld:{fd.id}"], kategorie_ids=[cat.id], trenn=";")
|
||||
v.name = "Kabelexport"
|
||||
v.datei_name = "kabel.lbx"
|
||||
v.datei = _zip(["label.xml"])
|
||||
db.commit()
|
||||
|
||||
inhalt = export_backup_json(db=db, _=admin).body
|
||||
daten = _json.loads(inhalt)
|
||||
gesichert = daten["etiketten_vorlagen"]
|
||||
assert [e["name"] for e in gesichert] == ["Kabelexport"]
|
||||
# Das Feld steht als Pfad::Label drin, nicht als ID.
|
||||
assert gesichert[0]["spalten"][1] == "feld:Kabel::Länge"
|
||||
|
||||
# In eine zweite, leere Instanz einspielen.
|
||||
from app.database import Base
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
zweiter_motor = create_engine(
|
||||
"sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool
|
||||
)
|
||||
Base.metadata.create_all(bind=zweiter_motor)
|
||||
zweite = sessionmaker(bind=zweiter_motor)()
|
||||
ensure_builtin_units(zweite)
|
||||
person = User(username="admin", password_hash="x", role=Role.admin)
|
||||
zweite.add(person)
|
||||
zweite.commit()
|
||||
|
||||
_import_json(zweite, inhalt, person, "add")
|
||||
|
||||
neu = zweite.query(EtikettenVorlage).filter(
|
||||
EtikettenVorlage.name == "Kabelexport"
|
||||
).one()
|
||||
konf = konfiguration_lesen(neu)
|
||||
assert konf["trennzeichen"] == ";"
|
||||
assert neu.datei_name == "kabel.lbx"
|
||||
assert neu.datei == _zip(["label.xml"])
|
||||
# Das Feld wurde ueber Pfad und Label wiedergefunden – mit NEUER ID.
|
||||
feld_spalte = [s for s in konf["spalten"] if s.startswith("feld:")][0]
|
||||
neue_fd = zweite.query(FieldDefinition).filter(FieldDefinition.label == "Länge").one()
|
||||
assert feld_spalte == f"feld:{neue_fd.id}"
|
||||
zweite.close()
|
||||
Reference in New Issue
Block a user