278 lines
9.6 KiB
Python
278 lines
9.6 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
import subprocess
|
||
from html import unescape
|
||
from html.parser import HTMLParser
|
||
from typing import Any
|
||
from urllib import error, request
|
||
|
||
|
||
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")
|
||
DURATION_RE = re.compile(r"^P(?:T(?:(?P<hours>\d+)H)?(?:(?P<minutes>\d+)M)?(?:(?P<seconds>\d+)S)?)?$")
|
||
|
||
|
||
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 fetch_url(url: str, timeout: int = 30) -> str:
|
||
req = request.Request(
|
||
url,
|
||
headers={
|
||
"User-Agent": "Mozilla/5.0 (Doris Kitchen)",
|
||
"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 Kitchen)",
|
||
"-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,
|
||
],
|
||
capture_output=True,
|
||
check=False,
|
||
)
|
||
if curl.returncode != 0:
|
||
raise RuntimeError(f"Failed to fetch URL: {url}")
|
||
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:
|
||
recipe_type = node.get("@type")
|
||
if isinstance(recipe_type, list):
|
||
return "Recipe" in recipe_type
|
||
return recipe_type == "Recipe"
|
||
|
||
|
||
def choose_recipe_node(nodes: list[dict[str, Any]]) -> dict[str, Any] | None:
|
||
recipes = [node for node in nodes if is_recipe_node(node)]
|
||
if not recipes:
|
||
return None
|
||
recipes.sort(key=lambda node: len(node.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 coerce_text(value: Any) -> str:
|
||
if value is None:
|
||
return ""
|
||
if isinstance(value, list):
|
||
return "\n".join(part for part in (coerce_text(item) for item in value) if part)
|
||
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):
|
||
out.extend(part.strip() for part in re.split(r"[,;|]", value) if part.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 "", ""
|
||
tokens = original.replace("–", "-").replace("—", "-").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 == "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(" ,;")
|
||
description = " ".join(desc_tokens).strip(" ,;")
|
||
return name[:128], description[:512]
|
||
|
||
|
||
def parse_duration_minutes(value: Any) -> int | None:
|
||
if value is None:
|
||
return None
|
||
if isinstance(value, (int, float)):
|
||
return int(value)
|
||
text = str(value).strip()
|
||
if not text:
|
||
return None
|
||
match = DURATION_RE.match(text)
|
||
if match:
|
||
hours = int(match.group("hours") or 0)
|
||
minutes = int(match.group("minutes") or 0)
|
||
seconds = int(match.group("seconds") or 0)
|
||
total = hours * 60 + minutes + (1 if seconds >= 30 else 0)
|
||
return total or None
|
||
number = re.search(r"(\d+)", text)
|
||
return int(number.group(1)) if number else None
|
||
|
||
|
||
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)
|
||
desc = coerce_text(recipe_node.get("description"))
|
||
instructions = coerce_text(recipe_node.get("recipeInstructions"))
|
||
description = "\n\n".join(part for part in [desc, instructions] if part).strip()
|
||
items: list[dict[str, Any]] = []
|
||
seen_items: set[tuple[str, str]] = set()
|
||
for raw in recipe_node.get("recipeIngredient") or []:
|
||
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)
|
||
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)
|
||
yield_value = recipe_node.get("recipeYield")
|
||
if isinstance(yield_value, list) and yield_value:
|
||
yield_value = yield_value[0]
|
||
if yield_value is not None:
|
||
match = re.search(r"(\d+)", str(yield_value))
|
||
if match:
|
||
payload["yields"] = int(match.group(1))
|
||
return payload
|
||
|
||
|
||
def normalize_recipe_from_url(url: str, timeout: int = 30) -> dict[str, Any]:
|
||
html = fetch_url(url, timeout=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}")
|
||
return normalize_recipe(url, recipe_node, title)
|
||
|