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\d+)H)?(?:(?P\d+)M)?(?:(?P\d+)S)?)?$") TEXT_DURATION_PART_RE = re.compile( r"(?P\d+(?:\.\d+)?)\s*(?Phours?|hrs?|hr|minutes?|mins?|min|seconds?|secs?|sec|s)\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)) if textual_matches: total_minutes = 0.0 for part in textual_matches: amount = float(part.group("value")) unit = part.group("unit").lower() if unit.startswith(("hour", "hr")): total_minutes += amount * 60 elif unit.startswith(("minute", "min")): total_minutes += amount else: total_minutes += amount / 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)