eb6da73400
Continuing through the 2nd-pass audit findings.
dedupe_recipes.py CODE-2 (HIGH): _cluster_by_name dropped union-find
single-link agglomerative for the same pair-accumulator pattern
consolidate_foods._cluster already uses. Single-link chained weak
similarities through the recipe corpus the same way it did with foods,
producing one giant cluster on a 250+ corpus that Sonnet would refuse.
Now emits one 2-recipe pair per (i,j) above NAME_THRESHOLD.
consolidate_foods.py + dedupe_recipes.py CODE-1 (HIGH): added
in-loop cancel-poll to both _cluster passes. Polls cancel_check every
5K pair-comparisons so a user-initiated cancel can abort cleanly mid-
scan instead of waiting tens of seconds. Run-walk callers also
re-check _cancelled() right after clustering returns and bail.
server.py CVE-NEW-5 (MED): /login skips OIDC re-init when
session.get('user') exists. Closes the login-CSRF surface where a
malicious cross-origin link `<a href=…/login?next=/poison>` would
re-trigger the OIDC handshake on a logged-in user and silently change
their post-login landing.
server.py CVE-NEW-7 (LOW): Permissions-Policy now sends both
`interest-cohort=()` (FLoC, Chrome ≤94) and `browsing-topics=()`
(Topics API, Chrome ≥115) for opt-out across the lineage.
server.py CVE-NEW-8 (LOW): /auth/callback OAuthError branch no longer
echoes Authentik's raw error string to auth_retry.html. Detail is
logged server-side; users see a generic retry message. Closes the
information-disclosure on Authentik error codes.
server.py CODE-9/CODE-10 (LOW): wrapped int() of ?page= and
?per_page= in try/except so garbage args land on safe defaults
instead of surfacing ValueError as a 500.
Deferred to the proxy-vhost commit:
- CVE-NEW-6 (cauldron bind / ProxyFix trusted-peer filter) needs
paired Apache vhost config (RequestHeader unset X-Forwarded-*)
before the docker-side change is safe; landing it solo would
either break the LAN deploy or leave a half-broken trust chain.
318 lines
12 KiB
Python
318 lines
12 KiB
Python
"""Foods consolidator — clusters Mealie's food rows by similarity, asks
|
|
Sonnet which clusters are dupes, and uses Mealie's PUT /api/foods/merge
|
|
to consolidate. Operates household-scoped via the user's Mealie token.
|
|
|
|
The walk:
|
|
1. Fetch all foods via GET /api/foods (group-wide; the user's
|
|
household admin scope is enough)
|
|
2. Filter to the user's own household (where edit permission exists)
|
|
3. Cluster by rapidfuzz.token_set_ratio ≥ THRESHOLD (default 88, slightly
|
|
stricter than Mealie's parser threshold of 85 to reduce false positives)
|
|
4. Drop singleton clusters (no merge candidate)
|
|
5. For each cluster, call forge.cluster_decision(...) — Sonnet
|
|
decides merge/keep + picks the canonical survivor
|
|
6. Persist the proposal
|
|
|
|
The apply:
|
|
1. For each approved cluster: PUT /api/foods/merge from each discard_id
|
|
into canonical_id
|
|
2. After merges, optionally PUT /api/foods/{canonical_id} with the
|
|
accumulated aliases array so the survivor learns the variant names
|
|
|
|
Same daemon-thread + cancel-respecting + stuck-job-recovery pattern as
|
|
bulk_sterilize.py.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import threading
|
|
from typing import Callable, Optional
|
|
|
|
from rapidfuzz import fuzz, process
|
|
|
|
from .db import DB
|
|
from .forge import Forge, ForgeError
|
|
from .mealie import Mealie, MealieError
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
CLUSTER_THRESHOLD = 88
|
|
|
|
|
|
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 _all_foods(mealie: Mealie) -> list[dict]:
|
|
out: list[dict] = []
|
|
page = 1
|
|
while page <= 20:
|
|
resp = mealie._get("/api/foods", search="", perPage=2000, page=page)
|
|
items = resp.get("items") or resp.get("data") or []
|
|
for item in items:
|
|
out.append(item)
|
|
tp = resp.get("total_pages") or resp.get("totalPages") or 1
|
|
if not items or page >= tp:
|
|
break
|
|
page += 1
|
|
return out
|
|
|
|
|
|
def _foods_in_household(mealie: Mealie, household_id: str) -> list[dict]:
|
|
foods = _all_foods(mealie)
|
|
if not household_id:
|
|
return foods
|
|
out = []
|
|
for f in foods:
|
|
hh = f.get("householdId") or f.get("household_id")
|
|
if not hh:
|
|
h = f.get("household")
|
|
if isinstance(h, dict):
|
|
hh = h.get("id")
|
|
if hh and hh == household_id:
|
|
out.append(f)
|
|
elif not hh:
|
|
# Some Mealie versions don't tag foods with householdId at the
|
|
# food level — they're group-scoped instead. In that case, we
|
|
# consolidate across the whole group (Alice is admin so writes
|
|
# land regardless).
|
|
out.append(f)
|
|
return out
|
|
|
|
|
|
def _cluster(
|
|
foods: list[dict],
|
|
threshold: int = CLUSTER_THRESHOLD,
|
|
cancel_check: Callable[[], bool] | None = None,
|
|
) -> list[list[dict]]:
|
|
"""Pair-based: emit one 2-food candidate per (i, j) where token_set_ratio
|
|
>= threshold. Replaces the original single-link agglomerative which
|
|
produced a 50+ food megacluster on Alice's catalog by chaining weak
|
|
similarities (`2% milk` → `acai berry` → `acai berry juice` → ...).
|
|
|
|
Each emitted pair is a clean Sonnet-decision unit — easier prompt,
|
|
higher accuracy, uniform UI cards. The trade-off (3-way dupes get
|
|
split into 3 pairs that go through review separately) is fine —
|
|
Mealie's merge endpoint is per-pair anyway, and the apply path
|
|
defensively skips a pair whose canonical_id was already merged
|
|
away by an earlier pair.
|
|
|
|
For ~3000 foods this is ~4M comparisons in pure Python — runs in
|
|
a few seconds. For larger catalogs (10K+) the inner loop polls
|
|
cancel_check every 5K comparisons so a user-initiated cancel can
|
|
abort cleanly mid-scan rather than waiting tens of seconds. 2nd-pass
|
|
audit fix CODE-1 (2026-05-02 PM)."""
|
|
n = len(foods)
|
|
names = [(f.get("name") or "").strip().lower() for f in foods]
|
|
pairs: list[list[dict]] = []
|
|
poll_every = 5000
|
|
cmp_count = 0
|
|
for i in range(n):
|
|
if not names[i]:
|
|
continue
|
|
for j in range(i + 1, n):
|
|
if not names[j]:
|
|
continue
|
|
cmp_count += 1
|
|
if cancel_check is not None and cmp_count % poll_every == 0:
|
|
if cancel_check():
|
|
return pairs
|
|
if fuzz.token_set_ratio(names[i], names[j]) >= threshold:
|
|
pairs.append([foods[i], foods[j]])
|
|
return pairs
|
|
|
|
|
|
def _cluster_key(cluster: list[dict]) -> str:
|
|
"""Stable key for the cluster — sorted name + ids."""
|
|
names = sorted((f.get("name") or "").strip().lower() for f in cluster)
|
|
return "|".join(names)[:255]
|
|
|
|
|
|
def run_walk(*, db: DB, job_id: int, mealie: Mealie, forge: Forge) -> None:
|
|
log.info("[consolidate:%s] walk start", job_id)
|
|
|
|
def _cancelled() -> bool:
|
|
s = db.get_consolidate_job_state(job_id)
|
|
return s in ("cancelled", "failed", "done")
|
|
|
|
try:
|
|
hh = _household_id_for(mealie)
|
|
foods = _foods_in_household(mealie, hh)
|
|
log.info("[consolidate:%s] household=%s foods=%d", job_id, hh, len(foods))
|
|
|
|
clusters = _cluster(foods, cancel_check=_cancelled)
|
|
log.info("[consolidate:%s] clusters≥2: %d", job_id, len(clusters))
|
|
if _cancelled():
|
|
log.info("[consolidate:%s] walk aborted during clustering", job_id)
|
|
return
|
|
|
|
with db.conn() as c, c.cursor() as cur:
|
|
cur.execute(
|
|
"UPDATE cauldron_consolidate_jobs SET total_clusters=%s WHERE id=%s",
|
|
(len(clusters), job_id),
|
|
)
|
|
|
|
for cluster in clusters:
|
|
if _cancelled():
|
|
log.info("[consolidate:%s] walk aborted (state changed)", job_id)
|
|
return
|
|
key = _cluster_key(cluster)
|
|
db.update_consolidate_job_progress(job_id, current_cluster=key[:80])
|
|
|
|
try:
|
|
decision = forge.cluster_decision(cluster)
|
|
except (ForgeError, RuntimeError) as e:
|
|
msg = str(e)[:500]
|
|
log.warning("[consolidate:%s] cluster_decision(%s): %s", job_id, key[:60], msg)
|
|
db.insert_consolidate_proposal(
|
|
job_id=job_id, cluster_key=key,
|
|
cluster=cluster, decision=None, error=msg,
|
|
)
|
|
db.update_consolidate_job_progress(job_id, error_delta=1, last_error=msg)
|
|
continue
|
|
|
|
db.insert_consolidate_proposal(
|
|
job_id=job_id, cluster_key=key,
|
|
cluster=cluster, decision=decision, error=None,
|
|
)
|
|
db.update_consolidate_job_progress(job_id, processed_delta=1)
|
|
|
|
db.finalize_consolidate_job(job_id, state="review")
|
|
log.info("[consolidate:%s] walk done; awaiting review", job_id)
|
|
except Exception:
|
|
log.exception("[consolidate:%s] walk crashed", job_id)
|
|
try:
|
|
db.finalize_consolidate_job(job_id, state="failed")
|
|
except Exception:
|
|
log.exception("[consolidate:%s] couldn't mark failed", job_id)
|
|
|
|
|
|
def run_apply(*, db: DB, job_id: int, mealie: Mealie) -> None:
|
|
log.info("[consolidate:%s] apply start", job_id)
|
|
|
|
def _cancelled() -> bool:
|
|
s = db.get_consolidate_job_state(job_id)
|
|
return s in ("cancelled", "failed", "done")
|
|
|
|
try:
|
|
approved = db.list_approved_unapplied_consolidate(job_id)
|
|
for row in approved:
|
|
if _cancelled():
|
|
log.info("[consolidate:%s] apply aborted (state changed)", job_id)
|
|
return
|
|
decision = row.get("sonnet_decision") or {}
|
|
if isinstance(decision, str):
|
|
try:
|
|
decision = json.loads(decision)
|
|
except Exception:
|
|
decision = {}
|
|
|
|
if not decision.get("merge"):
|
|
# User shouldn't have approved a non-merge cluster; defensive
|
|
db.mark_consolidate_proposal_applied(
|
|
row["id"], error="cluster decision was 'no merge' but row was approved",
|
|
)
|
|
continue
|
|
|
|
canonical_id = decision.get("canonical_id") or ""
|
|
discard_ids = decision.get("discard_ids") or []
|
|
alias_additions = decision.get("alias_additions") or []
|
|
if not canonical_id or not discard_ids:
|
|
db.mark_consolidate_proposal_applied(
|
|
row["id"], error="missing canonical_id or discard_ids",
|
|
)
|
|
continue
|
|
|
|
db.update_consolidate_job_progress(
|
|
job_id, current_cluster=row.get("cluster_key", "")[:80]
|
|
)
|
|
|
|
err: str | None = None
|
|
for did in discard_ids:
|
|
if did == canonical_id:
|
|
continue
|
|
try:
|
|
mealie.merge_foods(from_id=did, to_id=canonical_id)
|
|
except MealieError as e:
|
|
msg = str(e)
|
|
# Pair-based clustering can emit overlapping pairs:
|
|
# if (A,B) was already approved+merged, a later (A,C)
|
|
# pair has stale A. Mealie returns 404 — treat that
|
|
# as already-handled, not an error.
|
|
if "404" in msg or "not found" in msg.lower():
|
|
log.info("[consolidate:%s] merge %s → %s: stale (already merged elsewhere)", job_id, did, canonical_id)
|
|
continue
|
|
err = f"merge {did} → {canonical_id}: {e}"
|
|
log.warning("[consolidate:%s] %s", job_id, err)
|
|
break
|
|
|
|
if err is None and alias_additions:
|
|
# After successful merges, push the aliases onto the survivor
|
|
try:
|
|
survivor = mealie._get(f"/api/foods/{canonical_id}")
|
|
existing = survivor.get("aliases") or []
|
|
existing_names: set = set()
|
|
for a in existing:
|
|
if isinstance(a, str):
|
|
existing_names.add(a.strip().lower())
|
|
elif isinstance(a, dict):
|
|
existing_names.add((a.get("name") or "").strip().lower())
|
|
new_aliases: list[dict] = list(existing)
|
|
for n in alias_additions:
|
|
if n.strip().lower() not in existing_names and n.strip().lower() != (survivor.get("name") or "").strip().lower():
|
|
new_aliases.append({"name": n.strip()})
|
|
existing_names.add(n.strip().lower())
|
|
if new_aliases != existing:
|
|
body = dict(survivor)
|
|
body["aliases"] = new_aliases
|
|
mealie.update_food(canonical_id, body)
|
|
except MealieError as e:
|
|
err = f"alias update on {canonical_id}: {e}"
|
|
log.warning("[consolidate:%s] %s", job_id, err)
|
|
|
|
if err:
|
|
db.mark_consolidate_proposal_applied(row["id"], error=err)
|
|
db.update_consolidate_job_progress(
|
|
job_id, error_delta=1, last_error=err,
|
|
)
|
|
else:
|
|
db.mark_consolidate_proposal_applied(row["id"])
|
|
db.update_consolidate_job_progress(job_id, merged_delta=1)
|
|
|
|
db.finalize_consolidate_job(job_id, state="done")
|
|
log.info("[consolidate:%s] apply done", job_id)
|
|
except Exception:
|
|
log.exception("[consolidate:%s] apply crashed", job_id)
|
|
try:
|
|
db.finalize_consolidate_job(job_id, state="failed")
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def spawn_walk_thread(*, db: DB, job_id: int, mealie: Mealie, forge: Forge) -> threading.Thread:
|
|
t = threading.Thread(
|
|
target=run_walk,
|
|
kwargs={"db": db, "job_id": job_id, "mealie": mealie, "forge": forge},
|
|
name=f"consolidate-walk-{job_id}",
|
|
daemon=True,
|
|
)
|
|
t.start()
|
|
return t
|
|
|
|
|
|
def spawn_apply_thread(*, db: DB, job_id: int, mealie: Mealie) -> threading.Thread:
|
|
t = threading.Thread(
|
|
target=run_apply,
|
|
kwargs={"db": db, "job_id": job_id, "mealie": mealie},
|
|
name=f"consolidate-apply-{job_id}",
|
|
daemon=True,
|
|
)
|
|
t.start()
|
|
return t
|