"""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)