f2c657ff31
Statt 'fehlt 500 Gramm' jetzt die Produktgroesse als Leitangabe:
'fehlt 3 Glaeser (500 g)'. Auf ganze Gebinde aufgerundet, weil man nur
ganze Glaeser/Dosen kauft; Basiseinheit als kleiner Hinweis.
Backend: neues ShoppingNeed-Objekt an jeder Einkaufslisten-Zeile (Produkte
+ Gruppen, gesamt + je Ort) - damit die Angabe auch direkt ueber die API
kommt ('x Glaeser (y g)'), inkl. Pluralisierung aus der Gebinde-Tabelle.
Gruppen leiten das Gebinde ab, wenn alle passenden Produkte dasselbe haben,
sonst bleibt es bei der Basiseinheit. Bestehende Felder unveraendert.
Frontend: gemeinsame Komponente ShoppingNeedText fuer Karte und volle Seite.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
511 lines
19 KiB
Python
511 lines
19 KiB
Python
import math
|
||
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
|
||
|
||
from ..database import get_db
|
||
from ..deps import get_current_user
|
||
from ..models import (
|
||
Group,
|
||
GroupLocationMinStock,
|
||
Item,
|
||
Location,
|
||
Lot,
|
||
Movement,
|
||
PackageType,
|
||
Product,
|
||
ProductLocationMinStock,
|
||
User,
|
||
)
|
||
from ..schemas import (
|
||
BaseUnit,
|
||
ExpiringItem,
|
||
GroupShoppingItem,
|
||
LocationContentEntry,
|
||
LocationContents,
|
||
LocationNeedGroup,
|
||
LocationNeedProduct,
|
||
LocationNeeds,
|
||
MovementOut,
|
||
ShoppingItem,
|
||
ShoppingListAll,
|
||
ShoppingNeed,
|
||
)
|
||
from ..services.conversion import BASE_OF_KIND, article_unit, display_unit_info
|
||
from ..services.stock import (
|
||
current_stock,
|
||
descendant_location_ids,
|
||
location_subtree_stock_base,
|
||
)
|
||
from .settings import get_expiry_warning_days
|
||
|
||
router = APIRouter(tags=["views"])
|
||
|
||
|
||
# ---- Bedarf lesbar aufbereiten (Gebinde als Leitangabe) --------------------
|
||
_UNIT_SHORT: dict[BaseUnit, str] = {
|
||
BaseUnit.piece: "Stk",
|
||
BaseUnit.gram: "g",
|
||
BaseUnit.milliliter: "ml",
|
||
}
|
||
|
||
|
||
def _de_num(x: float) -> str:
|
||
"""Deutsche Kurzzahl ohne unnoetige Nullen: 500, 1, 1,4."""
|
||
r = round(float(x), 2)
|
||
if r == int(r):
|
||
return str(int(r))
|
||
return f"{r:.2f}".rstrip("0").rstrip(".").replace(".", ",")
|
||
|
||
|
||
def _package_plural(db: Session) -> dict[str, str]:
|
||
"""Einzahl -> Mehrzahl der Gebinde (Glas -> Gläser), aus der Gebinde-Tabelle."""
|
||
return {pt.singular: pt.plural for pt in db.query(PackageType).all()}
|
||
|
||
|
||
def _uniform_package(products: list[Product]) -> tuple[float, str] | None:
|
||
"""Gemeinsames Gebinde einer Produktmenge – nur wenn *alle* dasselbe haben.
|
||
|
||
Für eine Gruppe lässt sich „x Gläser" nur dann eindeutig sagen, wenn jedes
|
||
passende Produkt dieselbe Packungsgröße und -bezeichnung trägt. Sonst bleibt
|
||
es bei der Basiseinheit."""
|
||
if not products:
|
||
return None
|
||
combos = {(p.package_size, p.package_label or "Packung") for p in products}
|
||
if len(combos) == 1:
|
||
size, label = next(iter(combos))
|
||
if size and size > 0:
|
||
return float(size), label
|
||
return None
|
||
|
||
|
||
def _build_need(
|
||
*,
|
||
deficit_base: float,
|
||
stock_base: float,
|
||
min_base: float,
|
||
factor: float,
|
||
singular: str,
|
||
is_package: bool,
|
||
base_unit: BaseUnit | None,
|
||
plural: dict[str, str],
|
||
) -> ShoppingNeed:
|
||
"""Rechnet Basiseinheiten in Gebinde/Artikeleinheiten um und baut die Texte.
|
||
|
||
``factor`` = Basiseinheiten je Gebinde (Packungsgröße bzw. Einheitenfaktor).
|
||
Bei zählbaren Packungen wird die Kaufmenge auf ganze Gebinde aufgerundet."""
|
||
f = factor or 1.0
|
||
count = math.ceil(deficit_base / f - 1e-9) if is_package else round(deficit_base / f, 3)
|
||
label = singular if abs(count) == 1 else plural.get(singular, singular)
|
||
text = f"{_de_num(count)} {label}".strip()
|
||
hint = ""
|
||
base_amount: float | None = None
|
||
if (is_package or f != 1.0) and base_unit is not None:
|
||
base_amount = round(deficit_base, 3)
|
||
hint = f"{_de_num(deficit_base)} {_UNIT_SHORT.get(base_unit, base_unit.value)}"
|
||
return ShoppingNeed(
|
||
text=text,
|
||
hint=hint,
|
||
count=count,
|
||
label=label,
|
||
singular=singular,
|
||
is_package=is_package,
|
||
base_amount=base_amount,
|
||
base_unit=base_unit if base_amount is not None else None,
|
||
stock=round(stock_base / f, 3),
|
||
min_stock=round(min_base / f, 3),
|
||
)
|
||
|
||
|
||
def _gruppen_bedarf(
|
||
*,
|
||
deficit_unit: float,
|
||
stock_unit: float,
|
||
min_unit: float,
|
||
divisor: float,
|
||
unit_name: str,
|
||
base_unit: BaseUnit | None,
|
||
products: list[Product],
|
||
plural: dict[str, str],
|
||
) -> ShoppingNeed:
|
||
"""Bedarf einer Gruppe. Mengen kommen in der Gruppen-Einheit (``unit_name``);
|
||
``divisor`` rechnet sie in Basiseinheiten zurück. Haben alle passenden
|
||
Produkte dasselbe Gebinde, wird als Leitangabe dieses Gebinde (z.B. Gläser)
|
||
genutzt, sonst die Gruppen-Einheit selbst."""
|
||
pkg = _uniform_package(products) if base_unit is not None else None
|
||
if pkg is not None:
|
||
factor, singular, is_package = pkg[0], pkg[1], True
|
||
else:
|
||
factor, singular, is_package = (divisor or 1.0), unit_name, False
|
||
return _build_need(
|
||
deficit_base=deficit_unit * divisor,
|
||
stock_base=stock_unit * divisor,
|
||
min_base=min_unit * divisor,
|
||
factor=factor,
|
||
singular=singular,
|
||
is_package=is_package,
|
||
base_unit=base_unit,
|
||
plural=plural,
|
||
)
|
||
|
||
|
||
@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."""
|
||
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,
|
||
),
|
||
)
|
||
)
|
||
items.sort(key=lambda i: i.deficit, reverse=True)
|
||
return items
|
||
|
||
|
||
@router.get("/shopping-list/groups", response_model=list[GroupShoppingItem])
|
||
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:
|
||
unit = group.min_stock_unit
|
||
if unit is not None:
|
||
base = BASE_OF_KIND[unit.kind]
|
||
products = [p for p in group.products if p.base_unit == base]
|
||
divisor = unit.factor
|
||
stock = float(sum(current_stock(db, p.id) for p in products)) / divisor
|
||
unit_name = unit.name
|
||
base_unit: BaseUnit | None = base
|
||
else:
|
||
products = list(group.products)
|
||
divisor = 1.0
|
||
stock = float(sum(current_stock(db, p.id) for p in products))
|
||
unit_name = ""
|
||
base_unit = None
|
||
if stock < group.min_stock:
|
||
deficit = group.min_stock - stock
|
||
items.append(
|
||
GroupShoppingItem(
|
||
group_id=group.id,
|
||
name=group.name,
|
||
stock=stock,
|
||
min_stock=group.min_stock,
|
||
deficit=deficit,
|
||
unit_name=unit_name,
|
||
product_count=len(products),
|
||
need=_gruppen_bedarf(
|
||
deficit_unit=deficit,
|
||
stock_unit=stock,
|
||
min_unit=group.min_stock,
|
||
divisor=divisor,
|
||
unit_name=unit_name,
|
||
base_unit=base_unit,
|
||
products=products,
|
||
plural=plural,
|
||
),
|
||
)
|
||
)
|
||
items.sort(key=lambda i: i.deficit, reverse=True)
|
||
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."""
|
||
plural = _package_plural(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
|
||
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,
|
||
),
|
||
))
|
||
|
||
# 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:
|
||
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]
|
||
divisor, unit_name = unit.factor, unit.name
|
||
g_base_unit: BaseUnit | None = base
|
||
else:
|
||
matching, divisor, unit_name = list(group.products), 1.0, ""
|
||
g_base_unit = None
|
||
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),
|
||
need=_gruppen_bedarf(
|
||
deficit_unit=need,
|
||
stock_unit=bestand[loc],
|
||
min_unit=locs_min[loc],
|
||
divisor=divisor,
|
||
unit_name=unit_name,
|
||
base_unit=g_base_unit,
|
||
products=matching,
|
||
plural=plural,
|
||
),
|
||
))
|
||
|
||
loc_ids = set(prod_needs) | set(group_needs)
|
||
namen = {
|
||
loc.id: loc.name
|
||
for loc in db.query(Location).filter(Location.id.in_(loc_ids)).all()
|
||
}
|
||
result: list[LocationNeeds] = []
|
||
for loc_id in sorted(loc_ids, key=lambda i: namen.get(i, "")):
|
||
result.append(LocationNeeds(
|
||
location_id=loc_id,
|
||
location_name=namen.get(loc_id, "?"),
|
||
products=sorted(prod_needs.get(loc_id, []), key=lambda x: x.deficit, reverse=True),
|
||
groups=sorted(group_needs.get(loc_id, []), key=lambda x: x.deficit, reverse=True),
|
||
))
|
||
return result
|
||
|
||
|
||
@router.get("/shopping-list/all", response_model=ShoppingListAll)
|
||
def shopping_list_all(
|
||
db: Session = Depends(get_db), user: User = Depends(get_current_user)
|
||
) -> ShoppingListAll:
|
||
"""Komplette Einkaufsliste in einem Aufruf: Produkte, Gruppen und Bedarfe je
|
||
Lagerort – genau die drei Quellen, die auch die Oberfläche zusammenführt.
|
||
Erspart drei getrennte Abfragen."""
|
||
return ShoppingListAll(
|
||
products=shopping_list(db, user),
|
||
groups=group_shopping_list(db, user),
|
||
by_location=shopping_list_by_location(db, user),
|
||
)
|
||
|
||
|
||
@router.get("/location/{code}", response_model=LocationContents)
|
||
def location_contents(
|
||
code: str,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(get_current_user),
|
||
) -> LocationContents:
|
||
"""Alle Artikel, die an einem Lagerort liegen – inklusive der Unterorte.
|
||
|
||
Ziel des Lagerort-QR (`/l/<code>`) beim Nachschlagen: einmal scannen und
|
||
sehen, was im Regal/Fach (und allem darunter) steht. Zählt sowohl
|
||
Lot-Bestände (Lebensmittel/Objekte) als auch Einzelstücke.
|
||
"""
|
||
loc = db.get(Location, code)
|
||
if loc is None:
|
||
raise HTTPException(status.HTTP_404_NOT_FOUND, "Lagerort nicht gefunden")
|
||
|
||
ids = {code} | descendant_location_ids(db, code)
|
||
product_ids: set[int] = {
|
||
pid for (pid,) in db.query(Lot.product_id).filter(Lot.location_id.in_(ids)).distinct()
|
||
}
|
||
product_ids |= {
|
||
pid for (pid,) in db.query(Item.product_id).filter(Item.location_id.in_(ids)).distinct()
|
||
}
|
||
|
||
entries: list[LocationContentEntry] = []
|
||
for pid in product_ids:
|
||
product = db.get(Product, pid)
|
||
if product is None:
|
||
continue
|
||
faktor, label = article_unit(product)
|
||
stock = location_subtree_stock_base(db, product, code) / (faktor or 1.0)
|
||
if stock <= 0:
|
||
continue
|
||
entries.append(LocationContentEntry(
|
||
product_id=product.id, name=product.name, brand=product.brand,
|
||
stock=round(stock, 3), unit_label=label, individual=product.individual,
|
||
))
|
||
entries.sort(key=lambda e: e.name.lower())
|
||
return LocationContents(location_id=code, location_name=loc.name, products=entries)
|
||
|
||
|
||
@router.get("/expiring", response_model=list[ExpiringItem])
|
||
def expiring(
|
||
days: int | None = None,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(get_current_user),
|
||
) -> list[ExpiringItem]:
|
||
"""Chargen, die innerhalb der Warnfrist ablaufen (oder schon abgelaufen sind)."""
|
||
if days is None:
|
||
days = get_expiry_warning_days(db)
|
||
today = date.today()
|
||
threshold = today + timedelta(days=days)
|
||
|
||
lots = (
|
||
db.query(Lot)
|
||
.filter(Lot.best_before.isnot(None), Lot.best_before <= threshold, Lot.quantity > 0)
|
||
.order_by(Lot.best_before)
|
||
.all()
|
||
)
|
||
result: list[ExpiringItem] = []
|
||
for lot in lots:
|
||
product = lot.product
|
||
unit_name, unit_factor = display_unit_info(product)
|
||
result.append(
|
||
ExpiringItem(
|
||
lot_id=lot.id,
|
||
product_id=product.id,
|
||
product_name=product.name,
|
||
quantity=lot.quantity,
|
||
base_unit=product.base_unit,
|
||
best_before=lot.best_before,
|
||
best_before_precision=lot.best_before_precision,
|
||
days_left=(lot.best_before - today).days,
|
||
package_size=product.package_size,
|
||
package_label=product.package_label,
|
||
unit_name=unit_name,
|
||
unit_factor=unit_factor,
|
||
)
|
||
)
|
||
return result
|
||
|
||
|
||
@router.get("/movements", response_model=list[MovementOut])
|
||
def movements(
|
||
limit: int = 100,
|
||
product_id: int | None = None,
|
||
db: Session = Depends(get_db),
|
||
_: User = Depends(get_current_user),
|
||
) -> list[MovementOut]:
|
||
"""Bewegungsverlauf (wer hat wann was ein-/ausgelagert), neueste zuerst."""
|
||
limit = max(1, min(limit, 500))
|
||
query = (
|
||
db.query(Movement, Product, User)
|
||
.join(Product, Movement.product_id == Product.id)
|
||
.outerjoin(User, Movement.user_id == User.id)
|
||
)
|
||
if product_id is not None:
|
||
query = query.filter(Movement.product_id == product_id)
|
||
rows = query.order_by(Movement.created_at.desc(), Movement.id.desc()).limit(limit).all()
|
||
|
||
result: list[MovementOut] = []
|
||
for movement, product, user in rows:
|
||
unit_name, unit_factor = display_unit_info(product)
|
||
result.append(
|
||
MovementOut(
|
||
id=movement.id,
|
||
product_id=product.id,
|
||
product_name=product.name,
|
||
type=movement.type.value,
|
||
quantity=movement.quantity,
|
||
base_unit=product.base_unit,
|
||
unit_used=movement.unit_used,
|
||
username=user.username if user else None,
|
||
note=movement.note,
|
||
created_at=movement.created_at,
|
||
package_size=product.package_size,
|
||
package_label=product.package_label,
|
||
unit_name=unit_name,
|
||
unit_factor=unit_factor,
|
||
)
|
||
)
|
||
return result
|