guyg1474andClaude Opus 5.5
f0e17b06b5
- Abgehakte Produkte bleiben auf der Einkaufsliste, rutschen durchgestrichen
nach unten ("Im Wagen") und lassen sich zurücknehmen
- Button "Einkauf beenden (n)" verschiebt alle abgehakten Produkte auf einmal
in den Vorrat derselben Liste (Mengen werden zusammengeführt)
- Haken wird gespeichert (Spalte checked, Migration ergänzt sie) und ist für
alle sichtbar, die die Liste sehen; erneut hinzugefügte Produkte verlieren
den Haken
- Einträge gleiten per View Transition an ihren neuen Platz (aus bei
"Animation: aus")
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
363 lines
16 KiB
Python
363 lines
16 KiB
Python
"""Vorrat und Einkauf in mehreren Listen sowie Produktkatalog für einheitliche Namen.
|
||
|
||
Jede Liste (z. B. "Zuhause", "Urlaub") hat ihren eigenen Vorrat und ihre eigene Einkaufsliste.
|
||
Wer eine Liste anlegt, legt fest, ob alle Konten sie sehen oder nur ausgewählte. Ändern
|
||
(Name, Sichtbarkeit) und löschen dürfen der Ersteller und Administratoren. 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
|
||
MAX_LIST_NAME = 40
|
||
MAX_LISTS = 50
|
||
|
||
|
||
def _item(row) -> dict:
|
||
return {
|
||
"id": row["id"],
|
||
"name": row["name"],
|
||
"list": row["list"],
|
||
"listId": row["list_id"],
|
||
"quantity": _out_quantity(row["quantity"], row["unit"]),
|
||
"unit": row["unit"],
|
||
"checked": bool(row["checked"]),
|
||
"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, list_id: int, quantity, unit: str, user_id: int) -> None:
|
||
"""Produkt in den Vorrat bzw. Einkauf einer Liste setzen; steht es dort schon in derselben Einheit, wird die Menge erhöht.
|
||
|
||
Kommt ein schon abgehaktes Produkt erneut auf die Einkaufsliste, ist der Haken wieder weg (es fehlt ja noch etwas).
|
||
"""
|
||
conn.execute(
|
||
"""INSERT INTO household_items (key, name, list, list_id, quantity, unit, updated_at, updated_by)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
||
ON CONFLICT (list_id, key, list, unit) DO UPDATE SET
|
||
quantity = MIN(ROUND(quantity + excluded.quantity, 2), ?),
|
||
checked = 0,
|
||
updated_at = excluded.updated_at,
|
||
updated_by = excluded.updated_by""",
|
||
(_key(name), name, list_, list_id, quantity, unit, now_iso(), user_id, MAX_QUANTITY),
|
||
)
|
||
|
||
|
||
# ---------- Listen und wer sie sieht ----------
|
||
|
||
# Sichtbar: für alle freigegeben, selbst angelegt oder als Mitglied eingetragen.
|
||
# Listen eines gelöschten Kontos ohne Mitglieder sehen Administratoren, damit sie nicht verwaist bleiben.
|
||
_VISIBLE = """(l.everyone = 1 OR l.created_by = :uid
|
||
OR EXISTS (SELECT 1 FROM shopping_list_members m WHERE m.list_id = l.id AND m.user_id = :uid)
|
||
OR (:admin AND l.created_by IS NULL))"""
|
||
|
||
|
||
def _who(user: dict) -> dict:
|
||
return {"uid": user["id"], "admin": int(user["isAdmin"])}
|
||
|
||
|
||
def _visible_lists(conn, user: dict) -> list:
|
||
return conn.execute(
|
||
f"""SELECT l.*, u.display_name AS created_by_name FROM shopping_lists l
|
||
LEFT JOIN users u ON u.id = l.created_by
|
||
WHERE {_VISIBLE} ORDER BY l.position, l.id""",
|
||
_who(user),
|
||
).fetchall()
|
||
|
||
|
||
def _get_list(conn, list_id, user: dict):
|
||
if isinstance(list_id, bool) or not isinstance(list_id, int):
|
||
raise ValidationError("Bitte eine Liste wählen.")
|
||
row = conn.execute(f"SELECT l.* FROM shopping_lists l WHERE l.id = :id AND {_VISIBLE}", {"id": list_id, **_who(user)}).fetchone()
|
||
if not row:
|
||
raise HTTPException(404, "Liste nicht gefunden – vielleicht wurde sie gelöscht oder nicht mit dir geteilt.")
|
||
return row
|
||
|
||
|
||
def _can_manage(row, user: dict) -> bool:
|
||
return row["created_by"] == user["id"] or user["isAdmin"]
|
||
|
||
|
||
def _shopping_list(conn, row, user: dict) -> dict:
|
||
members = [r["user_id"] for r in conn.execute("SELECT user_id FROM shopping_list_members WHERE list_id = ?", (row["id"],))]
|
||
return {
|
||
"id": row["id"],
|
||
"name": row["name"],
|
||
"everyone": bool(row["everyone"]),
|
||
"members": members,
|
||
"createdBy": {"id": row["created_by"], "name": row["created_by_name"]} if row["created_by"] else None,
|
||
"canManage": _can_manage(row, user),
|
||
}
|
||
|
||
|
||
def _list_fields(conn, data: dict, user: dict, owner_id) -> tuple:
|
||
"""Name, "für alle" und Mitglieder prüfen. Der Ersteller sieht seine Liste immer und steht nicht bei den Mitgliedern."""
|
||
name = re.sub(r"\s+", " ", data.get("name")).strip() if isinstance(data.get("name"), str) else ""
|
||
if not name:
|
||
raise ValidationError("Name der Liste fehlt.")
|
||
if len(name) > MAX_LIST_NAME:
|
||
raise ValidationError(f"Name ist zu lang (max. {MAX_LIST_NAME} Zeichen).")
|
||
everyone = data.get("everyone", True)
|
||
if not isinstance(everyone, bool):
|
||
raise ValidationError("Ungültige Sichtbarkeit.")
|
||
members = data.get("members", [])
|
||
if not isinstance(members, list) or any(isinstance(m, bool) or not isinstance(m, int) for m in members):
|
||
raise ValidationError("Ungültige Auswahl der Konten.")
|
||
members = {m for m in members if m != owner_id}
|
||
if members:
|
||
known = {r["id"] for r in conn.execute(f"SELECT id FROM users WHERE id IN ({','.join('?' * len(members))})", list(members))}
|
||
if known != members:
|
||
raise ValidationError("Ein ausgewähltes Konto gibt es nicht mehr.")
|
||
return name, everyone, sorted(members) if not everyone else []
|
||
|
||
|
||
def _set_members(conn, list_id: int, members: list) -> None:
|
||
conn.execute("DELETE FROM shopping_list_members WHERE list_id = ?", (list_id,))
|
||
conn.executemany("INSERT INTO shopping_list_members (list_id, user_id) VALUES (?, ?)", [(list_id, m) for m in members])
|
||
|
||
|
||
def _get_item(conn, item_id: int, user: dict):
|
||
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.")
|
||
_get_list(conn, row["list_id"], user)
|
||
return row
|
||
|
||
|
||
def _state(conn, user: dict) -> dict:
|
||
lists = _visible_lists(conn, user)
|
||
ids = [r["id"] for r in lists]
|
||
items = conn.execute(
|
||
f"""SELECT i.*, u.display_name AS updated_by_name FROM household_items i
|
||
LEFT JOIN users u ON u.id = i.updated_by
|
||
WHERE i.list_id IN ({','.join('?' * len(ids))})""",
|
||
ids,
|
||
).fetchall()
|
||
# Häufig genutzte Produkte zuerst vorschlagen
|
||
products = conn.execute("SELECT name, uses FROM products").fetchall()
|
||
users = conn.execute("SELECT id, display_name FROM users ORDER BY display_name COLLATE NOCASE").fetchall()
|
||
return {
|
||
"items": [_item(r) for r in sorted(items, key=lambda r: _sort_key(r["name"]))],
|
||
"lists": [_shopping_list(conn, r, user) for r in lists],
|
||
"users": [{"id": u["id"], "name": u["display_name"]} for u in users],
|
||
"products": [r["name"] for r in sorted(products, key=lambda r: (-r["uses"], _sort_key(r["name"])))],
|
||
}
|
||
|
||
|
||
@router.get("/api/haushalt")
|
||
def get_household(user: dict = Depends(current_user)):
|
||
with transaction() as conn:
|
||
return _state(conn, user)
|
||
|
||
|
||
@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:
|
||
list_id = _get_list(conn, data.get("listId"), user)["id"]
|
||
_add(conn, _canonical_name(conn, name), list_, list_id, quantity, unit, user["id"])
|
||
return _state(conn, user)
|
||
|
||
|
||
@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 bzw. abhaken. Beim Einheitenwechsel wird mit einer bestehenden Zeile zusammengeführt."""
|
||
with transaction() as conn:
|
||
row = _get_item(conn, item_id, user)
|
||
if "checked" in data:
|
||
# Abhaken: liegt im Wagen, bleibt aber bis "Einkauf beenden" auf der Einkaufsliste
|
||
if not isinstance(data["checked"], bool) or row["list"] != "einkauf":
|
||
raise ValidationError("Nur Einträge auf der Einkaufsliste lassen sich abhaken.")
|
||
conn.execute(
|
||
"UPDATE household_items SET checked = ?, updated_at = ?, updated_by = ? WHERE id = ?",
|
||
(int(data["checked"]), now_iso(), user["id"], item_id),
|
||
)
|
||
if len(data) == 1:
|
||
return _state(conn, user)
|
||
row = _get_item(conn, item_id, user)
|
||
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 list_id = ? AND key = ? AND list = ? AND unit = ? AND id != ?",
|
||
(row["list_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"], row["list_id"], 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, user)
|
||
|
||
|
||
@router.post("/api/haushalt/items/{item_id}/move")
|
||
def move_item(item_id: int, user: dict = Depends(current_user)):
|
||
"""Innerhalb einer Liste: Einkauf → Vorrat ("gekauft") bzw. Vorrat → Einkauf ("aufgebraucht"), mit Zusammenführen."""
|
||
with transaction() as conn:
|
||
row = _get_item(conn, item_id, user)
|
||
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, row["list_id"], quantity, row["unit"], user["id"])
|
||
return _state(conn, user)
|
||
|
||
|
||
@router.post("/api/haushalt/listen/{list_id}/einkauf-beenden")
|
||
def finish_shopping(list_id: int, user: dict = Depends(current_user)):
|
||
"""Alle abgehakten Einträge der Einkaufsliste auf einmal in den Vorrat dieser Liste übernehmen."""
|
||
with transaction() as conn:
|
||
_get_list(conn, list_id, user)
|
||
rows = conn.execute(
|
||
"SELECT * FROM household_items WHERE list_id = ? AND list = 'einkauf' AND checked = 1", (list_id,)
|
||
).fetchall()
|
||
if not rows:
|
||
raise ValidationError("Nichts abgehakt – vielleicht hat gerade jemand anderes den Einkauf beendet.")
|
||
for row in rows:
|
||
conn.execute("DELETE FROM household_items WHERE id = ?", (row["id"],))
|
||
_add(conn, row["name"], "vorrat", list_id, row["quantity"], row["unit"], user["id"])
|
||
return {**_state(conn, user), "moved": len(rows)}
|
||
|
||
|
||
@router.delete("/api/haushalt/items/{item_id}")
|
||
def delete_item(item_id: int, user: dict = Depends(current_user)):
|
||
with transaction() as conn:
|
||
_get_item(conn, item_id, user)
|
||
conn.execute("DELETE FROM household_items WHERE id = ?", (item_id,))
|
||
return _state(conn, user)
|
||
|
||
|
||
@router.post("/api/haushalt/listen", status_code=201)
|
||
def create_list(data: dict = Body(...), user: dict = Depends(current_user)):
|
||
with transaction() as conn:
|
||
name, everyone, members = _list_fields(conn, data, user, user["id"])
|
||
if conn.execute("SELECT COUNT(*) FROM shopping_lists").fetchone()[0] >= MAX_LISTS:
|
||
raise ValidationError(f"Es gibt schon {MAX_LISTS} Listen. Bitte erst alte löschen.")
|
||
position = conn.execute("SELECT COALESCE(MAX(position), 0) + 1 FROM shopping_lists").fetchone()[0]
|
||
list_id = conn.execute(
|
||
"INSERT INTO shopping_lists (name, everyone, position, created_at, created_by) VALUES (?, ?, ?, ?, ?)",
|
||
(name, int(everyone), position, now_iso(), user["id"]),
|
||
).lastrowid
|
||
_set_members(conn, list_id, members)
|
||
return {**_state(conn, user), "created": list_id}
|
||
|
||
|
||
@router.patch("/api/haushalt/listen/{list_id}")
|
||
def update_list(list_id: int, data: dict = Body(...), user: dict = Depends(current_user)):
|
||
with transaction() as conn:
|
||
row = _get_list(conn, list_id, user)
|
||
if not _can_manage(row, user):
|
||
raise HTTPException(403, "Nur wer die Liste angelegt hat (oder ein Administrator) kann sie ändern.")
|
||
name, everyone, members = _list_fields(conn, data, user, row["created_by"])
|
||
conn.execute("UPDATE shopping_lists SET name = ?, everyone = ? WHERE id = ?", (name, int(everyone), list_id))
|
||
_set_members(conn, list_id, members)
|
||
return _state(conn, user)
|
||
|
||
|
||
@router.delete("/api/haushalt/listen/{list_id}")
|
||
def delete_list(list_id: int, user: dict = Depends(current_user)):
|
||
"""Liste samt Vorrat und Einkauf löschen (ON DELETE CASCADE)."""
|
||
with transaction() as conn:
|
||
row = _get_list(conn, list_id, user)
|
||
if not _can_manage(row, user):
|
||
raise HTTPException(403, "Nur wer die Liste angelegt hat (oder ein Administrator) kann sie löschen.")
|
||
conn.execute("DELETE FROM shopping_lists WHERE id = ?", (list_id,))
|
||
return _state(conn, user)
|
||
|
||
|
||
@router.delete("/api/haushalt/produkte")
|
||
def delete_product(name: str, user: 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, user)
|