759 lines
30 KiB
Python
759 lines
30 KiB
Python
import json
|
|
from typing import Optional, Iterable, List, Tuple, Dict
|
|
|
|
import httpx
|
|
from bs4 import BeautifulSoup
|
|
import html as _html
|
|
|
|
# POLICY: Strict ld+json-only recipe extraction
|
|
# ---------------------------------------------
|
|
# This scraper MUST NOT perform complex HTML-based ingredient parsing.
|
|
# Only two things are allowed when parsing HTML:
|
|
# 1) Discover additional, more-friendly variants of the same page (e.g., amp/print)
|
|
# 2) Locate and parse <script type="application/ld+json"> blocks that contain a Recipe
|
|
# Do NOT extract from application/json, __NEXT_DATA__, plugin markup, microdata, or generic
|
|
# DOM heuristics. This keeps behavior predictable and aligned with sites' structured data.
|
|
|
|
# Determine brotli support to avoid unreadable responses when the runtime lacks a decoder.
|
|
_HAS_BROTLI = False
|
|
try: # brotli or brotlicffi
|
|
import brotli as _brotli # type: ignore
|
|
_HAS_BROTLI = True
|
|
except Exception:
|
|
try:
|
|
import brotlicffi as _brotlicffi # type: ignore
|
|
_HAS_BROTLI = True
|
|
except Exception:
|
|
_HAS_BROTLI = False
|
|
|
|
# Base headers shared across profiles; specific profiles add UA and optional fetch headers.
|
|
DEFAULT_HEADERS = {
|
|
"Accept": "text/html,application/xhtml+xml;q=0.9,*/*;q=0.8",
|
|
"Accept-Language": "en-US,en;q=0.8",
|
|
# Only advertise br when we can decode it, otherwise prefer gzip/deflate for reliability
|
|
"Accept-Encoding": "gzip, deflate, br" if _HAS_BROTLI else "gzip, deflate",
|
|
"Connection": "keep-alive",
|
|
}
|
|
|
|
# Header profiles: start with an automation-forward identity, then fall back to browser-like
|
|
HEADER_PROFILES = [
|
|
{
|
|
"name": "automation",
|
|
"headers": {
|
|
**DEFAULT_HEADERS,
|
|
"User-Agent": "MunchEaseRecipeBot/1.0 (+https://example.com/bot)",
|
|
"From": "bot@example.com",
|
|
},
|
|
"add_referer": False,
|
|
},
|
|
{
|
|
"name": "chrome-desktop",
|
|
"headers": {
|
|
**DEFAULT_HEADERS,
|
|
"Upgrade-Insecure-Requests": "1",
|
|
"Sec-Fetch-Dest": "document",
|
|
"Sec-Fetch-Mode": "navigate",
|
|
"Sec-Fetch-Site": "none",
|
|
"Sec-Fetch-User": "?1",
|
|
# Client hints
|
|
"sec-ch-ua": '"Chromium";v="127", "Not=A?Brand";v="24", "Google Chrome";v="127"',
|
|
"sec-ch-ua-mobile": "?0",
|
|
"sec-ch-ua-platform": '"Linux"',
|
|
"User-Agent": (
|
|
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) "
|
|
"Chrome/127.0.0.0 Safari/537.36"
|
|
),
|
|
},
|
|
"add_referer": True,
|
|
},
|
|
{
|
|
"name": "chrome-mobile",
|
|
"headers": {
|
|
**DEFAULT_HEADERS,
|
|
"Upgrade-Insecure-Requests": "1",
|
|
"Sec-Fetch-Dest": "document",
|
|
"Sec-Fetch-Mode": "navigate",
|
|
"Sec-Fetch-Site": "none",
|
|
"Sec-Fetch-User": "?1",
|
|
"sec-ch-ua": '"Chromium";v="127", "Not=A?Brand";v="24", "Google Chrome";v="127"',
|
|
"sec-ch-ua-mobile": "?1",
|
|
"sec-ch-ua-platform": '"Android"',
|
|
"User-Agent": (
|
|
"Mozilla/5.0 (Linux; Android 12; Pixel 5) AppleWebKit/537.36 (KHTML, like Gecko) "
|
|
"Chrome/127.0.0.0 Mobile Safari/537.36"
|
|
),
|
|
},
|
|
"add_referer": True,
|
|
},
|
|
{
|
|
"name": "chrome-full",
|
|
"headers": {
|
|
**DEFAULT_HEADERS,
|
|
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
|
|
"Upgrade-Insecure-Requests": "1",
|
|
"Sec-Fetch-Dest": "document",
|
|
"Sec-Fetch-Mode": "navigate",
|
|
"Sec-Fetch-Site": "none",
|
|
"Sec-Fetch-User": "?1",
|
|
"sec-ch-ua": '"Chromium";v="127", "Google Chrome";v="127", ";Not A Brand";v="99"',
|
|
"sec-ch-ua-mobile": "?0",
|
|
"sec-ch-ua-platform": '"Linux"',
|
|
"sec-ch-ua-platform-version": '"6.8.0"',
|
|
"DNT": "1",
|
|
"Priority": "u=0, i",
|
|
"User-Agent": (
|
|
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) "
|
|
"Chrome/127.0.0.0 Safari/537.36"
|
|
),
|
|
},
|
|
"add_referer": True,
|
|
},
|
|
{
|
|
"name": "safari-mac",
|
|
"headers": {
|
|
**DEFAULT_HEADERS,
|
|
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
|
"Upgrade-Insecure-Requests": "1",
|
|
"User-Agent": (
|
|
"Mozilla/5.0 (Macintosh; Intel Mac OS X 14_0) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Safari/605.1.15"
|
|
),
|
|
},
|
|
"add_referer": True,
|
|
},
|
|
{
|
|
"name": "firefox-desktop",
|
|
"headers": {
|
|
**DEFAULT_HEADERS,
|
|
"User-Agent": (
|
|
"Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:128.0) Gecko/20100101 Firefox/128.0"
|
|
),
|
|
},
|
|
"add_referer": True,
|
|
},
|
|
{
|
|
"name": "safari-ios",
|
|
"headers": {
|
|
**DEFAULT_HEADERS,
|
|
"User-Agent": (
|
|
"Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) "
|
|
"AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1"
|
|
),
|
|
},
|
|
"add_referer": True,
|
|
},
|
|
{
|
|
"name": "edge-desktop",
|
|
"headers": {
|
|
**DEFAULT_HEADERS,
|
|
"Upgrade-Insecure-Requests": "1",
|
|
"Sec-Fetch-Dest": "document",
|
|
"Sec-Fetch-Mode": "navigate",
|
|
"Sec-Fetch-Site": "none",
|
|
"Sec-Fetch-User": "?1",
|
|
"sec-ch-ua": '"Chromium";v="127", "Not=A?Brand";v="24", "Microsoft Edge";v="127"',
|
|
"sec-ch-ua-mobile": "?0",
|
|
"sec-ch-ua-platform": '"Linux"',
|
|
"User-Agent": (
|
|
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) "
|
|
"Chrome/127.0.0.0 Safari/537.36 Edg/127.0.0.0"
|
|
),
|
|
},
|
|
"add_referer": True,
|
|
},
|
|
]
|
|
|
|
|
|
def looks_like_jsonld(html: str) -> bool:
|
|
"""Lightweight check for presence of JSON-LD Recipe data in an HTML string.
|
|
|
|
We only check for ld+json tags or @context+schema.org signatures.
|
|
This is an optimization hint and does not parse JSON.
|
|
"""
|
|
if not html:
|
|
return False
|
|
low = html.lower()
|
|
if "application/ld+json" in low:
|
|
return True
|
|
if "@context" in low and "schema.org" in low:
|
|
return True
|
|
return False
|
|
|
|
|
|
def _force_safe_accept_encoding(headers: Dict[str, str]) -> Dict[str, str]:
|
|
"""Return a copy of headers where Accept-Encoding excludes brotli.
|
|
|
|
Use this in contexts where brotli support may be missing to avoid unreadable payloads.
|
|
"""
|
|
h = dict(headers)
|
|
h["Accept-Encoding"] = "gzip, deflate"
|
|
return h
|
|
|
|
|
|
async def fetch_first_2xx_html(
|
|
client: httpx.AsyncClient,
|
|
url: str,
|
|
log=None,
|
|
safe_accept_encoding: bool = True,
|
|
) -> Tuple[Optional[str], Dict[str, Optional[str]]]:
|
|
"""Try header profiles against a URL and return the first 2xx HTML and metadata.
|
|
|
|
Returns (html, meta) where meta includes:
|
|
- status, final_url, content_type, content_encoding
|
|
- profile (name), request_headers (effective request headers)
|
|
If no profile returns 2xx, returns (None, meta_with_last_error_or_status).
|
|
"""
|
|
def _log(msg: str) -> None:
|
|
if log:
|
|
try:
|
|
log(msg)
|
|
except Exception:
|
|
pass
|
|
|
|
from urllib.parse import urlparse
|
|
parsed = urlparse(url)
|
|
origin = f"{parsed.scheme}://{parsed.netloc}/"
|
|
last_meta: Dict[str, Optional[str]] = {
|
|
"status": None,
|
|
"final_url": None,
|
|
"content_type": None,
|
|
"content_encoding": None,
|
|
"profile": None,
|
|
"error": None,
|
|
"request_headers": None,
|
|
}
|
|
last_html: Optional[str] = None
|
|
for prof in HEADER_PROFILES:
|
|
headers = dict(prof["headers"]) # copy
|
|
if prof.get("add_referer"):
|
|
headers.setdefault("Referer", origin)
|
|
if safe_accept_encoding:
|
|
headers = _force_safe_accept_encoding(headers)
|
|
try:
|
|
resp = await client.get(url, headers=headers, follow_redirects=True)
|
|
except Exception as e:
|
|
last_meta.update({"error": f"{type(e).__name__}: {e}", "profile": prof["name"]})
|
|
_log(f"GET[{prof['name']}] {url} -> EXC {type(e).__name__}: {e}")
|
|
continue
|
|
_log(f"GET[{prof['name']}] {url} -> {resp.status_code}")
|
|
meta = {
|
|
"status": str(resp.status_code),
|
|
"final_url": str(resp.url),
|
|
"content_type": resp.headers.get("content-type"),
|
|
"content_encoding": resp.headers.get("content-encoding"),
|
|
"profile": prof["name"],
|
|
"error": None,
|
|
"request_headers": json.dumps(headers),
|
|
}
|
|
last_meta = meta
|
|
if 200 <= resp.status_code < 300:
|
|
html = resp.text
|
|
last_html = html
|
|
# Prefer early return if JSON-LD signature seems present
|
|
if looks_like_jsonld(html):
|
|
return html, meta
|
|
# else keep searching other profiles for a more suitable variant
|
|
continue
|
|
return last_html, last_meta
|
|
|
|
|
|
def _is_recipe_ldata(ldata_node) -> bool:
|
|
"""Return True when a JSON-LD node represents a Recipe.
|
|
|
|
Handles cases where @type is a string or a list (order-insensitive).
|
|
"""
|
|
if "@type" not in ldata_node:
|
|
return False
|
|
typ = ldata_node["@type"]
|
|
if isinstance(typ, str):
|
|
return typ.lower() == "recipe"
|
|
if isinstance(typ, list):
|
|
for t in typ:
|
|
if isinstance(t, str) and t.lower() == "recipe":
|
|
return True
|
|
return False
|
|
|
|
|
|
def _fallback_urls(url: str) -> Iterable[str]:
|
|
"""Generate fallback URLs to try if the primary request is blocked.
|
|
|
|
Strategy:
|
|
- original URL
|
|
- add `?output=amp` if no existing query
|
|
- add `&output=amp` if query exists
|
|
- try `/amp` path suffix if not already present
|
|
- try `?amp=1` and bare `?amp` which some sites honor as AMP toggles
|
|
"""
|
|
yield url
|
|
try:
|
|
from urllib.parse import urlparse, urlunparse, parse_qsl, urlencode
|
|
|
|
parsed = urlparse(url)
|
|
q = dict(parse_qsl(parsed.query, keep_blank_values=True))
|
|
if q.get("output") != "amp":
|
|
q["output"] = "amp"
|
|
amp_url = urlunparse(parsed._replace(query=urlencode(q, doseq=True)))
|
|
if amp_url != url:
|
|
yield amp_url
|
|
|
|
# Try a path-based AMP fallback
|
|
if not parsed.path.endswith("/amp"):
|
|
amp_path = parsed.path.rstrip("/") + "/amp"
|
|
amp2 = urlunparse(parsed._replace(path=amp_path))
|
|
if amp2 != url:
|
|
yield amp2
|
|
# Additional common AMP toggles
|
|
q_amp = dict(parse_qsl(parsed.query, keep_blank_values=True))
|
|
if q_amp.get("amp") != "1":
|
|
q_amp["amp"] = "1"
|
|
amp1 = urlunparse(parsed._replace(query=urlencode(q_amp, doseq=True)))
|
|
if amp1 != url:
|
|
yield amp1
|
|
if "amp" not in (q_amp or {}):
|
|
bare = urlunparse(parsed._replace(query=(parsed.query + ("&" if parsed.query else "") + "amp")))
|
|
if bare != url:
|
|
yield bare
|
|
|
|
# HTTP scheme fallbacks when original is HTTPS (helps sites with misconfigured TLS)
|
|
if parsed.scheme == "https":
|
|
http_base = urlunparse(parsed._replace(scheme="http"))
|
|
if http_base != url:
|
|
yield http_base
|
|
# http + output=amp
|
|
q2 = dict(parse_qsl(parsed.query, keep_blank_values=True))
|
|
if q2.get("output") != "amp":
|
|
q2["output"] = "amp"
|
|
http_amp = urlunparse(parsed._replace(scheme="http", query=urlencode(q2, doseq=True)))
|
|
if http_amp != url:
|
|
yield http_amp
|
|
# http + amp=1
|
|
q3 = dict(parse_qsl(parsed.query, keep_blank_values=True))
|
|
if q3.get("amp") != "1":
|
|
q3["amp"] = "1"
|
|
http_amp1 = urlunparse(parsed._replace(scheme="http", query=urlencode(q3, doseq=True)))
|
|
if http_amp1 != url:
|
|
yield http_amp1
|
|
# http path amp
|
|
if not parsed.path.endswith("/amp"):
|
|
http_amp_path = urlunparse(parsed._replace(scheme="http", path=parsed.path.rstrip("/") + "/amp"))
|
|
if http_amp_path != url:
|
|
yield http_amp_path
|
|
except Exception:
|
|
# Be conservative if URL parsing fails
|
|
pass
|
|
|
|
|
|
BLOCK_STATUSES = {403, 406, 429, 460}
|
|
|
|
|
|
async def scrape_recipe_ldata(url: str, log=None, dump_dir: Optional[str] = None) -> Optional[dict]:
|
|
"""Return best-effort recipe JSON-LD (or heuristic dict) for the URL.
|
|
|
|
If 'log' is provided (callable taking a string), diagnostic messages are emitted
|
|
during scraping. No environment toggles are used; behavior matches production.
|
|
"""
|
|
|
|
def _log(msg: str) -> None:
|
|
if log:
|
|
try:
|
|
log(msg)
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
async with httpx.AsyncClient(http2=True, timeout=20.0) as client:
|
|
visited: set[str] = set()
|
|
|
|
# Queue of URLs to try, seeded with simple fallbacks; we'll append amp/print variants we discover.
|
|
to_try: List[str] = list(_fallback_urls(url))
|
|
while to_try:
|
|
candidate = to_try.pop(0)
|
|
if candidate in visited:
|
|
continue
|
|
visited.add(candidate)
|
|
# Try each header profile and attempt extraction per profile
|
|
any_2xx = False
|
|
from urllib.parse import urlparse, urlunparse
|
|
parsed = urlparse(candidate)
|
|
origin = f"{parsed.scheme}://{parsed.netloc}/"
|
|
for prof in HEADER_PROFILES:
|
|
prof_name = prof["name"]
|
|
headers = dict(prof["headers"]) # copy
|
|
if prof.get("add_referer"):
|
|
headers.setdefault("Referer", origin)
|
|
try:
|
|
resp = await client.get(candidate, headers=headers, follow_redirects=True)
|
|
except Exception as e:
|
|
_log(f"GET[{prof_name}] {candidate} -> EXC {type(e).__name__}: {e}")
|
|
continue
|
|
_log(f"GET[{prof_name}] {candidate} -> {resp.status_code}")
|
|
if resp.status_code in BLOCK_STATUSES or resp.status_code >= 300:
|
|
continue
|
|
any_2xx = True
|
|
html = resp.text
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
# Optional debug dump
|
|
if dump_dir:
|
|
try:
|
|
_dump_response(dump_dir, candidate, prof_name, resp)
|
|
except Exception as e:
|
|
_log(f"dump error: {type(e).__name__}: {e}")
|
|
# Detect common bot protection pages early
|
|
if (soup.title and "just a moment" in _extract_text(soup.title).lower()) or "challenge-platform" in html:
|
|
_log("Detected Cloudflare-like challenge; trying next profile")
|
|
continue
|
|
# FRIENDLY DISCOVERY: follow declared amphtml and obvious print links
|
|
try:
|
|
# rel="amphtml"
|
|
amp_link = None
|
|
for link in soup.find_all("link"):
|
|
rel = link.get("rel")
|
|
href = link.get("href")
|
|
if not href:
|
|
continue
|
|
if isinstance(rel, list) and any(r.lower() == "amphtml" for r in rel):
|
|
amp_link = href
|
|
break
|
|
if isinstance(rel, str) and rel.lower() == "amphtml":
|
|
amp_link = href
|
|
break
|
|
if amp_link:
|
|
from urllib.parse import urljoin
|
|
amp_abs = urljoin(candidate, amp_link)
|
|
if amp_abs not in visited and amp_abs not in to_try:
|
|
to_try.append(amp_abs)
|
|
_log(f"discovered amphtml link: {amp_abs}")
|
|
|
|
# print links (plugin or generic)
|
|
from urllib.parse import urljoin, parse_qsl, urlencode
|
|
for a in soup.find_all("a", href=True):
|
|
href = a["href"]
|
|
low = href.lower()
|
|
if any(k in low for k in ["print", "wprm-print", "tasty-recipes-print", "/print/"]):
|
|
absu = urljoin(candidate, href)
|
|
if absu not in visited and absu not in to_try:
|
|
to_try.append(absu)
|
|
# linked JSON-LD files
|
|
for link in soup.find_all("link"):
|
|
rel = link.get("rel")
|
|
typ = (link.get("type") or link.get("as") or "").lower()
|
|
href = link.get("href")
|
|
if not href:
|
|
continue
|
|
if (isinstance(rel, list) and any(r.lower() == "alternate" for r in rel)) or (
|
|
isinstance(rel, str) and rel.lower() == "alternate"
|
|
):
|
|
if "ld+json" in typ:
|
|
from urllib.parse import urljoin
|
|
absu = urljoin(candidate, href)
|
|
if absu not in visited and absu not in to_try:
|
|
to_try.append(absu)
|
|
_log(f"discovered linked ld+json: {absu}")
|
|
# query param prints: add print=1 once if missing
|
|
parsed_url = parsed
|
|
q_items = list(parse_qsl(parsed_url.query, keep_blank_values=True))
|
|
has_print = any(k == "print" for k, _ in q_items)
|
|
if not has_print:
|
|
q_items.append(("print", "1"))
|
|
print_url = urlunparse(parsed_url._replace(query=urlencode(q_items, doseq=True)))
|
|
if print_url not in visited and print_url not in to_try:
|
|
to_try.append(print_url)
|
|
_log(f"queued print variant: {print_url}")
|
|
except Exception as e:
|
|
_log(f"discovery error: {type(e).__name__}: {e}")
|
|
|
|
found = _extract_ldata_from_soup(soup, candidate, _log)
|
|
if found:
|
|
return found
|
|
|
|
if not any_2xx:
|
|
_log("All header profiles blocked or non-2xx; trying next fallback")
|
|
continue
|
|
|
|
_log("No recipe data found after all fallbacks")
|
|
return None
|
|
except Exception as e:
|
|
_log(f"EXC during scraping: {type(e).__name__}: {e}")
|
|
return None
|
|
|
|
|
|
def _extract_ldata_from_soup(soup: BeautifulSoup, candidate: str, log=None) -> Optional[dict]:
|
|
def _log(msg: str) -> None:
|
|
if log:
|
|
try:
|
|
log(msg)
|
|
except Exception:
|
|
pass
|
|
|
|
import re as _re
|
|
|
|
def _strip_to_json_candidate(text: str) -> str:
|
|
# Remove HTML/JS comment wrappers and CDATA markers
|
|
t = text.strip()
|
|
t = _re.sub(r"<!--|-->", "", t)
|
|
t = _re.sub(r"/\*.*?\*/", "", t, flags=_re.S)
|
|
t = _re.sub(r"(^|\s)//.*$", "", t, flags=_re.M)
|
|
# Trim to outermost JSON-like braces/brackets
|
|
first_brace = t.find("{")
|
|
first_brack = t.find("[")
|
|
starts = [i for i in [first_brace, first_brack] if i != -1]
|
|
if not starts:
|
|
return t
|
|
start = min(starts)
|
|
last_brace = t.rfind("}")
|
|
last_brack = t.rfind("]")
|
|
ends = [i for i in [last_brace, last_brack] if i != -1]
|
|
if not ends:
|
|
return t
|
|
end = max(ends)
|
|
return t[start : end + 1]
|
|
|
|
def _remove_trailing_commas(s: str) -> str:
|
|
prev = None
|
|
curr = s
|
|
# Iteratively remove trailing commas before } or ]
|
|
for _ in range(3): # limit passes
|
|
prev = curr
|
|
curr = _re.sub(r",\s*([}\]])", r"\1", curr)
|
|
if curr == prev:
|
|
break
|
|
return curr
|
|
|
|
def _parse_json_lenient(text: str) -> Optional[object]:
|
|
# Try strict first
|
|
try:
|
|
return json.loads(text)
|
|
except Exception:
|
|
pass
|
|
# Clean wrappers and comments
|
|
t = _strip_to_json_candidate(text)
|
|
# Sometimes multiple roots are concatenated; try to form a list conservatively
|
|
if t.count("{") > 1 and "}\n{" in t:
|
|
parts = [p for p in t.split("\n") if p.strip()]
|
|
maybe = "[" + ",".join(parts) + "]"
|
|
try:
|
|
return json.loads(maybe)
|
|
except Exception:
|
|
pass
|
|
# Remove trailing commas
|
|
t2 = _remove_trailing_commas(t)
|
|
try:
|
|
return json.loads(t2)
|
|
except Exception:
|
|
pass
|
|
# As a last resort, if no double quotes exist but single quotes do, try naive conversion
|
|
if '"' not in t2 and "'" in t2:
|
|
t3 = t2.replace("'", '"')
|
|
try:
|
|
return json.loads(t3)
|
|
except Exception:
|
|
pass
|
|
return None
|
|
# 1) JSON-LD scripts (strict mode: only source of truth)
|
|
ld_nodes = soup.find_all("script", attrs={"type": _re.compile(r"ld\+json", _re.I)})
|
|
_log(f"Found {len(ld_nodes)} ld+json scripts")
|
|
for idx, ld in enumerate(ld_nodes):
|
|
try:
|
|
text = ld.text.strip()
|
|
data = None
|
|
try:
|
|
data = json.loads(text)
|
|
except json.decoder.JSONDecodeError:
|
|
# Try lenient parsing strategies
|
|
data = _parse_json_lenient(text)
|
|
if data is not None:
|
|
_log(f"ld[{idx}]: parsed with lenient JSON repair")
|
|
else:
|
|
raise
|
|
|
|
if isinstance(data, dict) and _is_recipe_ldata(data):
|
|
_log(f"ld[{idx}]: direct @type Recipe found")
|
|
return data
|
|
|
|
if isinstance(data, dict) and "@graph" in data:
|
|
for gidx, item in enumerate(data["@graph"]):
|
|
if _is_recipe_ldata(item):
|
|
_log(f"ld[{idx}]: @graph item {gidx} is Recipe")
|
|
return item
|
|
|
|
if isinstance(data, list):
|
|
for lidx, item in enumerate(data):
|
|
if _is_recipe_ldata(item):
|
|
_log(f"ld[{idx}]: list item {lidx} is Recipe")
|
|
return item
|
|
|
|
except (json.decoder.JSONDecodeError, KeyError) as e:
|
|
_log(f"ld[{idx}]: JSON parse/Key error: {type(e).__name__}: {e}")
|
|
|
|
# 1b) JSON-LD within <noscript> blocks
|
|
try:
|
|
ns_count = 0
|
|
for ns in soup.find_all("noscript"):
|
|
payload = ns.get_text(strip=False) or ns.decode_contents(formatter="html") or ""
|
|
if not payload:
|
|
continue
|
|
nsoup = BeautifulSoup(payload, "html.parser")
|
|
nodes = nsoup.find_all("script", attrs={"type": _re.compile(r"ld\+json", _re.I)})
|
|
ns_count += len(nodes)
|
|
for nidx, node in enumerate(nodes):
|
|
try:
|
|
text = (node.text or node.get_text() or "").strip()
|
|
if not text:
|
|
continue
|
|
data = _parse_json_lenient(text) or {}
|
|
if isinstance(data, dict) and _is_recipe_ldata(data):
|
|
_log(f"noscript.ld[{nidx}]: direct Recipe found")
|
|
return data
|
|
if isinstance(data, dict) and "@graph" in data:
|
|
for gidx, item in enumerate(data.get("@graph") or []):
|
|
if _is_recipe_ldata(item):
|
|
_log(f"noscript.ld[{nidx}]: @graph item {gidx} is Recipe")
|
|
return item
|
|
if isinstance(data, list):
|
|
for lidx, item in enumerate(data):
|
|
if _is_recipe_ldata(item):
|
|
_log(f"noscript.ld[{nidx}]: list item {lidx} is Recipe")
|
|
return item
|
|
except Exception as e:
|
|
_log(f"noscript.ld[{nidx}]: parse error: {type(e).__name__}: {e}")
|
|
if ns_count:
|
|
_log(f"noscript: scanned {ns_count} ld+json scripts")
|
|
except Exception as e:
|
|
_log(f"noscript scan error: {type(e).__name__}: {e}")
|
|
|
|
# 1c) JSON-LD by content signature in any script or HTML when type is missing
|
|
def _extract_json_objects_around(text: str) -> List[str]:
|
|
out: List[str] = []
|
|
if not text:
|
|
return out
|
|
t = _html.unescape(text)
|
|
key = "@context"
|
|
pos = 0
|
|
while True:
|
|
at = t.find(key, pos)
|
|
if at == -1:
|
|
break
|
|
# find preceding '{'
|
|
lb = t.rfind("{", 0, at)
|
|
if lb == -1:
|
|
pos = at + len(key)
|
|
continue
|
|
i = lb
|
|
depth = 0
|
|
in_str = False
|
|
esc = False
|
|
end = -1
|
|
while i < len(t):
|
|
ch = t[i]
|
|
if in_str:
|
|
if esc:
|
|
esc = False
|
|
elif ch == "\\":
|
|
esc = True
|
|
elif ch == '"':
|
|
in_str = False
|
|
else:
|
|
if ch == '"':
|
|
in_str = True
|
|
elif ch == '{':
|
|
depth += 1
|
|
elif ch == '}':
|
|
depth -= 1
|
|
if depth == 0:
|
|
end = i
|
|
break
|
|
i += 1
|
|
if end != -1:
|
|
out.append(t[lb:end+1])
|
|
pos = end + 1
|
|
else:
|
|
pos = at + len(key)
|
|
return out
|
|
|
|
def _try_parse_jsonld_candidates(cands: List[str], origin: str) -> Optional[dict]:
|
|
for cidx, cand in enumerate(cands):
|
|
data = _parse_json_lenient(cand)
|
|
if data is None:
|
|
continue
|
|
if isinstance(data, dict) and _is_recipe_ldata(data):
|
|
_log(f"{origin}[{cidx}]: Recipe found")
|
|
return data
|
|
if isinstance(data, dict) and "@graph" in data:
|
|
for gidx, item in enumerate(data.get("@graph") or []):
|
|
if _is_recipe_ldata(item):
|
|
_log(f"{origin}[{cidx}]: @graph item {gidx} is Recipe")
|
|
return item
|
|
if isinstance(data, list):
|
|
for lidx, item in enumerate(data):
|
|
if _is_recipe_ldata(item):
|
|
_log(f"{origin}[{cidx}]: list item {lidx} is Recipe")
|
|
return item
|
|
return None
|
|
|
|
# Scan all <script> tags for JSON-LD signature when nothing found
|
|
try:
|
|
texts: List[str] = []
|
|
for sc in soup.find_all("script"):
|
|
txt = (sc.string or sc.get_text() or "").strip()
|
|
if not txt:
|
|
continue
|
|
if "@context" not in txt or "schema.org" not in txt:
|
|
continue
|
|
texts.append(txt)
|
|
if texts:
|
|
cands: List[str] = []
|
|
for t in texts:
|
|
cands.extend(_extract_json_objects_around(t))
|
|
found = _try_parse_jsonld_candidates(cands, origin="script-scan")
|
|
if found:
|
|
return found
|
|
except Exception as e:
|
|
_log(f"script content scan error: {type(e).__name__}: {e}")
|
|
|
|
# Finally, scan the full HTML string
|
|
try:
|
|
html_str = str(soup)
|
|
if "@context" in html_str and "schema.org" in html_str:
|
|
cands = _extract_json_objects_around(html_str)
|
|
found = _try_parse_jsonld_candidates(cands, origin="html-scan")
|
|
if found:
|
|
return found
|
|
except Exception as e:
|
|
_log(f"html scan error: {type(e).__name__}: {e}")
|
|
return None
|
|
|
|
|
|
def _dump_response(dump_dir: str, url: str, prof: str, resp: httpx.Response) -> None:
|
|
import os
|
|
from urllib.parse import urlparse
|
|
os.makedirs(dump_dir, exist_ok=True)
|
|
p = urlparse(url)
|
|
base = f"{p.netloc}{p.path}"
|
|
if not base or base.endswith("/"):
|
|
base += "index"
|
|
safe = base.replace("/", "_").replace("?", "_").replace("&", "_")
|
|
fname = f"{safe}__{prof}__{resp.status_code}.html"
|
|
meta = f"{safe}__{prof}__{resp.status_code}.meta"
|
|
fpath = os.path.join(dump_dir, fname)
|
|
mpath = os.path.join(dump_dir, meta)
|
|
with open(fpath, "w", encoding=resp.encoding or "utf-8", errors="ignore") as f:
|
|
f.write(resp.text)
|
|
with open(mpath, "w", encoding="utf-8") as f:
|
|
f.write(f"URL: {url}\n")
|
|
f.write(f"Profile: {prof}\n")
|
|
f.write(f"Status: {resp.status_code}\n")
|
|
f.write(f"Content-Type: {resp.headers.get('content-type','')}\n")
|
|
|
|
|
|
def scrape_recipe_ldata_from_html(html: str, base_url: str, log=None) -> Optional[dict]:
|
|
"""Extract recipe JSON-LD from raw HTML (strict ld+json-only).
|
|
|
|
base_url is used for relative URL resolution and as a fallback name/link context.
|
|
"""
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
return _extract_ldata_from_soup(soup, base_url, log)
|
|
|
|
|
|
|
|
def _extract_text(el) -> str:
|
|
return " ".join(el.get_text(" ", strip=True).split()) if el else ""
|
|
|