291 lines
9.8 KiB
Python
291 lines
9.8 KiB
Python
import json
|
|
from typing import Optional, Iterable, List
|
|
|
|
import httpx
|
|
from bs4 import BeautifulSoup
|
|
|
|
# A realistic browser header profile improves success rates against some CDNs/bot protections.
|
|
DEFAULT_HEADERS = {
|
|
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
|
|
"Accept-Language": "en-US,en;q=0.9",
|
|
"Accept-Encoding": "gzip, deflate, br",
|
|
"Connection": "keep-alive",
|
|
"Upgrade-Insecure-Requests": "1",
|
|
"Sec-Fetch-Dest": "document",
|
|
"Sec-Fetch-Mode": "navigate",
|
|
"Sec-Fetch-Site": "none",
|
|
"Sec-Fetch-User": "?1",
|
|
# A modern desktop Chrome UA with platform tokens; not tied to any user data.
|
|
"User-Agent": (
|
|
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) "
|
|
"Chrome/127.0.0.0 Safari/537.36"
|
|
),
|
|
}
|
|
|
|
|
|
def _is_recipe_ldata(ldata_node) -> bool:
|
|
if "@type" in ldata_node:
|
|
typ = ldata_node["@type"]
|
|
if isinstance(typ, list):
|
|
typ = typ[0]
|
|
|
|
if isinstance(typ, str) and typ.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
|
|
"""
|
|
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
|
|
except Exception:
|
|
# Be conservative if URL parsing fails
|
|
pass
|
|
|
|
|
|
BLOCK_STATUSES = {403, 406, 429, 460}
|
|
|
|
|
|
async def scrape_recipe_ldata(url: str) -> Optional[dict]:
|
|
# Try the URL with browser-like headers and fallback strategies when blocked.
|
|
async with httpx.AsyncClient() as client:
|
|
for candidate in _fallback_urls(url):
|
|
# Some CDNs prefer a referer; provide same-origin referer as a harmless hint.
|
|
headers = dict(DEFAULT_HEADERS)
|
|
headers.setdefault("Referer", candidate)
|
|
response = await client.get(candidate, headers=headers, follow_redirects=True)
|
|
if response.status_code in BLOCK_STATUSES:
|
|
# Try next fallback
|
|
continue
|
|
if response.status_code >= 300:
|
|
# Try next fallback on non-2xx
|
|
continue
|
|
|
|
# Extract the recipe ld+json data from this candidate
|
|
soup = BeautifulSoup(response.text, "html.parser")
|
|
for ld in soup.find_all("script", type="application/ld+json"):
|
|
try:
|
|
text = ld.text.strip()
|
|
# Some sites embed multiple JSON objects without an array. Try to coerce if needed.
|
|
try:
|
|
data = json.loads(text)
|
|
except json.decoder.JSONDecodeError:
|
|
# Attempt to split objects and wrap in a list (best-effort)
|
|
# Very conservative: only try if it looks like multiple root objects.
|
|
if text.count("{") > 1 and "}\n{" in text:
|
|
parts = [p for p in text.split("\n") if p.strip()]
|
|
maybe = "[" + ",".join(parts) + "]"
|
|
data = json.loads(maybe)
|
|
else:
|
|
raise
|
|
# _dump_json_data_to_log(data)
|
|
if _is_recipe_ldata(data):
|
|
return data
|
|
|
|
if "@graph" in data:
|
|
for item in data["@graph"]:
|
|
if _is_recipe_ldata(item):
|
|
return item
|
|
|
|
if isinstance(data, list):
|
|
for item in data:
|
|
if _is_recipe_ldata(item):
|
|
return item
|
|
|
|
except (json.decoder.JSONDecodeError, KeyError):
|
|
pass
|
|
|
|
# Fallback 1: microdata (itemtype Recipe, itemprop recipeIngredient)
|
|
md = _extract_recipe_from_microdata(soup)
|
|
if md:
|
|
return md
|
|
|
|
# Fallback 2: heuristic DOM extraction around "Ingredients" section
|
|
hd = _extract_recipe_heuristic(soup, candidate)
|
|
if hd:
|
|
return hd
|
|
|
|
return None
|
|
|
|
# Fallback return to satisfy static analysis
|
|
return None
|
|
|
|
|
|
def _dump_json_data_to_log(data: dict) -> str:
|
|
import os
|
|
import re
|
|
|
|
dir = "./data/dump"
|
|
if not os.path.exists(dir):
|
|
os.makedirs(dir)
|
|
|
|
prefix = "ldata_"
|
|
suffix = ".json"
|
|
file_ids = [
|
|
int(re.findall(r"\d+", f)[0])
|
|
for f in os.listdir(dir)
|
|
if re.match(prefix + r"\d+" + suffix, f)
|
|
]
|
|
id = max(file_ids) + 1 if file_ids else 0
|
|
filename = f"{prefix}{id}{suffix}"
|
|
full_path = os.path.join(dir, filename)
|
|
with open(full_path, "w") as f:
|
|
json.dump(data, f, indent=4)
|
|
return full_path
|
|
|
|
|
|
def _extract_text(el) -> str:
|
|
return " ".join(el.get_text(" ", strip=True).split()) if el else ""
|
|
|
|
|
|
def _extract_recipe_from_microdata(soup: BeautifulSoup) -> Optional[dict]:
|
|
import re
|
|
|
|
candidates = []
|
|
for node in soup.find_all(attrs={"itemtype": re.compile(r"schema\.org/Recipe", re.I)}):
|
|
candidates.append(node)
|
|
# Some sites omit itemtype but use itemprop markers globally
|
|
if not candidates:
|
|
candidates = [soup]
|
|
|
|
def get_itemprop(node, prop) -> Optional[str]:
|
|
el = node.find(attrs={"itemprop": prop})
|
|
if not el:
|
|
return None
|
|
# Support <meta content>, <img src>, or text
|
|
if el.has_attr("content"):
|
|
return el["content"]
|
|
if el.has_attr("src"):
|
|
return el["src"]
|
|
if el.has_attr("href"):
|
|
return el["href"]
|
|
return _extract_text(el)
|
|
|
|
for node in candidates:
|
|
# Gather ingredients by itemprop
|
|
ingredients: List[str] = []
|
|
for prop in ("recipeIngredient", "ingredients"):
|
|
for ing_el in node.find_all(attrs={"itemprop": prop}):
|
|
t = _extract_text(ing_el)
|
|
if t:
|
|
ingredients.append(t)
|
|
|
|
name = get_itemprop(node, "name") or get_itemprop(soup, "og:title") or None
|
|
image = (
|
|
get_itemprop(node, "image")
|
|
or get_itemprop(soup, "og:image")
|
|
or get_itemprop(soup, "twitter:image")
|
|
)
|
|
if image:
|
|
images = [image]
|
|
else:
|
|
images = []
|
|
ry = get_itemprop(node, "recipeYield") or get_itemprop(soup, "recipeYield")
|
|
|
|
# Minimal viability: at least one ingredient and a name
|
|
if ingredients and (name or len(ingredients) >= 2):
|
|
out = {"@type": "Recipe", "recipeIngredient": ingredients}
|
|
if name:
|
|
out["name"] = name
|
|
if images:
|
|
out["image"] = images
|
|
if ry:
|
|
out["recipeYield"] = ry
|
|
return out
|
|
return None
|
|
|
|
|
|
def _extract_recipe_heuristic(soup: BeautifulSoup, page_url: str) -> Optional[dict]:
|
|
"""Best-effort heuristic extraction when no structured data is present.
|
|
|
|
- Finds an Ingredients heading and the nearest list (ul/ol)
|
|
- Or any element with class/id containing 'ingredient' and list items
|
|
- Derives name from og:title or <title>
|
|
- Derives image from og:image if available
|
|
"""
|
|
import re
|
|
|
|
# Try explicit ingredients container by class/id
|
|
candidates = []
|
|
for el in soup.find_all(True, attrs={"class": re.compile("ingredient", re.I)}):
|
|
candidates.append(el)
|
|
for el in soup.find_all(True, attrs={"id": re.compile("ingredient", re.I)}):
|
|
candidates.append(el)
|
|
|
|
def collect_list_items(node) -> List[str]:
|
|
items: List[str] = []
|
|
for ul in node.find_all(["ul", "ol"]):
|
|
for li in ul.find_all("li"):
|
|
t = _extract_text(li)
|
|
if t:
|
|
items.append(t)
|
|
return items
|
|
|
|
ingredients: List[str] = []
|
|
for node in candidates:
|
|
ingredients = collect_list_items(node)
|
|
if ingredients:
|
|
break
|
|
|
|
# Try heading-based approach
|
|
if not ingredients:
|
|
for h in soup.find_all(["h1", "h2", "h3", "h4", "h5", "h6"]):
|
|
if "ingredient" in _extract_text(h).lower():
|
|
# Try next sibling lists or parent lists
|
|
sib = h.find_next_sibling()
|
|
tries = [sib, h.parent]
|
|
for t in tries:
|
|
if not t:
|
|
continue
|
|
lst = collect_list_items(t)
|
|
if lst:
|
|
ingredients = lst
|
|
break
|
|
if ingredients:
|
|
break
|
|
|
|
if not ingredients:
|
|
return None
|
|
|
|
# Name and image fallbacks
|
|
name = None
|
|
og_title = soup.find("meta", attrs={"property": "og:title"})
|
|
if og_title and og_title.has_attr("content"):
|
|
name = og_title["content"]
|
|
if not name and soup.title:
|
|
name = _extract_text(soup.title)
|
|
|
|
image = None
|
|
og_image = soup.find("meta", attrs={"property": "og:image"})
|
|
if og_image and og_image.has_attr("content"):
|
|
image = og_image["content"]
|
|
|
|
out = {"@type": "Recipe", "recipeIngredient": ingredients}
|
|
out["name"] = name or page_url
|
|
if image:
|
|
out["image"] = [image]
|
|
return out
|