cauldron/cauldron/enrich_recipes.py

"""Recipe metadata enrichment — once per recipe, persist forever.

Walks the user's household recipes, calls forge.enrich_recipe(recipe)
on each one, persists the structured metadata to cauldron_recipe_meta
keyed by (household_id, recipe_slug).

No review/apply step — the metadata is purely additive. The plan
generator reads it next time it runs.

Idempotent: skips recipes already enriched at the current
db.DB.ENRICH_VERSION. Bumping the version (when the prompt or schema
changes) forces a re-walk.

Same daemon-thread + cancel + stuck-recovery pattern as the rest.
"""
from __future__ import annotations

import json
import logging
import threading

from .db import DB
from .forge import Forge, ForgeError
from .mealie import Mealie, MealieError

log = logging.getLogger(__name__)


def _household_id_for(mealie: Mealie) -> str | None:
    me = mealie.who_am_i()
    hid = me.get("householdId") or me.get("household_id")
    if not hid:
        h = me.get("household")
        if isinstance(h, dict):
            hid = h.get("id")
    return hid


def _recipe_household_id(recipe: dict) -> str | None:
    hid = recipe.get("householdId") or recipe.get("household_id")
    if hid:
        return hid
    h = recipe.get("household")
    if isinstance(h, dict):
        return h.get("id")
    return None


def run_enrich(
    *,
    db: DB,
    job_id: int,
    household_id: int,
    mealie: Mealie,
    forge: Forge,
    force: bool = False,
) -> None:
    """Walk all recipes in the user's household, enrich each via Sonnet,
    persist. Runs in a daemon thread; respects external cancel."""
    log.info("[enrich:%s] start (force=%s)", job_id, force)

    def _cancelled() -> bool:
        s = db.get_enrich_job_state(job_id)
        return s in ("cancelled", "failed", "done")

    try:
        user_household = _household_id_for(mealie)

        # Pull every recipe slug from Mealie (paginated)
        slugs: list[tuple[str, str]] = []
        page = 1
        while page <= 50:
            resp = mealie.list_recipes(page=page, per_page=100)
            items = resp.get("items") or []
            for r in items:
                slug = r.get("slug")
                name = r.get("name") or slug or ""
                if slug:
                    slugs.append((slug, name))
            tp = resp.get("total_pages") or resp.get("totalPages") or 1
            if not items or page >= tp:
                break
            page += 1

        with db.conn() as c, c.cursor() as cur:
            cur.execute(
                "UPDATE cauldron_enrich_jobs SET total_recipes=%s WHERE id=%s",
                (len(slugs), job_id),
            )

        for slug, name in slugs:
            if _cancelled():
                log.info("[enrich:%s] aborted (state changed)", job_id)
                return

            # Skip cross-household — only enrich what the user owns
            try:
                recipe = mealie.get_recipe(slug)
            except MealieError as e:
                msg = str(e)[:500]
                log.warning("[enrich:%s] get_recipe(%s): %s", job_id, slug, msg)
                db.update_enrich_job_progress(
                    job_id, error_delta=1, current_slug=slug, last_error=msg
                )
                continue

            if user_household:
                rec_hh = _recipe_household_id(recipe)
                if rec_hh and rec_hh != user_household:
                    db.update_enrich_job_progress(
                        job_id, skipped_delta=1, current_slug=slug
                    )
                    continue

            # Skip if already enriched at the current version (unless forced)
            if not force:
                existing = db.get_recipe_meta(household_id, slug)
                if existing and existing.get("enrich_version") == db.ENRICH_VERSION:
                    db.update_enrich_job_progress(
                        job_id, skipped_delta=1, current_slug=slug
                    )
                    continue

            db.update_enrich_job_progress(job_id, current_slug=slug)
            try:
                meta = forge.enrich_recipe(recipe)
            except (ForgeError, RuntimeError) as e:
                msg = str(e)[:500]
                log.warning("[enrich:%s] enrich_recipe(%s): %s", job_id, slug, msg)
                db.update_enrich_job_progress(
                    job_id, error_delta=1, current_slug=slug, last_error=msg
                )
                continue

            try:
                db.upsert_recipe_meta(
                    household_id=household_id,
                    recipe_slug=slug,
                    meta_json=json.dumps(meta, ensure_ascii=False),
                    version=db.ENRICH_VERSION,
                )
                db.update_enrich_job_progress(job_id, enriched_delta=1)
            except Exception as e:
                msg = str(e)[:500]
                log.warning("[enrich:%s] persist(%s): %s", job_id, slug, msg)
                db.update_enrich_job_progress(
                    job_id, error_delta=1, current_slug=slug, last_error=msg
                )

        db.finalize_enrich_job(job_id, state="done")
        log.info("[enrich:%s] done", job_id)
    except Exception:
        log.exception("[enrich:%s] crashed", job_id)
        try:
            db.finalize_enrich_job(job_id, state="failed")
        except Exception:
            pass


def spawn_thread(
    *,
    db: DB,
    job_id: int,
    household_id: int,
    mealie: Mealie,
    forge: Forge,
    force: bool = False,
) -> threading.Thread:
    t = threading.Thread(
        target=run_enrich,
        kwargs={
            "db": db, "job_id": job_id, "household_id": household_id,
            "mealie": mealie, "forge": forge, "force": force,
        },
        name=f"enrich-recipes-{job_id}",
        daemon=True,
    )
    t.start()
    return t