506 lines
18 KiB
Python
506 lines
18 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
|
||
|
||
from app.services.tagging import curate_recipe_tags
|
||
|
||
|
||
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)?)?$")
|
||
TEXT_DURATION_PART_RE = re.compile(
|
||
r"(?P<value>\d+(?:\.\d+)?)\s*(?P<unit>days?|day|hours?|hrs?|hr|minutes?|mins?|min|seconds?|secs?|sec|s)\b",
|
||
re.IGNORECASE,
|
||
)
|
||
OVERNIGHT_RE = re.compile(r"\bovernight\b", re.IGNORECASE)
|
||
|
||
|
||
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
|
||
|
||
|
||
|
||
|
||
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 = 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 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
|
||
|
||
|
||
|
||
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 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 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):
|
||
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
|
||
textual_matches = list(TEXT_DURATION_PART_RE.finditer(text))
|
||
overnight_count = len(OVERNIGHT_RE.findall(text))
|
||
if textual_matches or overnight_count:
|
||
total_minutes = 0.0
|
||
for part in textual_matches:
|
||
amount = float(part.group("value"))
|
||
unit = part.group("unit").lower()
|
||
if unit.startswith("day"):
|
||
total_minutes += amount * 24 * 60
|
||
elif unit.startswith(("hour", "hr")):
|
||
total_minutes += amount * 60
|
||
elif unit.startswith(("minute", "min")):
|
||
total_minutes += amount
|
||
else:
|
||
total_minutes += amount / 60
|
||
total_minutes += overnight_count * 12 * 60
|
||
rounded = int(total_minutes + 0.5)
|
||
return rounded 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)
|
||
description = build_description(recipe_node)
|
||
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 = curate_recipe_tags(
|
||
name=name,
|
||
description=description,
|
||
items=items,
|
||
source=url,
|
||
existing_tags=tags,
|
||
)
|
||
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)
|
||
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}")
|
||
recipe_node = enrich_recipe_node_from_dom(recipe_node, html)
|
||
return normalize_recipe(url, recipe_node, title)
|