Add Doris KitchenOwl recipe importer
This commit is contained in:
Fizzlepoof
475
automation/bin/kitchenowl_recipe_import.py
Executable file
475
automation/bin/kitchenowl_recipe_import.py
Executable file
@@ -0,0 +1,475 @@
|
||||
#!/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())
|
||||
|
||||
|
||||
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)
|
||||
|
||||
|
||||
DURATION_RE = re.compile(r'^P(?:T(?:(?P<hours>\d+)H)?(?:(?P<minutes>\d+)M)?(?:(?P<seconds>\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 slugify_tag(value: str) -> str:
|
||||
value = value.strip().lower()
|
||||
value = re.sub(r'[^a-z0-9]+', '-', value)
|
||||
return value.strip('-')[:48]
|
||||
|
||||
|
||||
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_parts: list[str] = []
|
||||
desc = coerce_text(recipe_node.get('description'))
|
||||
instructions = recipe_node.get('recipeInstructions')
|
||||
instructions_text = coerce_text(instructions)
|
||||
if desc:
|
||||
description_parts.append(desc)
|
||||
if instructions_text:
|
||||
description_parts.append(instructions_text)
|
||||
description = '\n\n'.join(part for part in description_parts if part).strip()
|
||||
|
||||
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],
|
||||
}
|
||||
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}')
|
||||
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())
|
||||
Reference in New Issue
Block a user