guyg1474andClaude Opus 5.5
dd58b3d1f8
FastAPI-Backend und React-Frontend mit Konten, Sendungsverfolgung, Projekt-Artikeln, Vorrat & Einkauf mit Rezeptabgleich, Reise-Weltkarte, Kanban-Projekten mit Priorität, Server-Monitoring und den Designs Matrix und Blackwall. Betrieb per Docker hinter nginx (siehe DEPLOY.md). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
206 lines
8.2 KiB
Python
206 lines
8.2 KiB
Python
"""Geteilte Vorrats- und Einkaufsliste mit Produktkatalog für einheitliche Namen.
|
||
|
||
Alle Konten sehen und bearbeiten dieselben Listen. Jedes Produkt steht pro Liste und Einheit
|
||
(Anzahl, kg, l) nur einmal; erneutes Hinzufügen in derselben Einheit erhöht die Menge. Neue Produktnamen landen im Katalog und werden als
|
||
Vorschläge angeboten – "milch" wird dabei automatisch zum bereits bekannten "Milch".
|
||
"""
|
||
import math
|
||
import re
|
||
|
||
from fastapi import APIRouter, Body, Depends, HTTPException
|
||
|
||
from .auth import current_user
|
||
from .db import now_iso, transaction
|
||
from .links import ValidationError
|
||
|
||
router = APIRouter()
|
||
|
||
LISTS = ("vorrat", "einkauf")
|
||
UNITS = ("stk", "kg", "l")
|
||
MAX_NAME = 60
|
||
MAX_QUANTITY = 999
|
||
MAX_PRODUCTS = 2000
|
||
|
||
|
||
def _item(row) -> dict:
|
||
return {
|
||
"id": row["id"],
|
||
"name": row["name"],
|
||
"list": row["list"],
|
||
"quantity": _out_quantity(row["quantity"], row["unit"]),
|
||
"unit": row["unit"],
|
||
"updatedAt": row["updated_at"],
|
||
"updatedBy": {"id": row["updated_by"], "name": row["updated_by_name"]} if row["updated_by"] else None,
|
||
}
|
||
|
||
|
||
def _out_quantity(value, unit: str):
|
||
return int(value) if unit == "stk" else round(value, 2)
|
||
|
||
|
||
def _unit(value) -> str:
|
||
if value is None:
|
||
return "stk"
|
||
if value not in UNITS:
|
||
raise ValidationError("Unbekannte Einheit.")
|
||
return value
|
||
|
||
|
||
def _quantity(value, unit: str = "stk"):
|
||
"""Anzahl: ganze Zahl ab 1. kg/l: Kommazahl ab 0,01 (zwei Nachkommastellen)."""
|
||
if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
|
||
raise ValidationError("Menge fehlt.")
|
||
if unit == "stk":
|
||
if value != int(value) or not 1 <= value <= MAX_QUANTITY:
|
||
raise ValidationError(f"Anzahl muss eine ganze Zahl zwischen 1 und {MAX_QUANTITY} sein.")
|
||
return int(value)
|
||
value = round(value, 2)
|
||
if not 0 < value <= MAX_QUANTITY:
|
||
raise ValidationError(f"Menge muss zwischen 0,01 und {MAX_QUANTITY} liegen.")
|
||
return value
|
||
|
||
|
||
def _clean_name(value) -> str:
|
||
name = re.sub(r"\s+", " ", value).strip() if isinstance(value, str) else ""
|
||
if not name:
|
||
raise ValidationError("Produktname fehlt.")
|
||
if len(name) > MAX_NAME:
|
||
raise ValidationError(f"Produktname ist zu lang (max. {MAX_NAME} Zeichen).")
|
||
return name
|
||
|
||
|
||
def _key(name: str) -> str:
|
||
# casefold() statt COLLATE NOCASE: SQLite vergleicht nur ASCII ohne Groß/klein ("Äpfel" ≠ "äpfel")
|
||
return name.casefold()
|
||
|
||
|
||
_UMLAUTS = str.maketrans({"ä": "a", "ö": "o", "ü": "u", "ß": "ss"})
|
||
|
||
|
||
def _sort_key(name: str) -> str:
|
||
# Deutsche Sortierung: "Äpfel" gehört zu A, nicht hinter Z
|
||
return _key(name).translate(_UMLAUTS)
|
||
|
||
|
||
def _canonical_name(conn, name: str) -> str:
|
||
"""Schreibweise aus dem Katalog übernehmen bzw. neues Produkt aufnehmen."""
|
||
row = conn.execute("SELECT name FROM products WHERE key = ?", (_key(name),)).fetchone()
|
||
if row:
|
||
conn.execute("UPDATE products SET uses = uses + 1, last_used = ? WHERE key = ?", (now_iso(), _key(name)))
|
||
return row["name"]
|
||
if conn.execute("SELECT COUNT(*) FROM products").fetchone()[0] >= MAX_PRODUCTS:
|
||
raise ValidationError("Der Produktkatalog ist voll. Bitte alte Vorschläge löschen.")
|
||
# Neue Produkte mit großem Anfangsbuchstaben, damit "milch" und "Milch" nicht doppelt entstehen
|
||
name = name[0].upper() + name[1:]
|
||
conn.execute("INSERT INTO products (key, name, uses, last_used) VALUES (?, ?, 1, ?)", (_key(name), name, now_iso()))
|
||
return name
|
||
|
||
|
||
def _add(conn, name: str, list_: str, quantity, unit: str, user_id: int) -> None:
|
||
"""Produkt auf eine Liste setzen; steht es dort schon in derselben Einheit, wird die Menge erhöht."""
|
||
conn.execute(
|
||
"""INSERT INTO household_items (key, name, list, quantity, unit, updated_at, updated_by)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||
ON CONFLICT (key, list, unit) DO UPDATE SET
|
||
quantity = MIN(ROUND(quantity + excluded.quantity, 2), ?),
|
||
updated_at = excluded.updated_at,
|
||
updated_by = excluded.updated_by""",
|
||
(_key(name), name, list_, quantity, unit, now_iso(), user_id, MAX_QUANTITY),
|
||
)
|
||
|
||
|
||
def _get_item(conn, item_id: int):
|
||
row = conn.execute("SELECT * FROM household_items WHERE id = ?", (item_id,)).fetchone()
|
||
if not row:
|
||
raise HTTPException(404, "Eintrag nicht gefunden – vielleicht hat ihn gerade jemand anderes geändert.")
|
||
return row
|
||
|
||
|
||
def _state(conn) -> dict:
|
||
items = conn.execute(
|
||
"""SELECT i.*, u.display_name AS updated_by_name FROM household_items i
|
||
LEFT JOIN users u ON u.id = i.updated_by"""
|
||
).fetchall()
|
||
# Häufig genutzte Produkte zuerst vorschlagen
|
||
products = conn.execute("SELECT name, uses FROM products").fetchall()
|
||
return {
|
||
"items": [_item(r) for r in sorted(items, key=lambda r: _sort_key(r["name"]))],
|
||
"products": [r["name"] for r in sorted(products, key=lambda r: (-r["uses"], _sort_key(r["name"])))],
|
||
}
|
||
|
||
|
||
@router.get("/api/haushalt")
|
||
def get_household(_: dict = Depends(current_user)):
|
||
with transaction() as conn:
|
||
return _state(conn)
|
||
|
||
|
||
@router.post("/api/haushalt/items", status_code=201)
|
||
def add_item(data: dict = Body(...), user: dict = Depends(current_user)):
|
||
list_ = data.get("list")
|
||
if list_ not in LISTS:
|
||
raise ValidationError("Unbekannte Liste.")
|
||
name = _clean_name(data.get("name"))
|
||
unit = _unit(data.get("unit"))
|
||
quantity = _quantity(data.get("quantity", 1), unit)
|
||
with transaction() as conn:
|
||
_add(conn, _canonical_name(conn, name), list_, quantity, unit, user["id"])
|
||
return _state(conn)
|
||
|
||
|
||
@router.patch("/api/haushalt/items/{item_id}")
|
||
def update_item(item_id: int, data: dict = Body(...), user: dict = Depends(current_user)):
|
||
"""Menge und/oder Einheit ändern. Beim Einheitenwechsel wird mit einer bestehenden Zeile zusammengeführt."""
|
||
with transaction() as conn:
|
||
row = _get_item(conn, item_id)
|
||
unit = _unit(data.get("unit", row["unit"]))
|
||
if "quantity" in data:
|
||
quantity = _quantity(data["quantity"], unit)
|
||
else:
|
||
# Nur Einheit gewechselt: Zahl bleibt, bei "Anzahl" auf ganze Stück aufrunden
|
||
quantity = max(1, math.ceil(row["quantity"])) if unit == "stk" else row["quantity"]
|
||
other = conn.execute(
|
||
"SELECT id FROM household_items WHERE key = ? AND list = ? AND unit = ? AND id != ?",
|
||
(row["key"], row["list"], unit, item_id),
|
||
).fetchone()
|
||
if other:
|
||
conn.execute("DELETE FROM household_items WHERE id = ?", (item_id,))
|
||
_add(conn, row["name"], row["list"], quantity, unit, user["id"])
|
||
else:
|
||
conn.execute(
|
||
"UPDATE household_items SET quantity = ?, unit = ?, updated_at = ?, updated_by = ? WHERE id = ?",
|
||
(quantity, unit, now_iso(), user["id"], item_id),
|
||
)
|
||
return _state(conn)
|
||
|
||
|
||
@router.post("/api/haushalt/items/{item_id}/move")
|
||
def move_item(item_id: int, user: dict = Depends(current_user)):
|
||
"""Einkauf → Vorrat ("gekauft") bzw. Vorrat → Einkauf ("aufgebraucht"), mit Zusammenführen."""
|
||
with transaction() as conn:
|
||
row = _get_item(conn, item_id)
|
||
target = "vorrat" if row["list"] == "einkauf" else "einkauf"
|
||
# Aufgebraucht heißt: einmal nachkaufen (1 Stück, 1 kg, 1 l), egal wie viel vorher da war
|
||
quantity = row["quantity"] if target == "vorrat" else 1
|
||
conn.execute("DELETE FROM household_items WHERE id = ?", (item_id,))
|
||
_add(conn, row["name"], target, quantity, row["unit"], user["id"])
|
||
return _state(conn)
|
||
|
||
|
||
@router.delete("/api/haushalt/items/{item_id}")
|
||
def delete_item(item_id: int, _: dict = Depends(current_user)):
|
||
with transaction() as conn:
|
||
_get_item(conn, item_id)
|
||
conn.execute("DELETE FROM household_items WHERE id = ?", (item_id,))
|
||
return _state(conn)
|
||
|
||
|
||
@router.delete("/api/haushalt/produkte")
|
||
def delete_product(name: str, _: dict = Depends(current_user)):
|
||
"""Vorschlag aus dem Katalog entfernen (z. B. Tippfehler). Einträge auf den Listen bleiben."""
|
||
with transaction() as conn:
|
||
cur = conn.execute("DELETE FROM products WHERE key = ?", (_key(name),))
|
||
if not cur.rowcount:
|
||
raise HTTPException(404, "Vorschlag nicht gefunden.")
|
||
return _state(conn)
|