from __future__ import annotations import ipaddress import json import os import socket from dataclasses import dataclass from typing import Any from urllib.parse import urljoin, urlparse import requests from bs4 import BeautifulSoup MAX_REDIRECTS = 5 DEFAULT_MAX_BYTES = 3_000_000 DEFAULT_TIMEOUT = 20.0 class UnsafeSourceUrl(ValueError): pass @dataclass class ExtractedSource: final_url: str title: str image_url: str | None structured_recipe: dict[str, Any] | None visible_text: str content_type: str def to_dict(self) -> dict[str, Any]: return { "final_url": self.final_url, "title": self.title, "image_url": self.image_url, "structured_recipe": self.structured_recipe, "visible_text": self.visible_text, "content_type": self.content_type, } def _allow_private_sources() -> bool: return os.environ.get("ALLOW_PRIVATE_SOURCE_URLS", "").casefold() in { "1", "true", "yes", } def _verify_source_tls() -> bool: return os.environ.get("SOURCE_VERIFY_TLS", "true").casefold() not in { "0", "false", "no", } def _validate_public_host(hostname: str) -> None: if _allow_private_sources(): return try: infos = socket.getaddrinfo(hostname, None) except socket.gaierror as exc: raise UnsafeSourceUrl(f"Host kann nicht aufgelöst werden: {hostname}") from exc for info in infos: address = info[4][0] ip = ipaddress.ip_address(address) if ( ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved or ip.is_unspecified ): raise UnsafeSourceUrl( f"Private oder lokale Zieladresse ist nicht erlaubt: {address}" ) def validate_source_url(url: str) -> str: candidate = url.strip() if "://" not in candidate: candidate = f"https://{candidate}" parsed = urlparse(candidate) if parsed.scheme not in {"http", "https"}: raise UnsafeSourceUrl("Nur http- und https-URLs sind erlaubt.") if not parsed.hostname: raise UnsafeSourceUrl("Die URL enthält keinen gültigen Host.") if parsed.username or parsed.password: raise UnsafeSourceUrl("Zugangsdaten in der URL sind nicht erlaubt.") _validate_public_host(parsed.hostname) return parsed.geturl() def safe_fetch(url: str) -> tuple[str, bytes, str]: current = validate_source_url(url) max_bytes = int(os.environ.get("FETCH_MAX_BYTES", DEFAULT_MAX_BYTES)) timeout = float(os.environ.get("SOURCE_TIMEOUT", DEFAULT_TIMEOUT)) session = requests.Session() headers = { "User-Agent": ( "Mozilla/5.0 (compatible; TandoorAIRecipeImporter/1.0; " "+local-recipe-import)" ), "Accept": "text/html,application/xhtml+xml,application/json;q=0.9,*/*;q=0.5", } for _ in range(MAX_REDIRECTS + 1): response = session.get( current, headers=headers, timeout=timeout, verify=_verify_source_tls(), allow_redirects=False, stream=True, ) try: if 300 <= response.status_code < 400: location = response.headers.get("Location") if not location: raise RuntimeError("Redirect ohne Location-Header.") current = validate_source_url(urljoin(current, location)) continue response.raise_for_status() content_type = response.headers.get("Content-Type", "").split(";", 1)[0] if content_type not in { "text/html", "application/xhtml+xml", "application/json", "text/plain", "", }: raise RuntimeError( f"Nicht unterstützter Quelltyp: {content_type or 'unbekannt'}" ) chunks: list[bytes] = [] size = 0 for chunk in response.iter_content(64 * 1024): if not chunk: continue size += len(chunk) if size > max_bytes: raise RuntimeError( f"Die Quellseite überschreitet {max_bytes} Bytes." ) chunks.append(chunk) return current, b"".join(chunks), content_type finally: response.close() raise RuntimeError(f"Mehr als {MAX_REDIRECTS} Redirects.") def _iter_json_objects(value: Any): if isinstance(value, dict): yield value graph = value.get("@graph") if isinstance(graph, list): for item in graph: yield from _iter_json_objects(item) elif isinstance(value, list): for item in value: yield from _iter_json_objects(item) def _is_recipe_type(value: Any) -> bool: if isinstance(value, str): return value.casefold() == "recipe" if isinstance(value, list): return any(_is_recipe_type(item) for item in value) return False def _extract_json_ld_recipe(soup: BeautifulSoup) -> dict[str, Any] | None: for script in soup.find_all("script", attrs={"type": "application/ld+json"}): raw = script.string or script.get_text() if not raw.strip(): continue try: payload = json.loads(raw) except json.JSONDecodeError: continue for candidate in _iter_json_objects(payload): if _is_recipe_type(candidate.get("@type")): return candidate return None def _meta_content(soup: BeautifulSoup, *selectors: tuple[str, str]) -> str | None: for attribute, value in selectors: element = soup.find("meta", attrs={attribute: value}) if element and element.get("content"): return str(element["content"]).strip() return None def extract_from_html(final_url: str, html: str, content_type: str = "text/html") -> ExtractedSource: soup = BeautifulSoup(html, "html.parser") structured = _extract_json_ld_recipe(soup) title = "" if structured and structured.get("name"): title = str(structured["name"]).strip() if not title: title = ( _meta_content( soup, ("property", "og:title"), ("name", "twitter:title"), ) or (soup.title.get_text(" ", strip=True) if soup.title else "") ) image_url: str | None = None if structured: image = structured.get("image") if isinstance(image, str): image_url = image elif isinstance(image, list) and image: first = image[0] image_url = first if isinstance(first, str) else first.get("url") elif isinstance(image, dict): image_url = image.get("url") or image.get("contentUrl") image_url = image_url or _meta_content( soup, ("property", "og:image"), ("name", "twitter:image"), ) if image_url: image_url = urljoin(final_url, image_url) for tag in soup( ["script", "style", "noscript", "svg", "nav", "footer", "header", "aside"] ): tag.decompose() visible_text = "\n".join( line.strip() for line in soup.get_text("\n").splitlines() if line.strip() ) visible_text = visible_text[:60_000] return ExtractedSource( final_url=final_url, title=title[:500], image_url=image_url, structured_recipe=structured, visible_text=visible_text, content_type=content_type, ) def extract_source(url: str) -> ExtractedSource: final_url, body, content_type = safe_fetch(url) charset = "utf-8" html = body.decode(charset, errors="replace") return extract_from_html(final_url, html, content_type)