From 3f333595a243f15d88438a920d4bec61467632a7 Mon Sep 17 00:00:00 2001 From: Fizzlepoof Date: Thu, 21 May 2026 23:31:32 +0000 Subject: [PATCH] feat: improve KitchenOwl import parsing and tagging --- automation/bin/kitchenowl_recipe_import.py | 107 +++++++- .../app/services/recipe_importer.py | 109 +++++++- home/doris-kitchen/app/services/tagging.py | 245 ++++++++++++++---- home/doris-kitchen/tests/test_tagging.py | 121 +++++++++ 4 files changed, 528 insertions(+), 54 deletions(-) create mode 100644 home/doris-kitchen/tests/test_tagging.py diff --git a/automation/bin/kitchenowl_recipe_import.py b/automation/bin/kitchenowl_recipe_import.py index 1973f10..03eca13 100755 --- a/automation/bin/kitchenowl_recipe_import.py +++ b/automation/bin/kitchenowl_recipe_import.py @@ -120,6 +120,91 @@ def api_request(method: str, url: str, token: str, *, params: dict[str, Any] | N 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, @@ -197,6 +282,17 @@ def parse_json_ld_recipe(html: str) -> tuple[dict[str, Any] | None, str | None]: 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)?)?$') @@ -310,12 +406,18 @@ def build_description(recipe_node: dict[str, Any]) -> str: 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: - rendered_steps.append(f'{step_number}. {step}') - step_number += 1 + 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:])) @@ -499,6 +601,7 @@ def import_one(args: argparse.Namespace, base_url: str, token: str, household_id 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: diff --git a/home/doris-kitchen/app/services/recipe_importer.py b/home/doris-kitchen/app/services/recipe_importer.py index 0bed5e7..0a742bf 100644 --- a/home/doris-kitchen/app/services/recipe_importer.py +++ b/home/doris-kitchen/app/services/recipe_importer.py @@ -57,6 +57,91 @@ class ScriptExtractor(HTMLParser): 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, @@ -134,6 +219,17 @@ def parse_json_ld_recipe(html: str) -> tuple[dict[str, Any] | None, str | None]: 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: @@ -223,12 +319,18 @@ def build_description(recipe_node: dict[str, Any]) -> str: if steps: rendered_steps = [] step_number = 1 - for idx, step in enumerate(steps, start=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: - rendered_steps.append(f"{step_number}. {step}") - step_number += 1 + 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:])) @@ -376,4 +478,5 @@ def normalize_recipe_from_url(url: str, timeout: int = 30) -> dict[str, Any]: 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) diff --git a/home/doris-kitchen/app/services/tagging.py b/home/doris-kitchen/app/services/tagging.py index 97b7d1c..467fcab 100644 --- a/home/doris-kitchen/app/services/tagging.py +++ b/home/doris-kitchen/app/services/tagging.py @@ -2,6 +2,7 @@ from __future__ import annotations import re from typing import Any +from urllib.parse import urlparse def slugify_tag(value: str) -> str: @@ -44,24 +45,39 @@ COURSE_HINTS = { "main dish": "main-course", "salad": "salad", "side dish": "side-dish", + "snack": "snack", "soup": "soup", } METHOD_HINTS = { "baked": "baked", + "fried": "fried", + "fry": "fried", "grilled": "grilled", "noodle": "noodles", "ramen": "noodles", "rice": "rice", - "skillet": "skillet", "slow cooker": "slow-cooker", "stir fry": "stir-fry", "stir-fry": "stir-fry", } +DISH_HINTS = { + "tater tot": {"tater-tots", "potato", "side-dish"}, + "tater tots": {"tater-tots", "potato", "side-dish"}, + "potato": {"potato"}, + "potatoes": {"potato"}, +} + +CUISINE_PHRASE_HINTS = { + "halal": {"middle-eastern"}, + "harissa": {"middle-eastern"}, + "pita": {"middle-eastern"}, +} + SWEET_HINTS = ( - "brown sugar", "cake", "caramel", "chocolate", "cookie", "dessert", "frosting", - "honey", "icing", "maple syrup", "molasses", "muffin", "pie", "pudding", "sweet", + "brown sugar", "cake", "caramel", "chocolate", "coconut milk", "cookie", "dessert", "frosting", + "honey", "icing", "mango", "maple syrup", "molasses", "muffin", "pie", "pudding", "sugar", "sweet", "vanilla", "whipped cream", ) @@ -71,6 +87,41 @@ SAVORY_HINTS = ( "rice", "salt", "savory", "soy sauce", "stock", "thyme", ) +SOURCE_NOISE_WORDS = { + "and", + "better", + "chef", + "com", + "ever", + "guys", + "https", + "joshua", + "recipe", + "recipes", + "the", + "weissman", + "www", +} + +PROTEIN_EXCLUSION_PATTERNS = { + "beef": re.compile(r"\bbeef\s+(?:tallow|stock|broth|bouillon)\b"), + "chicken": re.compile(r"\bchicken\s+(?:stock|broth|bouillon)\b"), + "pork": re.compile(r"\bpork\s+(?:fat|lard|stock|broth)\b"), +} + +MANAGED_TAGS = { + *(tag for expanded in CUISINE_TAGS.values() for tag in expanded), + *(tag for expanded in CUISINE_PHRASE_HINTS.values() for tag in expanded), + *COURSE_HINTS.values(), + *METHOD_HINTS.values(), + *(tag for expanded in DISH_HINTS.values() for tag in expanded), + *PROTEIN_TAGS.values(), + "curry", + "pasta", + "savory", + "sweet", +} + def _flatten_texts(items: list[dict[str, Any]] | None) -> str: flattened: list[str] = [] @@ -80,6 +131,54 @@ def _flatten_texts(items: list[dict[str, Any]] | None) -> str: return " ".join(flattened) +def _tokenize_slug(value: str) -> set[str]: + return {part for part in slugify_tag(value).split("-") if part} + + +def _source_tokens(source: str) -> set[str]: + parsed = urlparse(source or "") + parts = re.split(r"[^a-z0-9]+", f"{parsed.netloc} {parsed.path}".lower()) + return {part for part in parts if part} + + +def _is_boilerplate_existing_tag(tag: str, name: str, source: str) -> bool: + slug = slugify_tag(tag) + if not slug: + return True + + if slug == slugify_tag(name): + return True + + tag_tokens = set(slug.split("-")) + if not tag_tokens: + return True + + combined_noise = _tokenize_slug(name) | _source_tokens(source) | SOURCE_NOISE_WORDS + meaningful_tokens = {token for token in tag_tokens if token not in combined_noise} + + if not meaningful_tokens and len(tag_tokens) >= 3: + return True + + if {"chef", "recipe"}.issubset(tag_tokens): + return True + + return False + + +def _has_phrase(text: str, phrase: str) -> bool: + return re.search(rf"\b{re.escape(phrase)}\b", text) is not None + + +def _has_protein(text: str, protein: str) -> bool: + if not _has_phrase(text, protein): + return False + exclusion = PROTEIN_EXCLUSION_PATTERNS.get(protein) + if exclusion and exclusion.search(text): + stripped = exclusion.sub("", text) + return _has_phrase(stripped, protein) + return True + + def curate_recipe_tags( *, name: str, @@ -88,70 +187,118 @@ def curate_recipe_tags( source: str = "", existing_tags: list[str] | None = None, ) -> list[str]: + item_text = _flatten_texts(items) haystack = " ".join( part for part in [ str(name or ""), str(description or ""), - _flatten_texts(items), - " ".join(existing_tags or []), + item_text, + str(source or ""), + ] + if part + ).lower() + protein_haystack = " ".join( + part for part in [ + str(name or ""), + item_text, str(source or ""), ] if part ).lower() + inferred: list[str] = [] + inferred_seen: set[str] = set() + + def add_inferred(tag: str) -> None: + slug = slugify_tag(tag) + if slug and slug not in inferred_seen: + inferred_seen.add(slug) + inferred.append(slug) + + for key, expanded in CUISINE_TAGS.items(): + if _has_phrase(haystack, key): + for tag in expanded: + add_inferred(tag) + + for key, expanded in CUISINE_PHRASE_HINTS.items(): + if _has_phrase(haystack, key): + for tag in expanded: + add_inferred(tag) + + for key, tag in COURSE_HINTS.items(): + if _has_phrase(haystack, key): + add_inferred(tag) + + for key, tag in METHOD_HINTS.items(): + if _has_phrase(haystack, key): + add_inferred(tag) + + for key, expanded in DISH_HINTS.items(): + if _has_phrase(haystack, key): + for tag in expanded: + add_inferred(tag) + + for key, tag in PROTEIN_TAGS.items(): + if _has_protein(protein_haystack, key): + add_inferred(tag) + + if "mango sticky rice" in haystack: + add_inferred("thai") + add_inferred("dessert") + add_inferred("sweet") + add_inferred("rice") + + if "pasta chips" in haystack: + add_inferred("snack") + add_inferred("pasta") + add_inferred("savory") + + if "kibbeh" in haystack: + add_inferred("middle-eastern") + add_inferred("lebanese") + add_inferred("main-course") + + if "larb" in haystack or "laab" in haystack: + add_inferred("thai") + add_inferred("main-course") + + if "curry" in haystack: + add_inferred("curry") + add_inferred("main-course") + + if "pasta" in haystack: + add_inferred("pasta") + + sweet_score = sum(1 for hint in SWEET_HINTS if hint in haystack) + savory_score = sum(1 for hint in SAVORY_HINTS if hint in haystack) + if sweet_score > savory_score and sweet_score > 0: + add_inferred("sweet") + elif savory_score > 0: + add_inferred("savory") + + if "dessert" not in inferred_seen and "sweet" not in inferred_seen and not any( + tag in inferred_seen for tag in COURSE_HINTS.values() + ): + add_inferred("main-course") + curated: list[str] = [] seen: set[str] = set() - def add(tag: str) -> None: + def add_final(tag: str) -> None: slug = slugify_tag(tag) if slug and slug not in seen: seen.add(slug) curated.append(slug) for tag in existing_tags or []: - add(tag) + if _is_boilerplate_existing_tag(tag, name, source): + continue + slug = slugify_tag(tag) + if slug in MANAGED_TAGS and slug not in inferred_seen: + continue + add_final(tag) - for key, expanded in CUISINE_TAGS.items(): - if key in haystack: - for tag in expanded: - add(tag) - - for key, tag in COURSE_HINTS.items(): - if key in haystack: - add(tag) - - for key, tag in METHOD_HINTS.items(): - if key in haystack: - add(tag) - - for key, tag in PROTEIN_TAGS.items(): - if key in haystack: - add(tag) - - if "kibbeh" in haystack: - add("middle-eastern") - add("lebanese") - add("main-course") - - if "larb" in haystack or "laab" in haystack: - add("thai") - add("main-course") - - if "curry" in haystack: - add("curry") - add("main-course") - - if "pasta" in haystack: - add("pasta") - - sweet_score = sum(1 for hint in SWEET_HINTS if hint in haystack) - savory_score = sum(1 for hint in SAVORY_HINTS if hint in haystack) - if sweet_score > savory_score and sweet_score > 0: - add("sweet") - elif savory_score > 0: - add("savory") - - if "dessert" not in seen and "sweet" not in seen and "main-course" not in seen and "side-dish" not in seen: - add("main-course") + for tag in inferred: + add_final(tag) return curated[:20] diff --git a/home/doris-kitchen/tests/test_tagging.py b/home/doris-kitchen/tests/test_tagging.py new file mode 100644 index 0000000..ee1cee5 --- /dev/null +++ b/home/doris-kitchen/tests/test_tagging.py @@ -0,0 +1,121 @@ +import sys +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from app.services.tagging import curate_recipe_tags +from app.services.recipe_importer import normalize_recipe_from_url + + +class TaggingTests(unittest.TestCase): + def test_tater_tots_drop_boilerplate_and_gain_meaningful_tags(self): + payload = normalize_recipe_from_url( + 'https://www.joshuaweissman.com/recipes/the-best-tater-tots-ever-recipe' + ) + + self.assertNotIn('the-best-tater-tots-ever', payload['tags']) + self.assertNotIn('chef-joshua-weissman-recipe', payload['tags']) + self.assertIn('potato', payload['tags']) + self.assertIn('tater-tots', payload['tags']) + self.assertIn('side-dish', payload['tags']) + self.assertNotIn('beef', payload['tags']) + + def test_halal_chicken_rice_drops_boilerplate_and_infers_cuisine(self): + payload = normalize_recipe_from_url( + 'https://www.joshuaweissman.com/recipes/halal-guys-chicken-over-rice-but-better-recipe' + ) + + self.assertNotIn('halal-guys-chicken-over-rice-but-better', payload['tags']) + self.assertNotIn('chef-joshua-weissman-recipe', payload['tags']) + self.assertIn('middle-eastern', payload['tags']) + self.assertIn('chicken', payload['tags']) + self.assertIn('rice', payload['tags']) + self.assertNotIn('skillet', payload['tags']) + + def test_existing_machine_tags_can_be_replaced_when_recipe_content_disagrees(self): + tags = curate_recipe_tags( + name='The Best Tater Tots Ever', + description='Homemade tater tots worth making. Fry until crisp.', + items=[ + {'name': 'Russet potatoes', 'description': 'peeled', 'optional': False}, + {'name': 'beef tallow', 'description': 'for frying', 'optional': False}, + ], + source='https://www.joshuaweissman.com/recipes/the-best-tater-tots-ever-recipe', + existing_tags=['beef', 'main-course', 'savory'], + ) + + self.assertNotIn('beef', tags) + self.assertNotIn('main-course', tags) + self.assertIn('side-dish', tags) + self.assertIn('savory', tags) + + def test_existing_human_curated_tags_are_preserved(self): + tags = curate_recipe_tags( + name='Simple Roast Chicken', + description='A chicken dinner with herbs.', + items=[{'name': 'chicken', 'description': '', 'optional': False}], + source='https://example.com/roast-chicken', + existing_tags=['family-favorite', 'weeknight'], + ) + + self.assertIn('family-favorite', tags) + self.assertIn('weeknight', tags) + + def test_mango_sticky_rice_is_treated_as_dessert_not_savory_main(self): + tags = curate_recipe_tags( + name='Perfect Mango Sticky Rice At Home', + description='Mango sticky rice is something everyone can, and should, make.', + items=[ + {'name': 'glutinous rice', 'description': '1.5 cups', 'optional': False}, + {'name': 'full-fat coconut milk', 'description': '1 can', 'optional': False}, + {'name': 'mangos', 'description': '2', 'optional': False}, + {'name': 'granulated sugar', 'description': '1/2 cup', 'optional': False}, + ], + source='https://www.joshuaweissman.com/recipes/perfect-mango-sticky-rice-recipe', + existing_tags=['chef-joshua-weissman-recipe', 'main-course', 'savory'], + ) + + self.assertIn('dessert', tags) + self.assertIn('sweet', tags) + self.assertIn('thai', tags) + self.assertNotIn('main-course', tags) + self.assertNotIn('savory', tags) + + def test_description_only_chicken_reference_does_not_force_chicken_tag(self): + tags = curate_recipe_tags( + name='Quick Asian Beef Ramen Noodles', + description='Try the chicken version next time if you want a variation.', + items=[ + {'name': 'ground beef', 'description': '200g', 'optional': False}, + {'name': 'ramen noodles', 'description': '2 packs', 'optional': False}, + {'name': 'soy sauce', 'description': '1 tbsp', 'optional': False}, + ], + source='https://www.recipetineats.com/asian-beef-and-noodles/', + existing_tags=['asian', 'noodles'], + ) + + self.assertIn('beef', tags) + self.assertNotIn('chicken', tags) + + def test_pasta_chips_are_snack_not_main_course(self): + tags = curate_recipe_tags( + name='Air Fryer Pasta Chips', + description='This could be one of the greatest homemade snacks of all time.', + items=[ + {'name': 'farfalle pasta', 'description': '8 oz', 'optional': False}, + {'name': 'pecorino romano', 'description': 'to taste', 'optional': False}, + {'name': 'olive oil', 'description': 'drizzle', 'optional': False}, + ], + source='https://www.joshuaweissman.com/recipes/air-fryer-pasta-chips-recipe', + existing_tags=['italian', 'main-course', 'pasta', 'savory'], + ) + + self.assertIn('snack', tags) + self.assertIn('pasta', tags) + self.assertNotIn('main-course', tags) + + +if __name__ == '__main__': + unittest.main()