140 lines
4.4 KiB
Python
140 lines
4.4 KiB
Python
import json
|
|
from typing import Optional, Iterable
|
|
|
|
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:
|
|
data = json.loads(ld.text)
|
|
# _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
|
|
|
|
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
|