Backend: verschachtelte Ort-Mindestbestaende hierarchisch verrechnen

shopping_list_by_location zaehlte Ober- und Unterort unabhaengig, obwohl der
Ober-Subtree den Unterort schon enthaelt - Bedarf wurde doppelt gemeldet. Jetzt
werden Bedarfe je Produkt/Gruppe von unten nach oben verrechnet (_netted_topups):
was in einen Unterort gekauft wird, deckt den Oberort mit. Im Mehl-Beispiel (Lemgo
5, Kueche 2, je 1 fehlend) meldet der Server nur noch 1 (Kueche) statt 2.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Scarriffle
2026-07-30 09:32:45 +02:00
parent 39803dfb82
commit f78ae924de
2 changed files with 110 additions and 21 deletions

View File

@@ -1,5 +1,6 @@
from collections import defaultdict
from datetime import date, timedelta
from typing import Callable
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
@@ -109,47 +110,87 @@ def group_shopping_list(
return items
def _netted_topups(
db: Session, locs_min: dict[str, float], stock_of: Callable[[str], float]
) -> dict[str, float]:
"""Bedarf je Ort mit verschachtelten Orten verrechnet: Was in einen Unterort
gekauft wird, liegt auch im Subtree des Oberorts und deckt dessen Bedarf mit.
``topup(ort)`` ist die je Ort ZUSÄTZLICH nötige Menge über die Käufe in den
Unterorten hinaus. So kostet „Lemgo braucht 5, Küche braucht 2" bei je 1 fehlend
nur 1 (in die Küche), nicht 2."""
locs = list(locs_min)
# Nachkommen-Bedarfsorte je Ort (im Lagerort-Baum), memoisiert von unten nach oben.
desc = {
loc: [d for d in locs if d != loc and d in descendant_location_ids(db, loc)]
for loc in locs
}
memo: dict[str, float] = {}
def topup(loc: str) -> float:
if loc not in memo:
committed = sum(topup(d) for d in desc[loc])
memo[loc] = max(0.0, locs_min[loc] - (stock_of(loc) + committed))
return memo[loc]
return {loc: topup(loc) for loc in locs}
@router.get("/shopping-list/by-location", response_model=list[LocationNeeds])
def shopping_list_by_location(
db: Session = Depends(get_db), _: User = Depends(get_current_user)
) -> list[LocationNeeds]:
"""Bedarfe je Lagerort: Produkte und Gruppen, deren Bestand AN DIESEM ORT
unter dem dort hinterlegten Mindestbestand liegt."""
prod_needs: dict[int, list[LocationNeedProduct]] = defaultdict(list)
group_needs: dict[int, list[LocationNeedGroup]] = defaultdict(list)
prod_needs: dict[str, list[LocationNeedProduct]] = defaultdict(list)
group_needs: dict[str, list[LocationNeedGroup]] = defaultdict(list)
# Je Produkt alle Ort-Mindestbestände sammeln und hierarchisch verrechnen.
prod_by_id: dict[int, list[ProductLocationMinStock]] = defaultdict(list)
for e in db.query(ProductLocationMinStock).all():
product = db.get(Product, e.product_id)
prod_by_id[e.product_id].append(e)
for product_id, entries in prod_by_id.items():
product = db.get(Product, product_id)
if product is None:
continue
faktor, label = article_unit(product)
stock = location_subtree_stock_base(db, product, e.location_id) / (faktor or 1.0)
if stock < e.min_stock:
prod_needs[e.location_id].append(LocationNeedProduct(
product_id=product.id, name=product.name, unit_label=label,
stock=round(stock, 3), min_stock=e.min_stock,
deficit=round(e.min_stock - stock, 3),
))
faktor = faktor or 1.0
locs_min = {e.location_id: e.min_stock for e in entries}
bestand = {loc: location_subtree_stock_base(db, product, loc) / faktor for loc in locs_min}
for loc, need in _netted_topups(db, locs_min, bestand.__getitem__).items():
if need > 1e-9:
prod_needs[loc].append(LocationNeedProduct(
product_id=product.id, name=product.name, unit_label=label,
stock=round(bestand[loc], 3), min_stock=locs_min[loc],
deficit=round(need, 3),
))
# Je Gruppe genauso Bestand je Ort ist die Summe der passenden Produkte im Subtree.
group_by_id: dict[int, list[GroupLocationMinStock]] = defaultdict(list)
for e in db.query(GroupLocationMinStock).all():
group = db.get(Group, e.group_id)
group_by_id[e.group_id].append(e)
for group_id, entries in group_by_id.items():
group = db.get(Group, group_id)
if group is None:
continue
unit = group.min_stock_unit
if unit is not None:
base = BASE_OF_KIND[unit.kind]
matching = [p for p in group.products if p.base_unit == base]
stock = sum(location_subtree_stock_base(db, p, e.location_id) for p in matching) / unit.factor
unit_name = unit.name
divisor, unit_name = unit.factor, unit.name
else:
stock = float(sum(location_subtree_stock_base(db, p, e.location_id) for p in group.products))
unit_name = ""
if stock < e.min_stock:
group_needs[e.location_id].append(LocationNeedGroup(
group_id=group.id, name=group.name, unit_name=unit_name,
stock=round(stock, 3), min_stock=e.min_stock,
deficit=round(e.min_stock - stock, 3),
))
matching, divisor, unit_name = list(group.products), 1.0, ""
locs_min = {e.location_id: e.min_stock for e in entries}
bestand = {
loc: sum(location_subtree_stock_base(db, p, loc) for p in matching) / divisor
for loc in locs_min
}
for loc, need in _netted_topups(db, locs_min, bestand.__getitem__).items():
if need > 1e-9:
group_needs[loc].append(LocationNeedGroup(
group_id=group.id, name=group.name, unit_name=unit_name,
stock=round(bestand[loc], 3), min_stock=locs_min[loc],
deficit=round(need, 3),
))
loc_ids = set(prod_needs) | set(group_needs)
namen = {