5ce4411d1e
Gegenstands-Produkte koennen als Einzelstuecke gefuehrt werden (Product.individual):
jedes physische Stueck ist ein Item mit eigener kurzer UID (fuer QR), eigenem
Lagerort, Kaufdatum, Garantie, Bezugsquelle und Notiz. So laesst sich dasselbe
Modell mehrfach getrennt fuehren (Powerbank 2024 + 2025).
Neu: models.Item + Product.individual (+Migration), Schemas, services/items.py
(UID-Erzeugung, Anreicherung), routers/items.py (CRUD, /items/by-uid/{uid} zur
QR-Aufloesung, /items/{id}/remove mit Grund als Bewegung). current_stock zaehlt
bei Einzelstueck-Produkten die Items. Tests + HTTP-Smoke gruen.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
352 lines
10 KiB
Python
352 lines
10 KiB
Python
"""Chargen-Logik: Einlagern erzeugt Lots, Auslagern bucht per FEFO ab."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import date
|
||
|
||
from sqlalchemy import asc
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..models import DatePrecision, Item, Lot, Movement, MovementType, Product, User
|
||
from .conversion import to_base
|
||
from .dates import clean_precision, normalize_best_before
|
||
|
||
|
||
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:
|
||
"""Bestand eines Produkts (in Basiseinheiten).
|
||
|
||
Einzelstück-Produkte zählen die Anzahl ihrer Items, alle anderen summieren
|
||
die Lot-Mengen.
|
||
"""
|
||
product = db.get(Product, product_id)
|
||
if product is not None and product.individual:
|
||
return float(db.query(Item).filter(Item.product_id == product_id).count())
|
||
total = (
|
||
db.query(Lot).with_entities(Lot.quantity).filter(Lot.product_id == product_id).all()
|
||
)
|
||
return float(sum(q for (q,) in total))
|
||
|
||
|
||
def check_in(
|
||
db: Session,
|
||
product: Product,
|
||
quantity: float,
|
||
unit: str,
|
||
best_before: date | None,
|
||
location_id: int | None,
|
||
user: User | None,
|
||
note: str | None = None,
|
||
best_before_precision: str | None = DatePrecision.day.value,
|
||
) -> Lot:
|
||
"""Legt eine neue Charge an und protokolliert die Bewegung.
|
||
|
||
Ist nur Monat/Jahr angegeben, wird das MHD auf den Monatsletzten gelegt;
|
||
die Genauigkeit wird an der Charge vermerkt, damit die Anzeige "09/2026"
|
||
statt "30.09.2026" schreiben kann.
|
||
"""
|
||
quantity_base = to_base(db, product, quantity, unit)
|
||
precision = clean_precision(best_before_precision)
|
||
|
||
lot = Lot(
|
||
product_id=product.id,
|
||
quantity=quantity_base,
|
||
best_before=normalize_best_before(best_before, precision),
|
||
best_before_precision=precision,
|
||
location_id=location_id,
|
||
)
|
||
db.add(lot)
|
||
db.flush() # lot.id verfügbar machen
|
||
|
||
db.add(
|
||
Movement(
|
||
product_id=product.id,
|
||
lot_id=lot.id,
|
||
user_id=user.id if user else None,
|
||
type=MovementType.in_,
|
||
quantity=quantity_base,
|
||
unit_used=unit,
|
||
note=note,
|
||
)
|
||
)
|
||
return lot
|
||
|
||
|
||
def check_out_lot(
|
||
db: Session,
|
||
product: Product,
|
||
lot: Lot,
|
||
quantity: float,
|
||
unit: str,
|
||
user: User | None,
|
||
note: str | None = None,
|
||
) -> list[dict]:
|
||
"""Bucht gezielt von EINER Charge ab (manuelle Auswahl statt FEFO)."""
|
||
needed = to_base(db, product, quantity, unit)
|
||
if needed > lot.quantity + 1e-9:
|
||
raise StockError(
|
||
f"Diese Charge hat nur {lot.quantity:g} {product.base_unit.value} "
|
||
f"(benötigt {needed:g})."
|
||
)
|
||
|
||
lot.quantity -= needed
|
||
db.add(
|
||
Movement(
|
||
product_id=product.id,
|
||
lot_id=lot.id,
|
||
user_id=user.id if user else None,
|
||
type=MovementType.out,
|
||
quantity=needed,
|
||
unit_used=unit,
|
||
note=note,
|
||
)
|
||
)
|
||
affected = [{"lot_id": lot.id, "quantity": needed}]
|
||
if lot.quantity <= 1e-9:
|
||
db.delete(lot)
|
||
return affected
|
||
|
||
|
||
def _fefo_lots(db: Session, product_id: int) -> list[Lot]:
|
||
"""Lots eines Produkts, sortiert nach Ablaufdatum (NULL zuletzt), dann Alter."""
|
||
lots = (
|
||
db.query(Lot)
|
||
.filter(Lot.product_id == product_id, Lot.quantity > 0)
|
||
.order_by(asc(Lot.created_at))
|
||
.all()
|
||
)
|
||
# NULL-best_before ans Ende (nach Datum aufsteigend). In Python sortieren, damit
|
||
# es über SQLite und Postgres identisch funktioniert.
|
||
return sorted(
|
||
lots,
|
||
key=lambda lot: (lot.best_before is None, lot.best_before or date.max, lot.id),
|
||
)
|
||
|
||
|
||
def check_out(
|
||
db: Session,
|
||
product: Product,
|
||
quantity: float,
|
||
unit: str,
|
||
user: User | None,
|
||
note: str | None = None,
|
||
) -> list[dict]:
|
||
"""Bucht ``quantity`` (in ``unit``) per FEFO von den Chargen ab.
|
||
|
||
Gibt die Liste der betroffenen Chargen mit abgebuchter Menge zurück.
|
||
Wirft StockError, wenn der Gesamtbestand nicht ausreicht.
|
||
"""
|
||
needed = to_base(db, product, quantity, unit)
|
||
available = current_stock(db, product.id)
|
||
if needed > available + 1e-9:
|
||
raise StockError(
|
||
f"Nicht genug Bestand: benötigt {needed:g}, verfügbar {available:g} "
|
||
f"{product.base_unit.value}"
|
||
)
|
||
|
||
affected: list[dict] = []
|
||
remaining = needed
|
||
for lot in _fefo_lots(db, product.id):
|
||
if remaining <= 1e-9:
|
||
break
|
||
take = min(lot.quantity, remaining)
|
||
lot.quantity -= take
|
||
remaining -= take
|
||
|
||
db.add(
|
||
Movement(
|
||
product_id=product.id,
|
||
lot_id=lot.id,
|
||
user_id=user.id if user else None,
|
||
type=MovementType.out,
|
||
quantity=take,
|
||
unit_used=unit,
|
||
note=note,
|
||
)
|
||
)
|
||
affected.append({"lot_id": lot.id, "quantity": take})
|
||
|
||
# Leere Charge entfernen, damit das "MHD-Array" sauber bleibt.
|
||
if lot.quantity <= 1e-9:
|
||
db.delete(lot)
|
||
|
||
return affected
|