f78ae924de
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>
347 lines
13 KiB
Python
347 lines
13 KiB
Python
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,
|
||
Product,
|
||
ProductLocationMinStock,
|
||
User,
|
||
)
|
||
from ..schemas import (
|
||
ExpiringItem,
|
||
GroupShoppingItem,
|
||
LocationContentEntry,
|
||
LocationContents,
|
||
LocationNeedGroup,
|
||
LocationNeedProduct,
|
||
LocationNeeds,
|
||
MovementOut,
|
||
ShoppingItem,
|
||
ShoppingListAll,
|
||
)
|
||
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"])
|
||
|
||
|
||
@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."""
|
||
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:
|
||
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,
|
||
)
|
||
)
|
||
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.
|
||
"""
|
||
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]
|
||
stock = float(sum(current_stock(db, p.id) for p in products)) / unit.factor
|
||
unit_name = unit.name
|
||
else:
|
||
products = list(group.products)
|
||
stock = float(sum(current_stock(db, p.id) for p in products))
|
||
unit_name = ""
|
||
if stock < group.min_stock:
|
||
items.append(
|
||
GroupShoppingItem(
|
||
group_id=group.id,
|
||
name=group.name,
|
||
stock=stock,
|
||
min_stock=group.min_stock,
|
||
deficit=group.min_stock - stock,
|
||
unit_name=unit_name,
|
||
product_count=len(products),
|
||
)
|
||
)
|
||
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."""
|
||
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
|
||
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_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
|
||
else:
|
||
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 = {
|
||
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
|