Gegenstands-Verwaltung (Non-Food) neben Lebensmitteln
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>
This commit is contained in:
Scarriffle
139
backend/app/services/fields.py
Normal file
139
backend/app/services/fields.py
Normal file
@@ -0,0 +1,139 @@
|
||||
"""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
|
||||
)
|
||||
)
|
||||
@@ -16,6 +16,181 @@ class StockError(ValueError):
|
||||
"""Fachlicher Fehler beim Ein-/Auslagern (z.B. zu wenig Bestand)."""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Gegenstände (Non-Food): Menge je Lagerort statt Chargen mit MHD.
|
||||
#
|
||||
# Technisch wird dieselbe Lot-Tabelle genutzt – je (Produkt, Lagerort) genau
|
||||
# eine Zeile mit ``best_before = NULL``. So laufen Bestands-Summe, Bewegungslog
|
||||
# und Export unverändert weiter; nur MHD/FEFO entfällt. Die Lebensmittel-Logik
|
||||
# oben (check_in/check_out) bleibt davon unberührt.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _object_lot(db: Session, product_id: int, location_id: int | None) -> Lot | None:
|
||||
"""Die (einzige) Bestandszeile eines Gegenstands an einem Lagerort."""
|
||||
query = db.query(Lot).filter(
|
||||
Lot.product_id == product_id, Lot.best_before.is_(None)
|
||||
)
|
||||
if location_id is None:
|
||||
query = query.filter(Lot.location_id.is_(None))
|
||||
else:
|
||||
query = query.filter(Lot.location_id == location_id)
|
||||
return query.first()
|
||||
|
||||
|
||||
def object_add(
|
||||
db: Session,
|
||||
product: Product,
|
||||
quantity: float,
|
||||
location_id: int | None,
|
||||
user: User | None,
|
||||
note: str | None = None,
|
||||
) -> Lot:
|
||||
"""Erhöht die Menge eines Gegenstands an einem Lagerort."""
|
||||
lot = _object_lot(db, product.id, location_id)
|
||||
if lot is None:
|
||||
lot = Lot(
|
||||
product_id=product.id,
|
||||
quantity=0.0,
|
||||
best_before=None,
|
||||
best_before_precision=DatePrecision.day.value,
|
||||
location_id=location_id,
|
||||
)
|
||||
db.add(lot)
|
||||
db.flush()
|
||||
lot.quantity += quantity
|
||||
db.add(
|
||||
Movement(
|
||||
product_id=product.id,
|
||||
lot_id=lot.id,
|
||||
user_id=user.id if user else None,
|
||||
type=MovementType.in_,
|
||||
quantity=quantity,
|
||||
unit_used=product.base_unit.value,
|
||||
note=note,
|
||||
location_id=location_id,
|
||||
)
|
||||
)
|
||||
return lot
|
||||
|
||||
|
||||
def object_remove(
|
||||
db: Session,
|
||||
product: Product,
|
||||
quantity: float,
|
||||
location_id: int | None,
|
||||
reason: str,
|
||||
user: User | None,
|
||||
note: str | None = None,
|
||||
) -> None:
|
||||
"""Entfernt eine Menge mit Grund (verloren/kaputt/…) aus dem Bestand."""
|
||||
lot = _object_lot(db, product.id, location_id)
|
||||
have = lot.quantity if lot else 0.0
|
||||
if quantity > have + 1e-9:
|
||||
raise StockError(
|
||||
f"Am Lagerort sind nur {have:g} {product.base_unit.value} vorhanden "
|
||||
f"(benötigt {quantity:g})."
|
||||
)
|
||||
lot.quantity -= quantity
|
||||
db.add(
|
||||
Movement(
|
||||
product_id=product.id,
|
||||
lot_id=lot.id,
|
||||
user_id=user.id if user else None,
|
||||
type=MovementType.out,
|
||||
quantity=quantity,
|
||||
unit_used=product.base_unit.value,
|
||||
note=note,
|
||||
location_id=location_id,
|
||||
reason=reason,
|
||||
)
|
||||
)
|
||||
if lot.quantity <= 1e-9:
|
||||
db.delete(lot)
|
||||
|
||||
|
||||
def object_relocate(
|
||||
db: Session,
|
||||
product: Product,
|
||||
quantity: float,
|
||||
from_location_id: int | None,
|
||||
to_location_id: int | None,
|
||||
user: User | None,
|
||||
note: str | None = None,
|
||||
) -> None:
|
||||
"""Bucht eine Menge von einem Lagerort zum anderen um (ohne Grund)."""
|
||||
if from_location_id == to_location_id:
|
||||
raise StockError("Quell- und Ziel-Lagerort sind identisch.")
|
||||
src = _object_lot(db, product.id, from_location_id)
|
||||
have = src.quantity if src else 0.0
|
||||
if quantity > have + 1e-9:
|
||||
raise StockError(
|
||||
f"Am Quell-Lagerort sind nur {have:g} {product.base_unit.value} "
|
||||
f"vorhanden (benötigt {quantity:g})."
|
||||
)
|
||||
beleg = note or "Umlagerung"
|
||||
src.quantity -= quantity
|
||||
# Als neutrale Korrektur (adjust) protokollieren, damit Umlagerungen die
|
||||
# Ein-/Auslager-Statistiken nicht verfälschen.
|
||||
db.add(
|
||||
Movement(
|
||||
product_id=product.id,
|
||||
lot_id=src.id,
|
||||
user_id=user.id if user else None,
|
||||
type=MovementType.adjust,
|
||||
quantity=-quantity,
|
||||
unit_used=product.base_unit.value,
|
||||
note=beleg,
|
||||
location_id=from_location_id,
|
||||
)
|
||||
)
|
||||
if src.quantity <= 1e-9:
|
||||
db.delete(src)
|
||||
|
||||
dest = _object_lot(db, product.id, to_location_id)
|
||||
if dest is None:
|
||||
dest = Lot(
|
||||
product_id=product.id,
|
||||
quantity=0.0,
|
||||
best_before=None,
|
||||
best_before_precision=DatePrecision.day.value,
|
||||
location_id=to_location_id,
|
||||
)
|
||||
db.add(dest)
|
||||
db.flush()
|
||||
dest.quantity += quantity
|
||||
db.add(
|
||||
Movement(
|
||||
product_id=product.id,
|
||||
lot_id=dest.id,
|
||||
user_id=user.id if user else None,
|
||||
type=MovementType.adjust,
|
||||
quantity=quantity,
|
||||
unit_used=product.base_unit.value,
|
||||
note=beleg,
|
||||
location_id=to_location_id,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def removal_stats(db: Session, product_id: int) -> dict[str, dict]:
|
||||
"""Entnahmen je Grund summieren: {reason: {quantity, count}}."""
|
||||
rows = (
|
||||
db.query(Movement)
|
||||
.filter(
|
||||
Movement.product_id == product_id,
|
||||
Movement.type == MovementType.out,
|
||||
Movement.reason.isnot(None),
|
||||
)
|
||||
.all()
|
||||
)
|
||||
stats: dict[str, dict] = {}
|
||||
for m in rows:
|
||||
eintrag = stats.setdefault(m.reason, {"quantity": 0.0, "count": 0})
|
||||
eintrag["quantity"] += m.quantity
|
||||
eintrag["count"] += 1
|
||||
return stats
|
||||
|
||||
|
||||
def current_stock(db: Session, product_id: int) -> float:
|
||||
"""Summe der Lot-Mengen eines Produkts (in Basiseinheiten)."""
|
||||
total = (
|
||||
|
||||
Reference in New Issue
Block a user