Die Kategorie bestimmt die Verwaltungsart (food/object). Gegenstaende: Menge je Lagerort statt Chargen/MHD, Umlagern (ohne Grund) und Entfernen mit Pflicht-Grund samt Entnahme-Statistik. Beliebig viele eigene Felder je Kategorie (vererbt an Unterkategorien), "gekauft bei" ueber eine verwaltbare Shop-Liste und ein Produktlink. Barcode-Lookup zusaetzlich ueber Open Products Facts. Der Lebensmittel-Teil (Chargen/MHD/FEFO) bleibt unveraendert. Umgesetzt in Backend (FastAPI, +Tests gruen), Web (React) und iOS (SwiftUI). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
140 lines
4.6 KiB
Python
140 lines
4.6 KiB
Python
"""Selbst definierte Felder je Kategorie: Vererbung, Validierung, Werte.
|
||
|
||
Felder hängen an einer Kategorie und gelten – über den Kategorie-Baum – auch für
|
||
deren Unterkategorien. Der Wert je Artikel steht in ProductFieldValue (immer als
|
||
Text; typgerecht geprüft wird hier).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
import unicodedata
|
||
from datetime import date
|
||
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..models import Category, FieldDefinition, FieldType, ProductFieldValue
|
||
|
||
|
||
class FieldError(ValueError):
|
||
"""Ungültige Feld-Definition oder ein Wert, der nicht zum Feldtyp passt."""
|
||
|
||
|
||
def slugify(label: str) -> str:
|
||
"""Maschinenlesbaren Schlüssel aus einem Label ableiten (z.B. „Kapazität“ → „kapazitaet“)."""
|
||
text = unicodedata.normalize("NFKD", label)
|
||
text = text.encode("ascii", "ignore").decode("ascii").lower()
|
||
text = re.sub(r"[^a-z0-9]+", "_", text).strip("_")
|
||
return text or "feld"
|
||
|
||
|
||
def options_list(fd: FieldDefinition) -> list[str]:
|
||
"""Auswahlmöglichkeiten eines select-Felds als Liste (leer, wenn keine)."""
|
||
if not fd.options:
|
||
return []
|
||
try:
|
||
data = json.loads(fd.options)
|
||
except (ValueError, TypeError):
|
||
return []
|
||
return [str(o) for o in data] if isinstance(data, list) else []
|
||
|
||
|
||
def _ancestry(db: Session, category_id: int) -> list[Category]:
|
||
"""Kategorien von der Wurzel bis zur angegebenen (einschließlich)."""
|
||
chain: list[Category] = []
|
||
seen: set[int] = set()
|
||
cid: int | None = category_id
|
||
while cid is not None and cid not in seen:
|
||
seen.add(cid)
|
||
cat = db.get(Category, cid)
|
||
if cat is None:
|
||
break
|
||
chain.append(cat)
|
||
cid = cat.parent_id
|
||
chain.reverse() # Wurzel zuerst, damit deren Felder oben stehen
|
||
return chain
|
||
|
||
|
||
def effective_field_definitions(
|
||
db: Session, category_id: int | None
|
||
) -> list[tuple[FieldDefinition, bool]]:
|
||
"""Alle für eine Kategorie geltenden Felder inkl. der geerbten.
|
||
|
||
Rückgabe: Liste aus (Felddefinition, geerbt?) – geerbt=True, wenn das Feld
|
||
von einer Oberkategorie stammt. Reihenfolge: Oberkategorien zuerst.
|
||
"""
|
||
if category_id is None:
|
||
return []
|
||
result: list[tuple[FieldDefinition, bool]] = []
|
||
for cat in _ancestry(db, category_id):
|
||
defs = (
|
||
db.query(FieldDefinition)
|
||
.filter(FieldDefinition.category_id == cat.id)
|
||
.order_by(FieldDefinition.position, FieldDefinition.id)
|
||
.all()
|
||
)
|
||
for d in defs:
|
||
result.append((d, cat.id != category_id))
|
||
return result
|
||
|
||
|
||
def coerce_value(fd: FieldDefinition, raw) -> str | None:
|
||
"""Einen Rohwert typgerecht prüfen und als Text zurückgeben (oder None = leer)."""
|
||
if raw is None:
|
||
return None
|
||
text = str(raw).strip()
|
||
if text == "":
|
||
return None
|
||
|
||
if fd.field_type == FieldType.number.value:
|
||
try:
|
||
float(text.replace(",", "."))
|
||
except ValueError as exc:
|
||
raise FieldError(f"„{fd.label}“ erwartet eine Zahl.") from exc
|
||
return text
|
||
if fd.field_type == FieldType.date.value:
|
||
try:
|
||
date.fromisoformat(text)
|
||
except ValueError as exc:
|
||
raise FieldError(f"„{fd.label}“ erwartet ein Datum (JJJJ-MM-TT).") from exc
|
||
return text
|
||
if fd.field_type == FieldType.boolean.value:
|
||
return "true" if text.lower() in ("1", "true", "ja", "yes", "on") else "false"
|
||
if fd.field_type == FieldType.select.value:
|
||
opts = options_list(fd)
|
||
if opts and text not in opts:
|
||
raise FieldError(f"„{text}“ ist keine gültige Auswahl für „{fd.label}“.")
|
||
return text
|
||
return text # text / textarea
|
||
|
||
|
||
def apply_field_values(
|
||
db: Session, product, values: dict[int, str | None] | None
|
||
) -> None:
|
||
"""Feldwerte eines Artikels setzen/ändern/löschen (nur die übergebenen Felder)."""
|
||
if not values:
|
||
return
|
||
existing = {
|
||
pfv.field_definition_id: pfv
|
||
for pfv in db.query(ProductFieldValue)
|
||
.filter(ProductFieldValue.product_id == product.id)
|
||
.all()
|
||
}
|
||
for fid, raw in values.items():
|
||
fd = db.get(FieldDefinition, fid)
|
||
if fd is None:
|
||
raise FieldError(f"Feld {fid} existiert nicht.")
|
||
val = coerce_value(fd, raw)
|
||
if fid in existing:
|
||
if val is None:
|
||
db.delete(existing[fid])
|
||
else:
|
||
existing[fid].value = val
|
||
elif val is not None:
|
||
db.add(
|
||
ProductFieldValue(
|
||
product_id=product.id, field_definition_id=fid, value=val
|
||
)
|
||
)
|