05b54dc7ac
Nach dem Umlagern/Aufteilen entstanden Dubletten: zwei Chargen desselben Artikels mit gleichem MHD am gleichen Lagerort, obwohl das fuer den Nutzer eine Charge ist. Neu: consolidate_lot_group fasst solche Chargen zusammen (Mengen addieren, Bewegungen auf die aeltere Charge umhaengen, Rest loeschen) - aufgerufen nach split, bulk-location und update_lot. Bestand und Historie bleiben unveraendert. Beim Start raeumt consolidate_duplicate_lots einmalig bestehende Dubletten auf. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
481 lines
15 KiB
Python
481 lines
15 KiB
Python
"""Chargen-Logik: Einlagern erzeugt Lots, Auslagern bucht per FEFO ab."""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import date
|
||
|
||
from sqlalchemy import asc, func
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..models import (
|
||
DatePrecision,
|
||
Item,
|
||
Location,
|
||
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: str | 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: str | 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: str | 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: str | None,
|
||
to_location_id: str | 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 location_stock_base(db: Session, product: Product, location_id: str) -> float:
|
||
"""Bestand eines Produkts an EINEM Lagerort (in Basiseinheiten).
|
||
|
||
Einzelstücke zählen die Items an diesem Ort, sonst werden die Lot-Mengen des
|
||
Ortes summiert.
|
||
"""
|
||
if product.individual:
|
||
return float(
|
||
db.query(Item)
|
||
.filter(Item.product_id == product.id, Item.location_id == location_id)
|
||
.count()
|
||
)
|
||
total = (
|
||
db.query(Lot.quantity)
|
||
.filter(Lot.product_id == product.id, Lot.location_id == location_id)
|
||
.all()
|
||
)
|
||
return float(sum(q for (q,) in total))
|
||
|
||
|
||
def descendant_location_ids(db: Session, location_id: str) -> set[str]:
|
||
"""Alle Unter-Lagerorte (rekursiv) eines Lagerorts."""
|
||
result: set[str] = set()
|
||
stack = [location_id]
|
||
while stack:
|
||
cur = stack.pop()
|
||
for (lid,) in db.query(Location.id).filter(Location.parent_id == cur).all():
|
||
if lid not in result:
|
||
result.add(lid)
|
||
stack.append(lid)
|
||
return result
|
||
|
||
|
||
def location_subtree_stock_base(db: Session, product: Product, location_id: str) -> float:
|
||
"""Bestand an einem Lagerort INKL. aller Unter-Lagerorte (Basiseinheiten).
|
||
|
||
So gilt ein Mindestbestand auf „Hedingen" als gedeckt, wenn der Vorrat
|
||
irgendwo darunter liegt (z.B. „Hedingen → Keller").
|
||
"""
|
||
ids = {location_id} | descendant_location_ids(db, location_id)
|
||
return float(sum(location_stock_base(db, product, lid) for lid in ids))
|
||
|
||
|
||
def check_in(
|
||
db: Session,
|
||
product: Product,
|
||
quantity: float,
|
||
unit: str,
|
||
best_before: date | None,
|
||
location_id: str | 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 consolidate_lot_group(
|
||
db: Session,
|
||
product_id: int,
|
||
best_before: date | None,
|
||
best_before_precision: str,
|
||
location_id: str | None,
|
||
) -> Lot | None:
|
||
"""Chargen mit gleichem Artikel, MHD, Genauigkeit und Lagerort zu EINER
|
||
zusammenfassen: Mengen addieren, Bewegungen auf die verbleibende (älteste)
|
||
Charge umhängen, die übrigen löschen. Gibt die verbleibende Charge zurück.
|
||
|
||
Wird nach jedem Umlagern/Aufteilen aufgerufen – zwei gleich gelagerte Chargen
|
||
mit demselben MHD sind für den Nutzer dieselbe Charge und sollen als eine
|
||
erscheinen. Der Bestand bleibt unverändert, die Historie (Bewegungen) auch.
|
||
"""
|
||
query = db.query(Lot).filter(
|
||
Lot.product_id == product_id,
|
||
Lot.best_before_precision == best_before_precision,
|
||
)
|
||
query = (
|
||
query.filter(Lot.best_before.is_(None))
|
||
if best_before is None
|
||
else query.filter(Lot.best_before == best_before)
|
||
)
|
||
query = (
|
||
query.filter(Lot.location_id.is_(None))
|
||
if location_id is None
|
||
else query.filter(Lot.location_id == location_id)
|
||
)
|
||
lots = query.order_by(asc(Lot.created_at), asc(Lot.id)).all()
|
||
if len(lots) <= 1:
|
||
return lots[0] if lots else None
|
||
|
||
survivor = lots[0]
|
||
for extra in lots[1:]:
|
||
survivor.quantity += extra.quantity
|
||
db.query(Movement).filter(Movement.lot_id == extra.id).update(
|
||
{Movement.lot_id: survivor.id}, synchronize_session=False
|
||
)
|
||
db.delete(extra)
|
||
db.flush()
|
||
return survivor
|
||
|
||
|
||
def consolidate_duplicate_lots(db: Session) -> int:
|
||
"""Alle bereits vorhandenen Dubletten einmalig zusammenfassen (beim Start).
|
||
Idempotent: Nach dem ersten Lauf gibt es keine Gruppen mit mehr als einer
|
||
Charge mehr. Gibt die Anzahl entfernter Chargen zurück."""
|
||
gruppen = (
|
||
db.query(
|
||
Lot.product_id, Lot.best_before, Lot.best_before_precision, Lot.location_id
|
||
)
|
||
.group_by(
|
||
Lot.product_id, Lot.best_before, Lot.best_before_precision, Lot.location_id
|
||
)
|
||
.having(func.count(Lot.id) > 1)
|
||
.all()
|
||
)
|
||
entfernt = 0
|
||
for product_id, best_before, precision, location_id in gruppen:
|
||
vorher = (
|
||
db.query(func.count(Lot.id))
|
||
.filter(
|
||
Lot.product_id == product_id,
|
||
Lot.best_before_precision == precision,
|
||
Lot.best_before.is_(None) if best_before is None else Lot.best_before == best_before,
|
||
Lot.location_id.is_(None) if location_id is None else Lot.location_id == location_id,
|
||
)
|
||
.scalar()
|
||
)
|
||
consolidate_lot_group(db, product_id, best_before, precision, location_id)
|
||
entfernt += max(0, (vorher or 0) - 1)
|
||
if entfernt:
|
||
db.commit()
|
||
return entfernt
|
||
|
||
|
||
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
|