guyg1474andClaude Opus 5.5
b30acb652a
Vorrat & Einkauf: - Mehrere Listen (z. B. "Zuhause", "Urlaub"), jede mit eigenem Vorrat und eigener Einkaufsliste; Auswahl oben auf der Seite, gemerkt pro Browser - Beim Anlegen festlegen, wer die Liste sieht: alle Konten oder nur ausgewählte (Ersteller sieht sie immer); ändern/löschen nur Ersteller und Administratoren - Gekauft, aufgebraucht, Nachkaufen und Rezeptabgleich wirken innerhalb der gewählten Liste - Migration: bisheriger Vorrat und Einkauf werden zur Liste "Zuhause" (für alle sichtbar) Reisen: - Schalter "Städte anderer zeigen": Städte der übrigen Konten in je einer eigenen Farbe, einzelne Personen ein-/ausblendbar, nur lesend - Info-Karte zeigt, wer noch in einer Stadt bzw. einem Land war Design "Fabrik" (Industrie-Inventar à la Satisfactory): - Stahlplatten mit Schrauben, Titelleisten mit Fenstersymbol, gelb leuchtende Hauptbuttons, orange Mengen-Zähler, Warnstreifen, Riffelblech und drehende Zahnräder im Hintergrund - Schriften Roboto / Roboto Condensed (OFL), lokal eingebunden Doku: README verlinkt das zugehörige Repository P42-S.M. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
359 lines
15 KiB
Python
359 lines
15 KiB
Python
"""Rezepte abgleichen: Zutaten von einer Rezeptseite (schema.org/Recipe) oder aus eingefügtem Text lesen,
|
||
den Produkten im Katalog zuordnen und Fehlendes auf die Einkaufsliste setzen.
|
||
|
||
Abgerufen wird nur die eine Seite, deren Link eingefügt wurde. Chefkoch & Co. betten die Zutaten als
|
||
JSON-LD ("recipeIngredient") ein; dafür braucht es keine API und keinen HTML-Parser.
|
||
"""
|
||
import html
|
||
import ipaddress
|
||
import json
|
||
import re
|
||
import socket
|
||
from urllib.parse import urljoin, urlsplit
|
||
|
||
import httpx
|
||
from fastapi import APIRouter, Body, Depends, HTTPException
|
||
|
||
from .auth import current_user
|
||
from .db import transaction
|
||
from .haushalt import _add, _canonical_name, _clean_name, _get_list, _key, _quantity, _state, _unit
|
||
from .links import ValidationError
|
||
|
||
router = APIRouter()
|
||
|
||
MAX_PAGE = 3 * 1024 * 1024
|
||
MAX_REDIRECTS = 3
|
||
TIMEOUT = 10
|
||
MAX_INGREDIENTS = 80
|
||
MAX_TEXT = 10_000
|
||
|
||
# ---------- Seite abrufen ----------
|
||
|
||
|
||
def _check_host(url: str) -> None:
|
||
"""Nur öffentliche Adressen abrufen – sonst ließe sich über den Server das Heimnetz abfragen."""
|
||
parts = urlsplit(url)
|
||
if parts.scheme not in ("http", "https") or not parts.hostname:
|
||
raise ValidationError("Bitte einen Link mit http:// oder https:// einfügen.")
|
||
try:
|
||
infos = socket.getaddrinfo(parts.hostname, parts.port or (443 if parts.scheme == "https" else 80))
|
||
except socket.gaierror:
|
||
raise ValidationError("Die Seite wurde nicht gefunden. Stimmt der Link?")
|
||
for info in infos:
|
||
if not ipaddress.ip_address(info[4][0].split("%")[0]).is_global:
|
||
raise ValidationError("Links ins eigene Netzwerk werden nicht abgerufen.")
|
||
|
||
|
||
async def _fetch(url: str) -> str:
|
||
headers = {"User-Agent": "Mozilla/5.0 (P42-Dashboard; Rezeptabgleich)", "Accept": "text/html"}
|
||
async with httpx.AsyncClient(timeout=TIMEOUT, follow_redirects=False, headers=headers) as client:
|
||
for _ in range(MAX_REDIRECTS + 1):
|
||
_check_host(url)
|
||
try:
|
||
async with client.stream("GET", url) as res:
|
||
if res.is_redirect and res.headers.get("location"):
|
||
url = urljoin(url, res.headers["location"])
|
||
continue
|
||
if res.status_code >= 400:
|
||
raise ValidationError(f"Die Seite antwortet mit Fehler {res.status_code}.")
|
||
data = bytearray()
|
||
async for chunk in res.aiter_bytes():
|
||
data += chunk
|
||
if len(data) > MAX_PAGE:
|
||
raise ValidationError("Die Seite ist zu groß.")
|
||
return data.decode(res.encoding or "utf-8", errors="replace")
|
||
except httpx.TimeoutException:
|
||
raise ValidationError("Die Seite hat nicht rechtzeitig geantwortet.")
|
||
except httpx.HTTPError:
|
||
raise ValidationError("Die Seite konnte nicht abgerufen werden.")
|
||
raise ValidationError("Zu viele Weiterleitungen.")
|
||
|
||
|
||
def _find_recipe(page: str):
|
||
for m in re.finditer(r"<script[^>]*application/ld\+json[^>]*>(.*?)</script>", page, re.S | re.I):
|
||
try:
|
||
data = json.loads(m.group(1).strip())
|
||
except ValueError:
|
||
continue
|
||
stack = [data]
|
||
while stack:
|
||
node = stack.pop()
|
||
if isinstance(node, list):
|
||
stack.extend(node)
|
||
elif isinstance(node, dict):
|
||
kind = node.get("@type")
|
||
if kind == "Recipe" or (isinstance(kind, list) and "Recipe" in kind):
|
||
return node
|
||
stack.extend(v for k, v in node.items() if k in ("@graph", "mainEntity", "itemListElement"))
|
||
return None
|
||
|
||
|
||
def _servings(value):
|
||
"""recipeYield kommt als Zahl, Text ("4 Portionen") oder Liste davon."""
|
||
for v in value if isinstance(value, list) else [value]:
|
||
m = re.search(r"\d+", str(v or ""))
|
||
if m and 0 < int(m.group()) <= 100:
|
||
return int(m.group())
|
||
return None
|
||
|
||
|
||
# ---------- Zutaten zerlegen ----------
|
||
|
||
FRACTIONS = {"½": 0.5, "¼": 0.25, "¾": 0.75, "⅓": 1 / 3, "⅔": 2 / 3, "⅛": 0.125}
|
||
NUMBER = r"(?:\d*[½¼¾⅓⅔⅛]|\d+(?:[.,]\d+)?(?:\s+\d+/\d+)?|\d+/\d+)"
|
||
AMOUNT = re.compile(rf"^(?P<a>{NUMBER})(?:\s*[-–]\s*(?P<b>{NUMBER}))?\s*")
|
||
|
||
UNITS = {
|
||
"g", "gr", "gramm", "kg", "mg", "ml", "l", "liter", "cl", "dl", "el", "tl", "essl", "teel", "msp",
|
||
"messerspitze", "prise", "pck", "päckchen", "packung", "pkt", "dose", "becher", "bund", "zehe", "stück",
|
||
"stk", "tasse", "glas", "gläser", "scheibe", "schuss", "spritzer", "handvoll", "hand", "blatt", "blätter",
|
||
"stange", "stängel", "kopf", "würfel", "beutel", "tropfen", "zweig", "knolle", "tüte", "flasche", "rolle",
|
||
"cm", "etwas", "wenig", "einige", "tafel", "riegel", "netz", "schale", "kugel",
|
||
}
|
||
OPTIONAL = re.compile(r"^(?:evtl\.?|eventuell|nach belieben|n\.\s?b\.?|optional|ggf\.?)\s*", re.I)
|
||
SIZE_WORDS = re.compile(
|
||
r"^(?:(?:sehr\s+)?(?:kl\.|gr\.|m\.-?\s?groß(?:e|er|es|en)?|"
|
||
r"(?:groß|klein|mittelgroß|frisch|reif|weich|kalt|warm|lauwarm|gehackt|gemahlen)(?:e|er|es|en)?)\s+)+",
|
||
re.I,
|
||
)
|
||
# Eiweiß, Eigelb & Co. kauft man als ganze Eier
|
||
EGG_PARTS = {"eiweiß": "weiß", "eiweiss": "weiß", "eiklar": "weiß", "eigelb": "gelb", "eidotter": "gelb", "dotter": "gelb"}
|
||
EGG_PART = re.compile(r"^(eiweiß|eiweiss|eiklar|eigelb|eidotter|dotter)(?:e|en|s)?(?=\s|$)")
|
||
|
||
# Selbstverständliches, das meist im Haus ist – nicht vorab ankreuzen
|
||
STAPLES = {"salz", "pfeffer", "wasser", "salz und pfeffer", "eiswasser", "leitungswasser", "öl zum braten"}
|
||
|
||
|
||
def _number(text: str) -> float:
|
||
text = text.strip().replace(",", ".")
|
||
if text[-1] in FRACTIONS:
|
||
return (float(text[:-1]) if text[:-1] else 0) + FRACTIONS[text[-1]]
|
||
total = 0.0
|
||
for part in text.split():
|
||
num, _, den = part.partition("/")
|
||
total += float(num) / float(den) if den else float(num)
|
||
return total
|
||
|
||
|
||
def _recipe_unit(token: str):
|
||
t = re.sub(r"\((?:n|s|e|en|er)\)$|/(?:n|s|e|en)$", "", token.casefold().rstrip(".,;:"))
|
||
t = t.rstrip(".")
|
||
return t if t in UNITS else None
|
||
|
||
|
||
def _plural(name: str) -> str:
|
||
# "Ei(er)" → "Eier", "Zwiebel(n)" → "Zwiebeln", "Zehe/n" → "Zehen"
|
||
return re.sub(r"(\w)(?:\((\w{1,3})\)|/(\w{1,2}))(?=\W|$)", lambda m: m[1] + (m[2] or m[3]), name)
|
||
|
||
|
||
def parse_ingredient(raw: str):
|
||
text = re.sub(r"\s+", " ", html.unescape(str(raw))).strip()
|
||
if not text:
|
||
return None
|
||
rest = text
|
||
optional = bool(OPTIONAL.match(rest))
|
||
rest = OPTIONAL.sub("", rest)
|
||
|
||
amount = None
|
||
m = AMOUNT.match(rest)
|
||
if m:
|
||
# Bei "2-3" lieber die größere Menge besorgen
|
||
amount = _number(m["b"] or m["a"])
|
||
rest = rest[m.end():]
|
||
|
||
rest = SIZE_WORDS.sub("", rest) # "1 kl. Dose Mais" → Einheit Dose
|
||
unit = None
|
||
first, _, tail = rest.partition(" ")
|
||
if tail and _recipe_unit(first):
|
||
unit, rest = _recipe_unit(first), tail
|
||
if first.endswith(","): # "1 TL, gestr. Zimt" → Zusatz hinter dem Komma ist keine Zutat
|
||
rest = re.sub(r"^[^ ]*\.\s*", "", rest) if re.match(r"^\w+\.\s", rest) else rest
|
||
|
||
# Klammern: "(oder Butter)" ist eine Alternative, alles andere ein Hinweis
|
||
alternatives = [a.strip() for a in re.findall(r"\(oder ([^)]+)\)", rest, re.I)]
|
||
name = re.sub(r"\s*\([^)]*\)", "", _plural(rest))
|
||
name = name.partition(",")[0] # Hinweise hinter dem Komma ("Saft davon") gehören nicht zum Namen
|
||
# "Schmand oder Crème fraîche" → Schmand, Crème fraîche als Alternative
|
||
name, *others = re.split(r"\s+oder\s+", name.strip(), flags=re.I)
|
||
alternatives += [o.strip() for o in others if o.strip()]
|
||
name = SIZE_WORDS.sub("", name.strip()).strip(" .-")
|
||
if not name:
|
||
return None
|
||
name = name[0].upper() + name[1:]
|
||
egg = EGG_PART.match(name.casefold())
|
||
if egg or name.casefold() == "ei":
|
||
name = "Eier"
|
||
return {
|
||
"raw": text,
|
||
"amount": round(amount, 3) if amount is not None else None,
|
||
"unit": unit,
|
||
"name": name[:60],
|
||
"alternatives": [_plural(a)[:60] for a in alternatives],
|
||
"optional": optional,
|
||
"staple": name.casefold() in STAPLES,
|
||
"eggPart": EGG_PARTS[egg[1]] if egg else None,
|
||
}
|
||
|
||
|
||
def merge(ingredients: list) -> list:
|
||
"""Doppelte Zutaten (z. B. Mehl für Teig und Streusel) zusammenfassen, wenn die Einheit passt.
|
||
|
||
Eier: ganze Eier zählen voll, von Eigelb und Eiweiß nur der größere Teil –
|
||
"3 Eigelb + 3 Eiweiß" sind 3 getrennte Eier, nicht 6.
|
||
"""
|
||
merged = {}
|
||
for ing in ingredients:
|
||
key = (_key(ing["name"]), ing["unit"])
|
||
part = ing.pop("eggPart", None)
|
||
amount = ing["amount"] if ing["amount"] is not None else (1 if part else None)
|
||
if key not in merged:
|
||
merged[key] = {**ing, "amount": None, "raws": [], "whole": 0.0, "parts": {}, "counted": False}
|
||
entry = merged[key]
|
||
entry["raws"].append(ing["raw"])
|
||
entry["optional"] = entry["optional"] and ing["optional"]
|
||
if amount is None:
|
||
continue
|
||
entry["counted"] = True
|
||
if part:
|
||
entry["parts"][part] = entry["parts"].get(part, 0) + amount
|
||
else:
|
||
entry["whole"] += amount
|
||
result = []
|
||
for entry in merged.values():
|
||
if entry["counted"]:
|
||
entry["amount"] = round(entry["whole"] + max(entry["parts"].values(), default=0), 3)
|
||
entry["raw"] = " + ".join(entry["raws"])
|
||
for k in ("raws", "whole", "parts", "counted"):
|
||
del entry[k]
|
||
result.append(entry)
|
||
return result
|
||
|
||
|
||
# ---------- Zuordnung zu Produkten ----------
|
||
|
||
_UMLAUTS = str.maketrans({"ä": "a", "ö": "o", "ü": "u", "ß": "ss"})
|
||
|
||
|
||
def _stem(name: str) -> str:
|
||
"""Grobe Grundform, damit "Apfel"/"Äpfel", "Ei"/"Eier", "Tomate"/"Tomaten" zusammenfinden."""
|
||
s = name.casefold().translate(_UMLAUTS)
|
||
if s in ("ei", "eier"): # zu kurz für die Endungsregel; "Eis" soll nicht zu "Eier" werden
|
||
return "ei"
|
||
for ending in ("en", "er", "n", "e", "s"):
|
||
if s.endswith(ending) and len(s) - len(ending) >= 3:
|
||
return s[: -len(ending)]
|
||
return s
|
||
|
||
|
||
def _matcher(conn):
|
||
aliases = {r["key"]: r["product"] for r in conn.execute("SELECT key, product FROM ingredient_aliases")}
|
||
products = [r["name"] for r in conn.execute("SELECT name FROM products")]
|
||
exact = {_key(p): p for p in products}
|
||
stems = {}
|
||
for p in products:
|
||
stems.setdefault(_stem(p), p)
|
||
|
||
def lookup(name: str):
|
||
k = _key(name)
|
||
if k in aliases:
|
||
return aliases[k]
|
||
if k in exact:
|
||
return exact[k]
|
||
if _stem(name) in stems:
|
||
return stems[_stem(name)]
|
||
# "brauner Zucker" → "Zucker"
|
||
last = name.split()[-1]
|
||
if last != name:
|
||
return exact.get(_key(last)) or stems.get(_stem(last))
|
||
return None
|
||
|
||
def match(ing: dict) -> dict:
|
||
found = lookup(ing["name"])
|
||
if not found:
|
||
for alt in ing["alternatives"]:
|
||
if found := lookup(alt):
|
||
break
|
||
return {**ing, "product": found or ing["name"], "known": bool(found)}
|
||
|
||
return match
|
||
|
||
|
||
def _result(conn, title, source, servings, raw_ingredients) -> dict:
|
||
parsed = [p for p in (parse_ingredient(r) for r in raw_ingredients[:MAX_INGREDIENTS]) if p]
|
||
if not parsed:
|
||
raise ValidationError("Keine Zutaten gefunden.")
|
||
match = _matcher(conn)
|
||
return {
|
||
"title": title,
|
||
"source": source,
|
||
"servings": servings,
|
||
"ingredients": [match(i) for i in merge(parsed)],
|
||
}
|
||
|
||
|
||
# ---------- API ----------
|
||
|
||
|
||
@router.post("/api/haushalt/rezept")
|
||
async def load_recipe(data: dict = Body(...), _: dict = Depends(current_user)):
|
||
"""Zutaten aus einem Rezept-Link ({url}) oder eingefügtem Text ({text, servings}) lesen."""
|
||
url = data.get("url").strip() if isinstance(data.get("url"), str) else ""
|
||
text = data.get("text") if isinstance(data.get("text"), str) else ""
|
||
if url:
|
||
if not re.match(r"^https?://", url, re.I):
|
||
url = "https://" + url
|
||
recipe = _find_recipe(await _fetch(url))
|
||
if not recipe or not recipe.get("recipeIngredient"):
|
||
raise ValidationError(
|
||
"Auf dieser Seite wurden keine Rezeptdaten gefunden. Du kannst die Zutaten stattdessen einfügen."
|
||
)
|
||
ingredients = recipe["recipeIngredient"]
|
||
if isinstance(ingredients, str):
|
||
ingredients = ingredients.splitlines()
|
||
title = html.unescape(str(recipe.get("name") or "Rezept"))[:200]
|
||
with transaction() as conn:
|
||
return _result(conn, title, url, _servings(recipe.get("recipeYield")), ingredients)
|
||
if text.strip():
|
||
if len(text) > MAX_TEXT:
|
||
raise ValidationError("Der Text ist zu lang.")
|
||
servings = data.get("servings")
|
||
servings = servings if isinstance(servings, int) and 0 < servings <= 100 else None
|
||
lines = [line.strip(" \t-•*·") for line in text.splitlines()]
|
||
with transaction() as conn:
|
||
return _result(conn, "Eigene Zutatenliste", None, servings, [line for line in lines if line])
|
||
raise ValidationError("Bitte einen Rezept-Link oder eine Zutatenliste angeben.")
|
||
|
||
|
||
@router.post("/api/haushalt/rezept/uebernehmen")
|
||
def take_over(data: dict = Body(...), user: dict = Depends(current_user)):
|
||
"""Ausgewählte Zutaten auf die Einkaufsliste setzen und korrigierte Zuordnungen merken."""
|
||
items = data.get("items") if isinstance(data.get("items"), list) else []
|
||
aliases = data.get("aliases") if isinstance(data.get("aliases"), list) else []
|
||
if not items and not aliases:
|
||
raise ValidationError("Nichts ausgewählt.")
|
||
if len(items) > MAX_INGREDIENTS or len(aliases) > MAX_INGREDIENTS:
|
||
raise ValidationError("Zu viele Zutaten.")
|
||
# Erst alles prüfen, dann in einem Rutsch speichern
|
||
checked = []
|
||
for i in items:
|
||
if not isinstance(i, dict):
|
||
raise ValidationError("Ungültige Zutat.")
|
||
unit = _unit(i.get("unit"))
|
||
checked.append((_clean_name(i.get("name")), _quantity(i.get("quantity"), unit), unit))
|
||
learned = [(_clean_name(a.get("ingredient")), _clean_name(a.get("product"))) for a in aliases if isinstance(a, dict)]
|
||
with transaction() as conn:
|
||
list_id = _get_list(conn, data.get("listId"), user)["id"] if checked else None # Einkauf dieser Liste
|
||
for ingredient, product in learned:
|
||
product = _canonical_name(conn, product)
|
||
if _key(ingredient) == _key(product):
|
||
conn.execute("DELETE FROM ingredient_aliases WHERE key = ?", (_key(ingredient),))
|
||
else:
|
||
conn.execute(
|
||
"INSERT INTO ingredient_aliases (key, product) VALUES (?, ?) "
|
||
"ON CONFLICT (key) DO UPDATE SET product = excluded.product",
|
||
(_key(ingredient), product),
|
||
)
|
||
for name, quantity, unit in checked:
|
||
_add(conn, _canonical_name(conn, name), "einkauf", list_id, quantity, unit, user["id"])
|
||
return _state(conn, user)
|