Expand Doris Kitchen import, failures, and tag curation

home/doris-kitchen/bin/curate-kitchenowl-tags.py

@@ -0,0 +1,119 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import json
+import os
+import sys
+from pathlib import Path
+from urllib import parse, request
+
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from app.services.tagging import curate_recipe_tags  # noqa: E402
+
+
+def load_env_file(path: Path) -> None:
+    if not path.exists():
+        return
+    for line in path.read_text().splitlines():
+        line = line.strip()
+        if not line or line.startswith("#") or "=" not in line:
+            continue
+        key, value = line.split("=", 1)
+        if key and key not in os.environ:
+            os.environ[key] = value.strip().strip('"').strip("'")
+
+
+def api_request(method: str, base_url: str, token: str, path: str, body: dict | None = None, timeout: int = 60):
+    data = json.dumps(body).encode("utf-8") if body is not None else None
+    req = request.Request(
+        f"{base_url}{path}",
+        data=data,
+        headers={
+            "Authorization": f"Bearer {token}",
+            "Accept": "application/json",
+            "Content-Type": "application/json",
+            "User-Agent": "doris-kitchen-tag-curator/1.0",
+        },
+        method=method,
+    )
+    with request.urlopen(req, timeout=timeout) as resp:
+        raw = resp.read().decode("utf-8", "replace")
+        return resp.status, json.loads(raw) if raw else None
+
+
+def normalize_existing_tags(tags: list) -> list[str]:
+    normalized: list[str] = []
+    for tag in tags or []:
+        if isinstance(tag, dict):
+            name = str(tag.get("name") or "").strip()
+        else:
+            name = str(tag or "").strip()
+        if name:
+            normalized.append(name)
+    return normalized
+
+
+def main() -> int:
+    env_file = Path(os.environ.get("DORIS_KITCHEN_ENV_FILE", ROOT / ".env"))
+    load_env_file(env_file)
+
+    base_url = os.environ.get("KITCHENOWL_BASE_URL", "").rstrip("/")
+    token = os.environ.get("KITCHENOWL_API_TOKEN", "")
+    household = os.environ.get("KITCHENOWL_HOUSEHOLD_ID", "1")
+    if not base_url or not token:
+        print("Missing KitchenOwl credentials.", file=sys.stderr)
+        return 2
+
+    _, recipes = api_request("GET", base_url, token, f"/api/household/{parse.quote(str(household))}/recipe")
+    changed: list[dict] = []
+    for recipe in recipes or []:
+        existing_tags = normalize_existing_tags(recipe.get("tags") or [])
+        curated_tags = curate_recipe_tags(
+            name=str(recipe.get("name") or ""),
+            description=str(recipe.get("description") or ""),
+            items=recipe.get("items") or [],
+            source=str(recipe.get("source") or ""),
+            existing_tags=existing_tags,
+        )
+        if curated_tags == existing_tags:
+            continue
+        body = {
+            "name": recipe.get("name"),
+            "description": recipe.get("description") or "",
+            "source": recipe.get("source"),
+            "visibility": int(recipe.get("visibility", 0) or 0),
+            "items": [
+                {
+                    "name": item.get("name"),
+                    "description": item.get("description", ""),
+                    "optional": bool(item.get("optional", False)),
+                }
+                for item in (recipe.get("items") or [])
+            ],
+            "tags": curated_tags,
+            "cook_time": recipe.get("cook_time"),
+            "prep_time": recipe.get("prep_time"),
+            "time": recipe.get("time"),
+            "yields": recipe.get("yields"),
+            "photo": recipe.get("photo"),
+            "planned": bool(recipe.get("planned", False)),
+            "planned_cooking_dates": recipe.get("planned_cooking_dates") or [],
+            "planned_days": recipe.get("planned_days") or [],
+        }
+        api_request("POST", base_url, token, f"/api/recipe/{recipe['id']}", body)
+        changed.append({
+            "id": recipe.get("id"),
+            "name": recipe.get("name"),
+            "before": existing_tags,
+            "after": curated_tags,
+        })
+
+    print(json.dumps({"changed": changed, "changed_count": len(changed)}, ensure_ascii=False, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())