ScarriffleandClaude Opus 5
fa5a2bc4fa
Eine Teilmenge auszulagern ging rechnerisch schon, aber es blieb keine Spur davon: Eine Charge mit 900 von urspruenglich 1000 ml war von einer frisch eingelagerten 900-ml-Charge nicht zu unterscheiden, und eine geoeffnete Milch galt weiter bis zum aufgedruckten MHD. Anbrechen spaltet jetzt eine Packung ab. Aus 4 x 1 l wird eine offene Charge mit 1 l und eine ungeoeffnete mit 3 l - sonst truegen die restlichen 3 l das kurze Verbrauchsdatum mit, obwohl sie noch zu sind. Der Gesamtbestand aendert sich dabei nicht, es ist eine Umbuchung wie das Aufteilen und schreibt wie dieses keine Bewegung. Verbrauchen geht in einem Schritt: Wer "100 ml verbraucht" eintraegt, bricht die Charge dabei an (abschaltbar ueber auto_open). Gerechnet wird vor dem Anbrechen - scheitert die Einheit, soll keine Packung grundlos offen stehen. MHD und "zu verbrauchen bis" stehen nebeneinander und bleiben beide sichtbar; gewarnt wird nach dem FRUEHEREN (Lot.faellig_am). Dafuer musste in /expiring die Vorbedingung "best_before IS NOT NULL" weg, sonst fehlte ausgerechnet die angebrochene Packung ohne aufgedrucktes MHD. ExpiringItem traegt beide Daten plus warn_date und warn_source, und best_before ist dort jetzt optional. Die Frist kommt aus drei Ebenen: Artikel schlaegt Kategorie, Kategorie schlaegt ueber ihre Oberkategorien hinweg den globalen Wert. Der Helfer dafuer (services/vererbung.py) liest die Kategoriekette von services/fields.py rueckwaerts - dort werden Felder ALLER Vorfahren gesammelt, hier gewinnt der naechste. Dieselbe Mechanik traegt die Hausmasse. Hausmasse sind der zweite Teil: "2 EL Mehl" liess sich bisher nicht buchen, weil eine globale Einheit "Essloeffel" bei Mehl und Zucker zwangslaeufig gleich schwer waere. Der Name ist deshalb global, die Menge haengt an Kategorie und Artikel. to_base loest sie nur mit hausmasse=True auf - beim Auslagern und Verbrauchen ja, beim Einlagern und bei Mindestbestaenden ausdruecklich nicht, denn ein Loeffel ist ein Schaetzwert. Verwaltete Einheiten werden zuerst geprueft, damit ein Mass namens "Liter" die echte Einheit nicht verdeckt. Zwei Stellen waeren sonst still kaputtgegangen: - Die Konsolidierung haette die offene Packung beim naechsten Einlagern am selben Ort einkassiert und ihr Verbrauchsdatum verschluckt. Angebrochene Chargen werden deshalb nie zusammengefasst - in consolidate_lot_group und im Start-Aufraeumer, beide bisher ohne jeden Test. Das Netz dafuer ist jetzt da (test_konsolidierung.py). - FEFO sortiert nach dem wirksamen Datum statt nur nach dem MHD. Eine offene Packung kommt damit meist von selbst zuerst, aber eine ungeoeffnete, die morgen ablaeuft, geht vor - Verderb schlaegt Bequemlichkeit. Erst bei gleichem Datum wird das Angebrochene vorgezogen. Dashboard bekommt "angebrochen" als fuenften Zustand, mit Vorrang abgelaufen > bald > angebrochen > in Ordnung: Was offen ist und morgen ablaeuft, gehoert in die Warnung, nicht in den ruhigeren Zustand. Oberflaechen in Web und App folgen in eigenen Commits. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
626 lines
23 KiB
Python
626 lines
23 KiB
Python
import math
|
||
from collections import defaultdict
|
||
from collections.abc import Iterable
|
||
from datetime import date, timedelta
|
||
from typing import Callable, TypeVar
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, status
|
||
from sqlalchemy import and_, or_
|
||
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,
|
||
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_gewichtet,
|
||
summe_bestand_im_subtree_gewichtet,
|
||
)
|
||
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] = {
|
||
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 _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 _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]:
|
||
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
|
||
# Nur zwischen Gruppen verrechnen, die in DERSELBEN Basiseinheit
|
||
# zaehlen. Frueher folgte das nebenbei aus der Teilmengen-Bedingung:
|
||
# Artikel verschiedener Arten waren zwangslaeufig disjunkt. Seit die
|
||
# Zweiteinheit einen Stueck-Artikel auch in einer Gramm-Gruppe
|
||
# mitzaehlen laesst, gilt das nicht mehr – und ein Bedarf in Stueck
|
||
# duerfte keinen Bedarf in Gramm decken (siehe _netted_topups:
|
||
# „Alle Mengen muessen in DERSELBEN Einheit vorliegen").
|
||
and ctxs[h.id].base_unit == ctxs[gid].base_unit
|
||
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_gewichtet(db, ctx.matching, ctx.faktoren)
|
||
if e.location_id is None
|
||
else summe_bestand_im_subtree_gewichtet(
|
||
db, ctx.matching, e.location_id, ctx.faktoren)
|
||
)
|
||
|
||
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)
|
||
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
|
||
|
||
|
||
@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]:
|
||
return einkaufsliste_gruppen(db)
|
||
|
||
|
||
def _netted_topups(
|
||
minima: dict[K, float],
|
||
nachfahren: Callable[[K], set[K]],
|
||
stock_of: Callable[[K], float],
|
||
) -> dict[K, float]:
|
||
"""Bedarfe entlang ihrer Hierarchie verrechnen.
|
||
|
||
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.
|
||
|
||
``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: 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)
|
||
|
||
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)
|
||
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,
|
||
),
|
||
))
|
||
|
||
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 = 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 = {
|
||
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)
|
||
|
||
# Gewarnt wird nach dem FRÜHEREN von MHD und Verbrauchsdatum. Die frühere
|
||
# Vorbedingung „best_before IS NOT NULL" musste weg: Eine angebrochene
|
||
# Packung ohne aufgedrucktes MHD hat nur ein Verbrauchsdatum – und genau um
|
||
# die geht es hier.
|
||
lots = (
|
||
db.query(Lot)
|
||
.filter(
|
||
Lot.quantity > 0,
|
||
or_(
|
||
and_(Lot.best_before.isnot(None), Lot.best_before <= threshold),
|
||
and_(Lot.consume_by.isnot(None), Lot.consume_by <= threshold),
|
||
),
|
||
)
|
||
.all()
|
||
)
|
||
result: list[ExpiringItem] = []
|
||
for lot in lots:
|
||
faellig = lot.faellig_am
|
||
if faellig is None:
|
||
continue # kann nicht vorkommen, der Filter oben schliesst es aus
|
||
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,
|
||
opened_on=lot.opened_on,
|
||
consume_by=lot.consume_by,
|
||
warn_date=faellig,
|
||
warn_source=(
|
||
"consume_by" if lot.consume_by is not None and faellig == lot.consume_by
|
||
else "best_before"
|
||
),
|
||
days_left=(faellig - today).days,
|
||
package_size=product.package_size,
|
||
package_label=product.package_label,
|
||
unit_name=unit_name,
|
||
unit_factor=unit_factor,
|
||
)
|
||
)
|
||
# Nach dem wirksamen Datum sortieren – in Python, wie in _fefo_lots, damit
|
||
# SQLite und Postgres dasselbe liefern.
|
||
result.sort(key=lambda e: (e.warn_date, e.lot_id))
|
||
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
|