Scarriffle/Vorrania
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Gruppen: Obergruppen + Mindestbestaende nur noch je Lagerort (Backend)

Browse Source 
Zwei zusammenhaengende Umbauten, weil sie dieselben Stellen betreffen.

Obergruppen: Gruppen bilden jetzt einen gerichteten azyklischen Graphen statt
einer flachen Liste. Eine Gruppe darf unter MEHREREN Obergruppen haengen -
"Grillwurst" unter "Wurst" UND unter "Grillgut"; mit einem einzelnen parent_id
waere genau das nicht abbildbar. Bestand und Mindestbestand einer Gruppe zaehlen
den gesamten Untergraphen, wobei eine ueber zwei Wege erreichbare Untergruppe
nur einmal zaehlt (services/gruppen.py arbeitet durchgaengig mit Mengen).
Product.group_id bleibt unveraendert - ein Artikel haengt weiter an genau einer
Gruppe.

Mindestbestaende: der separate Gesamt-Mindestbestand entfaellt. Er wird zur
Zeile mit location_id NULL ("Ueberall") und ist damit die Wurzel ueber allen
Lagerorten - dieselbe Verrechnung wie bei verschachtelten Orten greift jetzt
auch zwischen Ueberall und Kueche, wodurch derselbe Artikel nicht mehr doppelt
in der Einkaufsliste steht. Alle Werte liegen einheitlich in Basiseinheiten
statt in drei verschiedenen Einheiten nebeneinander; das Umrechnen beim
Umschalten der Erfassungseinheit entfaellt dadurch ersatzlos.

_netted_topups nimmt die Hierarchie jetzt als Parameter und faltet damit
Lagerort-Baum und Gruppen-Graph. Verrechnet wird zwischen zwei Gruppen nur,
wenn die zaehlenden Artikel der Untergruppe eine Teilmenge der Obergruppe sind -
zaehlt die Obergruppe in Kilogramm und die Untergruppe in Stueck, kommt ein Kauf
dort oben nicht an.

Die vierfach kopierte Bestandssumme wandert in Sammelabfragen
(summe_bestand_base), sonst vervielfacht der transitive Teilgraph die Abfragen.

Einmalige Datenwanderung beim Start (Merker in den Einstellungen), 18 neue
Tests - darunter Doppelzaehlung ueber zwei Wege, Ringschutz und die bewusst
offene Grenze bei zwei Obergruppen mit gemeinsamer Untergruppe.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Scarriffle
2026-08-15 21:48:14 +02:00
parent df65d9583c
commit eaacfd03e5
16 changed files with 1327 additions and 238 deletions
Download Patch File Download Diff File Expand all files Collapse all files

446
backend/app/routers/views.py
View File

@@ -1,7 +1,8 @@
import math
from collections import defaultdict
from collections.abc import Iterable
from datetime import date, timedelta
from typing import Callable
from typing import Callable, TypeVar
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
@@ -40,15 +41,22 @@ from ..services.conversion import (
display_unit_info,
group_min_context,
)
from ..services import gruppen as gruppen_graph
from ..services.stock import (
current_stock,
descendant_location_ids,
location_subtree_stock_base,
summe_bestand_base,
summe_bestand_im_subtree_base,
)
from .settings import get_expiry_warning_days
router = APIRouter(tags=["views"])
#: Schluesseltyp der Bedarfs-Verrechnung: ein Lagerort (bzw. None = „Ueberall")
#: oder ein Paar aus Gruppe und Ort.
K = TypeVar("K")
# ---- Bedarf lesbar aufbereiten (Gebinde als Leitangabe) --------------------
_UNIT_SHORT: dict[BaseUnit, str] = {
@@ -109,43 +117,187 @@ def _build_need(
)
def _artikel_bedarfe(
db: Session, ort_desc: dict[str, set[str]]
) -> dict[int, tuple[Product, dict, dict, dict]]:
"""Je Artikel Mindestbestand, Bestand und verrechneter Bedarf – je Ort.
Schluessel ``None`` ist „Ueberall" und liegt ueber allen Lagerorten. Alle
Mengen in Basiseinheiten.
"""
nach_artikel: dict[int, list[ProductLocationMinStock]] = defaultdict(list)
for e in db.query(ProductLocationMinStock).all():
if e.min_stock > 0:
nach_artikel[e.product_id].append(e)
ergebnis: dict[int, tuple[Product, dict, dict, dict]] = {}
for product_id, rows in nach_artikel.items():
product = db.get(Product, product_id)
if product is None:
continue
mins = {e.location_id: e.min_stock for e in rows}
bestand = {
loc: (
current_stock(db, product.id)
if loc is None
else location_subtree_stock_base(db, product, loc)
)
for loc in mins
}
needs = _netted_topups(mins, _ort_nachfahren(mins, ort_desc), bestand.__getitem__)
ergebnis[product_id] = (product, mins, bestand, needs)
return ergebnis
def einkaufsliste_artikel(db: Session) -> list[ShoppingItem]:
"""Artikel, bei denen „Überall" etwas fehlt – egal wo im Haus.
Käufe für einzelne Lagerorte sind bereits abgezogen: derselbe Artikel steht
dadurch nicht mehr doppelt in der Liste (einmal „Gesamt", einmal je Ort).
"""
plural = _package_plural(db)
ort_desc = _ort_nachfahren_tabelle(db)
items: list[ShoppingItem] = []
for product, mins, bestand, needs in _artikel_bedarfe(db, ort_desc).values():
if None not in mins:
continue # nur Ort-Bedarfe, kein „Überall"
fehlt = needs.get(None, 0.0)
if fehlt <= 1e-9:
continue
factor, singular = article_unit(product)
items.append(
ShoppingItem(
product_id=product.id,
name=product.name,
base_unit=product.base_unit,
package_size=product.package_size,
stock=round(bestand[None], 3),
min_stock=mins[None],
deficit=round(fehlt, 3),
need=_build_need(
deficit_base=fehlt,
stock_base=bestand[None],
min_base=mins[None],
factor=factor,
singular=singular,
is_package=bool(product.package_size and product.package_size > 0),
base_unit=product.base_unit,
plural=plural,
),
)
)
items.sort(key=lambda i: i.deficit, reverse=True)
return items
@router.get("/shopping-list", response_model=list[ShoppingItem])
def shopping_list(
db: Session = Depends(get_db), _: User = Depends(get_current_user)
) -> list[ShoppingItem]:
"""Produkte, deren Bestand unter dem Mindestbestand liegt."""
return einkaufsliste_artikel(db)
def _gruppen_bedarfe(db: Session, ort_desc: dict[str, set[str]]) -> tuple[dict, dict, dict, dict, dict]:
"""Je (Gruppe, Ort) Mindestbestand, Bestand und verrechneter Bedarf.
Hier wirken ZWEI Hierarchien zusammen: der Gruppen-Graph und der
Lagerort-Baum. „1 kg Grillwurst in die Küche" deckt auch „Wurst in Lemgo".
Deshalb ist ein Schlüssel ein Paar, und ``(h, m)`` gilt als Nachfahre von
``(g, l)``, wenn h unter g und m unter l liegt (oder gleich ist).
Verrechnet wird nur, wo die zählenden Artikel der Untergruppe eine Teilmenge
der Obergruppe sind: zählt „Wurst" in Kilogramm, ihre Untergruppe aber in
Stück, kommt ein Kauf dort oben gar nicht an (siehe die Einheiten-Filterung
in ``group_min_context``).
Alle Mengen in Basiseinheiten – nur darin lassen sich Gruppen mit
verschiedenen Erfassungseinheiten überhaupt gegeneinander verrechnen.
"""
eintraege = [e for e in db.query(GroupLocationMinStock).all() if e.min_stock > 0]
gruppen: dict[int, Group] = {}
for e in eintraege:
if e.group_id not in gruppen:
g = db.get(Group, e.group_id)
if g is not None:
gruppen[e.group_id] = g
ctxs = {gid: group_min_context(g) for gid, g in gruppen.items()}
artikel = {gid: {p.id for p in c.matching} for gid, c in ctxs.items()}
gruppen_desc = {
gid: {
h.id
for h in gruppen_graph.teilgraph(g)
if h.id != gid and h.id in artikel and artikel[h.id] <= artikel[gid]
}
for gid, g in gruppen.items()
}
minima: dict[tuple[int, str | None], float] = {}
bestand: dict[tuple[int, str | None], float] = {}
for e in eintraege:
ctx = ctxs.get(e.group_id)
if ctx is None:
continue
schluessel = (e.group_id, e.location_id)
minima[schluessel] = e.min_stock
bestand[schluessel] = (
summe_bestand_base(db, ctx.matching)
if e.location_id is None
else summe_bestand_im_subtree_base(db, ctx.matching, e.location_id)
)
alle_orte = {loc for (_, loc) in minima if loc is not None}
def nachfahren(k: tuple[int, str | None]) -> set[tuple[int, str | None]]:
gid, loc = k
unten = gruppen_desc.get(gid, set()) | {gid}
# „Überall" (None) liegt über allen Lagerorten.
orte = (alle_orte if loc is None else ort_desc.get(loc, set())) | {loc}
return {(h, m) for h in unten for m in orte if (h, m) in minima} - {k}
needs = _netted_topups(minima, nachfahren, bestand.__getitem__)
return gruppen, ctxs, minima, bestand, needs
def einkaufsliste_gruppen(db: Session) -> list[GroupShoppingItem]:
"""Gruppen, bei denen „Überall" etwas fehlt.
Gruppen-Bestand = Summe der Artikelbestände der Gruppe UND ihrer
Untergruppen (Basiseinheiten). Käufe für Untergruppen und für einzelne
Lagerorte sind bereits abgezogen.
"""
plural = _package_plural(db)
items: list[ShoppingItem] = []
products = (
db.query(Product)
.filter(Product.min_stock.isnot(None), Product.min_stock > 0)
.all()
)
for product in products:
stock = current_stock(db, product.id)
if stock < product.min_stock:
factor, singular = article_unit(product)
items.append(
ShoppingItem(
product_id=product.id,
name=product.name,
base_unit=product.base_unit,
package_size=product.package_size,
stock=stock,
min_stock=product.min_stock,
deficit=product.min_stock - stock,
need=_build_need(
deficit_base=product.min_stock - stock,
stock_base=stock,
min_base=product.min_stock,
factor=factor,
singular=singular,
is_package=bool(product.package_size and product.package_size > 0),
base_unit=product.base_unit,
plural=plural,
),
)
ort_desc = _ort_nachfahren_tabelle(db)
gruppen, ctxs, minima, bestand, needs = _gruppen_bedarfe(db, ort_desc)
items: list[GroupShoppingItem] = []
for (gid, loc), fehlt in needs.items():
if loc is not None or fehlt <= 1e-9:
continue
ctx = ctxs[gid]
schluessel = (gid, None)
items.append(
GroupShoppingItem(
group_id=gid,
name=gruppen[gid].name,
stock=round(bestand[schluessel] / ctx.divisor, 3),
min_stock=round(minima[schluessel] / ctx.divisor, 3),
deficit=round(fehlt / ctx.divisor, 3),
unit_name=ctx.label,
product_count=len(ctx.matching),
subgroup_count=len(gruppen_graph.nachfahren_ids(gruppen[gid])),
need=_build_need(
deficit_base=fehlt,
stock_base=bestand[schluessel],
min_base=minima[schluessel],
factor=ctx.divisor,
singular=ctx.label,
is_package=ctx.is_package,
base_unit=ctx.base_unit,
plural=plural,
),
)
)
items.sort(key=lambda i: i.deficit, reverse=True)
return items
@@ -154,146 +306,136 @@ def shopping_list(
def group_shopping_list(
db: Session = Depends(get_db), _: User = Depends(get_current_user)
) -> list[GroupShoppingItem]:
"""Gruppen, deren Gesamtbestand unter dem Gruppen-Mindestbestand liegt.
Gruppen-Bestand = Summe der Produktbestände in der Gruppe (in Basiseinheiten).
Sinnvoll, wenn die Produkte einer Gruppe dieselbe Basiseinheit teilen.
"""
plural = _package_plural(db)
items: list[GroupShoppingItem] = []
groups = (
db.query(Group).filter(Group.min_stock.isnot(None), Group.min_stock > 0).all()
)
for group in groups:
ctx = group_min_context(group)
stock_base = float(sum(current_stock(db, p.id) for p in ctx.matching))
min_base = group.min_stock * ctx.divisor
if stock_base < min_base:
deficit_base = min_base - stock_base
items.append(
GroupShoppingItem(
group_id=group.id,
name=group.name,
stock=round(stock_base / ctx.divisor, 3),
min_stock=group.min_stock,
deficit=round(deficit_base / ctx.divisor, 3),
unit_name=ctx.label,
product_count=len(ctx.matching),
need=_build_need(
deficit_base=deficit_base,
stock_base=stock_base,
min_base=min_base,
factor=ctx.divisor,
singular=ctx.label,
is_package=ctx.is_package,
base_unit=ctx.base_unit,
plural=plural,
),
)
)
items.sort(key=lambda i: i.deficit, reverse=True)
return items
return einkaufsliste_gruppen(db)
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] = {}
minima: dict[K, float],
nachfahren: Callable[[K], set[K]],
stock_of: Callable[[K], float],
) -> dict[K, float]:
"""Bedarfe entlang ihrer Hierarchie verrechnen.
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]
Was fuer einen Nachfahren gekauft wird, liegt auch bei dessen Vorfahren und
deckt deren Bedarf mit. ``topup(x)`` ist die ZUSAETZLICH noetige Menge –
ueber die Kaeufe fuer die Nachfahren hinaus. So kostet „Lemgo braucht 5,
Kueche braucht 2" bei je 1 fehlend nur 1 (in die Kueche), nicht 2.
return {loc: topup(loc) for loc in locs}
``nachfahren`` liefert die TRANSITIVE Nachfahren-MENGE, nicht die direkten
Kinder. Beim Gruppen-Graphen ist eine Untergruppe ueber mehrere Wege
erreichbar – als Menge zaehlt sie in ``committed`` trotzdem nur einmal.
Alle Mengen muessen in DERSELBEN Einheit vorliegen (hier: Basiseinheiten),
sonst wird Aepfel mit Birnen verrechnet.
"""
schluessel = list(minima)
desc = {k: [d for d in schluessel if d != k and d in nachfahren(k)] for k in schluessel}
memo: dict[K, float] = {}
def topup(k: K) -> float:
if k not in memo:
# Vorbelegen schuetzt vor einem Ring in den Daten: der wuerde sonst
# endlos rekursieren (die API laesst keinen zu, ein Import schon).
memo[k] = 0.0
committed = sum(topup(d) for d in desc[k])
memo[k] = max(0.0, minima[k] - (stock_of(k) + committed))
return memo[k]
return {k: topup(k) for k in schluessel}
def _ort_nachfahren(
schluessel: Iterable[str | None], ort_desc: dict[str, set[str]]
) -> Callable[[str | None], set[str | None]]:
"""Nachfahren-Funktion fuer die Ort-Hierarchie eines Bedarfssatzes.
„Ueberall" (``None``) liegt ueber ALLEN Lagerorten. Dadurch verrechnet
dieselbe Faltung auch Ueberall gegen die einzelnen Orte – der frueher
getrennte Gesamt-Mindestbestand erzeugt keinen zweiten Eintrag mehr.
"""
orte = list(schluessel)
def nachfahren(k: str | None) -> set[str | None]:
if k is None:
return {d for d in orte if d is not None}
return {d for d in orte if d is not None and d in ort_desc.get(k, set())}
return nachfahren
def _ort_nachfahren_tabelle(db: Session) -> dict[str, set[str]]:
"""Unterorte je Lagerort, einmal je Anfrage statt je Artikel und Gruppe.
``descendant_location_ids`` geht je Ebene an die Datenbank; frueher wurde es
fuer jeden Bedarfssatz neu aufgerufen.
"""
return {loc.id: descendant_location_ids(db, loc.id) for loc in db.query(Location).all()}
@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."""
"""Bedarfe je Lagerort: Artikel und Gruppen, denen AN DIESEM ORT etwas fehlt.
„Überall" gehört nicht hierher – das liefern ``/shopping-list`` und
``/shopping-list/groups``. Verrechnet wird über beide Listen hinweg, ein
Artikel steht also nur einmal drin.
"""
plural = _package_plural(db)
ort_desc = _ort_nachfahren_tabelle(db)
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():
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
for product, mins, bestand, needs in _artikel_bedarfe(db, ort_desc).values():
faktor, label = article_unit(product)
faktor = faktor or 1.0
ist_gebinde = bool(product.package_size and product.package_size > 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),
# Mengen liegen hier in Artikeleinheiten -> * faktor = Basiseinheiten.
need=_build_need(
deficit_base=need * faktor,
stock_base=bestand[loc] * faktor,
min_base=locs_min[loc] * faktor,
factor=faktor,
singular=label,
is_package=ist_gebinde,
base_unit=product.base_unit,
plural=plural,
),
))
for loc, need in needs.items():
if loc is None or need <= 1e-9:
continue
prod_needs[loc].append(LocationNeedProduct(
product_id=product.id, name=product.name, unit_label=label,
# Nach aussen weiterhin in Artikeleinheiten – gespeichert und
# gerechnet wird intern in Basiseinheiten.
stock=round(bestand[loc] / faktor, 3),
min_stock=round(mins[loc] / faktor, 3),
deficit=round(need / faktor, 3),
need=_build_need(
deficit_base=need,
stock_base=bestand[loc],
min_base=mins[loc],
factor=faktor,
singular=label,
is_package=ist_gebinde,
base_unit=product.base_unit,
plural=plural,
),
))
# 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_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:
gruppen, ctxs, minima, bestand_g, needs_g = _gruppen_bedarfe(db, ort_desc)
for (gid, loc), need in needs_g.items():
if loc is None or need <= 1e-9:
continue
ctx = group_min_context(group)
# Mengen in der Mindestbestand-Einheit (Gebinde ODER verwaltete Einheit).
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 ctx.matching) / ctx.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=ctx.label,
stock=round(bestand[loc], 3), min_stock=locs_min[loc],
deficit=round(need, 3),
need=_build_need(
deficit_base=need * ctx.divisor,
stock_base=bestand[loc] * ctx.divisor,
min_base=locs_min[loc] * ctx.divisor,
factor=ctx.divisor,
singular=ctx.label,
is_package=ctx.is_package,
base_unit=ctx.base_unit,
plural=plural,
),
))
ctx = ctxs[gid]
schluessel = (gid, loc)
group_needs[loc].append(LocationNeedGroup(
group_id=gid, name=gruppen[gid].name, unit_name=ctx.label,
stock=round(bestand_g[schluessel] / ctx.divisor, 3),
min_stock=round(minima[schluessel] / ctx.divisor, 3),
deficit=round(need / ctx.divisor, 3),
subgroup_count=len(gruppen_graph.nachfahren_ids(gruppen[gid])),
need=_build_need(
deficit_base=need,
stock_base=bestand_g[schluessel],
min_base=minima[schluessel],
factor=ctx.divisor,
singular=ctx.label,
is_package=ctx.is_package,
base_unit=ctx.base_unit,
plural=plural,
),
))
loc_ids = set(prod_needs) | set(group_needs)
namen = {