#!/usr/bin/env python3 """ Export all Tandoor ingredients (foods) with their properties to SQLite. Run: python3 export_tandoor_to_sqlite.py """ import sqlite3 import json import os import sys # ── config ────────────────────────────────────────────────────────────────── SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) SKILL_DIR = os.path.dirname(SCRIPT_DIR) # …/tandoor-manager/ DB_PATH = os.path.join(SKILL_DIR, "tandoor_foods.db") PREV_PATH = os.path.join(SKILL_DIR, "tandoor_foods_previous.db") CURR_PATH = os.path.join(SKILL_DIR, "tandoor_foods_current.db") KEY_FILE = os.path.expanduser("~/.openclaw/workspace/.secrets") SERVER = "https://rezept.papst.beer" API_KEY = None with open(KEY_FILE) as f: for line in f: line = line.strip() if line.startswith("TANDOOR_API_KEY="): API_KEY = line.split("=", 1)[1].strip().strip('"') break if API_KEY is None: # fallback: first non-empty line f.seek(0) for line in f: line = line.strip() if line: API_KEY = line break HEADERS = { "Authorization": f"Bearer {API_KEY}", "Referer": SERVER, } # ── helpers ────────────────────────────────────────────────────────────────── import urllib.request import urllib.parse def api(endpoint: str, method="GET", data=None): url = SERVER + endpoint body = json.dumps(data).encode() if data else None req = urllib.request.Request(url, data=body, method=method, headers=HEADERS) if data: req.add_header("Content-Type", "application/json") with urllib.request.urlopen(req, timeout=30) as resp: return json.loads(resp.read()) def paginate(endpoint: str, page_size=250): """Yield all items from a paginated endpoint.""" page = 1 while True: result = api(f"{endpoint}?page={page}&page_size={page_size}") results = result.get("results", result) if isinstance(result, dict) else result yield from results if isinstance(result, dict) and not result.get("next"): break page += 1 # ── main ───────────────────────────────────────────────────────────────────── def main(): print("Fetching foods from Tandoor...") foods = list(paginate("/api/food/")) print(f" → {len(foods)} foods fetched") print("Fetching property types...") prop_types = {p["id"]: p["name"] for p in paginate("/api/property-type/")} print(f" → {len(prop_types)} property types") # Also fetch units units = {u["id"]: u["name"] for u in paginate("/api/unit/")} print(f" → {len(units)} units") # Rotate: previous.db → (if exists) delete, current.db → previous.db if os.path.exists(PREV_PATH): os.remove(PREV_PATH) if os.path.exists(CURR_PATH): os.rename(CURR_PATH, PREV_PATH) conn = sqlite3.connect(CURR_PATH) conn.row_factory = sqlite3.Row cur = conn.cursor() # ── schema ──────────────────────────────────────────────────────────────── for tbl in ("foods", "properties", "property_types", "units", "meta"): cur.execute(f"DROP TABLE IF EXISTS {tbl}") cur.execute(""" CREATE TABLE property_types ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, kind TEXT, unit TEXT, description TEXT )""") cur.execute(""" CREATE TABLE units ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, plural_name TEXT, abbreviation TEXT, description TEXT )""") cur.execute(""" CREATE TABLE foods ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, name_plural TEXT, description TEXT, fdc_id INTEGER, food_type TEXT, ingredient_code TEXT, parent_id INTEGER, openapi_codename TEXT, property_type_ids TEXT, -- JSON list of property_type ids this food has properties_food_amount REAL, properties_food_unit_id INTEGER, supermarket_category_id INTEGER, supermarket_category_name TEXT, created_at TEXT, updated_at TEXT )""") cur.execute(""" CREATE TABLE properties ( id INTEGER PRIMARY KEY, food_id INTEGER NOT NULL, property_type_id INTEGER NOT NULL, property_type_name TEXT, property_amount REAL, FOREIGN KEY (food_id) REFERENCES foods(id), FOREIGN KEY (property_type_id) REFERENCES property_types(id) )""") # ── insert property types & units ──────────────────────────────────────── pt_rows = list(paginate("/api/property-type/")) for p in pt_rows: cur.execute(""" INSERT INTO property_types (id, name, kind, unit, description) VALUES (?, ?, ?, ?, ?)""", (p["id"], p["name"], p.get("kind"), p.get("unit"), p.get("description"))) prop_types = {p["id"]: p["name"] for p in pt_rows} u_rows = list(paginate("/api/unit/")) for u in u_rows: cur.execute(""" INSERT INTO units (id, name, plural_name, abbreviation, description) VALUES (?, ?, ?, ?, ?)""", (u["id"], u["name"], u.get("name_plural"), u.get("abbreviation"), u.get("description"))) units = {u["id"]: u["name"] for u in u_rows} # ── insert foods + properties ───────────────────────────────────────────── for food in foods: fid = food["id"] pt_ids = [p["property_type"]["id"] for p in food.get("properties", [])] unit_id = (food.get("properties_food_unit") or {}).get("id") cat = food.get("supermarket_category") or {} cur.execute(""" INSERT INTO foods ( id, name, name_plural, description, fdc_id, food_type, ingredient_code, parent_id, openapi_codename, property_type_ids, properties_food_amount, properties_food_unit_id, supermarket_category_id, supermarket_category_name, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( fid, food.get("name"), food.get("name_plural"), food.get("description"), food.get("fdc_id"), food.get("food_type"), food.get("ingredient_code"), food.get("parent"), food.get("openapi_codename"), json.dumps(pt_ids), food.get("properties_food_amount"), unit_id, cat.get("id"), cat.get("name"), food.get("created_at"), food.get("updated_at"), )) for prop in food.get("properties", []): pt = prop.get("property_type", {}) cur.execute(""" INSERT INTO properties (id, food_id, property_type_id, property_type_name, property_amount) VALUES (?, ?, ?, ?, ?)""", ( prop["id"], fid, pt.get("id"), pt.get("name"), prop.get("property_amount"), )) conn.commit() conn.execute("CREATE INDEX idx_properties_food_id ON properties(food_id)") conn.execute("CREATE INDEX idx_properties_pt ON properties(property_type_id)") # ── meta table ───────────────────────────────────────────────────────── from datetime import datetime, timezone now = datetime.now(timezone.utc).isoformat() conn.execute("CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT)") conn.execute("INSERT INTO meta (key, value) VALUES ('last_export', ?)", (now,)) conn.execute("INSERT INTO meta (key, value) VALUES ('food_count', ?)", (len(foods),)) conn.commit() conn.close() # also keep a symlink/persistent copy for other tools if os.path.exists(DB_PATH): os.remove(DB_PATH) import shutil shutil.copy2(CURR_PATH, DB_PATH) print(f"\n✅ SQLite database written to: {CURR_PATH}") print(f" Previous: {PREV_PATH if os.path.exists(PREV_PATH) else 'none'}") print(f" Foods: {len(foods)}") print(f" Property types: {len(prop_types)}") print(f" Units: {len(units)}") print(f" Last export: {now}") # quick summary no_fdc = sum(1 for f in foods if not f.get("fdc_id")) if no_fdc > 0: print(f"\n Foods WITHOUT fdc_id: {no_fdc}") for f in foods: if not f.get("fdc_id"): print(f" - {f['name']} (id={f['id']})") if __name__ == "__main__": main()