268 lines
7.9 KiB
Python
268 lines
7.9 KiB
Python
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)
|