Kochbuch/recipes/schema_org.py

"""Extract schema.org/Recipe (JSON-LD) and page metadata from HTML without extra dependencies."""

import html
import json
import re
from html.parser import HTMLParser

from . import parsing

_DURATION_RE = re.compile(r"^P(?:(\d+)D)?(?:T(?:(\d+)H)?(?:(\d+)M)?(?:[\d.]+S)?)?$", re.IGNORECASE)
_TAG_RE = re.compile(r"<[^>]+>")
MAX_TEXT_CHARS = 15_000


class PageParser(HTMLParser):
    """Collects JSON-LD blocks, <meta> tags, the <title> and the visible text of a page."""

    SKIP_TEXT = {"script", "style", "noscript", "svg", "nav", "footer", "header", "form", "button", "iframe"}
    BLOCK = {"p", "div", "li", "br", "h1", "h2", "h3", "h4", "h5", "h6", "tr", "section", "article", "ol", "ul"}

    def __init__(self):
        super().__init__(convert_charrefs=True)
        self.json_ld, self.meta, self.title = [], {}, ""
        self._stack, self._buffer, self._text = [], None, []
        self._in_title = False

    def handle_starttag(self, tag, attrs):
        attrs = dict(attrs)
        if tag == "script" and (attrs.get("type") or "").lower().startswith("application/ld+json"):
            self._buffer = []
        if tag == "meta":
            key = (attrs.get("property") or attrs.get("name") or "").lower()
            if key and attrs.get("content") and key not in self.meta:
                self.meta[key] = attrs["content"]
        if tag == "title":
            self._in_title = True
        if tag in self.BLOCK:
            self._text.append("\n")
        if tag not in ("br", "meta", "img", "link", "input", "hr"):
            self._stack.append(tag)

    def handle_endtag(self, tag):
        if tag == "script" and self._buffer is not None:
            self.json_ld.append("".join(self._buffer))
            self._buffer = None
        if tag == "title":
            self._in_title = False
        if tag in self._stack:
            while self._stack and self._stack.pop() != tag:
                pass

    def handle_data(self, data):
        if self._buffer is not None:
            self._buffer.append(data)
            return
        if self._in_title:
            self.title += data
        if not any(t in self.SKIP_TEXT for t in self._stack):
            self._text.append(data)

    def visible_text(self):
        text = "".join(self._text)
        text = re.sub(r"[ \t\r\f\v]+", " ", text)
        text = re.sub(r"\n\s*\n+", "\n", text)
        return text.strip()[:MAX_TEXT_CHARS]


def parse_page(page_html):
    parser = PageParser()
    try:
        parser.feed(page_html)
        parser.close()
    except Exception:  # malformed HTML must never break the import
        pass
    return parser


def _types(node):
    value = node.get("@type", [])
    return {t.lower() for t in (value if isinstance(value, list) else [value]) if isinstance(t, str)}


def find_recipe(data):
    """Depth-first search for the first object whose @type includes Recipe."""
    if isinstance(data, list):
        for item in data:
            found = find_recipe(item)
            if found:
                return found
    elif isinstance(data, dict):
        if "recipe" in _types(data):
            return data
        for key in ("@graph", "mainEntity", "mainEntityOfPage", "itemListElement", "item"):
            if key in data:
                found = find_recipe(data[key])
                if found:
                    return found
    return None


def _clean(text):
    if not isinstance(text, str):
        return ""
    text = html.unescape(_TAG_RE.sub(" ", text))
    text = re.sub(r"\s+", " ", text).strip()
    return re.sub(r"\s+([.,;:!?])", r"\1", text)


def _duration_minutes(value):
    if not isinstance(value, str):
        return None
    match = _DURATION_RE.match(value.strip())
    if not match or not any(match.groups()):
        return None
    days, hours, minutes = (int(g) if g else 0 for g in match.groups())
    total = days * 1440 + hours * 60 + minutes
    return total if 0 < total <= 7 * 1440 else None


def _yield(value):
    if isinstance(value, list):
        value = next((v for v in value if isinstance(v, (str, int))), None)
    if isinstance(value, (int, float)):
        return int(value), ""
    if isinstance(value, str):
        match = re.search(r"(\d+)", value)
        if match:
            label = _clean(value[match.end():]).strip(" ()")
            return int(match.group(1)), label[:30]
    return None, ""


def _steps(value):
    steps = []
    if isinstance(value, str):
        return parsing.parse_steps(_clean_multiline(value))
    if isinstance(value, dict):
        value = [value]
    for item in value or []:
        if isinstance(item, str):
            steps.append(_clean(item))
        elif isinstance(item, dict):
            types = _types(item)
            if "howtosection" in types:
                steps.extend(_steps(item.get("itemListElement")))
            else:
                steps.append(_clean(item.get("text") or item.get("name") or ""))
    return [s for s in steps if s]


def _clean_multiline(text):
    text = re.sub(r"<br\s*/?>|</p>|</li>", "\n", text, flags=re.IGNORECASE)
    return html.unescape(_TAG_RE.sub("", text))


def _keywords(recipe):
    words = []
    for key in ("recipeCategory", "recipeCuisine", "keywords"):
        value = recipe.get(key)
        if isinstance(value, str):
            words.extend(value.split(","))
        elif isinstance(value, list):
            words.extend(v for v in value if isinstance(v, str))
    return parsing.parse_tags(",".join(_clean(w) for w in words if len(w.strip()) <= 30))[:6]


def _image(value):
    if isinstance(value, str):
        return value
    if isinstance(value, list) and value:
        return _image(value[0])
    if isinstance(value, dict):
        return value.get("url") or value.get("contentUrl") or ""
    return ""


def recipe_to_initial(recipe):
    """Map a schema.org Recipe dict to RecipeForm initial values (+ image URL)."""
    servings, label = _yield(recipe.get("recipeYield"))
    ingredients = recipe.get("recipeIngredient") or recipe.get("ingredients") or []
    if isinstance(ingredients, str):
        ingredients = [ingredients]
    prep = _duration_minutes(recipe.get("prepTime"))
    cook = _duration_minutes(recipe.get("cookTime"))
    if prep is None and cook is None:
        prep = _duration_minutes(recipe.get("totalTime"))
    initial = {
        "title": _clean(recipe.get("name"))[:200],
        "description": _clean(recipe.get("description")),
        "prep_minutes": prep,
        "cook_minutes": cook,
        "ingredients_text": "\n".join(line for line in (_clean(i) for i in ingredients if isinstance(i, str)) if line),
        "steps_text": parsing.steps_to_text(_steps(recipe.get("recipeInstructions"))),
        "tags_text": ", ".join(_keywords(recipe)),
    }
    if servings and 1 <= servings <= 100:
        initial["servings"] = servings
        initial["servings_label"] = label if label and len(label) < 30 else "Portionen"
    return initial, _image(recipe.get("image"))


def extract(page_html):
    """Return (initial, image_url, page) – initial is None when the page has no usable Recipe."""
    page = parse_page(page_html)
    for block in page.json_ld:
        try:
            data = json.loads(block.strip())
        except json.JSONDecodeError:
            # Some sites put several objects or trailing commas in one block; skip those.
            continue
        recipe = find_recipe(data)
        if recipe:
            initial, image = recipe_to_initial(recipe)
            if initial["ingredients_text"] or initial["steps_text"]:
                return initial, image or page.meta.get("og:image", ""), page
    return None, page.meta.get("og:image", ""), page