#!/usr/bin/env python3 """Doris-managed KitchenOwl recipe importer. Flow: 1. Try KitchenOwl's native scrape endpoint for supported sites. 2. If that fails, fetch the page and extract schema.org Recipe JSON-LD. 3. Normalize ingredients/tags into KitchenOwl's create-recipe payload. 4. Optionally create the recipe via the KitchenOwl API. Secrets stay in env vars or a local .env, never in git. """ from __future__ import annotations import argparse import json import os import re import subprocess import sys import textwrap from html import unescape from html.parser import HTMLParser from pathlib import Path from typing import Any from urllib import error, parse, request DEFAULT_TIMEOUT = 30 SCRIPT_DIR = Path(__file__).resolve().parent REPO_ROOT = SCRIPT_DIR.parent.parent DEFAULT_ENV_PATH = REPO_ROOT / 'automation' / '.env' UNIT_WORDS = { 'teaspoon','teaspoons','tsp','tsp.','tablespoon','tablespoons','tbsp','tbsp.','cup','cups', 'oz','ounce','ounces','lb','lbs','pound','pounds','gram','grams','g','kg','ml','l','pinch', 'dash','clove','cloves','can','cans','package','packages','pkg','slice','slices','stalk','stalks' } STOPWORDS = {'fresh','large','small','medium','optional','to','taste','for','the','a','an'} TAG_KEYS = ('recipeCategory', 'recipeCuisine', 'keywords') class ScriptExtractor(HTMLParser): def __init__(self) -> None: super().__init__() self.in_script = False self.script_type = None self._buf: list[str] = [] self.scripts: list[tuple[str | None, str]] = [] self.title: str | None = None self._in_title = False def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: attrs_d = dict(attrs) if tag == 'script': self.in_script = True self.script_type = attrs_d.get('type') self._buf = [] elif tag == 'title': self._in_title = True def handle_endtag(self, tag: str) -> None: if tag == 'script' and self.in_script: self.scripts.append((self.script_type, ''.join(self._buf))) self.in_script = False self.script_type = None self._buf = [] elif tag == 'title': self._in_title = False def handle_data(self, data: str) -> None: if self.in_script: self._buf.append(data) elif self._in_title: self.title = (self.title or '') + data def load_env_file(path: Path) -> None: if not path.exists(): return for raw_line in path.read_text(encoding='utf-8').splitlines(): line = raw_line.strip() if not line or line.startswith('#') or '=' not in line: continue key, value = line.split('=', 1) key = key.strip() value = value.strip().strip('"').strip("'") os.environ.setdefault(key, value) def env(name: str, default: str | None = None) -> str | None: return os.environ.get(name, default) def api_request(method: str, url: str, token: str, *, params: dict[str, Any] | None = None, body: dict[str, Any] | None = None, timeout: int = DEFAULT_TIMEOUT) -> tuple[int, Any, dict[str, str]]: if params: url += ('&' if '?' in url else '?') + parse.urlencode(params, doseq=True) data = None headers = { 'Authorization': f'Bearer {token}', 'User-Agent': 'doris-kitchenowl-importer/1.0', 'Accept': 'application/json', } if body is not None: data = json.dumps(body).encode('utf-8') headers['Content-Type'] = 'application/json' req = request.Request(url, data=data, method=method.upper(), headers=headers) try: with request.urlopen(req, timeout=timeout) as resp: raw = resp.read().decode('utf-8', 'replace') ctype = resp.headers.get('Content-Type', '') payload = json.loads(raw) if 'json' in ctype else raw return resp.status, payload, dict(resp.headers.items()) except error.HTTPError as exc: raw = exc.read().decode('utf-8', 'replace') try: payload = json.loads(raw) except Exception: payload = raw return exc.code, payload, dict(exc.headers.items()) class RecipeDomExtractor(HTMLParser): def __init__(self) -> None: super().__init__() self._stack: list[str] = [] self._ingredient_depth: int | None = None self._direction_depth: int | None = None self._current_li: list[str] | None = None self._current_heading: list[str] | None = None self.ingredients: list[str] = [] self.sections: list[dict[str, Any]] = [] self._current_section: dict[str, Any] | None = None def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: attrs_d = dict(attrs) classes = set((attrs_d.get("class") or "").split()) self._stack.append(tag) if "ingredients-list" in classes: self._ingredient_depth = len(self._stack) if "directions-list" in classes: self._direction_depth = len(self._stack) if self.in_ingredients and tag == "li": self._current_li = [] elif self.in_directions and tag == "li": self._current_li = [] if self.in_directions and tag == "p": self._current_heading = [] def handle_endtag(self, tag: str) -> None: if self.in_ingredients and tag == "li" and self._current_li is not None: text = normalize_whitespace("".join(self._current_li)) if text: self.ingredients.append(text) self._current_li = None elif self.in_directions and tag == "li" and self._current_li is not None: text = normalize_whitespace("".join(self._current_li)) if text: self._ensure_section().setdefault("itemListElement", []).append({"@type": "HowToStep", "text": text}) self._current_li = None if self.in_directions and tag == "p" and self._current_heading is not None: heading = normalize_whitespace("".join(self._current_heading)).rstrip(":*").strip() if heading: self._current_section = {"@type": "HowToSection", "name": heading, "itemListElement": []} self.sections.append(self._current_section) self._current_heading = None if self._stack: self._stack.pop() if self._ingredient_depth is not None and len(self._stack) < self._ingredient_depth: self._ingredient_depth = None if self._direction_depth is not None and len(self._stack) < self._direction_depth: self._direction_depth = None def handle_data(self, data: str) -> None: if self.in_ingredients and self._current_li is not None: self._current_li.append(data) return if not self.in_directions: return if self._current_li is not None: self._current_li.append(data) elif self._current_heading is not None: self._current_heading.append(data) @property def in_ingredients(self) -> bool: return self._ingredient_depth is not None @property def in_directions(self) -> bool: return self._direction_depth is not None def _ensure_section(self) -> dict[str, Any]: if self._current_section is None: self._current_section = {"@type": "HowToSection", "name": "Instructions", "itemListElement": []} self.sections.append(self._current_section) return self._current_section def fetch_url(url: str, timeout: int = DEFAULT_TIMEOUT) -> str: req = request.Request( url, headers={ 'User-Agent': 'Mozilla/5.0 (Doris KitchenOwl Importer)', 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', 'Accept-Language': 'en-US,en;q=0.9', }, ) try: with request.urlopen(req, timeout=timeout) as resp: charset = resp.headers.get_content_charset() or 'utf-8' return resp.read().decode(charset, 'replace') except error.HTTPError as exc: if exc.code not in {403, 406, 429}: raise curl = subprocess.run( [ 'curl', '-fsSL', '--max-time', str(timeout), '-A', 'Mozilla/5.0 (Doris KitchenOwl Importer)', '-H', 'Accept: text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', '-H', 'Accept-Language: en-US,en;q=0.9', url, ], check=False, capture_output=True, ) if curl.returncode != 0: raise RuntimeError(f'Failed to fetch URL via urllib/curl: {url} ({curl.stderr.decode("utf-8", "replace").strip()})') return curl.stdout.decode('utf-8', 'replace') def flatten_graph(node: Any) -> list[dict[str, Any]]: out: list[dict[str, Any]] = [] if isinstance(node, list): for item in node: out.extend(flatten_graph(item)) elif isinstance(node, dict): if '@graph' in node: out.extend(flatten_graph(node['@graph'])) else: out.append(node) return out def is_recipe_node(node: dict[str, Any]) -> bool: t = node.get('@type') if isinstance(t, list): return 'Recipe' in t return t == 'Recipe' def choose_recipe_node(nodes: list[dict[str, Any]]) -> dict[str, Any] | None: recipes = [n for n in nodes if is_recipe_node(n)] if not recipes: return None recipes.sort(key=lambda r: len(r.get('recipeIngredient') or []), reverse=True) return recipes[0] def parse_json_ld_recipe(html: str) -> tuple[dict[str, Any] | None, str | None]: parser = ScriptExtractor() parser.feed(html) nodes: list[dict[str, Any]] = [] for script_type, content in parser.scripts: if script_type and 'ld+json' not in script_type: continue content = content.strip() if not content: continue try: data = json.loads(content) except Exception: continue nodes.extend(flatten_graph(data)) return choose_recipe_node(nodes), (parser.title.strip() if parser.title else None) def enrich_recipe_node_from_dom(recipe_node: dict[str, Any], html: str) -> dict[str, Any]: extractor = RecipeDomExtractor() extractor.feed(html) enriched = dict(recipe_node) if extractor.ingredients and not (enriched.get("recipeIngredient") or []): enriched["recipeIngredient"] = extractor.ingredients if extractor.sections and not extract_instruction_steps(enriched.get("recipeInstructions")): enriched["recipeInstructions"] = extractor.sections return enriched DURATION_RE = re.compile(r'^P(?:T(?:(?P\d+)H)?(?:(?P\d+)M)?(?:(?P\d+)S)?)?$') def parse_duration_minutes(value: Any) -> int | None: if value is None: return None if isinstance(value, (int, float)): return int(value) if not isinstance(value, str): return None value = value.strip() if not value: return None m = DURATION_RE.match(value) if m: hours = int(m.group('hours') or 0) minutes = int(m.group('minutes') or 0) seconds = int(m.group('seconds') or 0) total = hours * 60 + minutes + (1 if seconds >= 30 else 0) return total or None num = re.search(r'(\d+)', value) return int(num.group(1)) if num else None def coerce_text(value: Any) -> str: if value is None: return '' if isinstance(value, list): return '\n'.join(coerce_text(v) for v in value if coerce_text(v)) if isinstance(value, dict): if 'text' in value: return coerce_text(value['text']) if '@value' in value: return coerce_text(value['@value']) return json.dumps(value, ensure_ascii=False) return unescape(str(value)).strip() def normalize_whitespace(value: str) -> str: cleaned = re.sub(r'\s+', ' ', value or '').strip() return re.sub(r'(?<=[.!?])(?=[A-Z])', ' ', cleaned) def split_paragraphs(value: str) -> list[str]: parts: list[str] = [] for chunk in re.split(r'\n\s*\n|\r\n\r\n', value or ''): cleaned = normalize_whitespace(chunk) if cleaned: parts.append(cleaned) return parts def extract_instruction_steps(value: Any) -> list[str]: steps: list[str] = [] def visit(node: Any) -> None: if node is None: return if isinstance(node, str): for chunk in re.split(r'\n+', node): cleaned = normalize_whitespace(chunk) if cleaned: steps.append(cleaned) return if isinstance(node, list): for item in node: visit(item) return if isinstance(node, dict): if node.get('@type') == 'HowToSection': name = normalize_whitespace(coerce_text(node.get('name'))) if name: steps.append(f'{name}:') visit(node.get('itemListElement')) return if node.get('@type') in {'HowToStep', 'ListItem'}: text = normalize_whitespace(coerce_text(node.get('text') or node.get('name') or node.get('item'))) if text: steps.append(text) elif node.get('itemListElement') is not None: visit(node.get('itemListElement')) return if node.get('itemListElement') is not None: visit(node.get('itemListElement')) return text = normalize_whitespace(coerce_text(node.get('text') or node.get('name'))) if text: steps.append(text) visit(value) deduped: list[str] = [] seen: set[str] = set() for step in steps: key = step.lower() if key in seen: continue seen.add(key) deduped.append(step) return deduped def build_description(recipe_node: dict[str, Any]) -> str: intro_parts = split_paragraphs(coerce_text(recipe_node.get('description'))) intro = intro_parts[0] if intro_parts else '' notes = intro_parts[1:] if len(intro_parts) > 1 else [] steps = extract_instruction_steps(recipe_node.get('recipeInstructions')) parts: list[str] = [] if intro: parts.append(intro) if steps: rendered_steps = [] step_number = 1 in_notes = False for step in steps: if step.endswith(':'): rendered_steps.append(step) in_notes = step[:-1].strip().lower() == 'notes' step_number = 1 else: if in_notes: rendered_steps.append(f'- {step}') else: rendered_steps.append(f'{step_number}. {step}') step_number += 1 parts.append('\n'.join(rendered_steps)) elif intro_parts[1:]: parts.append('\n'.join(intro_parts[1:])) if notes: parts.append('Notes:\n- ' + '\n- '.join(notes)) return '\n\n'.join(part for part in parts if part).strip() def slugify_tag(value: str) -> str: value = value.strip().lower() value = re.sub(r'[^a-z0-9]+', '-', value) return value.strip('-')[:48] def extract_image_url(value: Any) -> str | None: if isinstance(value, str): cleaned = value.strip() return cleaned or None if isinstance(value, list): for item in value: found = extract_image_url(item) if found: return found return None if isinstance(value, dict): for key in ('url', 'contentUrl'): found = extract_image_url(value.get(key)) if found: return found return None def normalize_keywords(value: Any) -> list[str]: out: list[str] = [] if isinstance(value, str): parts = re.split(r'[,;|]', value) out.extend(p.strip() for p in parts if p.strip()) elif isinstance(value, list): for item in value: out.extend(normalize_keywords(item)) return out def split_ingredient(line: str) -> tuple[str, str]: original = ' '.join(line.strip().split()) if not original: return '', '' cleaned = original.replace('–', '-').replace('—', '-') tokens = cleaned.split() desc_tokens: list[str] = [] name_tokens = tokens[:] while name_tokens: token = name_tokens[0].strip(',()').lower() if re.fullmatch(r'[\d/.-]+', token) or token in {'x', '×'}: desc_tokens.append(name_tokens.pop(0)) continue if token in UNIT_WORDS: desc_tokens.append(name_tokens.pop(0)) continue if token in {'of'} and desc_tokens: desc_tokens.append(name_tokens.pop(0)) continue break if not name_tokens: return original[:128], '' while name_tokens and name_tokens[0].strip(',').lower() in STOPWORDS and len(name_tokens) > 1: desc_tokens.append(name_tokens.pop(0)) name = ' '.join(name_tokens).strip(' ,;') if ',' in name: first, rest = [p.strip() for p in name.split(',', 1)] if first and len(first.split()) <= 5: name = first if rest: desc_tokens.append(rest) description = ' '.join(desc_tokens).strip(' ,;') return name[:128], description[:512] def normalize_recipe(url: str, recipe_node: dict[str, Any], title_fallback: str | None = None) -> dict[str, Any]: name = coerce_text(recipe_node.get('name')) or (title_fallback or url) description = build_description(recipe_node) ingredients = recipe_node.get('recipeIngredient') or [] items = [] seen_items: set[tuple[str, str]] = set() for raw in ingredients: line = coerce_text(raw) if not line: continue item_name, item_desc = split_ingredient(line) if not item_name: item_name = line[:128] key = (item_name.lower(), item_desc.lower()) if key in seen_items: continue seen_items.add(key) items.append({'name': item_name, 'description': item_desc, 'optional': False}) tags: list[str] = [] for key in TAG_KEYS: tags.extend(normalize_keywords(recipe_node.get(key))) deduped_tags: list[str] = [] seen_tags: set[str] = set() for tag in tags: slug = slugify_tag(tag) if slug and slug not in seen_tags: seen_tags.add(slug) deduped_tags.append(slug) yields_val = recipe_node.get('recipeYield') yields_num = None if isinstance(yields_val, list) and yields_val: yields_val = yields_val[0] if yields_val is not None: match = re.search(r'(\d+)', str(yields_val)) if match: yields_num = int(match.group(1)) payload = { 'name': name[:128], 'description': description[:20000], 'source': url, 'visibility': 0, 'items': items, 'tags': deduped_tags[:20], } image_url = extract_image_url(recipe_node.get('image')) if image_url: payload['photo'] = image_url cook_time = parse_duration_minutes(recipe_node.get('cookTime')) prep_time = parse_duration_minutes(recipe_node.get('prepTime')) total_time = parse_duration_minutes(recipe_node.get('totalTime')) if cook_time is not None: payload['cook_time'] = cook_time if prep_time is not None: payload['prep_time'] = prep_time if total_time is not None: payload['time'] = total_time elif cook_time is not None or prep_time is not None: payload['time'] = (cook_time or 0) + (prep_time or 0) if yields_num is not None: payload['yields'] = yields_num return payload def find_existing_recipe(base_url: str, token: str, household_id: int, payload: dict[str, Any]) -> dict[str, Any] | None: status, data, _ = api_request('GET', f'{base_url}/api/household/{household_id}/recipe', token) if status != 200 or not isinstance(data, list): return None want_source = (payload.get('source') or '').strip().lower() want_name = (payload.get('name') or '').strip().lower() for recipe in data: source = str(recipe.get('source') or '').strip().lower() name = str(recipe.get('name') or '').strip().lower() if want_source and source and source == want_source: return recipe if want_name and name == want_name: return recipe return None def import_one(args: argparse.Namespace, base_url: str, token: str, household_id: int, url: str) -> dict[str, Any]: result: dict[str, Any] = {'url': url} payload = None scrape_error = None if not args.skip_kitchenowl_scrape: status, data, _ = api_request('GET', f'{base_url}/api/household/{household_id}/recipe/scrape', token, params={'url': url}, timeout=args.timeout) if status == 200 and isinstance(data, dict): payload = data result['strategy'] = 'kitchenowl-scrape' else: scrape_error = {'status': status, 'detail': data} if payload is None: html = fetch_url(url, timeout=args.timeout) recipe_node, title = parse_json_ld_recipe(html) if not recipe_node: raise RuntimeError(f'No schema.org Recipe JSON-LD found for {url}') recipe_node = enrich_recipe_node_from_dom(recipe_node, html) payload = normalize_recipe(url, recipe_node, title) result['strategy'] = 'json-ld-fallback' if scrape_error: result['scrape_error'] = scrape_error payload['visibility'] = args.visibility result['payload'] = payload existing = None if args.allow_duplicates else find_existing_recipe(base_url, token, household_id, payload) if existing: result['existing'] = {'id': existing.get('id'), 'name': existing.get('name'), 'source': existing.get('source')} result['status'] = 'duplicate' return result if args.create: status, created, _ = api_request('POST', f'{base_url}/api/household/{household_id}/recipe', token, body=payload, timeout=args.timeout) result['create_status'] = status result['created'] = created result['status'] = 'created' if status == 200 else 'error' else: result['status'] = 'dry-run' return result def parse_args() -> argparse.Namespace: p = argparse.ArgumentParser(description='Import recipe links into KitchenOwl with Doris-managed fallback parsing.') p.add_argument('urls', nargs='+', help='Recipe URLs to normalize/import') p.add_argument('--env-file', default=str(DEFAULT_ENV_PATH), help='Optional .env file to load before reading env vars') p.add_argument('--base-url', default=None, help='KitchenOwl base URL, e.g. https://owl.paccoco.com') p.add_argument('--token', default=None, help='KitchenOwl bearer token (prefer env var instead)') p.add_argument('--household-id', type=int, default=None, help='KitchenOwl household id') p.add_argument('--create', action='store_true', help='Create recipes in KitchenOwl. Default is dry-run only.') p.add_argument('--allow-duplicates', action='store_true', help='Do not skip name/source duplicates.') p.add_argument('--skip-kitchenowl-scrape', action='store_true', help='Go straight to Doris JSON-LD parsing.') p.add_argument('--visibility', type=int, choices=[0,1,2], default=0, help='Recipe visibility: 0 private, 1 link, 2 public.') p.add_argument('--pretty', action='store_true', help='Pretty-print JSON output') p.add_argument('--timeout', type=int, default=DEFAULT_TIMEOUT, help='HTTP timeout in seconds') return p.parse_args() def main() -> int: args = parse_args() load_env_file(Path(args.env_file).expanduser()) base_url = (args.base_url or env('KITCHENOWL_BASE_URL') or env('KITCHENOWL_URL') or '').rstrip('/') token = args.token or env('KITCHENOWL_API_TOKEN') or '' household_raw = args.household_id or env('KITCHENOWL_HOUSEHOLD_ID') if not base_url: print('Missing KitchenOwl base URL. Set KITCHENOWL_BASE_URL or KITCHENOWL_URL.', file=sys.stderr) return 2 if not token: print('Missing KitchenOwl API token. Set KITCHENOWL_API_TOKEN.', file=sys.stderr) return 2 if not household_raw: print('Missing KitchenOwl household id. Set KITCHENOWL_HOUSEHOLD_ID.', file=sys.stderr) return 2 household_id = int(household_raw) results = [] failures = 0 for url in args.urls: try: results.append(import_one(args, base_url, token, household_id, url)) except Exception as exc: failures += 1 results.append({'url': url, 'status': 'error', 'error': str(exc)}) dump = json.dumps(results, ensure_ascii=False, indent=2 if args.pretty else None) print(dump) return 1 if failures else 0 if __name__ == '__main__': raise SystemExit(main())