hnzio/nouri-App
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 18 Wiki Activity
Files
v.1.3.3
nouri-App/nouri/db.py
T

1081 lines
43 KiB
Python
Raw Permalink Blame History

from __future__ import annotations
import sqlite3
from pathlib import Path
import click
from flask import Flask, current_app, g
from flask.cli import with_appcontext
from werkzeug.security import generate_password_hash
from .constants import (
DAYPARTS,
DAYPART_SLUG_TO_MEAL_TYPE,
DEFAULT_CATEGORIES,
DEFAULT_CATEGORY_BUILDERS,
)
CURRENT_SCHEMA_VERSION = "1.3.3"
ANIMAL_HINTS = (
"huhn",
"hähn",
"rind",
"hack",
"schwein",
"speck",
"salami",
"wurst",
"thunfisch",
"lachs",
"fisch",
"garnelen",
"shrimp",
"sardinen",
)
def normalize_name_for_profile(name: str | None) -> str:
return (name or "").strip().lower()
def infer_food_flavor_profile(
name: str | None,
category: str | None,
base_type: str | None = None,
suggestion_role: str | None = None,
) -> str:
normalized_name = normalize_name_for_profile(name)
normalized_category = (category or "").strip().lower()
normalized_base_type = (base_type or "").strip().lower()
normalized_role = (suggestion_role or "").strip().lower()
if any(token in normalized_name for token in ("proteinpulver", "eiweißpulver", "whey", "clear whey")):
return "neutral"
if any(token in normalized_name for token in ("schoko", "choco", "müsli", "granola", "cornflakes", "fruchtjoghurt", "vanillejoghurt", "pudding")):
return "sweet"
if any(token in normalized_name for token in ("banane", "apfel", "birne", "beeren", "himbeer", "erdbeer", "heidelbeer", "mango", "kiwi", "trauben")):
return "sweet"
if any(token in normalized_name for token in ("räucher", "tofu", "tempeh", "hack", "salami", "wurst", "thunfisch", "lachs", "fisch", "huhn", "hähn", "rind", "schwein", "aufstrich", "pesto", "humus", "hummus", "reisgericht", "chili", "curry")):
return "savory"
if any(token in normalized_name for token in ("naturjoghurt", "joghurt natur", "quark", "skyr", "haferflocken", "gurke", "karotte", "karotten", "kartoffel", "kartoffeln", "reis", "nudeln", "brot", "brötchen")):
return "neutral"
if "obst" in normalized_category or normalized_base_type == "fruit":
return "sweet"
if any(token in normalized_category for token in ("eiweiß", "protein")) or normalized_base_type == "protein":
return "savory"
if any(token in normalized_category for token in ("gemüse",)) or normalized_base_type in {"veg", "carb", "dairy", "nuts", "seeds"}:
return "neutral"
if normalized_role in {"topping", "cooking"}:
return "neutral"
return "neutral"
def infer_food_profile(name: str | None, category: str | None, energy_density: str | None) -> dict[str, object]:
normalized_name = normalize_name_for_profile(name)
normalized_category = (category or "").strip().lower()
base_type = "neutral"
if "eiweiß" in normalized_category or "protein" in normalized_category:
base_type = "protein"
elif "kohlenhyd" in normalized_category or "brot" in normalized_category or "getreide" in normalized_category:
base_type = "carb"
elif "milch" in normalized_category:
base_type = "dairy"
elif "obst" in normalized_category:
base_type = "fruit"
elif "gemüse" in normalized_category:
base_type = "veg"
elif "nüsse" in normalized_name or "nuss" in normalized_name:
base_type = "nuts"
elif "saat" in normalized_name or "leinsamen" in normalized_name or "chia" in normalized_name:
base_type = "seeds"
suggestion_role = "base"
suggestion_priority = "normal"
can_be_meal_core = 0
if any(token in normalized_name for token in ("proteinpulver", "eiweißpulver", "whey", "clear whey")):
return {
"base_type": "protein",
"suggestion_role": "complement",
"suggestion_priority": "rare",
"can_be_meal_core": 0,
}
if any(token in normalized_name for token in ("flohsamen", "flohsamenschalen", "leinsamen", "chia", "hanfsamen")):
return {
"base_type": "seeds",
"suggestion_role": "topping",
"suggestion_priority": "normal",
"can_be_meal_core": 0,
}
if "tomatenmark" in normalized_name:
return {
"base_type": "neutral",
"suggestion_role": "cooking",
"suggestion_priority": "rare",
"can_be_meal_core": 0,
}
if any(token in normalized_name for token in ("saure gurken", "essiggurken", "cornichons", "gurkenscheiben")):
return {
"base_type": "veg",
"suggestion_role": "complement",
"suggestion_priority": "rare",
"can_be_meal_core": 0,
}
if any(token in normalized_name for token in ("tofu", "tempeh", "vegane schnitzel", "vegane nuggets", "veganes hack", "sojageschnetzeltes")):
return {
"base_type": "protein",
"suggestion_role": "main",
"suggestion_priority": "prefer",
"can_be_meal_core": 1,
}
if any(token in normalized_name for token in ("thunfisch", "lachs", "fisch", "huhn", "hähn", "rind", "schwein", "hack")):
return {
"base_type": "protein",
"suggestion_role": "main",
"suggestion_priority": "rare",
"can_be_meal_core": 1,
}
if any(token in normalized_name for token in ("joghurt", "skyr", "quark", "hüttenkäse", "körniger frischkäse")):
return {
"base_type": "dairy",
"suggestion_role": "base",
"suggestion_priority": "prefer",
"can_be_meal_core": 1,
}
if any(token in normalized_name for token in ("müsli", "hafer", "porridge", "cornflakes", "brot", "brötchen", "reis", "nudel", "kartoffel", "wrap")):
return {
"base_type": "carb",
"suggestion_role": "base",
"suggestion_priority": "normal",
"can_be_meal_core": 1,
}
if any(token in normalized_name for token in ("beeren", "banane", "apfel", "obst", "birne", "trauben", "kiwi")):
return {
"base_type": "fruit",
"suggestion_role": "complement",
"suggestion_priority": "prefer",
"can_be_meal_core": 0,
}
if any(token in normalized_name for token in ("gemüse", "brokkoli", "spinat", "erbsen", "paprika", "karotte", "zucchini", "salat", "tomate", "tk gemüse")):
return {
"base_type": "veg",
"suggestion_role": "complement",
"suggestion_priority": "prefer",
"can_be_meal_core": 0,
}
if any(token in normalized_name for token in ("nussmus", "erdnuss", "mandeln", "walnüsse", "cashew")):
return {
"base_type": "nuts",
"suggestion_role": "topping",
"suggestion_priority": "normal",
"can_be_meal_core": 0,
}
if any(token in normalized_name for token in ("terrine", "5-minuten", "instant", "cup noodles")):
return {
"base_type": "carb" if (energy_density or "neutral") != "high" else "neutral",
"suggestion_role": "solo",
"suggestion_priority": "rare",
"can_be_meal_core": 1,
}
if base_type in {"protein", "carb", "dairy"}:
suggestion_role = "base"
can_be_meal_core = 1
elif base_type in {"veg", "fruit"}:
suggestion_role = "complement"
elif base_type in {"nuts", "seeds"}:
suggestion_role = "topping"
return {
"base_type": base_type,
"suggestion_role": suggestion_role,
"suggestion_priority": suggestion_priority,
"can_be_meal_core": can_be_meal_core,
}
def infer_meal_tags(name: str | None, legacy_category: str | None) -> str:
normalized_name = normalize_name_for_profile(name)
normalized_category = (legacy_category or "").strip().lower()
tags: list[str] = []
if normalized_category == "warmes":
tags.extend(["warm", "savory"])
if normalized_category == "kleines essen":
tags.extend(["simple", "quick"])
if normalized_category == "snack":
tags.append("simple")
if any(token in normalized_name for token in ("porridge", "müsli", "joghurt", "quark")):
tags.append("sweet")
if any(token in normalized_name for token in ("salat", "brot", "toast", "tofu", "reis", "nudel", "pfanne")):
tags.append("savory")
if any(token in normalized_name for token in ("to go", "unterwegs", "wrap")):
tags.append("portable")
if any(token in normalized_name for token in ("overnight", "vorbereitet", "meal prep")):
tags.append("prep")
if any(token in normalized_name for token in ("schnell", "5-minuten", "instant")):
tags.append("quick")
if any(token in normalized_name for token in ("einfach", "ruhig")):
tags.append("simple")
unique_tags: list[str] = []
for tag in tags:
if tag and tag not in unique_tags:
unique_tags.append(tag)
return ",".join(unique_tags)
def infer_meal_type_from_dayparts(database: sqlite3.Connection, item_id: int) -> str:
row = database.execute(
"""
SELECT dayparts.slug
FROM item_dayparts
JOIN dayparts ON dayparts.id = item_dayparts.daypart_id
WHERE item_dayparts.item_id = ?
ORDER BY dayparts.sort_order
LIMIT 1
""",
(item_id,),
).fetchone()
if row is None:
return "snack"
return DAYPART_SLUG_TO_MEAL_TYPE.get(row["slug"], "snack")
def migrate_item_profiles(database: sqlite3.Connection) -> None:
rows = database.execute(
"""
SELECT id, kind, name, category, energy_density
FROM items
ORDER BY id
"""
).fetchall()
for row in rows:
item_id = int(row["id"])
if row["kind"] == "food":
profile = infer_food_profile(row["name"], row["category"], row["energy_density"])
database.execute(
"""
UPDATE items
SET base_type = ?,
suggestion_role = ?,
suggestion_priority = ?,
can_be_meal_core = ?
WHERE id = ?
""",
(
profile["base_type"],
profile["suggestion_role"],
profile["suggestion_priority"],
profile["can_be_meal_core"],
item_id,
),
)
continue
meal_type = infer_meal_type_from_dayparts(database, item_id)
meal_tags = infer_meal_tags(row["name"], row["category"])
database.execute(
"""
UPDATE items
SET meal_type = COALESCE(NULLIF(meal_type, ''), ?),
meal_tags = CASE
WHEN meal_tags IS NULL OR meal_tags = '' THEN ?
ELSE meal_tags
END,
category = CASE
WHEN kind = 'meal' AND category IN ('Kohlenhydrate', 'Milchprodukt', 'Obst', 'Gemüse', 'Eiweißquelle', 'Snack', 'Warmes', 'Kleines Essen')
THEN NULL
ELSE category
END
WHERE id = ?
""",
(meal_type, meal_tags, item_id),
)
def migrate_food_flavor_profiles(database: sqlite3.Connection) -> None:
if get_meta(database, "food_flavor_profiles_migrated") == "1":
return
rows = database.execute(
"""
SELECT id, name, category, base_type, suggestion_role, flavor_profile
FROM items
WHERE kind = 'food'
ORDER BY id
"""
).fetchall()
for row in rows:
current_flavor = (row["flavor_profile"] or "").strip().lower()
if current_flavor in {"sweet", "savory"}:
continue
database.execute(
"""
UPDATE items
SET flavor_profile = ?
WHERE id = ?
""",
(
infer_food_flavor_profile(
row["name"],
row["category"],
row["base_type"],
row["suggestion_role"],
),
int(row["id"]),
),
)
set_meta(database, "food_flavor_profiles_migrated", "1")
def migrate_item_archive_state(database: sqlite3.Connection) -> None:
if get_meta(database, "item_archive_state_migrated") == "1":
return
if "is_archived" not in table_columns(database, "items"):
return
database.execute("UPDATE items SET is_archived = 0 WHERE is_archived IS NULL")
database.execute("UPDATE items SET is_archived = 1 WHERE availability_state = 'archived'")
database.execute("UPDATE items SET availability_state = 'idea' WHERE availability_state = 'archived'")
set_meta(database, "item_archive_state_migrated", "1")
def get_db() -> sqlite3.Connection:
if "db" not in g:
g.db = sqlite3.connect(
current_app.config["DATABASE_PATH"],
detect_types=sqlite3.PARSE_DECLTYPES,
timeout=30,
)
g.db.row_factory = sqlite3.Row
g.db.execute("PRAGMA foreign_keys = ON")
g.db.execute("PRAGMA busy_timeout = 30000")
return g.db
def close_db(_error=None) -> None:
database = g.pop("db", None)
if database is not None:
database.close()
def table_columns(database: sqlite3.Connection, table_name: str) -> set[str]:
rows = database.execute(f"PRAGMA table_info({table_name})").fetchall()
return {row["name"] for row in rows}
def table_exists(database: sqlite3.Connection, table_name: str) -> bool:
row = database.execute(
"SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?",
(table_name,),
).fetchone()
return row is not None
def add_column_if_missing(database: sqlite3.Connection, table_name: str, definition: str) -> None:
column_name = definition.split()[0]
if column_name not in table_columns(database, table_name):
database.execute(f"ALTER TABLE {table_name} ADD COLUMN {definition}")
def ensure_meta_table(database: sqlite3.Connection) -> None:
database.execute(
"""
CREATE TABLE IF NOT EXISTS app_meta (
key TEXT PRIMARY KEY,
value TEXT,
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
)
"""
)
def get_meta(database: sqlite3.Connection, key: str) -> str | None:
row = database.execute("SELECT value FROM app_meta WHERE key = ?", (key,)).fetchone()
return row["value"] if row else None
def set_meta(database: sqlite3.Connection, key: str, value: str) -> None:
database.execute(
"""
INSERT INTO app_meta (key, value, updated_at)
VALUES (?, ?, CURRENT_TIMESTAMP)
ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = CURRENT_TIMESTAMP
""",
(key, value),
)
def item_kind_constraint_supports_shopping(database: sqlite3.Connection) -> bool:
row = database.execute(
"""
SELECT sql
FROM sqlite_master
WHERE type = 'table' AND name = 'items'
"""
).fetchone()
return bool(row and row["sql"] and "'shopping'" in row["sql"])
def migrate_items_kind_constraint(database: sqlite3.Connection) -> None:
if not table_exists(database, "items") or item_kind_constraint_supports_shopping(database):
return
columns = table_columns(database, "items")
if "kind" not in columns:
return
foreign_keys_enabled = bool(database.execute("PRAGMA foreign_keys").fetchone()[0])
if foreign_keys_enabled:
database.execute("PRAGMA foreign_keys = OFF")
try:
database.execute("DROP TABLE IF EXISTS items_new")
database.execute(
"""
CREATE TABLE items_new (
id INTEGER PRIMARY KEY AUTOINCREMENT,
household_id INTEGER,
owner_user_id INTEGER,
target_user_id INTEGER,
visibility TEXT NOT NULL DEFAULT 'shared',
kind TEXT NOT NULL CHECK (kind IN ('food', 'meal', 'shopping')),
name TEXT NOT NULL,
category TEXT,
base_type TEXT NOT NULL DEFAULT 'neutral',
flavor_profile TEXT NOT NULL DEFAULT 'neutral',
suggestion_role TEXT NOT NULL DEFAULT 'base',
suggestion_priority TEXT NOT NULL DEFAULT 'normal',
can_be_meal_core INTEGER NOT NULL DEFAULT 0,
meal_type TEXT,
meal_tags TEXT NOT NULL DEFAULT '',
energy_density TEXT NOT NULL DEFAULT 'neutral',
note TEXT,
photo_filename TEXT,
availability_state TEXT NOT NULL DEFAULT 'idea' CHECK (availability_state IN ('idea', 'home', 'archived')),
is_archived INTEGER NOT NULL DEFAULT 0,
is_quick_added INTEGER NOT NULL DEFAULT 0,
created_by INTEGER,
updated_by INTEGER,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (household_id) REFERENCES households(id) ON DELETE CASCADE,
FOREIGN KEY (owner_user_id) REFERENCES users(id) ON DELETE SET NULL,
FOREIGN KEY (target_user_id) REFERENCES users(id) ON DELETE SET NULL,
FOREIGN KEY (created_by) REFERENCES users(id) ON DELETE SET NULL,
FOREIGN KEY (updated_by) REFERENCES users(id) ON DELETE SET NULL
)
"""
)
target_columns = [
"id",
"household_id",
"owner_user_id",
"target_user_id",
"visibility",
"kind",
"name",
"category",
"base_type",
"flavor_profile",
"suggestion_role",
"suggestion_priority",
"can_be_meal_core",
"meal_type",
"meal_tags",
"energy_density",
"note",
"photo_filename",
"availability_state",
"is_archived",
"is_quick_added",
"created_by",
"updated_by",
"created_at",
"updated_at",
]
copy_columns = [column for column in target_columns if column in columns]
quoted_columns = ", ".join(copy_columns)
database.execute(
f"""
INSERT INTO items_new ({quoted_columns})
SELECT {quoted_columns}
FROM items
"""
)
database.execute("DROP TABLE items")
database.execute("ALTER TABLE items_new RENAME TO items")
finally:
if foreign_keys_enabled:
database.execute("PRAGMA foreign_keys = ON")
def bootstrap_legacy_schema(database: sqlite3.Connection) -> None:
ensure_meta_table(database)
database.execute(
"""
CREATE TABLE IF NOT EXISTS households (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
shopping_weekday INTEGER NOT NULL DEFAULT 5,
shopping_prep_days INTEGER NOT NULL DEFAULT 1,
shopping_reminder_time TEXT NOT NULL DEFAULT '18:00',
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
)
"""
)
database.execute(
"""
CREATE TABLE IF NOT EXISTS household_categories (
id INTEGER PRIMARY KEY AUTOINCREMENT,
household_id INTEGER NOT NULL,
name TEXT NOT NULL,
builder_key TEXT NOT NULL DEFAULT 'neutral',
sort_order INTEGER NOT NULL DEFAULT 100,
is_active INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE (household_id, name)
)
"""
)
if table_exists(database, "households"):
add_column_if_missing(database, "households", "shopping_weekday INTEGER NOT NULL DEFAULT 5")
add_column_if_missing(database, "households", "shopping_prep_days INTEGER NOT NULL DEFAULT 1")
add_column_if_missing(database, "households", "shopping_reminder_time TEXT NOT NULL DEFAULT '18:00'")
if table_exists(database, "household_categories"):
add_column_if_missing(database, "household_categories", "builder_key TEXT NOT NULL DEFAULT 'neutral'")
if table_exists(database, "users"):
add_column_if_missing(database, "users", "household_id INTEGER")
add_column_if_missing(database, "users", "email TEXT")
add_column_if_missing(database, "users", "role TEXT NOT NULL DEFAULT 'member'")
add_column_if_missing(database, "users", "is_active INTEGER NOT NULL DEFAULT 1")
add_column_if_missing(database, "users", "updated_at TEXT")
if table_exists(database, "items"):
add_column_if_missing(database, "items", "household_id INTEGER")
add_column_if_missing(database, "items", "owner_user_id INTEGER")
add_column_if_missing(database, "items", "target_user_id INTEGER")
add_column_if_missing(database, "items", "visibility TEXT NOT NULL DEFAULT 'shared'")
add_column_if_missing(database, "items", "base_type TEXT NOT NULL DEFAULT 'neutral'")
add_column_if_missing(database, "items", "flavor_profile TEXT NOT NULL DEFAULT 'neutral'")
add_column_if_missing(database, "items", "suggestion_role TEXT NOT NULL DEFAULT 'base'")
add_column_if_missing(database, "items", "suggestion_priority TEXT NOT NULL DEFAULT 'normal'")
add_column_if_missing(database, "items", "can_be_meal_core INTEGER NOT NULL DEFAULT 0")
add_column_if_missing(database, "items", "meal_type TEXT")
add_column_if_missing(database, "items", "meal_tags TEXT NOT NULL DEFAULT ''")
add_column_if_missing(database, "items", "energy_density TEXT NOT NULL DEFAULT 'neutral'")
add_column_if_missing(database, "items", "is_archived INTEGER NOT NULL DEFAULT 0")
if table_exists(database, "shopping_entries"):
add_column_if_missing(database, "shopping_entries", "household_id INTEGER")
add_column_if_missing(database, "shopping_entries", "owner_user_id INTEGER")
add_column_if_missing(database, "shopping_entries", "visibility TEXT NOT NULL DEFAULT 'shared'")
add_column_if_missing(database, "shopping_entries", "shopping_note TEXT NOT NULL DEFAULT ''")
add_column_if_missing(database, "shopping_entries", "needed_for_date TEXT")
add_column_if_missing(database, "shopping_entries", "needed_for_daypart_id INTEGER")
if table_exists(database, "shopping_needs"):
add_column_if_missing(database, "shopping_needs", "source_item_id INTEGER")
add_column_if_missing(database, "shopping_needs", "activation_date TEXT")
add_column_if_missing(database, "shopping_needs", "is_activated INTEGER NOT NULL DEFAULT 0")
add_column_if_missing(database, "shopping_needs", "activated_at TEXT")
database.execute(
"""
CREATE TABLE IF NOT EXISTS user_settings (
user_id INTEGER PRIMARY KEY,
reminders_enabled INTEGER NOT NULL DEFAULT 1,
push_enabled INTEGER NOT NULL DEFAULT 0,
notification_channel TEXT NOT NULL DEFAULT 'in_app',
suggestion_style TEXT NOT NULL DEFAULT 'balanced',
energy_preference TEXT NOT NULL DEFAULT 'neutral',
remind_before_shopping INTEGER NOT NULL DEFAULT 1,
remind_on_shopping_day INTEGER NOT NULL DEFAULT 1,
show_missing_for_upcoming_week INTEGER NOT NULL DEFAULT 1,
show_planned_not_shopped INTEGER NOT NULL DEFAULT 1,
remind_tomorrow_if_sparse INTEGER NOT NULL DEFAULT 1,
remind_week_if_sparse INTEGER NOT NULL DEFAULT 1,
push_missing_breakfast INTEGER NOT NULL DEFAULT 0,
push_missing_lunch INTEGER NOT NULL DEFAULT 0,
push_missing_dinner INTEGER NOT NULL DEFAULT 0,
push_small_snack INTEGER NOT NULL DEFAULT 0,
suggest_home_for_today INTEGER NOT NULL DEFAULT 1,
remind_small_snack INTEGER NOT NULL DEFAULT 0,
remind_nuts INTEGER NOT NULL DEFAULT 0,
show_meal_balancing INTEGER NOT NULL DEFAULT 1,
suggest_templates INTEGER NOT NULL DEFAULT 1,
suggest_patterns INTEGER NOT NULL DEFAULT 1,
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
)
"""
)
database.execute(
"""
CREATE TABLE IF NOT EXISTS push_subscriptions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
endpoint TEXT NOT NULL UNIQUE,
p256dh TEXT NOT NULL,
auth TEXT NOT NULL,
user_agent TEXT,
is_active INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
last_test_at TEXT,
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
)
"""
)
database.execute(
"""
CREATE TABLE IF NOT EXISTS reminder_events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
event_key TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE (user_id, event_key),
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
)
"""
)
database.execute(
"""
CREATE TABLE IF NOT EXISTS hidden_generated_suggestions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
suggestion_key TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE (user_id, suggestion_key),
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
)
"""
)
database.execute(
"""
CREATE TABLE IF NOT EXISTS shopping_needs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
household_id INTEGER NOT NULL,
owner_user_id INTEGER,
visibility TEXT NOT NULL DEFAULT 'shared',
item_id INTEGER NOT NULL,
source_item_id INTEGER,
needed_for_date TEXT NOT NULL,
needed_for_daypart_id INTEGER,
activation_date TEXT NOT NULL,
is_activated INTEGER NOT NULL DEFAULT 0,
activated_at TEXT,
created_by INTEGER,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE (item_id, source_item_id, needed_for_date, needed_for_daypart_id, visibility),
FOREIGN KEY (household_id) REFERENCES households(id) ON DELETE CASCADE,
FOREIGN KEY (owner_user_id) REFERENCES users(id) ON DELETE SET NULL,
FOREIGN KEY (item_id) REFERENCES items(id) ON DELETE CASCADE,
FOREIGN KEY (source_item_id) REFERENCES items(id) ON DELETE SET NULL,
FOREIGN KEY (needed_for_daypart_id) REFERENCES dayparts(id) ON DELETE SET NULL,
FOREIGN KEY (created_by) REFERENCES users(id) ON DELETE SET NULL
)
"""
)
if table_exists(database, "plan_entries"):
add_column_if_missing(database, "plan_entries", "household_id INTEGER")
add_column_if_missing(database, "plan_entries", "owner_user_id INTEGER")
add_column_if_missing(database, "plan_entries", "visibility TEXT NOT NULL DEFAULT 'shared'")
if table_exists(database, "user_settings"):
add_column_if_missing(database, "user_settings", "suggestion_style TEXT NOT NULL DEFAULT 'balanced'")
add_column_if_missing(database, "user_settings", "energy_preference TEXT NOT NULL DEFAULT 'neutral'")
add_column_if_missing(database, "user_settings", "protein_preference TEXT NOT NULL DEFAULT 'mixed'")
add_column_if_missing(database, "user_settings", "push_missing_breakfast INTEGER NOT NULL DEFAULT 0")
add_column_if_missing(database, "user_settings", "push_missing_lunch INTEGER NOT NULL DEFAULT 0")
add_column_if_missing(database, "user_settings", "push_missing_dinner INTEGER NOT NULL DEFAULT 0")
add_column_if_missing(database, "user_settings", "push_small_snack INTEGER NOT NULL DEFAULT 0")
def ensure_default_household(database: sqlite3.Connection) -> int:
household = database.execute(
"SELECT id FROM households ORDER BY id LIMIT 1"
).fetchone()
if household:
return int(household["id"])
database.execute(
"INSERT INTO households (name) VALUES (?)",
("Unser Haushalt",),
)
return int(database.execute("SELECT id FROM households ORDER BY id LIMIT 1").fetchone()["id"])
def household_ids(database: sqlite3.Connection) -> list[int]:
return [int(row["id"]) for row in database.execute("SELECT id FROM households ORDER BY id").fetchall()]
def first_user_id(database: sqlite3.Connection) -> int | None:
row = database.execute("SELECT id FROM users ORDER BY id LIMIT 1").fetchone()
return int(row["id"]) if row else None
def sync_default_categories(database: sqlite3.Connection) -> None:
for household_id in household_ids(database):
legacy = database.execute(
"""
SELECT id
FROM household_categories
WHERE household_id = ? AND name = 'Brot & Getreide'
LIMIT 1
""",
(household_id,),
).fetchone()
updated = database.execute(
"""
SELECT id
FROM household_categories
WHERE household_id = ? AND name = 'Kohlenhydrate'
LIMIT 1
""",
(household_id,),
).fetchone()
if legacy and not updated:
database.execute(
"""
UPDATE household_categories
SET name = 'Kohlenhydrate', builder_key = 'carb'
WHERE id = ?
""",
(legacy["id"],),
)
database.execute(
"""
UPDATE items
SET category = 'Kohlenhydrate'
WHERE household_id = ? AND category = 'Brot & Getreide'
""",
(household_id,),
)
for sort_order, name in enumerate(DEFAULT_CATEGORIES, start=10):
database.execute(
"""
INSERT OR IGNORE INTO household_categories (household_id, name, builder_key, sort_order, is_active)
VALUES (?, ?, ?, ?, 1)
""",
(household_id, name, DEFAULT_CATEGORY_BUILDERS.get(name, "neutral"), sort_order),
)
database.execute(
"""
UPDATE household_categories
SET builder_key = COALESCE(NULLIF(builder_key, ''), ?)
WHERE household_id = ? AND name = ?
""",
(DEFAULT_CATEGORY_BUILDERS.get(name, "neutral"), household_id, name),
)
def ensure_schema_upgrades(database: sqlite3.Connection) -> None:
ensure_meta_table(database)
add_column_if_missing(database, "users", "household_id INTEGER")
add_column_if_missing(database, "users", "email TEXT")
add_column_if_missing(database, "users", "role TEXT NOT NULL DEFAULT 'member'")
add_column_if_missing(database, "users", "is_active INTEGER NOT NULL DEFAULT 1")
add_column_if_missing(database, "users", "updated_at TEXT")
default_household_id = ensure_default_household(database)
database.execute("UPDATE households SET shopping_weekday = 5 WHERE shopping_weekday IS NULL")
database.execute("UPDATE households SET shopping_prep_days = 1 WHERE shopping_prep_days IS NULL")
database.execute(
"UPDATE households SET shopping_reminder_time = '18:00' WHERE shopping_reminder_time IS NULL OR shopping_reminder_time = ''"
)
database.execute(
"UPDATE users SET household_id = ? WHERE household_id IS NULL",
(default_household_id,),
)
database.execute("UPDATE users SET role = 'member' WHERE role IS NULL OR role = ''")
database.execute("UPDATE users SET is_active = 1 WHERE is_active IS NULL")
database.execute("UPDATE users SET email = NULL WHERE TRIM(COALESCE(email, '')) = ''")
database.execute("UPDATE users SET updated_at = COALESCE(updated_at, created_at, CURRENT_TIMESTAMP)")
admin_row = database.execute(
"SELECT id FROM users WHERE role = 'admin' AND is_active = 1 ORDER BY id LIMIT 1"
).fetchone()
if admin_row is None:
first_id = first_user_id(database)
if first_id is not None:
database.execute("UPDATE users SET role = 'admin' WHERE id = ?", (first_id,))
default_owner_id = first_user_id(database)
for table_name in ("items", "shopping_entries", "plan_entries"):
add_column_if_missing(database, table_name, "household_id INTEGER")
add_column_if_missing(database, table_name, "owner_user_id INTEGER")
add_column_if_missing(database, table_name, "visibility TEXT NOT NULL DEFAULT 'shared'")
add_column_if_missing(database, "items", "target_user_id INTEGER")
add_column_if_missing(database, "items", "base_type TEXT NOT NULL DEFAULT 'neutral'")
add_column_if_missing(database, "items", "flavor_profile TEXT NOT NULL DEFAULT 'neutral'")
add_column_if_missing(database, "items", "suggestion_role TEXT NOT NULL DEFAULT 'base'")
add_column_if_missing(database, "items", "suggestion_priority TEXT NOT NULL DEFAULT 'normal'")
add_column_if_missing(database, "items", "can_be_meal_core INTEGER NOT NULL DEFAULT 0")
add_column_if_missing(database, "items", "meal_type TEXT")
add_column_if_missing(database, "items", "meal_tags TEXT NOT NULL DEFAULT ''")
add_column_if_missing(database, "items", "energy_density TEXT NOT NULL DEFAULT 'neutral'")
add_column_if_missing(database, "items", "is_archived INTEGER NOT NULL DEFAULT 0")
add_column_if_missing(database, "items", "is_quick_added INTEGER NOT NULL DEFAULT 0")
migrate_items_kind_constraint(database)
add_column_if_missing(database, "shopping_entries", "needed_for_date TEXT")
add_column_if_missing(database, "shopping_entries", "needed_for_daypart_id INTEGER")
add_column_if_missing(database, "shopping_entries", "shopping_note TEXT NOT NULL DEFAULT ''")
add_column_if_missing(database, "user_settings", "suggestion_style TEXT NOT NULL DEFAULT 'balanced'")
add_column_if_missing(database, "user_settings", "energy_preference TEXT NOT NULL DEFAULT 'neutral'")
add_column_if_missing(database, "user_settings", "protein_preference TEXT NOT NULL DEFAULT 'mixed'")
add_column_if_missing(database, "user_settings", "push_missing_breakfast INTEGER NOT NULL DEFAULT 0")
add_column_if_missing(database, "user_settings", "push_missing_lunch INTEGER NOT NULL DEFAULT 0")
add_column_if_missing(database, "user_settings", "push_missing_dinner INTEGER NOT NULL DEFAULT 0")
add_column_if_missing(database, "user_settings", "push_small_snack INTEGER NOT NULL DEFAULT 0")
if default_owner_id is not None:
database.execute(
"""
UPDATE items
SET household_id = COALESCE(household_id, ?),
owner_user_id = COALESCE(owner_user_id, created_by, ?),
visibility = CASE WHEN visibility IS NULL OR visibility = '' THEN 'shared' ELSE visibility END
WHERE household_id IS NULL OR owner_user_id IS NULL OR visibility IS NULL OR visibility = ''
""",
(default_household_id, default_owner_id),
)
database.execute(
"""
UPDATE shopping_entries
SET household_id = COALESCE(household_id, ?),
owner_user_id = COALESCE(owner_user_id, added_by, ?),
visibility = CASE WHEN visibility IS NULL OR visibility = '' THEN 'shared' ELSE visibility END
WHERE household_id IS NULL OR owner_user_id IS NULL OR visibility IS NULL OR visibility = ''
""",
(default_household_id, default_owner_id),
)
database.execute(
"""
UPDATE plan_entries
SET household_id = COALESCE(household_id, ?),
owner_user_id = COALESCE(owner_user_id, created_by, ?),
visibility = CASE WHEN visibility IS NULL OR visibility = '' THEN 'shared' ELSE visibility END
WHERE household_id IS NULL OR owner_user_id IS NULL OR visibility IS NULL OR visibility = ''
""",
(default_household_id, default_owner_id),
)
else:
database.execute("UPDATE items SET visibility = 'shared' WHERE visibility IS NULL OR visibility = ''")
database.execute("UPDATE shopping_entries SET visibility = 'shared' WHERE visibility IS NULL OR visibility = ''")
database.execute("UPDATE plan_entries SET visibility = 'shared' WHERE visibility IS NULL OR visibility = ''")
sync_default_categories(database)
migrate_item_profiles(database)
migrate_food_flavor_profiles(database)
migrate_item_archive_state(database)
database.execute(
"""
INSERT OR IGNORE INTO user_settings (user_id)
SELECT id FROM users
"""
)
database.execute("UPDATE items SET energy_density = 'neutral' WHERE energy_density IS NULL OR energy_density = ''")
database.execute("UPDATE items SET base_type = 'neutral' WHERE base_type IS NULL OR base_type = ''")
database.execute("UPDATE items SET flavor_profile = 'neutral' WHERE flavor_profile IS NULL OR flavor_profile = ''")
database.execute("UPDATE items SET suggestion_role = 'base' WHERE suggestion_role IS NULL OR suggestion_role = ''")
database.execute("UPDATE items SET suggestion_priority = 'normal' WHERE suggestion_priority IS NULL OR suggestion_priority = ''")
database.execute("UPDATE items SET can_be_meal_core = 0 WHERE can_be_meal_core IS NULL")
database.execute("UPDATE items SET meal_tags = '' WHERE meal_tags IS NULL")
database.execute("UPDATE items SET is_archived = 0 WHERE is_archived IS NULL")
database.execute("UPDATE items SET is_quick_added = 0 WHERE is_quick_added IS NULL")
database.execute("UPDATE shopping_entries SET shopping_note = '' WHERE shopping_note IS NULL")
database.execute("UPDATE user_settings SET suggestion_style = 'balanced' WHERE suggestion_style IS NULL OR suggestion_style = ''")
database.execute("UPDATE user_settings SET energy_preference = 'neutral' WHERE energy_preference IS NULL OR energy_preference = ''")
database.execute("UPDATE user_settings SET protein_preference = 'mixed' WHERE protein_preference IS NULL OR protein_preference = ''")
database.execute("UPDATE user_settings SET push_missing_breakfast = 0 WHERE push_missing_breakfast IS NULL")
database.execute("UPDATE user_settings SET push_missing_lunch = 0 WHERE push_missing_lunch IS NULL")
database.execute("UPDATE user_settings SET push_missing_dinner = 0 WHERE push_missing_dinner IS NULL")
database.execute("UPDATE user_settings SET push_small_snack = 0 WHERE push_small_snack IS NULL")
database.execute(
"""
CREATE UNIQUE INDEX IF NOT EXISTS idx_users_email_unique
ON users (email)
WHERE email IS NOT NULL AND email != ''
"""
)
database.execute(
"""
CREATE INDEX IF NOT EXISTS idx_items_kind_name
ON items (kind, name)
"""
)
database.execute(
"""
CREATE INDEX IF NOT EXISTS idx_items_household_visibility
ON items (household_id, visibility, availability_state)
"""
)
database.execute(
"""
CREATE INDEX IF NOT EXISTS idx_items_household_visibility_archived
ON items (household_id, visibility, is_archived, availability_state)
"""
)
database.execute(
"""
CREATE INDEX IF NOT EXISTS idx_items_target_user
ON items (target_user_id)
"""
)
database.execute(
"""
CREATE INDEX IF NOT EXISTS idx_plan_entries_household_visibility
ON plan_entries (household_id, visibility, plan_date)
"""
)
database.execute(
"""
CREATE INDEX IF NOT EXISTS idx_shopping_entries_household_visibility
ON shopping_entries (household_id, visibility, is_checked)
"""
)
database.execute("DROP INDEX IF EXISTS idx_shopping_entries_open_item")
database.execute(
"""
CREATE UNIQUE INDEX IF NOT EXISTS idx_shopping_entries_open_item
ON shopping_entries (item_id, COALESCE(shopping_note, ''))
WHERE is_checked = 0
"""
)
database.execute(
"""
CREATE INDEX IF NOT EXISTS idx_shopping_needs_household_activation
ON shopping_needs (household_id, activation_date, is_activated)
"""
)
database.execute(
"""
CREATE INDEX IF NOT EXISTS idx_hidden_generated_suggestions_user
ON hidden_generated_suggestions (user_id)
"""
)
set_meta(database, "schema_version", CURRENT_SCHEMA_VERSION)
def apply_schema(database: sqlite3.Connection) -> None:
bootstrap_legacy_schema(database)
schema_path = Path(__file__).with_name("schema.sql")
database.executescript(schema_path.read_text(encoding="utf-8"))
ensure_schema_upgrades(database)
sync_dayparts(database)
def init_db() -> None:
database = get_db()
apply_schema(database)
database.commit()
def sync_dayparts(database: sqlite3.Connection) -> None:
for entry in DAYPARTS:
database.execute(
"""
INSERT OR IGNORE INTO dayparts (slug, name, sort_order)
VALUES (?, ?, ?)
""",
(entry["slug"], entry["name"], entry["sort_order"]),
)
database.execute(
"""
UPDATE dayparts
SET name = ?, sort_order = ?
WHERE slug = ?
""",
(entry["name"], entry["sort_order"], entry["slug"]),
)
def init_db_if_needed(app: Flask) -> None:
with app.app_context():
init_db()
def user_count() -> int:
row = get_db().execute("SELECT COUNT(*) AS count FROM users").fetchone()
return int(row["count"])
def active_admin_count(household_id: int) -> int:
row = get_db().execute(
"""
SELECT COUNT(*) AS count
FROM users
WHERE household_id = ? AND role = 'admin' AND is_active = 1
""",
(household_id,),
).fetchone()
return int(row["count"])
@click.command("init-db")
@with_appcontext
def init_db_command() -> None:
init_db()
click.echo("Database initialized.")
@click.command("create-user")
@click.argument("username")
@click.argument("password")
@click.option("--display-name", default="", help="Friendly display name.")
@click.option("--email", default="", help="Optional email address.")
@click.option("--role", default="member", type=click.Choice(["admin", "member"]))
@with_appcontext
def create_user_command(username: str, password: str, display_name: str, email: str, role: str) -> None:
database = get_db()
household_id = ensure_default_household(database)
database.execute(
"""
INSERT INTO users (household_id, username, email, display_name, role, password_hash)
VALUES (?, ?, ?, ?, ?, ?)
""",
(
household_id,
username.strip().lower(),
email.strip().lower() or None,
display_name.strip(),
role,
generate_password_hash(password),
),
)
database.commit()
click.echo(f"User '{username}' created.")
def init_app(app: Flask) -> None:
app.teardown_appcontext(close_db)
app.cli.add_command(init_db_command)
app.cli.add_command(create_user_command)
Reference in New Issue View Git Blame Copy Permalink