guyg1474andClaude Opus 5.5
0fb3521c2f
- Freunde (freunde.py, Tabelle friendships): suchen, anfragen, annehmen, ablehnen, entfernen; neue Seite "Freunde" im Menü - Vorratslisten und Kanban-Projekte immer privat: sichtbar für Ersteller und Mitglieder; teilen nur mit Freunden (Admins: mit allen); Kanban-Mitglieder (Tabelle kanban_project_members), Bearbeiter nur aus dem Projekt, ausgetretene Mitglieder verlieren ihre Zuteilung - Artikel: lesen alle; bearbeiten Autor, Mitarbeitende verknüpfter Kanban-Projekte und Admins; verknüpfen nur bearbeitbare Artikel (sonst ließen sich Rechte erschleichen); Board-Liste auf der Artikelseite zeigt nur sichtbare Boards - Administratoren sehen und bearbeiten alles - Reisen: Städte nur von Freunden (Admins: alle); "Reiserouten zeigen" zeichnet die Strecken der Kilometer-Rechnung als Bögen - Migration Schritt 6: bisher gemeinsame Listen und alle Projekte bekommen die vorhandenen Konten als Mitglieder, ohne Ersteller → erster Admin; erste Liste einer neuen Installation gehört dem ersten Konto - Tests: test_freunde.py, test_zugriff.py, bestehende auf Privat-Modell umgestellt Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
600 lines
28 KiB
Python
600 lines
28 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.
|
||
Listen sind privat: Sehen dürfen sie der Ersteller und die Freunde, mit denen er sie teilt (Mitglieder).
|
||
Ändern (Name, Mitglieder) und löschen dürfen der Ersteller und Administratoren; Administratoren sehen alles. 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 json
|
||
import math
|
||
|
||
from fastapi import APIRouter, Body, Depends, HTTPException
|
||
|
||
from .auth import current_user
|
||
from .common import ValidationError, clean_text, now_iso, user_list
|
||
from .db import transaction
|
||
from .freunde import check_shareable, shareable
|
||
from .kategorien import CATEGORIES, KEYS as CATEGORY_KEYS, guess
|
||
|
||
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
|
||
MAX_STORES = 30
|
||
MAX_STORE_NAME = 40
|
||
|
||
|
||
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"]),
|
||
"category": row["category"] or guess(row["name"]),
|
||
"storeId": row["store_id"],
|
||
"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:
|
||
return clean_text(value, "Produktname", MAX_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, list_id=None) -> str:
|
||
"""Schreibweise aus dem Katalog übernehmen bzw. neues Produkt aufnehmen; Nutzung in der Liste mitzählen."""
|
||
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)))
|
||
name = row["name"]
|
||
else:
|
||
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, category) VALUES (?, ?, 1, ?, ?)",
|
||
(_key(name), name, now_iso(), guess(name)),
|
||
)
|
||
if list_id is not None:
|
||
conn.execute(
|
||
"""INSERT INTO list_products (list_id, key, uses, last_used) VALUES (?, ?, 1, ?)
|
||
ON CONFLICT (list_id, key) DO UPDATE SET uses = uses + 1, last_used = excluded.last_used""",
|
||
(list_id, _key(name), now_iso()),
|
||
)
|
||
return name
|
||
|
||
|
||
def _visible_product_rows(conn, user: dict) -> list:
|
||
"""Produkte aus den Listen, die das Konto sehen darf – häufig genutzte zuerst."""
|
||
rows = conn.execute(
|
||
f"""SELECT p.name, p.category, SUM(lp.uses) AS uses FROM list_products lp
|
||
JOIN products p ON p.key = lp.key
|
||
JOIN shopping_lists l ON l.id = lp.list_id
|
||
WHERE {_VISIBLE} GROUP BY p.key""",
|
||
_who(user),
|
||
).fetchall()
|
||
return sorted(rows, key=lambda r: (-r["uses"], _sort_key(r["name"])))
|
||
|
||
|
||
def visible_products(conn, user: dict) -> list:
|
||
return [r["name"] for r in _visible_product_rows(conn, user)]
|
||
|
||
|
||
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: selbst angelegt oder als Mitglied eingetragen; Administratoren sehen alle Listen.
|
||
_VISIBLE = """(:admin 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))"""
|
||
|
||
|
||
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"],
|
||
"members": members,
|
||
"createdBy": {"id": row["created_by"], "name": row["created_by_name"]} if row["created_by"] else None,
|
||
"canManage": _can_manage(row, user),
|
||
"categoryOrder": _category_order(row["category_order"]),
|
||
"stores": [
|
||
{"id": s["id"], "name": s["name"], "categoryOrder": _category_order(s["category_order"] or row["category_order"])}
|
||
for s in conn.execute("SELECT * FROM stores WHERE list_id = ? ORDER BY position, id", (row["id"],))
|
||
],
|
||
}
|
||
|
||
|
||
def _category_order(value) -> list:
|
||
"""Gespeicherte Reihenfolge, ergänzt um neu hinzugekommene Kategorien (am Ende)."""
|
||
try:
|
||
order = [k for k in json.loads(value) if k in CATEGORY_KEYS] if value else []
|
||
except ValueError:
|
||
order = []
|
||
return order + [k for k in CATEGORY_KEYS if k not in order]
|
||
|
||
|
||
def _list_fields(conn, data: dict, user: dict, owner_id, current=()) -> tuple:
|
||
"""Name und Mitglieder prüfen. Mitglieder müssen Freunde sein (Admins: beliebig); schon eingetragene dürfen bleiben.
|
||
Der Ersteller sieht seine Liste immer und steht nicht bei den Mitgliedern."""
|
||
name = clean_text(data.get("name"), "Name der Liste", MAX_LIST_NAME)
|
||
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.")
|
||
check_shareable(conn, user, members, keep=current)
|
||
return name, sorted(members)
|
||
|
||
|
||
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]
|
||
marks = ",".join("?" * len(ids))
|
||
items = conn.execute(
|
||
f"""SELECT i.*, u.display_name AS updated_by_name, p.category, sp.store_id FROM household_items i
|
||
LEFT JOIN users u ON u.id = i.updated_by
|
||
LEFT JOIN products p ON p.key = i.key
|
||
LEFT JOIN store_products sp ON sp.list_id = i.list_id AND sp.key = i.key
|
||
WHERE i.list_id IN ({marks})""",
|
||
ids,
|
||
).fetchall()
|
||
minimums = conn.execute(f"SELECT * FROM stock_minimums WHERE list_id IN ({marks}) ORDER BY name", ids).fetchall()
|
||
products = _visible_product_rows(conn, user)
|
||
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": user_list(conn), # Namen für die Anzeige der Mitglieder
|
||
"shareable": shareable(conn, user), # wählbar beim Teilen: Freunde (Admins: alle)
|
||
"products": [r["name"] for r in products],
|
||
"productCategories": {r["name"]: r["category"] or guess(r["name"]) for r in products},
|
||
"categories": [{"key": k, "title": t} for k, t in CATEGORIES],
|
||
"minimums": [
|
||
{"listId": r["list_id"], "key": r["key"], "name": r["name"], "unit": r["unit"], "minimum": _out_quantity(r["minimum"], r["unit"])}
|
||
for r in minimums
|
||
],
|
||
}
|
||
|
||
|
||
# ---------- Mindestbestand ----------
|
||
|
||
|
||
def _refill(conn, list_id: int, key: str, unit: str, user_id: int) -> None:
|
||
"""Liegt der Vorrat unter dem Mindestbestand, die Einkaufsliste bis zur fehlenden Menge auffüllen."""
|
||
minimum = conn.execute(
|
||
"SELECT name, minimum FROM stock_minimums WHERE list_id = ? AND key = ? AND unit = ?", (list_id, key, unit)
|
||
).fetchone()
|
||
if not minimum:
|
||
return
|
||
def amount(list_):
|
||
row = conn.execute(
|
||
"SELECT quantity FROM household_items WHERE list_id = ? AND key = ? AND list = ? AND unit = ?", (list_id, key, list_, unit)
|
||
).fetchone()
|
||
return row["quantity"] if row else 0
|
||
needed = minimum["minimum"] - amount("vorrat") - amount("einkauf")
|
||
if unit == "stk":
|
||
needed = math.ceil(needed)
|
||
needed = round(needed, 2)
|
||
if needed > 0:
|
||
_add(conn, minimum["name"], "einkauf", list_id, min(needed, MAX_QUANTITY), unit, user_id)
|
||
|
||
|
||
@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"]
|
||
name = _canonical_name(conn, name, list_id)
|
||
_add(conn, name, list_, list_id, quantity, unit, user["id"])
|
||
# Im Laden-Filter hinzugefügt: Produkt gleich diesem Laden zuordnen
|
||
if data.get("storeId") is not None:
|
||
_assign_store(conn, list_id, _key(name), _get_store(conn, data["storeId"], user, list_id)["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),
|
||
)
|
||
if row["list"] == "vorrat":
|
||
for u in {row["unit"], unit}: # bei Einheitenwechsel beide Einheiten prüfen
|
||
_refill(conn, row["list_id"], row["key"], u, user["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"])
|
||
_refill(conn, row["list_id"], row["key"], 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:
|
||
row = _get_item(conn, item_id, user)
|
||
conn.execute("DELETE FROM household_items WHERE id = ?", (item_id,))
|
||
if row["list"] == "vorrat":
|
||
_refill(conn, row["list_id"], row["key"], row["unit"], user["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, 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 (?, 0, ?, ?, ?)",
|
||
(name, 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.")
|
||
current = [r["user_id"] for r in conn.execute("SELECT user_id FROM shopping_list_members WHERE list_id = ?", (list_id,))]
|
||
name, members = _list_fields(conn, data, user, row["created_by"], current)
|
||
conn.execute("UPDATE shopping_lists SET name = ? WHERE id = ?", (name, 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 entfernen (z. B. Tippfehler) – nur aus den Listen, die das Konto sieht. Einträge auf den Listen bleiben."""
|
||
with transaction() as conn:
|
||
ids = [r["id"] for r in _visible_lists(conn, user)]
|
||
cur = conn.execute(
|
||
f"DELETE FROM list_products WHERE key = ? AND list_id IN ({','.join('?' * len(ids))})", (_key(name), *ids)
|
||
)
|
||
if not cur.rowcount:
|
||
raise HTTPException(404, "Vorschlag nicht gefunden.")
|
||
# Schreibweise nur löschen, wenn keine Liste das Produkt mehr kennt
|
||
conn.execute(
|
||
"DELETE FROM products WHERE key = ? AND NOT EXISTS (SELECT 1 FROM list_products WHERE key = ?)", (_key(name), _key(name))
|
||
)
|
||
return _state(conn, user)
|
||
|
||
|
||
|
||
@router.put("/api/haushalt/listen/{list_id}/mindestbestand")
|
||
def set_minimum(list_id: int, data: dict = Body(...), user: dict = Depends(current_user)):
|
||
"""Mindestbestand setzen ({name, unit, minimum}); minimum 0 oder null entfernt ihn."""
|
||
name = _clean_name(data.get("name"))
|
||
unit = _unit(data.get("unit"))
|
||
with transaction() as conn:
|
||
_get_list(conn, list_id, user)
|
||
if data.get("minimum") in (None, 0):
|
||
conn.execute("DELETE FROM stock_minimums WHERE list_id = ? AND key = ? AND unit = ?", (list_id, _key(name), unit))
|
||
return _state(conn, user)
|
||
minimum = _quantity(data.get("minimum"), unit)
|
||
name = _canonical_name(conn, name, list_id)
|
||
conn.execute(
|
||
"""INSERT INTO stock_minimums (list_id, key, name, unit, minimum) VALUES (?, ?, ?, ?, ?)
|
||
ON CONFLICT (list_id, key, unit) DO UPDATE SET minimum = excluded.minimum, name = excluded.name""",
|
||
(list_id, _key(name), name, unit, minimum),
|
||
)
|
||
_refill(conn, list_id, _key(name), unit, user["id"])
|
||
return _state(conn, user)
|
||
|
||
|
||
# ---------- Laden-Kategorien ----------
|
||
|
||
|
||
@router.put("/api/haushalt/produkte/kategorie")
|
||
def set_category(data: dict = Body(...), user: dict = Depends(current_user)):
|
||
"""Kategorie eines Produkts ändern – nur für Produkte aus Listen, die das Konto sieht."""
|
||
if data.get("category") not in CATEGORY_KEYS:
|
||
raise ValidationError("Unbekannte Kategorie.")
|
||
with transaction() as conn:
|
||
ids = [r["id"] for r in _visible_lists(conn, user)]
|
||
key = _key(_clean_name(data.get("name")))
|
||
visible = conn.execute(
|
||
f"SELECT 1 FROM household_items WHERE key = ? AND list_id IN ({','.join('?' * len(ids))}) "
|
||
f"UNION SELECT 1 FROM list_products WHERE key = ? AND list_id IN ({','.join('?' * len(ids))})",
|
||
(key, *ids, key, *ids),
|
||
).fetchone()
|
||
if not visible:
|
||
raise HTTPException(404, "Produkt nicht gefunden.")
|
||
conn.execute("UPDATE products SET category = ? WHERE key = ?", (data["category"], key))
|
||
return _state(conn, user)
|
||
|
||
|
||
@router.put("/api/haushalt/listen/{list_id}/reihenfolge")
|
||
def set_category_order(list_id: int, data: dict = Body(...), user: dict = Depends(current_user)):
|
||
"""Reihenfolge der Kategorien für den Laden dieser Liste (alle, die die Liste sehen, dürfen sie anpassen)."""
|
||
order = data.get("order")
|
||
if not isinstance(order, list) or sorted(order) != sorted(CATEGORY_KEYS):
|
||
raise ValidationError("Die Reihenfolge muss jede Kategorie genau einmal enthalten.")
|
||
with transaction() as conn:
|
||
_get_list(conn, list_id, user)
|
||
conn.execute("UPDATE shopping_lists SET category_order = ? WHERE id = ?", (json.dumps(order), list_id))
|
||
return _state(conn, user)
|
||
|
||
|
||
# ---------- Läden ----------
|
||
|
||
|
||
def _get_store(conn, store_id, user: dict, list_id=None):
|
||
"""Laden einer Liste, die das Konto sieht (optional: muss zu list_id gehören)."""
|
||
if isinstance(store_id, bool) or not isinstance(store_id, int):
|
||
raise ValidationError("Ungültiger Laden.")
|
||
row = conn.execute("SELECT * FROM stores WHERE id = ?", (store_id,)).fetchone()
|
||
if not row or (list_id is not None and row["list_id"] != list_id):
|
||
raise HTTPException(404, "Laden nicht gefunden.")
|
||
_get_list(conn, row["list_id"], user)
|
||
return row
|
||
|
||
|
||
def _assign_store(conn, list_id: int, key: str, store_id) -> None:
|
||
if store_id is None:
|
||
conn.execute("DELETE FROM store_products WHERE list_id = ? AND key = ?", (list_id, key))
|
||
else:
|
||
conn.execute(
|
||
"""INSERT INTO store_products (list_id, key, store_id) VALUES (?, ?, ?)
|
||
ON CONFLICT (list_id, key) DO UPDATE SET store_id = excluded.store_id""",
|
||
(list_id, key, store_id),
|
||
)
|
||
|
||
|
||
@router.post("/api/haushalt/listen/{list_id}/laeden", status_code=201)
|
||
def create_store(list_id: int, data: dict = Body(...), user: dict = Depends(current_user)):
|
||
"""Laden anlegen (alle, die die Liste sehen, dürfen Läden verwalten)."""
|
||
name = clean_text(data.get("name"), "Name des Ladens", MAX_STORE_NAME)
|
||
with transaction() as conn:
|
||
_get_list(conn, list_id, user)
|
||
if conn.execute("SELECT COUNT(*) FROM stores WHERE list_id = ?", (list_id,)).fetchone()[0] >= MAX_STORES:
|
||
raise ValidationError(f"Maximal {MAX_STORES} Läden pro Liste.")
|
||
if conn.execute("SELECT 1 FROM stores WHERE list_id = ? AND name = ? COLLATE NOCASE", (list_id, name)).fetchone():
|
||
raise ValidationError("Diesen Laden gibt es in der Liste schon.")
|
||
position = conn.execute("SELECT COALESCE(MAX(position), 0) + 1 FROM stores WHERE list_id = ?", (list_id,)).fetchone()[0]
|
||
store_id = conn.execute(
|
||
"INSERT INTO stores (list_id, name, position) VALUES (?, ?, ?)", (list_id, name, position)
|
||
).lastrowid
|
||
return {**_state(conn, user), "created": store_id}
|
||
|
||
|
||
@router.patch("/api/haushalt/laeden/{store_id}")
|
||
def update_store(store_id: int, data: dict = Body(...), user: dict = Depends(current_user)):
|
||
"""Laden umbenennen ({name}) oder seine Abteilungs-Reihenfolge setzen ({categoryOrder}, null = wie die Liste)."""
|
||
with transaction() as conn:
|
||
row = _get_store(conn, store_id, user)
|
||
if "name" in data:
|
||
name = clean_text(data.get("name"), "Name des Ladens", MAX_STORE_NAME)
|
||
if conn.execute(
|
||
"SELECT 1 FROM stores WHERE list_id = ? AND name = ? COLLATE NOCASE AND id != ?", (row["list_id"], name, store_id)
|
||
).fetchone():
|
||
raise ValidationError("Diesen Laden gibt es in der Liste schon.")
|
||
conn.execute("UPDATE stores SET name = ? WHERE id = ?", (name, store_id))
|
||
if "categoryOrder" in data:
|
||
order = data["categoryOrder"]
|
||
if order is not None and (not isinstance(order, list) or sorted(order) != sorted(CATEGORY_KEYS)):
|
||
raise ValidationError("Die Reihenfolge muss jede Kategorie genau einmal enthalten.")
|
||
conn.execute("UPDATE stores SET category_order = ? WHERE id = ?", (json.dumps(order) if order else None, store_id))
|
||
return _state(conn, user)
|
||
|
||
|
||
@router.put("/api/haushalt/listen/{list_id}/laeden-reihenfolge")
|
||
def order_stores(list_id: int, data: dict = Body(...), user: dict = Depends(current_user)):
|
||
"""Reihenfolge der Läden ({order: [ids]}) – so erscheinen sie in der Einkaufsliste."""
|
||
with transaction() as conn:
|
||
_get_list(conn, list_id, user)
|
||
ids = [r["id"] for r in conn.execute("SELECT id FROM stores WHERE list_id = ?", (list_id,))]
|
||
order = data.get("order")
|
||
if not isinstance(order, list) or sorted(order) != sorted(ids):
|
||
raise ValidationError("Die Reihenfolge muss jeden Laden genau einmal enthalten.")
|
||
conn.executemany("UPDATE stores SET position = ? WHERE id = ?", [(i, s) for i, s in enumerate(order)])
|
||
return _state(conn, user)
|
||
|
||
|
||
@router.delete("/api/haushalt/laeden/{store_id}")
|
||
def delete_store(store_id: int, user: dict = Depends(current_user)):
|
||
"""Laden löschen; seine Produkte bleiben auf der Liste, nur ohne Laden."""
|
||
with transaction() as conn:
|
||
_get_store(conn, store_id, user)
|
||
conn.execute("DELETE FROM stores WHERE id = ?", (store_id,))
|
||
return _state(conn, user)
|
||
|
||
|
||
@router.put("/api/haushalt/listen/{list_id}/produkt-laden")
|
||
def set_product_store(list_id: int, data: dict = Body(...), user: dict = Depends(current_user)):
|
||
"""Produkt einem Laden zuordnen ({name, storeId}); storeId null = kein Laden. Gilt für diese Liste."""
|
||
name = _clean_name(data.get("name"))
|
||
with transaction() as conn:
|
||
_get_list(conn, list_id, user)
|
||
store_id = data.get("storeId")
|
||
if store_id is not None:
|
||
store_id = _get_store(conn, store_id, user, list_id)["id"]
|
||
_assign_store(conn, list_id, _key(name), store_id)
|
||
return _state(conn, user)
|
||
|