c9603a07c1
- migrations 012 + 013: cauldron_meal_plan_slots + cauldron_pick_points - db: list_plan_slots, save_plan_slots, delete_plan_slots, mark_plan_generated, clear_plan_generated, award_pick_points, enrich_plan_with_slots; scoreboard extended with points (sum from pick_points) and weeks_locked alias - forge.generate_plan: sonnet prompt builds 7-day plan respecting picks, validates slot count + day uniqueness + slug-in-pool, fills picker_subs from ground-truth picks (model output is advisory) - POST /api/plan/generate: race-safe (existing slots → 409 with plan), lock-aware (locked → 409), idempotent - POST /api/plan/regenerate: re-roll for the original generator, gated by ownership + lock; wipes slots + pick_points then re-runs generate - plan.html: generate CTA + 7 day cards with picker chips + AI reason + re-roll button (generator-only, pre-lock); scoreboard now shows points + wins - /list: pulls plan slots, queries Mealie for ingredients, runs aggregator, renders 48px-tall checkbox shopping list with localStorage state per plan_id - tests: 13 new tests across forge.generate_plan + /api/plan/generate routes + /list view + scoreboard SQL inspection. conftest+_testenv stub pymysql/oidc/foods at import time so tests run against module-level app without a live DB. Both pytest and `unittest discover` paths green (27/27). Defers: bulk sterilizer admin (A1), foods dedupe (A2), Mealie shopping-list- export (button rendered but disabled). 7-slot count is fixed at the endpoint (no UI for slot-count selection yet). Spec: memory/spec-cauldron-v0.3.md
855 lines
36 KiB
Python
855 lines
36 KiB
Python
"""DB access + migrations against mariadb.
|
||
|
||
Uses PyMySQL with a tiny per-request connection (no pool) — Cauldron is
|
||
LAN-only family-internal, traffic is single-digit qps. If load ever grows
|
||
swap in DBUtils.PooledDB or SQLAlchemy.
|
||
"""
|
||
from contextlib import contextmanager
|
||
from pathlib import Path
|
||
|
||
import pymysql
|
||
import pymysql.cursors
|
||
|
||
|
||
MIGRATIONS = [
|
||
# 001 — bookkeeping
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS schema_migrations (
|
||
version VARCHAR(16) PRIMARY KEY,
|
||
applied_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
||
""",
|
||
# 002 — users (Authentik subject is the PK)
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS cauldron_users (
|
||
authentik_sub VARCHAR(190) PRIMARY KEY,
|
||
email VARCHAR(255) NOT NULL,
|
||
display_name VARCHAR(255),
|
||
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||
last_seen DATETIME,
|
||
INDEX idx_email (email)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
||
""",
|
||
# 003 — per-user encrypted Mealie tokens
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS cauldron_user_mealie_tokens (
|
||
authentik_sub VARCHAR(190) PRIMARY KEY,
|
||
encrypted_token BLOB NOT NULL,
|
||
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||
last_validated DATETIME,
|
||
last_failure_at DATETIME,
|
||
last_failure_reason VARCHAR(500),
|
||
FOREIGN KEY (authentik_sub) REFERENCES cauldron_users(authentik_sub)
|
||
ON DELETE CASCADE
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
||
""",
|
||
# 004 — chat / AI run log (joins to clawdforge runs server-side)
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS cauldron_chat_log (
|
||
id BIGINT PRIMARY KEY AUTO_INCREMENT,
|
||
authentik_sub VARCHAR(190) NOT NULL,
|
||
ts DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||
intent VARCHAR(64),
|
||
forge_duration_ms INT,
|
||
forge_model VARCHAR(64),
|
||
prompt_chars INT,
|
||
result_chars INT,
|
||
ok BOOLEAN NOT NULL DEFAULT TRUE,
|
||
error VARCHAR(500),
|
||
INDEX idx_user_ts (authentik_sub, ts)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
||
""",
|
||
# 005 — meal picks: per-user list of recipes the user wants in the next
|
||
# AI meal plan run. Pre-populated wishlist that the planner respects.
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS cauldron_meal_picks (
|
||
authentik_sub VARCHAR(190) NOT NULL,
|
||
recipe_slug VARCHAR(255) NOT NULL,
|
||
recipe_name VARCHAR(500) NOT NULL,
|
||
added_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||
PRIMARY KEY (authentik_sub, recipe_slug),
|
||
INDEX idx_user_added (authentik_sub, added_at)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
||
""",
|
||
# 006 — households (cached mirror of Mealie's household) + membership.
|
||
# Keyed by Mealie's UUID. Multiple cauldron users join via the same
|
||
# Mealie household to share picks/plans.
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS cauldron_households (
|
||
id BIGINT PRIMARY KEY AUTO_INCREMENT,
|
||
mealie_household_id VARCHAR(64) UNIQUE NOT NULL,
|
||
name VARCHAR(255) NOT NULL,
|
||
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
||
""",
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS cauldron_household_members (
|
||
household_id BIGINT NOT NULL,
|
||
authentik_sub VARCHAR(190) NOT NULL,
|
||
role VARCHAR(32) NOT NULL DEFAULT 'member',
|
||
joined_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||
PRIMARY KEY (household_id, authentik_sub),
|
||
FOREIGN KEY (household_id) REFERENCES cauldron_households(id) ON DELETE CASCADE,
|
||
FOREIGN KEY (authentik_sub) REFERENCES cauldron_users(authentik_sub) ON DELETE CASCADE
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
||
""",
|
||
# 007 — meal plans (per household per week). Lock state + race metadata.
|
||
# week_start = Monday (date) of the week.
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS cauldron_meal_plans (
|
||
id BIGINT PRIMARY KEY AUTO_INCREMENT,
|
||
household_id BIGINT NOT NULL,
|
||
week_start DATE NOT NULL,
|
||
generated_by_sub VARCHAR(190),
|
||
generated_at DATETIME,
|
||
locked_by_sub VARCHAR(190),
|
||
locked_at DATETIME,
|
||
locked_reason ENUM('user','auto') DEFAULT NULL,
|
||
UNIQUE KEY uk_household_week (household_id, week_start),
|
||
INDEX idx_locked_by (locked_by_sub),
|
||
FOREIGN KEY (household_id) REFERENCES cauldron_households(id) ON DELETE CASCADE
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
||
""",
|
||
# 008 — local recipe index for fast in-process search. Mirrors enough
|
||
# of Mealie's recipe shape to fuzzy-rank without round-tripping to
|
||
# Mealie on every keystroke. Refreshed on demand (on first /recipes
|
||
# load, after pin/unpin, every 5min, or on /me 'refresh' button).
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS cauldron_recipe_index (
|
||
household_id BIGINT NOT NULL,
|
||
slug VARCHAR(255) NOT NULL,
|
||
name VARCHAR(500) NOT NULL,
|
||
description TEXT,
|
||
tags_text TEXT,
|
||
cats_text TEXT,
|
||
foods_text TEXT,
|
||
ings_text TEXT,
|
||
date_updated DATETIME,
|
||
date_added DATETIME,
|
||
last_made DATETIME,
|
||
total_time VARCHAR(64),
|
||
recipe_yield VARCHAR(255),
|
||
raw_json JSON,
|
||
indexed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||
PRIMARY KEY (household_id, slug),
|
||
FULLTEXT KEY ft_text (name, description, tags_text, cats_text, foods_text),
|
||
INDEX idx_household (household_id),
|
||
FOREIGN KEY (household_id) REFERENCES cauldron_households(id) ON DELETE CASCADE
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
||
""",
|
||
# 009 — refresh state per household
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS cauldron_recipe_index_state (
|
||
household_id BIGINT PRIMARY KEY,
|
||
last_refreshed_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||
recipe_count INT NOT NULL DEFAULT 0,
|
||
FOREIGN KEY (household_id) REFERENCES cauldron_households(id) ON DELETE CASCADE
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
||
""",
|
||
# 010 — canonical foods table for the unit-aware aggregator. Each row is
|
||
# ONE food (e.g. "rice", "butter", "onion") with density + unit class.
|
||
# Seeded from USDA SR Legacy via scripts/build_foods_seed.py; will be
|
||
# extended with claude-curated entries in v0.3 step 2.
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS cauldron_foods (
|
||
id BIGINT PRIMARY KEY AUTO_INCREMENT,
|
||
canonical_name VARCHAR(255) NOT NULL,
|
||
plural_name VARCHAR(255),
|
||
category VARCHAR(64),
|
||
density_g_per_ml DECIMAL(6,3),
|
||
common_size_g DECIMAL(8,2),
|
||
default_unit_class ENUM('mass','volume','count','mixed') NOT NULL DEFAULT 'mass',
|
||
usda_fdc_id INT,
|
||
usda_description VARCHAR(500),
|
||
notes JSON,
|
||
source ENUM('usda','claude','manual') NOT NULL DEFAULT 'usda',
|
||
last_updated DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
|
||
UNIQUE KEY uk_canonical (canonical_name),
|
||
INDEX idx_category (category),
|
||
INDEX idx_usda (usda_fdc_id)
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
||
""",
|
||
# 011 — Mealie food_id → cauldron food_id mapping per household. The
|
||
# foods dedupe step (v0.3 A2) populates this. Aggregator joins through
|
||
# this to group ingredients across recipes by canonical food.
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS cauldron_food_mapping (
|
||
household_id BIGINT NOT NULL,
|
||
mealie_food_id VARCHAR(64) NOT NULL,
|
||
cauldron_food_id BIGINT NOT NULL,
|
||
confidence DECIMAL(4,2) NOT NULL DEFAULT 1.00,
|
||
mapped_by ENUM('exact','fuzzy','claude','manual') NOT NULL DEFAULT 'fuzzy',
|
||
mapped_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||
PRIMARY KEY (household_id, mealie_food_id),
|
||
INDEX idx_canonical (cauldron_food_id),
|
||
FOREIGN KEY (household_id) REFERENCES cauldron_households(id) ON DELETE CASCADE,
|
||
FOREIGN KEY (cauldron_food_id) REFERENCES cauldron_foods(id) ON DELETE CASCADE
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
||
""",
|
||
# 012 — AI-generated meal plan slots. One row per (plan, day). Created
|
||
# when a household member triggers /api/plan/generate. picker_subs JSON
|
||
# holds the authentik_subs of household members who pinned this slot's
|
||
# recipe (empty list if AI-chosen). reason is the AI's user-facing
|
||
# rationale. notes is reserved for future swap/edit history.
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS cauldron_meal_plan_slots (
|
||
id BIGINT PRIMARY KEY AUTO_INCREMENT,
|
||
plan_id BIGINT NOT NULL,
|
||
day VARCHAR(10) NOT NULL,
|
||
recipe_slug VARCHAR(255) NOT NULL,
|
||
recipe_name VARCHAR(500) NOT NULL,
|
||
source ENUM('mealie','pick') NOT NULL DEFAULT 'mealie',
|
||
picker_subs JSON,
|
||
reason VARCHAR(500),
|
||
notes JSON,
|
||
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||
UNIQUE KEY uk_plan_day (plan_id, day),
|
||
INDEX idx_plan (plan_id),
|
||
FOREIGN KEY (plan_id) REFERENCES cauldron_meal_plans(id) ON DELETE CASCADE
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
||
""",
|
||
# 013 — pick-points ledger. 1pt awarded when a member's pick lands in
|
||
# a generated plan ('pick_used'). Reserved reasons for v0.4: first-to-
|
||
# lock + streak bonuses. Joins to households + plans + users so a row
|
||
# disappears cleanly if any of them are removed.
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS cauldron_pick_points (
|
||
id BIGINT PRIMARY KEY AUTO_INCREMENT,
|
||
household_id BIGINT NOT NULL,
|
||
plan_id BIGINT NOT NULL,
|
||
authentik_sub VARCHAR(190) NOT NULL,
|
||
points INT NOT NULL,
|
||
reason ENUM('pick_used','first_to_lock','streak_bonus') NOT NULL,
|
||
awarded_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||
INDEX idx_household_user (household_id, authentik_sub),
|
||
INDEX idx_plan (plan_id),
|
||
FOREIGN KEY (household_id) REFERENCES cauldron_households(id) ON DELETE CASCADE,
|
||
FOREIGN KEY (plan_id) REFERENCES cauldron_meal_plans(id) ON DELETE CASCADE,
|
||
FOREIGN KEY (authentik_sub) REFERENCES cauldron_users(authentik_sub) ON DELETE CASCADE
|
||
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
|
||
""",
|
||
]
|
||
|
||
|
||
class DB:
|
||
def __init__(self, *, host: str, port: int, name: str, user: str, password: str):
|
||
self.kwargs = dict(
|
||
host=host,
|
||
port=port,
|
||
user=user,
|
||
password=password,
|
||
database=name,
|
||
charset="utf8mb4",
|
||
cursorclass=pymysql.cursors.DictCursor,
|
||
autocommit=False,
|
||
)
|
||
|
||
@contextmanager
|
||
def conn(self):
|
||
c = pymysql.connect(**self.kwargs)
|
||
try:
|
||
yield c
|
||
c.commit()
|
||
except Exception:
|
||
c.rollback()
|
||
raise
|
||
finally:
|
||
c.close()
|
||
|
||
def migrate(self) -> list[str]:
|
||
"""Apply pending migrations. Returns list of versions applied."""
|
||
applied: list[str] = []
|
||
with self.conn() as c:
|
||
with c.cursor() as cur:
|
||
cur.execute(MIGRATIONS[0]) # bootstrap migrations table
|
||
cur.execute("SELECT version FROM schema_migrations")
|
||
done = {r["version"] for r in cur.fetchall()}
|
||
for i, sql in enumerate(MIGRATIONS, start=1):
|
||
ver = f"{i:03d}"
|
||
if ver in done:
|
||
continue
|
||
cur.execute(sql)
|
||
# IGNORE to tolerate the multi-worker boot race where two
|
||
# gunicorn workers both bootstrap an empty migrations table
|
||
cur.execute(
|
||
"INSERT IGNORE INTO schema_migrations (version) VALUES (%s)", (ver,)
|
||
)
|
||
applied.append(ver)
|
||
return applied
|
||
|
||
# --- user ops -----------------------------------------------------------
|
||
|
||
def upsert_user(self, *, sub: str, email: str, display_name: str | None) -> None:
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO cauldron_users (authentik_sub, email, display_name, last_seen)
|
||
VALUES (%s, %s, %s, NOW())
|
||
ON DUPLICATE KEY UPDATE
|
||
email = VALUES(email),
|
||
display_name = COALESCE(VALUES(display_name), display_name),
|
||
last_seen = NOW()
|
||
""",
|
||
(sub, email, display_name),
|
||
)
|
||
|
||
def get_user(self, sub: str) -> dict | None:
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT authentik_sub, email, display_name, last_seen FROM cauldron_users WHERE authentik_sub=%s",
|
||
(sub,),
|
||
)
|
||
return cur.fetchone()
|
||
|
||
# --- mealie token ops ---------------------------------------------------
|
||
|
||
def get_user_mealie_token_blob(self, sub: str) -> bytes | None:
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT encrypted_token FROM cauldron_user_mealie_tokens WHERE authentik_sub=%s",
|
||
(sub,),
|
||
)
|
||
row = cur.fetchone()
|
||
return row["encrypted_token"] if row else None
|
||
|
||
def set_user_mealie_token_blob(self, sub: str, blob: bytes) -> None:
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO cauldron_user_mealie_tokens (authentik_sub, encrypted_token, last_validated)
|
||
VALUES (%s, %s, NOW())
|
||
ON DUPLICATE KEY UPDATE
|
||
encrypted_token = VALUES(encrypted_token),
|
||
last_validated = NOW(),
|
||
last_failure_at = NULL,
|
||
last_failure_reason = NULL
|
||
""",
|
||
(sub, blob),
|
||
)
|
||
|
||
def delete_user_mealie_token(self, sub: str) -> None:
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"DELETE FROM cauldron_user_mealie_tokens WHERE authentik_sub=%s",
|
||
(sub,),
|
||
)
|
||
|
||
def mark_user_mealie_token_failure(self, sub: str, reason: str) -> None:
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
UPDATE cauldron_user_mealie_tokens
|
||
SET last_failure_at = NOW(), last_failure_reason = %s
|
||
WHERE authentik_sub = %s
|
||
""",
|
||
(reason[:500], sub),
|
||
)
|
||
|
||
# --- households ---------------------------------------------------------
|
||
|
||
def upsert_household(self, *, mealie_household_id: str, name: str) -> int:
|
||
"""Create or update a household record. Returns local PK (id)."""
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO cauldron_households (mealie_household_id, name)
|
||
VALUES (%s, %s)
|
||
ON DUPLICATE KEY UPDATE name = VALUES(name), id = LAST_INSERT_ID(id)
|
||
""",
|
||
(mealie_household_id, name),
|
||
)
|
||
return cur.lastrowid
|
||
|
||
def add_household_member(self, household_id: int, sub: str, role: str = "member") -> None:
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
INSERT IGNORE INTO cauldron_household_members
|
||
(household_id, authentik_sub, role)
|
||
VALUES (%s, %s, %s)
|
||
""",
|
||
(household_id, sub, role),
|
||
)
|
||
|
||
def get_user_household_id(self, sub: str) -> int | None:
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT household_id FROM cauldron_household_members WHERE authentik_sub=%s LIMIT 1",
|
||
(sub,),
|
||
)
|
||
row = cur.fetchone()
|
||
return row["household_id"] if row else None
|
||
|
||
def list_household_member_subs(self, household_id: int) -> list[str]:
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT authentik_sub FROM cauldron_household_members WHERE household_id=%s",
|
||
(household_id,),
|
||
)
|
||
return [r["authentik_sub"] for r in cur.fetchall()]
|
||
|
||
# --- meal plans (per household per week) -------------------------------
|
||
|
||
def get_or_create_plan(self, household_id: int, week_start) -> dict:
|
||
"""Get the plan record for a (household, week_start), creating an
|
||
empty one if it doesn't exist. week_start is a date (Monday)."""
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
INSERT IGNORE INTO cauldron_meal_plans (household_id, week_start)
|
||
VALUES (%s, %s)
|
||
""",
|
||
(household_id, week_start),
|
||
)
|
||
cur.execute(
|
||
"SELECT * FROM cauldron_meal_plans WHERE household_id=%s AND week_start=%s",
|
||
(household_id, week_start),
|
||
)
|
||
return dict(cur.fetchone())
|
||
|
||
def lock_plan(self, plan_id: int, *, sub: str, reason: str = "user") -> dict:
|
||
"""Lock a plan if not already locked. Returns updated plan dict."""
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
UPDATE cauldron_meal_plans
|
||
SET locked_by_sub = %s, locked_at = NOW(), locked_reason = %s
|
||
WHERE id = %s AND locked_at IS NULL
|
||
""",
|
||
(sub, reason, plan_id),
|
||
)
|
||
cur.execute("SELECT * FROM cauldron_meal_plans WHERE id=%s", (plan_id,))
|
||
return dict(cur.fetchone())
|
||
|
||
def auto_lock_past_unlocked_plans(self, household_id: int, before_date) -> int:
|
||
"""Mark any past unlocked plans as auto-locked. Returns count."""
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
UPDATE cauldron_meal_plans
|
||
SET locked_at = NOW(), locked_reason = 'auto'
|
||
WHERE household_id = %s AND week_start < %s AND locked_at IS NULL
|
||
""",
|
||
(household_id, before_date),
|
||
)
|
||
return cur.rowcount
|
||
|
||
def household_scoreboard(self, household_id: int) -> list[dict]:
|
||
"""Per-user lock counts + pick-points + most recent lock time.
|
||
|
||
Three numbers per row:
|
||
wins — user-locked weeks (excludes auto-locks)
|
||
weeks_locked — alias for wins, preserved for older callers
|
||
points — sum of cauldron_pick_points for this user/household
|
||
|
||
Sort: points desc, then wins desc, then last_win desc — points are
|
||
the headline metric in v0.3 (every pick lands → matters daily).
|
||
|
||
We compute lock counts and points as separate scalar subqueries so
|
||
the JOIN doesn't blow up on the cartesian (members × plans × points).
|
||
"""
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
SELECT
|
||
u.authentik_sub AS sub,
|
||
u.email AS email,
|
||
u.display_name AS display_name,
|
||
COALESCE((
|
||
SELECT COUNT(*) FROM cauldron_meal_plans p
|
||
WHERE p.locked_by_sub = m.authentik_sub
|
||
AND p.household_id = m.household_id
|
||
AND p.locked_reason = 'user'
|
||
), 0) AS wins,
|
||
(
|
||
SELECT MAX(p.locked_at) FROM cauldron_meal_plans p
|
||
WHERE p.locked_by_sub = m.authentik_sub
|
||
AND p.household_id = m.household_id
|
||
AND p.locked_reason = 'user'
|
||
) AS last_win,
|
||
COALESCE((
|
||
SELECT SUM(pp.points) FROM cauldron_pick_points pp
|
||
WHERE pp.household_id = m.household_id
|
||
AND pp.authentik_sub = m.authentik_sub
|
||
), 0) AS points
|
||
FROM cauldron_household_members m
|
||
LEFT JOIN cauldron_users u
|
||
ON u.authentik_sub = m.authentik_sub
|
||
WHERE m.household_id = %s
|
||
ORDER BY points DESC, wins DESC, last_win DESC
|
||
""",
|
||
(household_id,),
|
||
)
|
||
out = []
|
||
for r in cur.fetchall():
|
||
d = dict(r)
|
||
d["points"] = int(d.get("points") or 0)
|
||
d["wins"] = int(d.get("wins") or 0)
|
||
d["weeks_locked"] = d["wins"]
|
||
out.append(d)
|
||
return out
|
||
|
||
def household_streak(self, household_id: int) -> dict | None:
|
||
"""Compute current win streak: walk back from most recent locked week,
|
||
counting consecutive weeks won by the same user. Returns
|
||
{sub, display_name, count} or None if no locks."""
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
SELECT p.week_start, p.locked_by_sub, u.display_name, u.email
|
||
FROM cauldron_meal_plans p
|
||
LEFT JOIN cauldron_users u ON u.authentik_sub = p.locked_by_sub
|
||
WHERE p.household_id = %s
|
||
AND p.locked_at IS NOT NULL
|
||
AND p.locked_reason = 'user'
|
||
ORDER BY p.week_start DESC
|
||
""",
|
||
(household_id,),
|
||
)
|
||
rows = cur.fetchall()
|
||
if not rows:
|
||
return None
|
||
leader = rows[0]["locked_by_sub"]
|
||
count = 0
|
||
for r in rows:
|
||
if r["locked_by_sub"] != leader:
|
||
break
|
||
count += 1
|
||
return {
|
||
"sub": leader,
|
||
"display_name": rows[0]["display_name"] or rows[0]["email"],
|
||
"count": count,
|
||
}
|
||
|
||
# --- plan slots + pick points (v0.3 A4) --------------------------------
|
||
|
||
# Day order is stable Mon..Sun. Used everywhere we need to render slots
|
||
# in calendar order.
|
||
PLAN_DAYS = ("monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday")
|
||
|
||
def list_plan_slots(self, plan_id: int) -> list[dict]:
|
||
"""All slots for a plan, ordered Mon..Sun. picker_subs is decoded
|
||
from JSON to a list (or [] if null)."""
|
||
import json as _json
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
SELECT id, plan_id, day, recipe_slug, recipe_name, source,
|
||
picker_subs, reason, notes, created_at
|
||
FROM cauldron_meal_plan_slots
|
||
WHERE plan_id = %s
|
||
""",
|
||
(plan_id,),
|
||
)
|
||
rows = [dict(r) for r in cur.fetchall()]
|
||
for r in rows:
|
||
ps = r.get("picker_subs")
|
||
if isinstance(ps, str):
|
||
try:
|
||
r["picker_subs"] = _json.loads(ps)
|
||
except Exception:
|
||
r["picker_subs"] = []
|
||
elif ps is None:
|
||
r["picker_subs"] = []
|
||
n = r.get("notes")
|
||
if isinstance(n, str):
|
||
try:
|
||
r["notes"] = _json.loads(n)
|
||
except Exception:
|
||
r["notes"] = None
|
||
order = {d: i for i, d in enumerate(self.PLAN_DAYS)}
|
||
rows.sort(key=lambda r: order.get((r.get("day") or "").lower(), 99))
|
||
return rows
|
||
|
||
def save_plan_slots(self, plan_id: int, slots: list[dict]) -> int:
|
||
"""INSERT IGNORE every slot. Returns count actually inserted —
|
||
callers can use this to detect race contention (zero rows = someone
|
||
else already saved this plan)."""
|
||
import json as _json
|
||
if not slots:
|
||
return 0
|
||
inserted = 0
|
||
with self.conn() as c, c.cursor() as cur:
|
||
for s in slots:
|
||
cur.execute(
|
||
"""
|
||
INSERT IGNORE INTO cauldron_meal_plan_slots
|
||
(plan_id, day, recipe_slug, recipe_name, source,
|
||
picker_subs, reason, notes)
|
||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
|
||
""",
|
||
(
|
||
plan_id,
|
||
(s.get("day") or "").lower()[:10],
|
||
s["recipe_slug"][:255],
|
||
(s.get("recipe_name") or s["recipe_slug"])[:500],
|
||
s.get("source") or ("pick" if s.get("picker_subs") else "mealie"),
|
||
_json.dumps(s.get("picker_subs") or []),
|
||
(s.get("reason") or "")[:500] or None,
|
||
_json.dumps(s["notes"]) if s.get("notes") is not None else None,
|
||
),
|
||
)
|
||
inserted += cur.rowcount
|
||
return inserted
|
||
|
||
def delete_plan_slots(self, plan_id: int) -> int:
|
||
"""Wipe slots for a plan (used by re-roll). Also nukes the matching
|
||
pick_points so the ledger doesn't double-count on regenerate."""
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute("DELETE FROM cauldron_meal_plan_slots WHERE plan_id=%s", (plan_id,))
|
||
slots_removed = cur.rowcount
|
||
cur.execute("DELETE FROM cauldron_pick_points WHERE plan_id=%s", (plan_id,))
|
||
return slots_removed
|
||
|
||
def mark_plan_generated(self, plan_id: int, sub: str) -> dict:
|
||
"""Set generated_by_sub + generated_at IF not already set. Returns
|
||
the post-update plan row. Idempotent for the same generator."""
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
UPDATE cauldron_meal_plans
|
||
SET generated_by_sub = %s,
|
||
generated_at = NOW()
|
||
WHERE id = %s AND generated_at IS NULL
|
||
""",
|
||
(sub, plan_id),
|
||
)
|
||
cur.execute("SELECT * FROM cauldron_meal_plans WHERE id=%s", (plan_id,))
|
||
return dict(cur.fetchone())
|
||
|
||
def clear_plan_generated(self, plan_id: int) -> None:
|
||
"""Re-roll path: clear the generated_by/at marker so the next
|
||
generate writes fresh metadata."""
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
UPDATE cauldron_meal_plans
|
||
SET generated_by_sub = NULL,
|
||
generated_at = NULL
|
||
WHERE id = %s
|
||
""",
|
||
(plan_id,),
|
||
)
|
||
|
||
def award_pick_points(
|
||
self,
|
||
household_id: int,
|
||
plan_id: int,
|
||
sub: str,
|
||
points: int,
|
||
reason: str = "pick_used",
|
||
) -> int:
|
||
"""Insert one ledger row. Returns the new row id. Reason must be one
|
||
of the ENUM values; we don't validate here — DB will reject bad ones."""
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO cauldron_pick_points
|
||
(household_id, plan_id, authentik_sub, points, reason)
|
||
VALUES (%s, %s, %s, %s, %s)
|
||
""",
|
||
(household_id, plan_id, sub, int(points), reason),
|
||
)
|
||
return cur.lastrowid
|
||
|
||
def enrich_plan_with_slots(self, plan: dict) -> dict:
|
||
"""In-place: add `slots` key to a plan dict. Returns the same dict
|
||
for chaining. Empty list if there are no slots yet."""
|
||
plan["slots"] = self.list_plan_slots(plan["id"]) if plan.get("id") else []
|
||
return plan
|
||
|
||
# --- meal picks ---------------------------------------------------------
|
||
|
||
def add_meal_pick(self, sub: str, slug: str, name: str) -> bool:
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
INSERT IGNORE INTO cauldron_meal_picks (authentik_sub, recipe_slug, recipe_name)
|
||
VALUES (%s, %s, %s)
|
||
""",
|
||
(sub, slug, name[:500]),
|
||
)
|
||
return cur.rowcount > 0
|
||
|
||
def remove_meal_pick(self, sub: str, slug: str) -> bool:
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"DELETE FROM cauldron_meal_picks WHERE authentik_sub=%s AND recipe_slug=%s",
|
||
(sub, slug),
|
||
)
|
||
return cur.rowcount > 0
|
||
|
||
def list_meal_picks(self, sub: str) -> list[dict]:
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT recipe_slug, recipe_name, added_at FROM cauldron_meal_picks "
|
||
"WHERE authentik_sub=%s ORDER BY added_at DESC",
|
||
(sub,),
|
||
)
|
||
return [dict(r) for r in cur.fetchall()]
|
||
|
||
def list_meal_pick_slugs(self, sub: str) -> set[str]:
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT recipe_slug FROM cauldron_meal_picks WHERE authentik_sub=%s",
|
||
(sub,),
|
||
)
|
||
return {r["recipe_slug"] for r in cur.fetchall()}
|
||
|
||
def list_household_pick_slugs(self, household_id: int) -> set[str]:
|
||
"""Union of picks across all members of the household."""
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
SELECT DISTINCT p.recipe_slug
|
||
FROM cauldron_meal_picks p
|
||
JOIN cauldron_household_members m ON m.authentik_sub = p.authentik_sub
|
||
WHERE m.household_id = %s
|
||
""",
|
||
(household_id,),
|
||
)
|
||
return {r["recipe_slug"] for r in cur.fetchall()}
|
||
|
||
def list_household_picks_with_pickers(self, household_id: int) -> list[dict]:
|
||
"""All picks across the household, grouped by slug, with the list of
|
||
members who picked each (so the UI can show 'pinned by Alice · Bob').
|
||
Latest pick added_at per slug for ordering."""
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
SELECT
|
||
p.recipe_slug AS slug,
|
||
MIN(p.recipe_name) AS name,
|
||
GROUP_CONCAT(
|
||
DISTINCT COALESCE(NULLIF(u.display_name, ''),
|
||
SUBSTRING_INDEX(u.email, '@', 1))
|
||
ORDER BY p.added_at ASC
|
||
SEPARATOR '|'
|
||
) AS pickers,
|
||
GROUP_CONCAT(
|
||
DISTINCT u.authentik_sub
|
||
ORDER BY p.added_at ASC
|
||
SEPARATOR '|'
|
||
) AS picker_subs,
|
||
MAX(p.added_at) AS last_pick_at,
|
||
COUNT(*) AS pick_count
|
||
FROM cauldron_meal_picks p
|
||
JOIN cauldron_household_members m ON m.authentik_sub = p.authentik_sub
|
||
LEFT JOIN cauldron_users u ON u.authentik_sub = p.authentik_sub
|
||
WHERE m.household_id = %s
|
||
GROUP BY p.recipe_slug
|
||
ORDER BY last_pick_at DESC
|
||
""",
|
||
(household_id,),
|
||
)
|
||
out = []
|
||
for r in cur.fetchall():
|
||
d = dict(r)
|
||
d["pickers"] = (d["pickers"] or "").split("|") if d["pickers"] else []
|
||
d["picker_subs"] = (d["picker_subs"] or "").split("|") if d["picker_subs"] else []
|
||
out.append(d)
|
||
return out
|
||
|
||
# --- recipe index -------------------------------------------------------
|
||
|
||
def get_index_state(self, household_id: int) -> dict | None:
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT last_refreshed_at, recipe_count FROM cauldron_recipe_index_state WHERE household_id=%s",
|
||
(household_id,),
|
||
)
|
||
return cur.fetchone()
|
||
|
||
def replace_recipe_index(self, household_id: int, rows: list[dict]) -> int:
|
||
"""Atomic-ish replace of the index for one household. Drops + reinserts."""
|
||
import json as _json
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute("DELETE FROM cauldron_recipe_index WHERE household_id=%s", (household_id,))
|
||
for r in rows:
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO cauldron_recipe_index
|
||
(household_id, slug, name, description, tags_text, cats_text,
|
||
foods_text, ings_text, date_updated, date_added, last_made,
|
||
total_time, recipe_yield, raw_json)
|
||
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
|
||
""",
|
||
(
|
||
household_id,
|
||
r["slug"],
|
||
r["name"][:500],
|
||
(r.get("description") or "")[:65000],
|
||
(r.get("tags_text") or "")[:65000],
|
||
(r.get("cats_text") or "")[:65000],
|
||
(r.get("foods_text") or "")[:65000],
|
||
(r.get("ings_text") or "")[:65000],
|
||
r.get("date_updated"),
|
||
r.get("date_added"),
|
||
r.get("last_made"),
|
||
(r.get("total_time") or "")[:64],
|
||
(r.get("recipe_yield") or "")[:255],
|
||
_json.dumps(r.get("raw") or {}, default=str),
|
||
),
|
||
)
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO cauldron_recipe_index_state (household_id, last_refreshed_at, recipe_count)
|
||
VALUES (%s, NOW(), %s)
|
||
ON DUPLICATE KEY UPDATE last_refreshed_at=NOW(), recipe_count=VALUES(recipe_count)
|
||
""",
|
||
(household_id, len(rows)),
|
||
)
|
||
return len(rows)
|
||
|
||
def list_indexed_recipes(self, household_id: int, *, category: str | None = None,
|
||
order_by: str = "date_added", order_dir: str = "desc",
|
||
limit: int = 1000, offset: int = 0) -> list[dict]:
|
||
"""Pull the indexed recipe rows. Used both for non-search browse + as
|
||
the candidate set for in-process fuzzy ranking on search."""
|
||
order_col = {
|
||
"date_added": "date_added",
|
||
"date_updated": "date_updated",
|
||
"last_made": "last_made",
|
||
"name": "name",
|
||
}.get(order_by, "date_added")
|
||
order_dir_sql = "DESC" if order_dir.lower() != "asc" else "ASC"
|
||
sql = f"""
|
||
SELECT slug, name, description, tags_text, cats_text, foods_text,
|
||
date_updated, date_added, last_made, total_time, recipe_yield, raw_json
|
||
FROM cauldron_recipe_index
|
||
WHERE household_id = %s
|
||
"""
|
||
params: list = [household_id]
|
||
if category:
|
||
sql += " AND cats_text LIKE %s"
|
||
params.append(f"%{category}%")
|
||
sql += f" ORDER BY {order_col} {order_dir_sql} LIMIT %s OFFSET %s"
|
||
params += [limit, offset]
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(sql, params)
|
||
return [dict(r) for r in cur.fetchall()]
|
||
|
||
# --- chat log -----------------------------------------------------------
|
||
|
||
def log_chat(
|
||
self,
|
||
*,
|
||
sub: str,
|
||
intent: str,
|
||
duration_ms: int,
|
||
model: str,
|
||
prompt_chars: int,
|
||
result_chars: int,
|
||
ok: bool,
|
||
error: str | None = None,
|
||
) -> None:
|
||
with self.conn() as c, c.cursor() as cur:
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO cauldron_chat_log
|
||
(authentik_sub, intent, forge_duration_ms, forge_model,
|
||
prompt_chars, result_chars, ok, error)
|
||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
|
||
""",
|
||
(sub, intent, duration_ms, model, prompt_chars, result_chars, ok, (error or "")[:500] or None),
|
||
)
|