1804 lines
78 KiB
Python
Executable File
1804 lines
78 KiB
Python
Executable File
import os
|
|
import re
|
|
import sqlite3
|
|
import secrets
|
|
from datetime import datetime, date, timedelta
|
|
from typing import List, Dict, Any, Optional, Tuple
|
|
|
|
try:
|
|
from dotenv import load_dotenv
|
|
load_dotenv()
|
|
except ImportError:
|
|
pass
|
|
|
|
try:
|
|
import pymysql
|
|
import pymysql.cursors
|
|
except ImportError:
|
|
pymysql = None
|
|
|
|
DB_PATH = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "data", "wtc.db")
|
|
|
|
DB_ENGINE = os.getenv("DB_ENGINE", "mariadb").lower()
|
|
MARIADB_HOST = os.getenv("MARIADB_HOST", "db.waiswuaschd.de")
|
|
MARIADB_PORT = int(os.getenv("MARIADB_PORT", "3306"))
|
|
MARIADB_USER = os.getenv("MARIADB_USER", "v2_usr_wtc_cookbook")
|
|
MARIADB_PASS = os.getenv("MARIADB_PASS", "XKikHf&j6Wj!rs5%A8#ZzyvE#*A1hT")
|
|
MARIADB_NAME = os.getenv("MARIADB_NAME", "v2_db_wtc_cookbook")
|
|
|
|
LEGACY_CATEGORIES = [
|
|
"Reis",
|
|
"Nudeln",
|
|
"Kartoffeln",
|
|
"Teigig",
|
|
"asiatisch",
|
|
"Hausmannskost",
|
|
"Suppen",
|
|
"Eintöpfe",
|
|
"Süßspeisen",
|
|
"Fisch"
|
|
]
|
|
|
|
# 10 Kochbuch-Kapitel aus dem Benutzerentwurf mit handgezeichneten Doodles & Icons
|
|
CHAPTERS = [
|
|
{"number": 1, "title": "Suppen & Eintöpfe", "icon": "🍲", "doodle_hint": "Dampfende Suppenschale", "illustration_url": "/static/img/chapters/chapter_1.png"},
|
|
{"number": 2, "title": "Traditionell & Basics", "icon": "🥟", "doodle_hint": "Klassiker & Brotzeit", "illustration_url": "/static/img/chapters/chapter_2.png"},
|
|
{"number": 3, "title": "Pasta & Nudelgerichte", "icon": "🍝", "doodle_hint": "Spaghetti & Pasta", "illustration_url": "/static/img/chapters/chapter_3.png"},
|
|
{"number": 4, "title": "Kartoffeln", "icon": "🥔", "doodle_hint": "Kartoffelauflauf & Klöße", "illustration_url": "/static/img/chapters/chapter_4.png"},
|
|
{"number": 5, "title": "Reis & Getreide", "icon": "🍚", "doodle_hint": "Reisschale & Körner", "illustration_url": "/static/img/chapters/chapter_5.png"},
|
|
{"number": 6, "title": "Teig & Gebackenes", "icon": "🍕", "doodle_hint": "Pizza, Flammkuchen & Brot", "illustration_url": "/static/img/chapters/chapter_6.png"},
|
|
{"number": 7, "title": "Asiatische Küche", "icon": "🥢", "doodle_hint": "Wok, Ramen & Thai-Curry", "illustration_url": "/static/img/chapters/chapter_7.png"},
|
|
{"number": 8, "title": "Süßspeisen", "icon": "🥞", "doodle_hint": "Kaiserschmarrn & Knödel", "illustration_url": "/static/img/chapters/chapter_8.png"},
|
|
{"number": 9, "title": "Desserts", "icon": "🍰", "doodle_hint": "Tiramisu & Mousse", "illustration_url": "/static/img/chapters/chapter_9.png"},
|
|
{"number": 10, "title": "Kuchen & Gebäck", "icon": "🎂", "doodle_hint": "Torten & Feingebäck", "illustration_url": "/static/img/chapters/chapter_10.png"}
|
|
]
|
|
|
|
CATEGORIES = [c["title"] for c in CHAPTERS] + [c for c in LEGACY_CATEGORIES if c not in [ch["title"] for ch in CHAPTERS]]
|
|
|
|
EFFORTS = ["gering", "mittel", "hoch"]
|
|
|
|
CHAPTER_MAP = {c["title"]: c["number"] for c in CHAPTERS}
|
|
for c in CHAPTERS:
|
|
CHAPTER_MAP[str(c["number"])] = c["number"]
|
|
|
|
# Standard-Kategorien für die Einkaufsliste (Supermarkt-Laufweg)
|
|
# Standard (Vollsortimenter z. B. Edeka / Rewe)
|
|
DEFAULT_SHOPPING_CATEGORIES = [
|
|
{"name": "Obst & Gemüse", "icon": "🥦", "shelf_type": "produce", "sort_order": 1},
|
|
{"name": "Brot & Backwaren", "icon": "🍞", "shelf_type": "bakery", "sort_order": 2},
|
|
{"name": "Vorrat & Nudeln", "icon": "🥫", "shelf_type": "pantry", "sort_order": 3},
|
|
{"name": "Gewürze & Öle", "icon": "🧂", "shelf_type": "spices", "sort_order": 4},
|
|
{"name": "Kühlregal & Käse", "icon": "🧀", "shelf_type": "dairy", "sort_order": 5},
|
|
{"name": "Fleisch & Fisch", "icon": "🥩", "shelf_type": "meat", "sort_order": 6},
|
|
{"name": "Tiefkühl", "icon": "❄️", "shelf_type": "frozen", "sort_order": 7},
|
|
{"name": "Getränke", "icon": "🧃", "shelf_type": "drinks", "sort_order": 8},
|
|
{"name": "Drogerie & Sonstiges", "icon": "🧺", "shelf_type": "drugstore", "sort_order": 9},
|
|
]
|
|
|
|
# Discounter-Preset (z. B. Aldi / Lidl: Kühlung & Fleisch früher, Vorrat/Aktion zentral)
|
|
DISCOUNTER_SHOPPING_CATEGORIES = [
|
|
{"name": "Obst & Gemüse", "icon": "🥦", "shelf_type": "produce", "sort_order": 1},
|
|
{"name": "Brot & Backautomat", "icon": "🍞", "shelf_type": "bakery", "sort_order": 2},
|
|
{"name": "Kühlregal & Käse", "icon": "🧀", "shelf_type": "dairy", "sort_order": 3},
|
|
{"name": "Frischfleisch & Wurst", "icon": "🥩", "shelf_type": "meat", "sort_order": 4},
|
|
{"name": "Vorrat, Nudeln & Konserven", "icon": "🥫", "shelf_type": "pantry", "sort_order": 5},
|
|
{"name": "Gewürze, Öle & Soßen", "icon": "🧂", "shelf_type": "spices", "sort_order": 6},
|
|
{"name": "Tiefkühl", "icon": "❄️", "shelf_type": "frozen", "sort_order": 7},
|
|
{"name": "Getränke", "icon": "🧃", "shelf_type": "drinks", "sort_order": 8},
|
|
{"name": "Drogerie & Haushalt", "icon": "🧺", "shelf_type": "drugstore", "sort_order": 9},
|
|
]
|
|
|
|
SHELF_TYPES = [
|
|
{"type": "produce", "label": "Obst & Gemüse", "icon": "🥦"},
|
|
{"type": "bakery", "label": "Brot & Backwaren", "icon": "🍞"},
|
|
{"type": "pantry", "label": "Vorrat & Nudeln", "icon": "🥫"},
|
|
{"type": "spices", "label": "Gewürze & Öle", "icon": "🧂"},
|
|
{"type": "dairy", "label": "Kühlregal & Käse", "icon": "🧀"},
|
|
{"type": "meat", "label": "Fleisch & Fisch", "icon": "🥩"},
|
|
{"type": "frozen", "label": "Tiefkühl", "icon": "❄️"},
|
|
{"type": "drinks", "label": "Getränke", "icon": "🧃"},
|
|
{"type": "drugstore", "label": "Drogerie & Haushalt", "icon": "🧺"},
|
|
{"type": "other", "label": "Sonstiges", "icon": "🛒"},
|
|
]
|
|
|
|
|
|
def clean_source_label(source: Optional[str]) -> str:
|
|
"""Extrahiert einen sauberen, kurzen Rezepttitel aus der Quellangabe (ohne Wochenplan-Präfixe)."""
|
|
if not source:
|
|
return ""
|
|
src = str(source).strip()
|
|
import re
|
|
m = re.search(r'\((.+)\)$', src)
|
|
if m:
|
|
return m.group(1).strip()
|
|
return src
|
|
|
|
|
|
def clean_instructions_text(text: Optional[str]) -> str:
|
|
"""Entfernt vorangestellte Schrittnummern (z. B. '1. ', 'Schritt 1: ') aus Zubereitungszeilen."""
|
|
if not text:
|
|
return ""
|
|
import re
|
|
cleaned_lines = []
|
|
for line in str(text).splitlines():
|
|
line_s = line.strip()
|
|
if not line_s:
|
|
continue
|
|
cleaned = re.sub(r'^(?:Schritt\s*\d+[:.]?|\d+[\.\)\:])\s*', '', line_s, flags=re.IGNORECASE)
|
|
cleaned_lines.append(cleaned)
|
|
return "\n".join(cleaned_lines)
|
|
|
|
|
|
def save_ingredient_category_mapping(ingredient_name: str, category_name: str):
|
|
"""Speichert oder aktualisiert die bevorzugte Supermarkt-Kategorie einer Zutat."""
|
|
from .ingredients import clean_ingredient_name, get_ingredient_key
|
|
clean = clean_ingredient_name(ingredient_name)
|
|
key = get_ingredient_key(clean)
|
|
cat = (category_name or "").strip()
|
|
if not key or not cat:
|
|
return
|
|
|
|
try:
|
|
if DB_ENGINE == "mariadb":
|
|
execute_query(
|
|
"""
|
|
INSERT INTO ingredient_category_mappings (ingredient_name, category_name)
|
|
VALUES (?, ?)
|
|
ON DUPLICATE KEY UPDATE category_name = VALUES(category_name), updated_at = CURRENT_TIMESTAMP
|
|
""",
|
|
(key, cat),
|
|
fetch="none",
|
|
commit=True
|
|
)
|
|
else:
|
|
execute_query(
|
|
"""
|
|
INSERT INTO ingredient_category_mappings (ingredient_name, category_name)
|
|
VALUES (?, ?)
|
|
ON CONFLICT(ingredient_name) DO UPDATE SET category_name = excluded.category_name, updated_at = CURRENT_TIMESTAMP
|
|
""",
|
|
(key, cat),
|
|
fetch="none",
|
|
commit=True
|
|
)
|
|
except Exception as e:
|
|
print(f"Fehler beim Speichern des Kategorie-Mappings für '{key}': {e}")
|
|
|
|
|
|
def get_ingredient_category_mapping(ingredient_name: str) -> Optional[str]:
|
|
"""Sucht nach einer gelernten Nutzer-Zuordnung für diese Zutat."""
|
|
from .ingredients import clean_ingredient_name, get_ingredient_key
|
|
clean = clean_ingredient_name(ingredient_name)
|
|
key = get_ingredient_key(clean)
|
|
if not key:
|
|
return None
|
|
try:
|
|
row = execute_query(
|
|
"SELECT category_name FROM ingredient_category_mappings WHERE ingredient_name = ? LIMIT 1",
|
|
(key,),
|
|
fetch="one"
|
|
)
|
|
if row and row.get("category_name"):
|
|
return str(row["category_name"]).strip()
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def guess_category_for_ingredient(name: str) -> str:
|
|
"""Bestimmt automatisch die passende Supermarkt-Kategorie dynamisch, semantisch über Regal-Typen (shelf_type) und lernfähig."""
|
|
if not name:
|
|
return "Drogerie & Sonstiges"
|
|
|
|
# 1. Gespeichertes Nutzer-Mapping prüfen
|
|
learned = get_ingredient_category_mapping(name)
|
|
try:
|
|
active_cats = get_shopping_categories()
|
|
except Exception:
|
|
active_cats = []
|
|
|
|
cat_names = [c["name"] for c in active_cats]
|
|
if learned and (not cat_names or learned in cat_names):
|
|
return learned
|
|
|
|
from .ingredients import clean_ingredient_name
|
|
clean = clean_ingredient_name(name).lower()
|
|
|
|
# Hilfsfunktion: Finde Kategorie passend zum Ziel-Regaltyp (shelf_type) oder Name
|
|
def resolve_category(target_type: str, search_terms: List[str], fallback_name: str) -> str:
|
|
# Prio A: Exakter Match auf shelf_type
|
|
for cat in active_cats:
|
|
if cat.get("shelf_type") == target_type:
|
|
return cat["name"]
|
|
# Prio B: Namens-Match gegen bestehende Kategorien
|
|
for cat_name in cat_names:
|
|
cat_lower = cat_name.lower()
|
|
if any(term in cat_lower for term in search_terms):
|
|
return cat_name
|
|
# Prio C: Gibt es den Fallback-Namen?
|
|
if fallback_name in cat_names:
|
|
return fallback_name
|
|
# Prio D: Erste Kategorie oder Sonstiges
|
|
return cat_names[-1] if cat_names else fallback_name
|
|
|
|
# 2. Tiefkühl (frozen)
|
|
if any(k in clean for k in [
|
|
"tk", "tiefkühl", "gefroren", "eiscreme", "speiseeis", "erbsen (tk)", "tk-", "tiefgefroren"
|
|
]):
|
|
return resolve_category("frozen", ["tiefkühl", "tk", "frost", "eis"], "Tiefkühl")
|
|
|
|
# 3. Obst & Gemüse (produce)
|
|
produce_keywords = [
|
|
"zwiebel", "knoblauch", "kartoffel", "karotte", "möhre", "tomate", "paprika",
|
|
"brokkoli", "blumenkohl", "spinat", "zucchini", "aubergine", "gurke", "salat",
|
|
"lauch", "porree", "frühlingszwiebel", "lauchzwiebel", "sellerie", "kohlrabi", "champignon",
|
|
"pilz", "steinpilz", "pfifferling", "shiitake", "austernpilz", "kräuterseitling",
|
|
"apfel", "äpfel", "birne", "birnen", "banane", "bananen", "avocado", "zitrone", "zitronen",
|
|
"limette", "limetten", "orange", "orangen", "beere", "erdbeere", "himbeere",
|
|
"heidelbeere", "blaubeere", "brombeere", "kirsche", "kirschen", "pflaume", "zwetschge",
|
|
"mango", "ananas", "maracuja", "pfirsich", "nektarine", "aprikose", "melone", "wassermelone",
|
|
"petersilie", "schnittlauch", "basilikum", "koriander", "kresse", "gartenkresse", "brunnenkresse",
|
|
"ingwer", "chili", "peperoni", "kürbis", "radieschen", "rucola", "rosmarin",
|
|
"minze", "dill", "salbei", "kräuter", "kohl", "wirsing", "rotkohl", "weißkohl", "spitzkohl",
|
|
"grünkohl", "rosenkohl", "edamame", "sprossen", "keimlinge", "bimi", "chinakohl", "pak choi",
|
|
"fenchel", "pastinake", "rote bete", "rote beete", "schalotte", "babyspinat", "feldsalat",
|
|
"kopfsalat", "eisbergsalat", "chicoree", "radicchio", "feige", "traube", "weintraube"
|
|
]
|
|
if any(re.search(r"\b" + re.escape(k), clean) for k in produce_keywords):
|
|
return resolve_category("produce", ["obst", "gemüse", "frische", "früchte"], "Obst & Gemüse")
|
|
|
|
# 4. Kühlregal & Käse (dairy)
|
|
dairy_keywords = [
|
|
"milch", "butter", "sahne", "schmand", "quark", "joghurt", "käse", "cheddar",
|
|
"parmesan", "mozzarella", "feta", "gouda", "bergkäse", "frischkäse", "emmentaler",
|
|
"gruyère", "gruyere", "pecorino", "manchego", "halloumi", "burrata", "mascarpone",
|
|
"ricotta", "crème fraîche", "creme fraiche", "ei", "eier", "eigelb", "eiweiß",
|
|
"tofu", "räuchertofu", "seidentofu", "tempeh", "sauerrahm", "buttermilch", "kefir",
|
|
"skyr", "hafermilch", "sojamilch", "mandelmilch", "streukäse", "ghee"
|
|
]
|
|
if any(re.search(r"\b" + re.escape(k), clean) for k in dairy_keywords):
|
|
return resolve_category("dairy", ["kühl", "käse", "milch", "molkerei"], "Kühlregal & Käse")
|
|
|
|
# 5. Fleisch & Fisch (meat)
|
|
meat_keywords = [
|
|
"hackfleisch", "hack", "rinderhack", "hähnchen", "huhn", "pute", "speck",
|
|
"schinken", "wurst", "würstchen", "lachs", "garnele", "garnelen", "fisch",
|
|
"forelle", "kabeljau", "rind", "schwein", "kalb", "lamm", "steak", "filet",
|
|
"bratwurst", "wiener", "bacon", "meeresfrüchte", "thunfisch"
|
|
]
|
|
if any(re.search(r"\b" + re.escape(k), clean) for k in meat_keywords):
|
|
return resolve_category("meat", ["fleisch", "fisch", "metzger", "wurst"], "Fleisch & Fisch")
|
|
|
|
# 6. Brot & Backwaren (bakery)
|
|
bakery_keywords = [
|
|
"brot", "brötchen", "semmel", "semmeln", "toast", "toastbrot", "baguette",
|
|
"ciabatta", "fladenbrot", "tortilla", "wrap", "wraps", "pita", "naan",
|
|
"panko", "paniermehl", "semmelbrösel", "blätterteig", "croissant", "brioche"
|
|
]
|
|
if any(re.search(r"\b" + re.escape(k), clean) for k in bakery_keywords):
|
|
return resolve_category("bakery", ["brot", "backwaren", "bäckerei"], "Brot & Backwaren")
|
|
|
|
# 7. Gewürze & Öle (spices)
|
|
spice_keywords = [
|
|
"olivenöl", "rapsöl", "sonnenblumenöl", "sesamöl", "bratöl", "kokosöl", "leinöl", "erdnussöl",
|
|
"essig", "balsamico", "apfelessig", "weißweinessig", "rotweinessig", "salz", "meersalz",
|
|
"pfeffer", "paprikapulver", "curry", "kreuzkümmel", "kumin", "kümmel", "oregano", "thymian",
|
|
"majoran", "muskat", "muskatnuss", "cayennepfeffer", "zimt", "zimtstange", "lorbeer",
|
|
"vanilleschote", "vanille", "kurkuma", "chiliflocken", "kardamom", "nelken", "piment",
|
|
"wacholder", "garam masala", "ras el hanout"
|
|
]
|
|
if re.search(r"\böl\b", clean) or any(re.search(r"\b" + re.escape(k), clean) for k in spice_keywords):
|
|
return resolve_category("spices", ["gewürz", "öl", "essig"], "Gewürze & Öle")
|
|
|
|
# 8. Vorrat & Nudeln (pantry)
|
|
pantry_keywords = [
|
|
"nudel", "nudeln", "spaghetti", "penne", "mie", "pasta", "reis", "basmati",
|
|
"jasmin", "linsen", "kichererbsen", "bohnen", "kidneybohnen", "mais",
|
|
"tomatenmark", "passierte tomaten", "gehackte tomaten", "dosentomaten",
|
|
"mehl", "zucker", "puderzucker", "rohrzucker", "haferflocken", "kokosmilch",
|
|
"brühe", "gemüsebrühe", "rinderbrühe", "hühnerbrühe", "fond", "couscous", "bulgur",
|
|
"quinoa", "polenta", "grieß", "erdnussbutter", "marmelade", "honig", "ahornsirup",
|
|
"agavendicksaft", "senf", "ketchup", "mayonnaise", "schokolade", "kuvertüre", "kakao",
|
|
"toffifee", "nüsse", "mandeln", "walnüsse", "haselnüsse", "cashew", "cashews",
|
|
"macadamia", "sesam", "chiasamen", "leinsamen", "pinienkerne", "sojasauce",
|
|
"tamari", "teriyaki", "hoisin", "backpulver", "natron", "vanillezucker", "trockenhefe",
|
|
"sahnesteif", "speisestärke", "stärke"
|
|
]
|
|
if any(re.search(r"\b" + re.escape(k), clean) for k in pantry_keywords):
|
|
return resolve_category("pantry", ["vorrat", "nudel", "konserve", "trocken"], "Vorrat & Nudeln")
|
|
|
|
# 9. Getränke (drinks)
|
|
drink_keywords = [
|
|
"wasser", "mineralwasser", "saft", "apfelsaft", "orangensaft", "wein",
|
|
"weißwein", "rotwein", "sekt", "prosecco", "bier", "kaffee", "espresso",
|
|
"tee", "cola", "limonade", "sirup", "tonic"
|
|
]
|
|
if any(re.search(r"\b" + re.escape(k), clean) for k in drink_keywords):
|
|
return resolve_category("drinks", ["getränk", "durst", "wein", "saft"], "Getränke")
|
|
|
|
# Fallback: Drogerie & Sonstiges (drugstore / other)
|
|
return resolve_category("drugstore", ["drogerie", "haushalt", "sonstig"], "Drogerie & Sonstiges")
|
|
|
|
|
|
def get_connection():
|
|
if DB_ENGINE == "mariadb":
|
|
if pymysql is None:
|
|
raise RuntimeError("PyMySQL ist nicht installiert, wird aber für MariaDB benötigt.")
|
|
return pymysql.connect(
|
|
host=MARIADB_HOST,
|
|
port=MARIADB_PORT,
|
|
user=MARIADB_USER,
|
|
password=MARIADB_PASS,
|
|
database=MARIADB_NAME,
|
|
charset="utf8mb4",
|
|
cursorclass=pymysql.cursors.DictCursor,
|
|
autocommit=True
|
|
)
|
|
else:
|
|
os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
|
|
conn = sqlite3.connect(DB_PATH)
|
|
conn.row_factory = sqlite3.Row
|
|
return conn
|
|
|
|
|
|
def execute_query(query: str, params: tuple = (), fetch: str = "all", commit: bool = False):
|
|
conn = get_connection()
|
|
try:
|
|
if DB_ENGINE == "mariadb":
|
|
# MariaDB uses %s instead of ?
|
|
maria_query = query.replace("?", "%s")
|
|
with conn.cursor() as cursor:
|
|
cursor.execute(maria_query, params)
|
|
last_id = cursor.lastrowid
|
|
rowcount = cursor.rowcount
|
|
if fetch == "all":
|
|
rows = cursor.fetchall()
|
|
return rows
|
|
elif fetch == "one":
|
|
return cursor.fetchone()
|
|
else:
|
|
return {"lastrowid": last_id, "rowcount": rowcount}
|
|
else:
|
|
cursor = conn.cursor()
|
|
cursor.execute(query, params)
|
|
if commit:
|
|
conn.commit()
|
|
last_id = cursor.lastrowid
|
|
rowcount = cursor.rowcount
|
|
if fetch == "all":
|
|
rows = [dict(r) for r in cursor.fetchall()]
|
|
return rows
|
|
elif fetch == "one":
|
|
row = cursor.fetchone()
|
|
return dict(row) if row else None
|
|
else:
|
|
return {"lastrowid": last_id, "rowcount": rowcount}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def init_db():
|
|
conn = get_connection()
|
|
try:
|
|
if DB_ENGINE == "mariadb":
|
|
with conn.cursor() as cursor:
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS dishes (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
name VARCHAR(255) NOT NULL UNIQUE,
|
|
category VARCHAR(100) NOT NULL,
|
|
effort VARCHAR(20) NOT NULL,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
INDEX idx_dish_cat (category),
|
|
INDEX idx_dish_eff (effort)
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
|
|
""")
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS recipes (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
title VARCHAR(255) NOT NULL,
|
|
category VARCHAR(100) NOT NULL,
|
|
chapter_number INT NOT NULL,
|
|
effort VARCHAR(20) NOT NULL DEFAULT 'mittel',
|
|
prep_time VARCHAR(50) DEFAULT '',
|
|
cook_time VARCHAR(50) DEFAULT '',
|
|
portions INT DEFAULT 4,
|
|
ingredients LONGTEXT NOT NULL,
|
|
instructions LONGTEXT NOT NULL,
|
|
notes TEXT,
|
|
source VARCHAR(255) DEFAULT '',
|
|
image_url VARCHAR(500) DEFAULT '',
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
|
INDEX idx_recipe_chapter (chapter_number),
|
|
INDEX idx_recipe_cat (category)
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
|
|
""")
|
|
# Migration check for MariaDB
|
|
cursor.execute("""
|
|
SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS
|
|
WHERE TABLE_SCHEMA = %s AND TABLE_NAME = 'recipes' AND COLUMN_NAME = 'source'
|
|
""", (MARIADB_NAME,))
|
|
if not cursor.fetchone():
|
|
cursor.execute("ALTER TABLE recipes ADD COLUMN source VARCHAR(255) DEFAULT ''")
|
|
|
|
cursor.execute("""
|
|
SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS
|
|
WHERE TABLE_SCHEMA = %s AND TABLE_NAME = 'recipes' AND COLUMN_NAME = 'printed_at'
|
|
""", (MARIADB_NAME,))
|
|
if not cursor.fetchone():
|
|
cursor.execute("ALTER TABLE recipes ADD COLUMN printed_at DATETIME NULL DEFAULT NULL AFTER image_url")
|
|
|
|
cursor.execute("""
|
|
SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS
|
|
WHERE TABLE_SCHEMA = %s AND TABLE_NAME = 'recipes' AND COLUMN_NAME = 'subtitle'
|
|
""", (MARIADB_NAME,))
|
|
if not cursor.fetchone():
|
|
cursor.execute("ALTER TABLE recipes ADD COLUMN subtitle VARCHAR(255) DEFAULT '' AFTER title")
|
|
|
|
cursor.execute("""
|
|
SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS
|
|
WHERE TABLE_SCHEMA = %s AND TABLE_NAME = 'recipes' AND COLUMN_NAME = 'has_portion_table'
|
|
""", (MARIADB_NAME,))
|
|
if not cursor.fetchone():
|
|
cursor.execute("ALTER TABLE recipes ADD COLUMN has_portion_table TINYINT(1) NOT NULL DEFAULT 0 AFTER portions")
|
|
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS meal_plans (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
year INT NOT NULL,
|
|
week INT NOT NULL,
|
|
day_of_week INT NOT NULL,
|
|
meal_type VARCHAR(20) NOT NULL,
|
|
recipe_id INT NULL,
|
|
custom_text VARCHAR(255) NULL,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
|
UNIQUE KEY uq_meal_slot (year, week, day_of_week, meal_type),
|
|
INDEX idx_plan_week (year, week)
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
|
|
""")
|
|
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS shopping_list (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
item_name VARCHAR(255) NOT NULL,
|
|
amount VARCHAR(50) DEFAULT '',
|
|
unit VARCHAR(50) DEFAULT '',
|
|
checked TINYINT(1) NOT NULL DEFAULT 0,
|
|
source VARCHAR(255) DEFAULT '',
|
|
category VARCHAR(100) DEFAULT 'Drogerie & Sonstiges',
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
|
INDEX idx_shop_checked (checked)
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
|
|
""")
|
|
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS shopping_categories (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
name VARCHAR(100) NOT NULL UNIQUE,
|
|
icon VARCHAR(50) DEFAULT '🛒',
|
|
shelf_type VARCHAR(50) DEFAULT 'other',
|
|
sort_order INT NOT NULL DEFAULT 0,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
|
|
""")
|
|
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS ingredient_category_mappings (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
ingredient_name VARCHAR(150) NOT NULL UNIQUE,
|
|
category_name VARCHAR(100) NOT NULL,
|
|
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
|
INDEX idx_ing_cat_name (ingredient_name)
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
|
|
""")
|
|
|
|
try:
|
|
cursor.execute("""
|
|
SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS
|
|
WHERE TABLE_SCHEMA = %s AND TABLE_NAME = 'shopping_categories' AND COLUMN_NAME = 'shelf_type'
|
|
""", (MARIADB_NAME,))
|
|
if not cursor.fetchone():
|
|
cursor.execute("ALTER TABLE shopping_categories ADD COLUMN shelf_type VARCHAR(50) DEFAULT 'other' AFTER icon")
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
cursor.execute("""
|
|
SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS
|
|
WHERE TABLE_SCHEMA = %s AND TABLE_NAME = 'shopping_list' AND COLUMN_NAME = 'category'
|
|
""", (MARIADB_NAME,))
|
|
if not cursor.fetchone():
|
|
cursor.execute("ALTER TABLE shopping_list ADD COLUMN category VARCHAR(100) DEFAULT 'Drogerie & Sonstiges' AFTER source")
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
cursor.execute("""
|
|
SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS
|
|
WHERE TABLE_SCHEMA = %s AND TABLE_NAME = 'shopping_list' AND COLUMN_NAME = 'note'
|
|
""", (MARIADB_NAME,))
|
|
if not cursor.fetchone():
|
|
cursor.execute("ALTER TABLE shopping_list ADD COLUMN note VARCHAR(255) DEFAULT '' AFTER category")
|
|
except Exception:
|
|
pass
|
|
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS recipe_shares (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
recipe_id INT NOT NULL,
|
|
token VARCHAR(64) NOT NULL UNIQUE,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
expires_at DATETIME NULL DEFAULT NULL,
|
|
views_count INT DEFAULT 0,
|
|
revoked TINYINT DEFAULT 0,
|
|
label VARCHAR(100) DEFAULT '',
|
|
INDEX idx_share_token (token),
|
|
INDEX idx_share_recipe (recipe_id)
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
|
|
""")
|
|
|
|
cursor.execute("DELETE FROM dishes")
|
|
else:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS dishes (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL UNIQUE COLLATE NOCASE,
|
|
category TEXT NOT NULL,
|
|
effort TEXT NOT NULL,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
""")
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS recipes (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
title TEXT NOT NULL,
|
|
subtitle TEXT DEFAULT '',
|
|
category TEXT NOT NULL,
|
|
chapter_number INTEGER NOT NULL,
|
|
effort TEXT NOT NULL DEFAULT 'mittel',
|
|
prep_time TEXT DEFAULT '',
|
|
cook_time TEXT DEFAULT '',
|
|
portions INTEGER DEFAULT 4,
|
|
has_portion_table INTEGER DEFAULT 0,
|
|
ingredients TEXT NOT NULL,
|
|
instructions TEXT NOT NULL,
|
|
notes TEXT DEFAULT '',
|
|
source TEXT DEFAULT '',
|
|
image_url TEXT DEFAULT '',
|
|
printed_at TIMESTAMP NULL DEFAULT NULL,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
""")
|
|
# Migration check for SQLite
|
|
cursor.execute("PRAGMA table_info(recipes)")
|
|
existing_cols = [c[1] for c in cursor.fetchall()]
|
|
if "subtitle" not in existing_cols:
|
|
cursor.execute("ALTER TABLE recipes ADD COLUMN subtitle TEXT DEFAULT ''")
|
|
if "source" not in existing_cols:
|
|
cursor.execute("ALTER TABLE recipes ADD COLUMN source TEXT DEFAULT ''")
|
|
if "printed_at" not in existing_cols:
|
|
cursor.execute("ALTER TABLE recipes ADD COLUMN printed_at TIMESTAMP NULL DEFAULT NULL")
|
|
if "has_portion_table" not in existing_cols:
|
|
cursor.execute("ALTER TABLE recipes ADD COLUMN has_portion_table INTEGER DEFAULT 0")
|
|
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS meal_plans (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
year INTEGER NOT NULL,
|
|
week INTEGER NOT NULL,
|
|
day_of_week INTEGER NOT NULL,
|
|
meal_type TEXT NOT NULL,
|
|
recipe_id INTEGER NULL,
|
|
custom_text TEXT NULL,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
UNIQUE (year, week, day_of_week, meal_type)
|
|
)
|
|
""")
|
|
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS shopping_list (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
item_name TEXT NOT NULL,
|
|
amount TEXT DEFAULT '',
|
|
unit TEXT DEFAULT '',
|
|
checked INTEGER NOT NULL DEFAULT 0,
|
|
source TEXT DEFAULT '',
|
|
category TEXT DEFAULT 'Drogerie & Sonstiges',
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
""")
|
|
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS shopping_categories (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL UNIQUE COLLATE NOCASE,
|
|
icon TEXT DEFAULT '🛒',
|
|
shelf_type TEXT DEFAULT 'other',
|
|
sort_order INTEGER NOT NULL DEFAULT 0,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
""")
|
|
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS ingredient_category_mappings (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
ingredient_name TEXT NOT NULL UNIQUE COLLATE NOCASE,
|
|
category_name TEXT NOT NULL,
|
|
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
""")
|
|
|
|
cursor.execute("PRAGMA table_info(shopping_categories)")
|
|
existing_cat_cols = [c[1] for c in cursor.fetchall()]
|
|
if "shelf_type" not in existing_cat_cols:
|
|
cursor.execute("ALTER TABLE shopping_categories ADD COLUMN shelf_type TEXT DEFAULT 'other'")
|
|
|
|
cursor.execute("PRAGMA table_info(shopping_list)")
|
|
existing_shop_cols = [c[1] for c in cursor.fetchall()]
|
|
if "category" not in existing_shop_cols:
|
|
cursor.execute("ALTER TABLE shopping_list ADD COLUMN category TEXT DEFAULT 'Drogerie & Sonstiges'")
|
|
if "note" not in existing_shop_cols:
|
|
cursor.execute("ALTER TABLE shopping_list ADD COLUMN note TEXT DEFAULT ''")
|
|
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS recipe_shares (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
recipe_id INTEGER NOT NULL,
|
|
token TEXT NOT NULL UNIQUE,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
expires_at DATETIME NULL,
|
|
views_count INTEGER DEFAULT 0,
|
|
revoked INTEGER DEFAULT 0,
|
|
label TEXT DEFAULT ''
|
|
)
|
|
""")
|
|
cursor.execute("CREATE INDEX IF NOT EXISTS idx_share_token ON recipe_shares (token)")
|
|
cursor.execute("CREATE INDEX IF NOT EXISTS idx_share_recipe ON recipe_shares (recipe_id)")
|
|
|
|
cursor.execute("DELETE FROM dishes")
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
# Standard-Kategorien initialisieren & Altdaten abgleichen
|
|
seed_default_shopping_categories_if_empty()
|
|
|
|
|
|
# --- Dishes (Zufallsgenerator) ---
|
|
|
|
def get_all_dishes(category: Optional[str] = None, effort: Optional[str] = None, search: Optional[str] = None) -> List[Dict[str, Any]]:
|
|
query = "SELECT id, name, category, effort, created_at FROM dishes WHERE 1=1"
|
|
params = []
|
|
|
|
if category and category.strip() and category != "alle":
|
|
query += " AND category = ?"
|
|
params.append(category.strip())
|
|
|
|
if effort and effort.strip() and effort != "alle":
|
|
query += " AND effort = ?"
|
|
params.append(effort.strip())
|
|
|
|
if search and search.strip():
|
|
query += " AND name LIKE ?"
|
|
params.append(f"%{search.strip()}%")
|
|
|
|
query += " ORDER BY category ASC, name ASC"
|
|
return execute_query(query, tuple(params), fetch="all")
|
|
|
|
|
|
def add_dish(name: str, category: str, effort: str) -> Dict[str, Any]:
|
|
name = name.strip()
|
|
category = category.strip()
|
|
effort = effort.strip()
|
|
|
|
if not name:
|
|
raise ValueError("Name darf nicht leer sein.")
|
|
if category not in CATEGORIES:
|
|
raise ValueError(f"Ungültige Kategorie '{category}'. Gültig sind: {', '.join(CATEGORIES)}")
|
|
if effort not in EFFORTS:
|
|
raise ValueError(f"Ungültiger Aufwand '{effort}'. Gültig sind: {', '.join(EFFORTS)}")
|
|
|
|
try:
|
|
res = execute_query(
|
|
"INSERT INTO dishes (name, category, effort) VALUES (?, ?, ?)",
|
|
(name, category, effort),
|
|
fetch="none",
|
|
commit=True
|
|
)
|
|
dish_id = res["lastrowid"]
|
|
return execute_query("SELECT id, name, category, effort, created_at FROM dishes WHERE id = ?", (dish_id,), fetch="one")
|
|
except Exception as e:
|
|
if "UNIQUE" in str(e).upper() or "DUPLICATE" in str(e).upper():
|
|
raise ValueError(f"Ein Gericht mit dem Namen '{name}' existiert bereits.")
|
|
raise
|
|
|
|
|
|
def update_dish(dish_id: int, name: str, category: str, effort: str) -> Dict[str, Any]:
|
|
name = name.strip()
|
|
category = category.strip()
|
|
effort = effort.strip()
|
|
|
|
if not name:
|
|
raise ValueError("Name darf nicht leer sein.")
|
|
if category not in CATEGORIES:
|
|
raise ValueError(f"Ungültige Kategorie '{category}'.")
|
|
if effort not in EFFORTS:
|
|
raise ValueError(f"Ungültiger Aufwand '{effort}'.")
|
|
|
|
try:
|
|
res = execute_query(
|
|
"UPDATE dishes SET name = ?, category = ?, effort = ? WHERE id = ?",
|
|
(name, category, effort, dish_id),
|
|
fetch="none",
|
|
commit=True
|
|
)
|
|
if res["rowcount"] == 0:
|
|
raise KeyError(f"Gericht #{dish_id} nicht gefunden.")
|
|
return execute_query("SELECT id, name, category, effort, created_at FROM dishes WHERE id = ?", (dish_id,), fetch="one")
|
|
except Exception as e:
|
|
if "UNIQUE" in str(e).upper() or "DUPLICATE" in str(e).upper():
|
|
raise ValueError(f"Ein Gericht mit dem Namen '{name}' existiert bereits.")
|
|
raise
|
|
|
|
|
|
def delete_dish(dish_id: int) -> bool:
|
|
res = execute_query("DELETE FROM dishes WHERE id = ?", (dish_id,), fetch="none", commit=True)
|
|
return res["rowcount"] > 0
|
|
|
|
|
|
def clear_all_dishes():
|
|
execute_query("DELETE FROM dishes", fetch="none", commit=True)
|
|
if DB_ENGINE == "sqlite":
|
|
try:
|
|
execute_query("DELETE FROM sqlite_sequence WHERE name = 'dishes'", fetch="none", commit=True)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def get_stats() -> Dict[str, Any]:
|
|
total_res = execute_query("SELECT COUNT(*) as total FROM dishes", fetch="one")
|
|
total = total_res["total"] if total_res else 0
|
|
|
|
cat_rows = execute_query("SELECT category, COUNT(*) as cnt FROM dishes GROUP BY category", fetch="all")
|
|
by_category = {row["category"]: row["cnt"] for row in cat_rows}
|
|
|
|
eff_rows = execute_query("SELECT effort, COUNT(*) as cnt FROM dishes GROUP BY effort", fetch="all")
|
|
by_effort = {row["effort"]: row["cnt"] for row in eff_rows}
|
|
|
|
recipe_count_res = execute_query("SELECT COUNT(*) as total FROM recipes", fetch="one")
|
|
total_recipes = recipe_count_res["total"] if recipe_count_res else 0
|
|
|
|
return {
|
|
"total": total,
|
|
"by_category": by_category,
|
|
"by_effort": by_effort,
|
|
"total_recipes": total_recipes
|
|
}
|
|
|
|
|
|
def resolve_category_terms(cat_input: Optional[str]) -> List[str]:
|
|
if not cat_input or str(cat_input).strip().lower() in ["", "alle", "beliebig", "none"]:
|
|
return []
|
|
s = str(cat_input).strip()
|
|
chap_num = None
|
|
if s.isdigit():
|
|
chap_num = int(s)
|
|
else:
|
|
for c in CHAPTERS:
|
|
if s.lower() == c["title"].lower() or s.lower() in c["title"].lower():
|
|
chap_num = c["number"]
|
|
break
|
|
|
|
chapter_terms = {
|
|
1: ["Suppen", "Eintöpfe", "Suppe", "Eintopf", "Suppen & Eintöpfe", "Chili", "Gulasch"],
|
|
2: ["Traditionell & Basics", "Hausmannskost", "Traditionell", "Basics", "Fisch", "Brotzeit", "Klassiker", "Schnitzel", "Rouladen", "Klopse", "Forelle", "Lachs"],
|
|
3: ["Pasta & Nudelgerichte", "Nudeln", "Pasta", "Spaghetti", "Penne", "Tagliatelle", "Carbonara", "Lasagne", "Spätzle"],
|
|
4: ["Kartoffeln", "Kartoffel", "Gratin", "Pommes", "Püree"],
|
|
5: ["Reis & Getreide", "Reis", "Getreide", "Risotto", "Paella"],
|
|
6: ["Teig & Gebackenes", "Teigig", "Teig", "Pizza", "Flammkuchen", "Quiche", "Calzone"],
|
|
7: ["Asiatische Küche", "asiatisch", "Asiatisch", "Curry", "Pad Thai", "Bami", "Wok", "Ramen", "Sushi"],
|
|
8: ["Süßspeisen", "Süßspeise", "Kaiserschmarrn", "Germknödel", "Dampfnudel", "Apfelküchle", "Milchreis"],
|
|
9: ["Desserts", "Dessert", "Tiramisu", "Mousse", "Pudding"],
|
|
10: ["Kuchen & Gebäck", "Kuchen", "Gebäck", "Torte"],
|
|
11: ["Plätzchen", "Kekse", "Keks"]
|
|
}
|
|
|
|
if chap_num and chap_num in chapter_terms:
|
|
return chapter_terms[chap_num]
|
|
|
|
# Legacy category names mapping
|
|
legacy_map = {
|
|
"nudeln": chapter_terms[3],
|
|
"pasta": chapter_terms[3],
|
|
"reis": chapter_terms[5],
|
|
"kartoffeln": chapter_terms[4],
|
|
"teigig": chapter_terms[6],
|
|
"suppen": chapter_terms[1],
|
|
"eintöpfe": chapter_terms[1],
|
|
"eintoepfe": chapter_terms[1],
|
|
"hausmannskost": chapter_terms[2],
|
|
"asiatisch": chapter_terms[7],
|
|
"süßspeisen": chapter_terms[8],
|
|
"suessspeisen": chapter_terms[8],
|
|
"fisch": ["Fisch", "Forelle", "Lachs", "Kabeljau", "Hausmannskost"]
|
|
}
|
|
if s.lower() in legacy_map:
|
|
return legacy_map[s.lower()]
|
|
|
|
return [s]
|
|
|
|
|
|
def get_chapter_info_for_dish(dish_category: str = "", dish_name: str = "", recipe_chapter: Optional[int] = None) -> Dict[str, Any]:
|
|
if recipe_chapter and 1 <= recipe_chapter <= len(CHAPTERS):
|
|
return CHAPTERS[recipe_chapter - 1]
|
|
|
|
combo = f"{dish_category or ''} {dish_name or ''}".lower()
|
|
|
|
if any(k in combo for k in ["suppe", "eintopf", "chili", "brühe"]):
|
|
return CHAPTERS[0] # 1: Suppen & Eintöpfe
|
|
if any(k in combo for k in ["nudel", "pasta", "spaghetti", "penne", "tagliatelle", "carbonara", "bolognese", "spätzle", "lasagne"]):
|
|
return CHAPTERS[2] # 3: Pasta & Nudelgerichte
|
|
if any(k in combo for k in ["kartoffel", "pommes", "gratin", "püree"]):
|
|
return CHAPTERS[3] # 4: Kartoffeln
|
|
if any(k in combo for k in ["curry", "pad thai", "bami", "wok", "ramen", "sushi", "asiatisch"]):
|
|
return CHAPTERS[6] # 7: Asiatische Küche
|
|
if any(k in combo for k in ["reis", "risotto", "paella", "djuvec"]):
|
|
return CHAPTERS[4] # 5: Reis & Getreide
|
|
if any(k in combo for k in ["pizza", "flammkuchen", "quiche", "calzone", "teig", "brot"]):
|
|
return CHAPTERS[5] # 6: Teig & Gebackenes
|
|
if any(k in combo for k in ["kuchen", "torte", "muffin", "gebäck"]):
|
|
return CHAPTERS[9] # 10: Kuchen & Gebäck
|
|
if any(k in combo for k in ["keks", "plätzchen"]):
|
|
return CHAPTERS[10] # 11: Plätzchen
|
|
if any(k in combo for k in ["tiramisu", "mousse", "pudding", "dessert"]):
|
|
return CHAPTERS[8] # 9: Desserts
|
|
if any(k in combo for k in ["kaiserschmarrn", "knödel", "apfelküchle", "dampfnudel", "milchreis", "süßspeis"]):
|
|
return CHAPTERS[7] # 8: Süßspeisen
|
|
|
|
return CHAPTERS[1] # 2: Traditionell & Basics
|
|
|
|
|
|
def query_recipe_candidates(chapter_number: Optional[int] = None, efforts: Optional[List[str]] = None) -> List[Dict[str, Any]]:
|
|
"""Gibt alle Rezepte aus der recipes-Tabelle zurück, die zu den Shuffle-Kriterien passen."""
|
|
all_recipes = get_all_recipes()
|
|
if not all_recipes:
|
|
return []
|
|
|
|
# 1. Nach Kapitel filtern (wenn spezifiziert)
|
|
if chapter_number is not None and chapter_number > 0:
|
|
pool = [r for r in all_recipes if r.get("chapter_number") == chapter_number]
|
|
else:
|
|
# Standard beim Zufallsgenerator: Kategorien Desserts (9), Kuchen & Gebäck (10) und Plätzchen (11) ausschließen
|
|
EXCLUDED_SHUFFLE_CHAPTERS = {9, 10, 11}
|
|
pool = [r for r in all_recipes if r.get("chapter_number") not in EXCLUDED_SHUFFLE_CHAPTERS]
|
|
|
|
if not pool:
|
|
return []
|
|
|
|
# 2. Nach Aufwand filtern (nur, wenn eine Teilmenge gewählt wurde)
|
|
if efforts and len(efforts) > 0 and len(efforts) < len(EFFORTS):
|
|
clean_efforts = [e.strip().lower() for e in efforts if e.strip() in EFFORTS]
|
|
effort_pool = [r for r in pool if (r.get("effort") or "").lower() in clean_efforts]
|
|
if effort_pool:
|
|
return effort_pool
|
|
|
|
return pool
|
|
|
|
|
|
def query_candidates(category: Optional[str], efforts: Optional[List[str]]) -> List[Dict[str, Any]]:
|
|
"""Fallback-Funktion für alte Dish-Abfragen"""
|
|
return query_recipe_candidates()
|
|
|
|
|
|
# --- Recipes (Kochbuch) ---
|
|
|
|
def format_printed_date(p_at: Any) -> str:
|
|
if not p_at:
|
|
return ""
|
|
if hasattr(p_at, "strftime"):
|
|
return p_at.strftime("%d.%m.%Y")
|
|
try:
|
|
parts = str(p_at).split(" ")[0].split("-")
|
|
if len(parts) == 3:
|
|
return f"{parts[2]}.{parts[1]}.{parts[0]}"
|
|
except Exception:
|
|
pass
|
|
return str(p_at)
|
|
|
|
|
|
def add_recipe(
|
|
title: str,
|
|
category: str,
|
|
chapter_number: int,
|
|
effort: str = "mittel",
|
|
prep_time: str = "",
|
|
cook_time: str = "",
|
|
portions: int = 4,
|
|
ingredients: str = "",
|
|
instructions: str = "",
|
|
notes: str = "",
|
|
source: str = "",
|
|
image_url: str = "",
|
|
printed_at: Optional[str] = None,
|
|
has_portion_table: int = 0,
|
|
subtitle: str = ""
|
|
) -> Dict[str, Any]:
|
|
title = title.strip()
|
|
subtitle = (subtitle or "").strip()
|
|
category = category.strip()
|
|
effort = effort.strip().lower()
|
|
source = source.strip()
|
|
|
|
if not title:
|
|
raise ValueError("Rezepttitel darf nicht leer sein.")
|
|
if not category:
|
|
raise ValueError("Kategorie darf nicht leer sein.")
|
|
if chapter_number not in [c["number"] for c in CHAPTERS]:
|
|
# Fallback to mapped chapter number if possible
|
|
chapter_number = CHAPTER_MAP.get(category, 1)
|
|
|
|
instructions = clean_instructions_text(instructions)
|
|
|
|
res = execute_query(
|
|
"""
|
|
INSERT INTO recipes (title, subtitle, category, chapter_number, effort, prep_time, cook_time, portions, has_portion_table, ingredients, instructions, notes, source, image_url, printed_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(title, subtitle, category, chapter_number, effort, prep_time, cook_time, portions, int(bool(has_portion_table)), ingredients, instructions, notes, source, image_url, printed_at),
|
|
fetch="none",
|
|
commit=True
|
|
)
|
|
recipe_id = res["lastrowid"]
|
|
return get_recipe(recipe_id)
|
|
|
|
|
|
def update_recipe(
|
|
recipe_id: int,
|
|
title: str,
|
|
category: str,
|
|
chapter_number: int,
|
|
effort: str = "mittel",
|
|
prep_time: str = "",
|
|
cook_time: str = "",
|
|
portions: int = 4,
|
|
ingredients: str = "",
|
|
instructions: str = "",
|
|
notes: str = "",
|
|
source: str = "",
|
|
image_url: Optional[str] = None,
|
|
printed_at: Optional[str] = None,
|
|
has_portion_table: Optional[int] = None,
|
|
subtitle: Optional[str] = None
|
|
) -> Dict[str, Any]:
|
|
title = title.strip()
|
|
category = category.strip()
|
|
effort = effort.strip().lower()
|
|
source = source.strip()
|
|
|
|
if not title:
|
|
raise ValueError("Rezepttitel darf nicht leer sein.")
|
|
|
|
existing = get_recipe(recipe_id)
|
|
if not existing:
|
|
raise KeyError(f"Rezept #{recipe_id} nicht gefunden.")
|
|
|
|
use_subtitle = (subtitle if subtitle is not None else existing.get("subtitle", "")).strip()
|
|
|
|
instructions = clean_instructions_text(instructions)
|
|
|
|
use_portion_table = int(bool(has_portion_table)) if has_portion_table is not None else existing.get("has_portion_table", 0)
|
|
|
|
if image_url is not None:
|
|
query = """
|
|
UPDATE recipes
|
|
SET title = ?, subtitle = ?, category = ?, chapter_number = ?, effort = ?, prep_time = ?, cook_time = ?, portions = ?, has_portion_table = ?, ingredients = ?, instructions = ?, notes = ?, source = ?, image_url = ?, printed_at = ?
|
|
WHERE id = ?
|
|
"""
|
|
params = (title, use_subtitle, category, chapter_number, effort, prep_time, cook_time, portions, use_portion_table, ingredients, instructions, notes, source, image_url, printed_at, recipe_id)
|
|
else:
|
|
query = """
|
|
UPDATE recipes
|
|
SET title = ?, subtitle = ?, category = ?, chapter_number = ?, effort = ?, prep_time = ?, cook_time = ?, portions = ?, has_portion_table = ?, ingredients = ?, instructions = ?, notes = ?, source = ?, printed_at = ?
|
|
WHERE id = ?
|
|
"""
|
|
params = (title, use_subtitle, category, chapter_number, effort, prep_time, cook_time, portions, use_portion_table, ingredients, instructions, notes, source, printed_at, recipe_id)
|
|
|
|
res = execute_query(query, params, fetch="none", commit=True)
|
|
return get_recipe(recipe_id)
|
|
|
|
|
|
def delete_recipe(recipe_id: int) -> bool:
|
|
res = execute_query("DELETE FROM recipes WHERE id = ?", (recipe_id,), fetch="none", commit=True)
|
|
return res["rowcount"] > 0
|
|
|
|
|
|
def get_recipe(recipe_id: int) -> Optional[Dict[str, Any]]:
|
|
row = execute_query("SELECT * FROM recipes WHERE id = ?", (recipe_id,), fetch="one")
|
|
if row:
|
|
row["printed_at_formatted"] = format_printed_date(row.get("printed_at"))
|
|
row["has_portion_table"] = bool(row.get("has_portion_table", 0))
|
|
return row
|
|
|
|
|
|
def get_all_recipes(
|
|
chapter_number: Optional[int] = None,
|
|
category: Optional[str] = None,
|
|
search: Optional[str] = None,
|
|
printed_status: Optional[str] = None
|
|
) -> List[Dict[str, Any]]:
|
|
query = "SELECT * FROM recipes WHERE 1=1"
|
|
params = []
|
|
|
|
if chapter_number is not None and chapter_number > 0:
|
|
query += " AND chapter_number = ?"
|
|
params.append(chapter_number)
|
|
|
|
if category and category.strip() and category != "alle":
|
|
query += " AND category = ?"
|
|
params.append(category.strip())
|
|
|
|
if search and search.strip():
|
|
query += " AND (title LIKE ? OR subtitle LIKE ? OR ingredients LIKE ? OR instructions LIKE ?)"
|
|
s = f"%{search.strip()}%"
|
|
params.extend([s, s, s, s])
|
|
|
|
if printed_status == "unprinted":
|
|
query += " AND printed_at IS NULL"
|
|
elif printed_status == "printed":
|
|
query += " AND printed_at IS NOT NULL"
|
|
|
|
query += " ORDER BY chapter_number ASC, title ASC"
|
|
rows = execute_query(query, tuple(params), fetch="all")
|
|
for r in rows:
|
|
r["printed_at_formatted"] = format_printed_date(r.get("printed_at"))
|
|
r["has_portion_table"] = bool(r.get("has_portion_table", 0))
|
|
return rows
|
|
|
|
|
|
def mark_recipes_printed(recipe_ids: List[int], print_date: Optional[str] = None) -> int:
|
|
"""Markiert die angegebenen Rezepte als gedruckt."""
|
|
if not recipe_ids:
|
|
return 0
|
|
date_val = print_date or datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
placeholders = ", ".join(["?"] * len(recipe_ids))
|
|
query = f"UPDATE recipes SET printed_at = ? WHERE id IN ({placeholders})"
|
|
params = [date_val] + list(recipe_ids)
|
|
res = execute_query(query, tuple(params), fetch="none", commit=True)
|
|
return res.get("rowcount", 0)
|
|
|
|
|
|
def unmark_recipes_printed(recipe_ids: List[int]) -> int:
|
|
"""Entfernt den Gedruckt-Status von den angegebenen Rezepten."""
|
|
if not recipe_ids:
|
|
return 0
|
|
placeholders = ", ".join(["?"] * len(recipe_ids))
|
|
query = f"UPDATE recipes SET printed_at = NULL WHERE id IN ({placeholders})"
|
|
res = execute_query(query, tuple(recipe_ids), fetch="none", commit=True)
|
|
return res.get("rowcount", 0)
|
|
|
|
|
|
def toggle_recipe_printed(recipe_id: int) -> Dict[str, Any]:
|
|
"""Schaltet den Druckstatus eines Rezepts um."""
|
|
rec = get_recipe(recipe_id)
|
|
if not rec:
|
|
raise KeyError(f"Rezept #{recipe_id} nicht gefunden.")
|
|
if rec.get("printed_at"):
|
|
unmark_recipes_printed([recipe_id])
|
|
else:
|
|
mark_recipes_printed([recipe_id])
|
|
return get_recipe(recipe_id)
|
|
|
|
|
|
# ==========================================
|
|
# WOCHENPLAN (MEAL PLANNER)
|
|
# ==========================================
|
|
|
|
def get_meal_plan(year: int, week: int) -> List[Dict[str, Any]]:
|
|
"""Gibt alle Einträge des Wochenplans für die angegebene KW zurück."""
|
|
query = """
|
|
SELECT
|
|
m.id, m.year, m.week, m.day_of_week, m.meal_type, m.recipe_id, m.custom_text,
|
|
r.title AS recipe_title, r.image_url AS recipe_image_url, r.chapter_number AS recipe_chapter_number,
|
|
r.category AS recipe_category
|
|
FROM meal_plans m
|
|
LEFT JOIN recipes r ON m.recipe_id = r.id
|
|
WHERE m.year = ? AND m.week = ?
|
|
ORDER BY m.day_of_week ASC,
|
|
CASE m.meal_type
|
|
WHEN 'frueh' THEN 1
|
|
WHEN 'mittag' THEN 2
|
|
WHEN 'kaffee' THEN 3
|
|
WHEN 'abend' THEN 4
|
|
ELSE 5
|
|
END ASC
|
|
"""
|
|
return execute_query(query, (year, week), fetch="all")
|
|
|
|
|
|
def set_meal_plan_slot(
|
|
year: int,
|
|
week: int,
|
|
day_of_week: int,
|
|
meal_type: str,
|
|
recipe_id: Optional[int] = None,
|
|
custom_text: Optional[str] = None
|
|
) -> Dict[str, Any]:
|
|
"""Setzt oder aktualisiert einen Slot im Wochenplan."""
|
|
if DB_ENGINE == "mariadb":
|
|
query = """
|
|
INSERT INTO meal_plans (year, week, day_of_week, meal_type, recipe_id, custom_text)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
ON DUPLICATE KEY UPDATE
|
|
recipe_id = VALUES(recipe_id),
|
|
custom_text = VALUES(custom_text),
|
|
updated_at = CURRENT_TIMESTAMP
|
|
"""
|
|
execute_query(query, (year, week, day_of_week, meal_type, recipe_id, custom_text), fetch="none", commit=True)
|
|
else:
|
|
query = """
|
|
INSERT INTO meal_plans (year, week, day_of_week, meal_type, recipe_id, custom_text, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP)
|
|
ON CONFLICT(year, week, day_of_week, meal_type) DO UPDATE SET
|
|
recipe_id = excluded.recipe_id,
|
|
custom_text = excluded.custom_text,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
"""
|
|
execute_query(query, (year, week, day_of_week, meal_type, recipe_id, custom_text), fetch="none", commit=True)
|
|
|
|
slot_query = """
|
|
SELECT
|
|
m.id, m.year, m.week, m.day_of_week, m.meal_type, m.recipe_id, m.custom_text,
|
|
r.title AS recipe_title, r.image_url AS recipe_image_url
|
|
FROM meal_plans m
|
|
LEFT JOIN recipes r ON m.recipe_id = r.id
|
|
WHERE m.year = ? AND m.week = ? AND m.day_of_week = ? AND m.meal_type = ?
|
|
"""
|
|
return execute_query(slot_query, (year, week, day_of_week, meal_type), fetch="one")
|
|
|
|
|
|
def delete_meal_plan_slot(year: int, week: int, day_of_week: int, meal_type: str) -> bool:
|
|
"""Löscht einen einzelnen Slot aus dem Wochenplan."""
|
|
query = "DELETE FROM meal_plans WHERE year = ? AND week = ? AND day_of_week = ? AND meal_type = ?"
|
|
res = execute_query(query, (year, week, day_of_week, meal_type), fetch="none", commit=True)
|
|
return res.get("rowcount", 0) > 0
|
|
|
|
|
|
def clear_meal_plan_week(year: int, week: int) -> int:
|
|
"""Leert den gesamten Wochenplan für eine Kalenderwoche."""
|
|
query = "DELETE FROM meal_plans WHERE year = ? AND week = ?"
|
|
res = execute_query(query, (year, week), fetch="none", commit=True)
|
|
return res.get("rowcount", 0)
|
|
|
|
|
|
# --- Shopping Categories & Shopping List (Einkaufsliste) ---
|
|
|
|
def seed_default_shopping_categories_if_empty(preset: str = "standard"):
|
|
"""Initialisiert die Standard-Kategorien, falls noch keine vorhanden sind, und kategorisiert alte Einträge."""
|
|
try:
|
|
cats = execute_query("SELECT COUNT(*) as cnt FROM shopping_categories", fetch="one")
|
|
cnt = cats["cnt"] if cats else 0
|
|
if cnt == 0:
|
|
cat_list = DISCOUNTER_SHOPPING_CATEGORIES if preset == "discounter" else DEFAULT_SHOPPING_CATEGORIES
|
|
for c in cat_list:
|
|
execute_query(
|
|
"INSERT INTO shopping_categories (name, icon, shelf_type, sort_order) VALUES (?, ?, ?, ?)",
|
|
(c["name"], c["icon"], c.get("shelf_type", "other"), c["sort_order"]),
|
|
fetch="none",
|
|
commit=True
|
|
)
|
|
# Bestehende Einträge ohne Kategorie automatisch zuweisen
|
|
items = execute_query("SELECT id, item_name, category FROM shopping_list", fetch="all")
|
|
for it in items:
|
|
cur_cat = it.get("category")
|
|
if not cur_cat or cur_cat in ("", "None", "null", "Sonstiges") or cur_cat == "Drogerie & Sonstiges":
|
|
guessed = guess_category_for_ingredient(it.get("item_name", ""))
|
|
if guessed != cur_cat:
|
|
execute_query("UPDATE shopping_list SET category = ? WHERE id = ?", (guessed, it["id"]), fetch="none", commit=True)
|
|
except Exception as e:
|
|
print(f"Hinweis beim Initialisieren der Shopping-Kategorien: {e}")
|
|
|
|
|
|
def get_shopping_categories() -> List[Dict[str, Any]]:
|
|
"""Gibt alle Supermarkt-Bereiche/Kategorien sortiert nach Reihenfolge zurück."""
|
|
query = "SELECT id, name, icon, shelf_type, sort_order, created_at FROM shopping_categories ORDER BY sort_order ASC, id ASC"
|
|
rows = execute_query(query, fetch="all")
|
|
if not rows:
|
|
seed_default_shopping_categories_if_empty()
|
|
rows = execute_query(query, fetch="all")
|
|
for r in rows:
|
|
if not r.get("shelf_type"):
|
|
r["shelf_type"] = "other"
|
|
if isinstance(r.get("created_at"), (datetime, date)):
|
|
r["created_at"] = r["created_at"].isoformat()
|
|
return rows
|
|
|
|
|
|
def add_shopping_category(name: str, icon: str = "🛒", shelf_type: str = "other", sort_order: Optional[int] = None) -> Dict[str, Any]:
|
|
"""Fügt einen neuen Supermarkt-Bereich hinzu."""
|
|
clean_name = name.strip()
|
|
if not clean_name:
|
|
raise ValueError("Name des Bereichs darf nicht leer sein.")
|
|
clean_icon = (icon or "🛒").strip()
|
|
clean_shelf = (shelf_type or "other").strip()
|
|
|
|
if sort_order is None:
|
|
max_row = execute_query("SELECT MAX(sort_order) as m FROM shopping_categories", fetch="one")
|
|
sort_order = ((max_row["m"] or 0) + 1) if max_row else 1
|
|
|
|
query = "INSERT INTO shopping_categories (name, icon, shelf_type, sort_order) VALUES (?, ?, ?, ?)"
|
|
res = execute_query(query, (clean_name, clean_icon, clean_shelf, sort_order), fetch="none", commit=True)
|
|
cat_id = res.get("lastrowid")
|
|
row = execute_query("SELECT * FROM shopping_categories WHERE id = ?", (cat_id,), fetch="one")
|
|
if row:
|
|
if not row.get("shelf_type"):
|
|
row["shelf_type"] = clean_shelf
|
|
if isinstance(row.get("created_at"), (datetime, date)):
|
|
row["created_at"] = row["created_at"].isoformat()
|
|
return row
|
|
|
|
|
|
def update_shopping_category(cat_id: int, name: str, icon: str, shelf_type: str = "other", sort_order: Optional[int] = None) -> Optional[Dict[str, Any]]:
|
|
"""Aktualisiert einen Supermarkt-Bereich und benennt vorhandene Zuweisungen um."""
|
|
old = execute_query("SELECT * FROM shopping_categories WHERE id = ?", (cat_id,), fetch="one")
|
|
if not old:
|
|
return None
|
|
|
|
clean_name = name.strip()
|
|
clean_icon = (icon or "🛒").strip()
|
|
clean_shelf = (shelf_type or old.get("shelf_type") or "other").strip()
|
|
order_val = sort_order if sort_order is not None else old.get("sort_order", 0)
|
|
|
|
query = "UPDATE shopping_categories SET name = ?, icon = ?, shelf_type = ?, sort_order = ? WHERE id = ?"
|
|
execute_query(query, (clean_name, clean_icon, clean_shelf, order_val, cat_id), fetch="none", commit=True)
|
|
|
|
# Falls der Name geändert wurde, aktualisiere die Items in shopping_list
|
|
if old["name"] != clean_name:
|
|
execute_query("UPDATE shopping_list SET category = ? WHERE category = ?", (clean_name, old["name"]), fetch="none", commit=True)
|
|
|
|
row = execute_query("SELECT * FROM shopping_categories WHERE id = ?", (cat_id,), fetch="one")
|
|
if row:
|
|
if not row.get("shelf_type"):
|
|
row["shelf_type"] = clean_shelf
|
|
if isinstance(row.get("created_at"), (datetime, date)):
|
|
row["created_at"] = row["created_at"].isoformat()
|
|
return row
|
|
|
|
|
|
def delete_shopping_category(cat_id: int) -> bool:
|
|
"""Löscht einen Bereich. Zugeordnete Items werden auf einen Ausweichbereich gelegt."""
|
|
old = execute_query("SELECT * FROM shopping_categories WHERE id = ?", (cat_id,), fetch="one")
|
|
if not old:
|
|
return False
|
|
|
|
fallback = execute_query("SELECT name FROM shopping_categories WHERE id != ? ORDER BY sort_order DESC LIMIT 1", (cat_id,), fetch="one")
|
|
fallback_name = fallback["name"] if fallback else "Drogerie & Sonstiges"
|
|
|
|
execute_query("UPDATE shopping_list SET category = ? WHERE category = ?", (fallback_name, old["name"]), fetch="none", commit=True)
|
|
res = execute_query("DELETE FROM shopping_categories WHERE id = ?", (cat_id,), fetch="none", commit=True)
|
|
return res.get("rowcount", 0) > 0
|
|
|
|
|
|
def reorder_shopping_categories(category_ids: List[int]) -> bool:
|
|
"""Aktualisiert die Reihenfolge der Supermarkt-Bereiche."""
|
|
for idx, cid in enumerate(category_ids):
|
|
execute_query("UPDATE shopping_categories SET sort_order = ? WHERE id = ?", (idx + 1, cid), fetch="none", commit=True)
|
|
return True
|
|
|
|
|
|
def reset_default_shopping_categories(preset: str = "standard") -> List[Dict[str, Any]]:
|
|
"""Setzt alle Bereiche auf das gewählte Preset zurück (standard=Vollsortimenter, discounter=Discounter)."""
|
|
execute_query("DELETE FROM shopping_categories", fetch="none", commit=True)
|
|
seed_default_shopping_categories_if_empty(preset=preset)
|
|
# Alle offenen Listeneinträge nach dem neuen Laufweg neu kategorisieren
|
|
try:
|
|
items = execute_query("SELECT id, item_name FROM shopping_list", fetch="all")
|
|
for it in items:
|
|
guessed = guess_category_for_ingredient(it.get("item_name", ""))
|
|
execute_query("UPDATE shopping_list SET category = ? WHERE id = ?", (guessed, it["id"]), fetch="none", commit=True)
|
|
except Exception:
|
|
pass
|
|
return get_shopping_categories()
|
|
|
|
|
|
def update_shopping_item(
|
|
item_id: int,
|
|
item_name: str,
|
|
amount: str = "",
|
|
unit: str = "",
|
|
category: Optional[str] = None,
|
|
note: str = ""
|
|
) -> Optional[Dict[str, Any]]:
|
|
"""Aktualisiert alle Eigenschaften eines Eintrags in der Einkaufsliste und lernt die Kategorie."""
|
|
from .ingredients import clean_ingredient_name
|
|
cleaned_name = clean_ingredient_name(item_name) or item_name.strip()
|
|
cat = (category or "").strip()
|
|
if not cat:
|
|
cat = guess_category_for_ingredient(cleaned_name)
|
|
else:
|
|
save_ingredient_category_mapping(cleaned_name, cat)
|
|
|
|
query = """
|
|
UPDATE shopping_list
|
|
SET item_name = ?, amount = ?, unit = ?, category = ?, note = ?
|
|
WHERE id = ?
|
|
"""
|
|
execute_query(query, (cleaned_name, amount.strip(), unit.strip(), cat, (note or "").strip(), item_id), fetch="none", commit=True)
|
|
row = execute_query("SELECT * FROM shopping_list WHERE id = ?", (item_id,), fetch="one")
|
|
if row:
|
|
row["checked"] = bool(row.get("checked"))
|
|
row["display_source"] = clean_source_label(row.get("source", ""))
|
|
row["note"] = str(row.get("note") or "").strip()
|
|
if isinstance(row.get("created_at"), (datetime, date)):
|
|
row["created_at"] = row["created_at"].isoformat()
|
|
if isinstance(row.get("updated_at"), (datetime, date)):
|
|
row["updated_at"] = row["updated_at"].isoformat()
|
|
return row
|
|
|
|
|
|
def update_shopping_item_category(item_id: int, category: str) -> Optional[Dict[str, Any]]:
|
|
"""Ändert die Kategorie/den Bereich eines einzelnen Listeneintrags und merkt sich die Zuordnung."""
|
|
clean_cat = category.strip()
|
|
curr = execute_query("SELECT item_name FROM shopping_list WHERE id = ?", (item_id,), fetch="one")
|
|
if curr and curr.get("item_name"):
|
|
save_ingredient_category_mapping(curr["item_name"], clean_cat)
|
|
|
|
query = "UPDATE shopping_list SET category = ? WHERE id = ?"
|
|
execute_query(query, (clean_cat, item_id), fetch="none", commit=True)
|
|
row = execute_query("SELECT * FROM shopping_list WHERE id = ?", (item_id,), fetch="one")
|
|
if row:
|
|
row["checked"] = bool(row.get("checked"))
|
|
row["display_source"] = clean_source_label(row.get("source", ""))
|
|
row["note"] = str(row.get("note") or "").strip()
|
|
if isinstance(row.get("created_at"), (datetime, date)):
|
|
row["created_at"] = row["created_at"].isoformat()
|
|
if isinstance(row.get("updated_at"), (datetime, date)):
|
|
row["updated_at"] = row["updated_at"].isoformat()
|
|
return row
|
|
|
|
|
|
def get_shopping_list() -> List[Dict[str, Any]]:
|
|
"""Gibt alle Einträge der Einkaufsliste zurück, sortiert nach unvollständig/erledigt."""
|
|
query = """
|
|
SELECT id, item_name, amount, unit, checked, source, category, note, created_at, updated_at
|
|
FROM shopping_list
|
|
ORDER BY checked ASC, id DESC
|
|
"""
|
|
rows = execute_query(query, fetch="all")
|
|
for r in rows:
|
|
r["checked"] = bool(r.get("checked"))
|
|
if not r.get("category"):
|
|
r["category"] = guess_category_for_ingredient(r.get("item_name", ""))
|
|
r["display_source"] = clean_source_label(r.get("source", ""))
|
|
r["note"] = str(r.get("note") or "").strip()
|
|
if isinstance(r.get("created_at"), (datetime, date)):
|
|
r["created_at"] = r["created_at"].isoformat()
|
|
if isinstance(r.get("updated_at"), (datetime, date)):
|
|
r["updated_at"] = r["updated_at"].isoformat()
|
|
return rows
|
|
|
|
|
|
def add_shopping_item(
|
|
item_name: str,
|
|
amount: str = "",
|
|
unit: str = "",
|
|
source: str = "",
|
|
category: Optional[str] = None,
|
|
note: str = ""
|
|
) -> Dict[str, Any]:
|
|
"""Fügt einen einzelnen Eintrag zur Einkaufsliste hinzu oder aggregiert ihn mit einem offenen Eintrag."""
|
|
from .ingredients import clean_ingredient_name, get_ingredient_key, add_ingredient_amounts
|
|
|
|
clean_name = clean_ingredient_name(item_name) or item_name.strip()
|
|
key = get_ingredient_key(clean_name)
|
|
amt = amount.strip()
|
|
u = unit.strip()
|
|
clean_src = clean_source_label(source)
|
|
clean_note = (note or "").strip()
|
|
|
|
cat = (category or "").strip()
|
|
if not cat:
|
|
cat = guess_category_for_ingredient(clean_name)
|
|
|
|
# Prüfen, ob bereits ein offener Eintrag mit demselben Schlüssel existiert
|
|
if key:
|
|
existing = execute_query(
|
|
"SELECT * FROM shopping_list WHERE checked = 0 ORDER BY id DESC",
|
|
fetch="all"
|
|
)
|
|
match = next((row for row in existing if get_ingredient_key(row.get("item_name", "")) == key), None)
|
|
if match:
|
|
new_amt, new_unit = add_ingredient_amounts(match.get("amount", ""), match.get("unit", ""), amt, u)
|
|
existing_sources = [s.strip() for s in (match.get("source") or "").split("&") if s.strip()]
|
|
if clean_src and clean_src not in existing_sources:
|
|
existing_sources.append(clean_src)
|
|
new_source = " & ".join(existing_sources)
|
|
|
|
existing_notes = [n.strip() for n in (match.get("note") or "").split(" | ") if n.strip()]
|
|
if clean_note and clean_note not in existing_notes:
|
|
existing_notes.append(clean_note)
|
|
new_note = " | ".join(existing_notes)
|
|
|
|
new_cat = category if category else (match.get("category") or cat)
|
|
execute_query(
|
|
"UPDATE shopping_list SET amount = ?, unit = ?, source = ?, category = ?, note = ? WHERE id = ?",
|
|
(new_amt, new_unit, new_source, new_cat, new_note, match["id"]),
|
|
fetch="none",
|
|
commit=True
|
|
)
|
|
updated_row = execute_query("SELECT * FROM shopping_list WHERE id = ?", (match["id"],), fetch="one")
|
|
if updated_row:
|
|
updated_row["checked"] = bool(updated_row.get("checked"))
|
|
updated_row["display_source"] = clean_source_label(updated_row.get("source", ""))
|
|
updated_row["note"] = str(updated_row.get("note") or "").strip()
|
|
if isinstance(updated_row.get("created_at"), (datetime, date)):
|
|
updated_row["created_at"] = updated_row["created_at"].isoformat()
|
|
if isinstance(updated_row.get("updated_at"), (datetime, date)):
|
|
updated_row["updated_at"] = updated_row["updated_at"].isoformat()
|
|
return updated_row
|
|
|
|
# Sonst neu anlegen
|
|
query = "INSERT INTO shopping_list (item_name, amount, unit, source, category, note, checked) VALUES (?, ?, ?, ?, ?, ?, 0)"
|
|
res = execute_query(query, (clean_name, amt, u, clean_src, cat, clean_note), fetch="none", commit=True)
|
|
item_id = res.get("lastrowid")
|
|
row = execute_query("SELECT * FROM shopping_list WHERE id = ?", (item_id,), fetch="one")
|
|
if row:
|
|
row["checked"] = bool(row.get("checked"))
|
|
row["display_source"] = clean_source_label(row.get("source", ""))
|
|
row["note"] = str(row.get("note") or "").strip()
|
|
if isinstance(row.get("created_at"), (datetime, date)):
|
|
row["created_at"] = row["created_at"].isoformat()
|
|
if isinstance(row.get("updated_at"), (datetime, date)):
|
|
row["updated_at"] = row["updated_at"].isoformat()
|
|
return row
|
|
|
|
|
|
def add_shopping_items_bulk(items: List[Dict[str, Any]]) -> int:
|
|
"""Fügt mehrere Zutaten/Einträge auf einmal zur Einkaufsliste hinzu, fasst Duplikate zusammen und summiert Mengen."""
|
|
if not items:
|
|
return 0
|
|
|
|
from .ingredients import clean_ingredient_name, get_ingredient_key, add_ingredient_amounts
|
|
|
|
# 1. Vorhandene offene (ungehakte) Einträge laden
|
|
existing_rows = execute_query(
|
|
"SELECT id, item_name, amount, unit, source, category, note, checked FROM shopping_list WHERE checked = 0",
|
|
fetch="all"
|
|
)
|
|
open_items_map: Dict[str, Dict[str, Any]] = {}
|
|
for row in existing_rows:
|
|
k = get_ingredient_key(row.get("item_name", ""))
|
|
if k and k not in open_items_map:
|
|
open_items_map[k] = row
|
|
|
|
count = 0
|
|
for it in items:
|
|
raw_name = (it.get("item_name") or it.get("name") or it.get("raw") or "").strip()
|
|
if not raw_name:
|
|
continue
|
|
clean_name = clean_ingredient_name(raw_name) or raw_name
|
|
key = get_ingredient_key(clean_name)
|
|
|
|
amt = str(it.get("amount") or "").strip()
|
|
unit = str(it.get("unit") or "").strip()
|
|
raw_src = str(it.get("source") or it.get("recipe_title") or "").strip()
|
|
src = clean_source_label(raw_src)
|
|
cat = str(it.get("category") or "").strip()
|
|
if not cat:
|
|
cat = guess_category_for_ingredient(clean_name)
|
|
note = str(it.get("note") or "").strip()
|
|
|
|
# Prüfen, ob Zutat bereits in der offenen Liste existiert (auch innerhalb dieses Batches)
|
|
if key in open_items_map:
|
|
target = open_items_map[key]
|
|
new_amt, new_unit = add_ingredient_amounts(
|
|
target.get("amount", ""), target.get("unit", ""),
|
|
amt, unit
|
|
)
|
|
|
|
existing_sources = [s.strip() for s in (target.get("source") or "").split("&") if s.strip()]
|
|
if src and src not in existing_sources:
|
|
existing_sources.append(src)
|
|
new_source = " & ".join(existing_sources)
|
|
|
|
existing_notes = [n.strip() for n in (target.get("note") or "").split(" | ") if n.strip()]
|
|
if note and note not in existing_notes:
|
|
existing_notes.append(note)
|
|
new_note = " | ".join(existing_notes)
|
|
|
|
new_cat = target.get("category") or cat
|
|
|
|
item_id = target["id"]
|
|
execute_query(
|
|
"UPDATE shopping_list SET amount = ?, unit = ?, source = ?, category = ?, note = ? WHERE id = ?",
|
|
(new_amt, new_unit, new_source, new_cat, new_note, item_id),
|
|
fetch="none",
|
|
commit=True
|
|
)
|
|
|
|
target["amount"] = new_amt
|
|
target["unit"] = new_unit
|
|
target["source"] = new_source
|
|
target["note"] = new_note
|
|
target["category"] = new_cat
|
|
count += 1
|
|
else:
|
|
query = "INSERT INTO shopping_list (item_name, amount, unit, source, category, note, checked) VALUES (?, ?, ?, ?, ?, ?, 0)"
|
|
res = execute_query(query, (clean_name, amt, unit, src, cat, note), fetch="none", commit=True)
|
|
new_id = res.get("lastrowid")
|
|
new_row = {
|
|
"id": new_id,
|
|
"item_name": clean_name,
|
|
"amount": amt,
|
|
"unit": unit,
|
|
"source": src,
|
|
"category": cat,
|
|
"note": note,
|
|
"checked": 0
|
|
}
|
|
if key:
|
|
open_items_map[key] = new_row
|
|
count += 1
|
|
|
|
return count
|
|
|
|
|
|
def toggle_shopping_item(item_id: int) -> Optional[Dict[str, Any]]:
|
|
"""Schaltet den Erledigt-Status eines Eintrags um."""
|
|
query = "UPDATE shopping_list SET checked = CASE WHEN checked = 1 THEN 0 ELSE 1 END WHERE id = ?"
|
|
execute_query(query, (item_id,), fetch="none", commit=True)
|
|
row = execute_query("SELECT * FROM shopping_list WHERE id = ?", (item_id,), fetch="one")
|
|
if row:
|
|
row["checked"] = bool(row.get("checked"))
|
|
row["display_source"] = clean_source_label(row.get("source", ""))
|
|
return row
|
|
|
|
|
|
def set_shopping_item_checked(item_id: int, checked: bool) -> Optional[Dict[str, Any]]:
|
|
"""Setzt den Erledigt-Status eines Eintrags explizit auf True oder False."""
|
|
val = 1 if checked else 0
|
|
query = "UPDATE shopping_list SET checked = ? WHERE id = ?"
|
|
execute_query(query, (val, item_id), fetch="none", commit=True)
|
|
row = execute_query("SELECT * FROM shopping_list WHERE id = ?", (item_id,), fetch="one")
|
|
if row:
|
|
row["checked"] = bool(row.get("checked"))
|
|
row["display_source"] = clean_source_label(row.get("source", ""))
|
|
return row
|
|
|
|
|
|
def delete_shopping_item(item_id: int) -> bool:
|
|
"""Löscht einen einzelnen Eintrag aus der Einkaufsliste."""
|
|
query = "DELETE FROM shopping_list WHERE id = ?"
|
|
res = execute_query(query, (item_id,), fetch="none", commit=True)
|
|
return res.get("rowcount", 0) > 0
|
|
|
|
|
|
def delete_checked_shopping_items() -> int:
|
|
"""Löscht alle bereits abgehakten Einträge aus der Einkaufsliste."""
|
|
query = "DELETE FROM shopping_list WHERE checked = 1"
|
|
res = execute_query(query, fetch="none", commit=True)
|
|
return res.get("rowcount", 0)
|
|
|
|
|
|
def clear_shopping_list() -> int:
|
|
"""Leert die gesamte Einkaufsliste."""
|
|
query = "DELETE FROM shopping_list"
|
|
res = execute_query(query, fetch="none", commit=True)
|
|
return res.get("rowcount", 0)
|
|
|
|
|
|
def get_meal_plan_ingredients(year: int, week: int) -> List[Dict[str, Any]]:
|
|
"""Gibt alle Zutaten aller Rezepte des Wochenplans strukturiert für die Einkaufsliste zurück."""
|
|
slots = get_meal_plan(year, week)
|
|
if not slots:
|
|
return []
|
|
|
|
from .ingredients import parse_ingredients_block, clean_ingredient_name
|
|
|
|
all_recipes = {r["id"]: r for r in get_all_recipes()}
|
|
day_names = ["Montag", "Dienstag", "Mittwoch", "Donnerstag", "Freitag", "Samstag", "Sonntag"]
|
|
meal_labels = {"frueh": "Früh", "mittag": "Mittag", "kaffee": "Kaffee & Kuchen", "abend": "Abend"}
|
|
|
|
recipe_groups = []
|
|
seen_slots = set()
|
|
|
|
for s in slots:
|
|
r_id = s.get("recipe_id")
|
|
if not r_id or r_id not in all_recipes:
|
|
continue
|
|
|
|
recipe = all_recipes[r_id]
|
|
d_idx = s.get("day_of_week", 0)
|
|
day_str = day_names[d_idx] if 0 <= d_idx < 7 else f"Tag {d_idx}"
|
|
m_type = s.get("meal_type", "")
|
|
m_label = meal_labels.get(m_type, m_type.capitalize())
|
|
slot_key = f"{d_idx}_{m_type}_{r_id}"
|
|
|
|
if slot_key in seen_slots:
|
|
continue
|
|
seen_slots.add(slot_key)
|
|
|
|
slot_label = f"{day_str} {m_label}"
|
|
parsed = parse_ingredients_block(recipe.get("ingredients", ""))
|
|
|
|
items = []
|
|
for line in parsed:
|
|
if line.get("is_header"):
|
|
continue
|
|
raw_text = line.get("raw", "").strip()
|
|
if not raw_text:
|
|
continue
|
|
|
|
name = line.get("name") or raw_text
|
|
clean_name = clean_ingredient_name(name) or name
|
|
amt = str(line.get("amount") or "") if line.get("amount") is not None else ""
|
|
unit = line.get("unit") or ""
|
|
|
|
items.append({
|
|
"recipe_id": recipe["id"],
|
|
"recipe_title": recipe["title"],
|
|
"slot_label": slot_label,
|
|
"raw": raw_text,
|
|
"name": clean_name,
|
|
"item_name": clean_name,
|
|
"amount": amt,
|
|
"unit": unit
|
|
})
|
|
|
|
if items:
|
|
recipe_groups.append({
|
|
"recipe_id": recipe["id"],
|
|
"recipe_title": recipe["title"],
|
|
"slot_label": slot_label,
|
|
"items": items,
|
|
"ingredients": items
|
|
})
|
|
|
|
return recipe_groups
|
|
|
|
|
|
|
|
|
|
|
|
# --- Rezept-Freigaben (Share Links) ---
|
|
|
|
def parse_db_datetime(dt_val: Any) -> Optional[datetime]:
|
|
if not dt_val:
|
|
return None
|
|
if isinstance(dt_val, datetime):
|
|
return dt_val
|
|
try:
|
|
return datetime.fromisoformat(str(dt_val))
|
|
except Exception:
|
|
try:
|
|
return datetime.strptime(str(dt_val).split(".")[0], "%Y-%m-%d %H:%M:%S")
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def format_share_expiry(dt_val: Any) -> Tuple[str, str, bool]:
|
|
"""
|
|
Gibt (formatted_date, relative_status, is_expired) zurück.
|
|
"""
|
|
dt = parse_db_datetime(dt_val)
|
|
if not dt:
|
|
return ("Unbegrenzt", "Unbegrenzt gültig", False)
|
|
|
|
now = datetime.now()
|
|
is_expired = now > dt
|
|
date_str = dt.strftime("%d.%m.%Y %H:%M")
|
|
|
|
if is_expired:
|
|
return (date_str, f"Abgelaufen am {date_str} Uhr", True)
|
|
|
|
delta = dt - now
|
|
if delta.days > 0:
|
|
days = delta.days
|
|
rel = f"Noch {days} {'Tag' if days == 1 else 'Tage'} gültig"
|
|
else:
|
|
hours = max(1, int(delta.total_seconds() // 3600))
|
|
rel = f"Noch {hours} {'Stunde' if hours == 1 else 'Stunden'} gültig"
|
|
|
|
return (date_str, rel, False)
|
|
|
|
|
|
def create_recipe_share(recipe_id: int, expires_in_seconds: Optional[int] = None, label: str = "") -> Dict[str, Any]:
|
|
token = secrets.token_urlsafe(16)
|
|
expires_at_dt = None
|
|
expires_at_sql = None
|
|
if expires_in_seconds is not None and expires_in_seconds > 0:
|
|
expires_at_dt = datetime.now() + timedelta(seconds=expires_in_seconds)
|
|
expires_at_sql = expires_at_dt.strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
res = execute_query(
|
|
"INSERT INTO recipe_shares (recipe_id, token, expires_at, label) VALUES (?, ?, ?, ?)",
|
|
(recipe_id, token, expires_at_sql, label.strip()),
|
|
fetch="none",
|
|
commit=True
|
|
)
|
|
last_id = res.get("lastrowid") if isinstance(res, dict) else None
|
|
|
|
fmt_date, rel_status, is_exp = format_share_expiry(expires_at_dt)
|
|
return {
|
|
"id": last_id,
|
|
"recipe_id": recipe_id,
|
|
"token": token,
|
|
"created_at": datetime.now().isoformat(),
|
|
"expires_at": expires_at_dt.isoformat() if expires_at_dt else None,
|
|
"expires_at_formatted": fmt_date,
|
|
"relative_status": rel_status,
|
|
"is_expired": is_exp,
|
|
"views_count": 0,
|
|
"revoked": False,
|
|
"label": label.strip()
|
|
}
|
|
|
|
|
|
def get_recipe_share_by_token(token: str, record_view: bool = False) -> Optional[Dict[str, Any]]:
|
|
row = execute_query(
|
|
"SELECT * FROM recipe_shares WHERE token = ? LIMIT 1",
|
|
(token.strip(),),
|
|
fetch="one"
|
|
)
|
|
if not row:
|
|
return None
|
|
|
|
fmt_date, rel_status, is_exp = format_share_expiry(row.get("expires_at"))
|
|
is_revoked = bool(row.get("revoked", 0))
|
|
|
|
if record_view and not is_revoked and not is_exp:
|
|
execute_query("UPDATE recipe_shares SET views_count = views_count + 1 WHERE id = ?", (row["id"],), commit=True)
|
|
row["views_count"] = row.get("views_count", 0) + 1
|
|
|
|
row["is_expired"] = is_exp
|
|
row["revoked"] = is_revoked
|
|
row["expires_at_formatted"] = fmt_date
|
|
row["relative_status"] = rel_status
|
|
row["is_active"] = (not is_revoked and not is_exp)
|
|
return row
|
|
|
|
|
|
def get_recipe_shares_for_recipe(recipe_id: int) -> List[Dict[str, Any]]:
|
|
rows = execute_query(
|
|
"SELECT * FROM recipe_shares WHERE recipe_id = ? ORDER BY id DESC",
|
|
(recipe_id,),
|
|
fetch="all"
|
|
)
|
|
results = []
|
|
for r in (rows or []):
|
|
fmt_date, rel_status, is_exp = format_share_expiry(r.get("expires_at"))
|
|
is_revoked = bool(r.get("revoked", 0))
|
|
r["is_expired"] = is_exp
|
|
r["revoked"] = is_revoked
|
|
r["expires_at_formatted"] = fmt_date
|
|
r["relative_status"] = rel_status
|
|
r["is_active"] = (not is_revoked and not is_exp)
|
|
results.append(r)
|
|
return results
|
|
|
|
|
|
def delete_recipe_share(share_id: int, recipe_id: int) -> bool:
|
|
res = execute_query(
|
|
"DELETE FROM recipe_shares WHERE id = ? AND recipe_id = ?",
|
|
(share_id, recipe_id),
|
|
fetch="none",
|
|
commit=True
|
|
)
|
|
return bool(res and res.get("rowcount", 0) > 0)
|
|
|
|
|
|
def revoke_recipe_share(share_id: int, recipe_id: int) -> bool:
|
|
res = execute_query(
|
|
"UPDATE recipe_shares SET revoked = 1 WHERE id = ? AND recipe_id = ?",
|
|
(share_id, recipe_id),
|
|
fetch="none",
|
|
commit=True
|
|
)
|
|
return bool(res and res.get("rowcount", 0) > 0)
|