755 lines
31 KiB
Python
Executable File
755 lines
31 KiB
Python
Executable File
import os
|
|
import sqlite3
|
|
from datetime import datetime
|
|
from typing import List, Dict, Any, Optional
|
|
|
|
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"
|
|
]
|
|
|
|
# 11 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"},
|
|
{"number": 11, "title": "Plätzchen", "icon": "🍪", "doodle_hint": "Weihnachtsplätzchen & Kekse", "illustration_url": "/static/img/chapters/chapter_11.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"]
|
|
|
|
|
|
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 = '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("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,
|
|
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 "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("DELETE FROM dishes")
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
# --- 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:
|
|
pool = list(all_recipes)
|
|
|
|
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
|
|
) -> 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.")
|
|
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)
|
|
|
|
res = execute_query(
|
|
"""
|
|
INSERT INTO recipes (title, category, chapter_number, effort, prep_time, cook_time, portions, has_portion_table, ingredients, instructions, notes, source, image_url, printed_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(title, 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
|
|
) -> 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_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 = ?, category = ?, chapter_number = ?, effort = ?, prep_time = ?, cook_time = ?, portions = ?, has_portion_table = ?, ingredients = ?, instructions = ?, notes = ?, source = ?, image_url = ?, printed_at = ?
|
|
WHERE id = ?
|
|
"""
|
|
params = (title, 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 = ?, category = ?, chapter_number = ?, effort = ?, prep_time = ?, cook_time = ?, portions = ?, has_portion_table = ?, ingredients = ?, instructions = ?, notes = ?, source = ?, printed_at = ?
|
|
WHERE id = ?
|
|
"""
|
|
params = (title, 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 ingredients LIKE ? OR instructions LIKE ?)"
|
|
s = f"%{search.strip()}%"
|
|
params.extend([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)
|
|
|
|
|