91 lines
2.5 KiB
Python
91 lines
2.5 KiB
Python
import json
|
|
from typing import Optional
|
|
|
|
import httpx
|
|
from bs4 import BeautifulSoup
|
|
|
|
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.5",
|
|
"DNT": "1",
|
|
"Sec-GPC": "1",
|
|
"Connection": "keep-alive",
|
|
"Upgrade-Insecure-Requests": "1",
|
|
"Sec-Fetch-Dest": "document",
|
|
"Sec-Fetch-Mode": "navigate",
|
|
"Sec-Fetch-Site": "none",
|
|
"Sec-Fetch-User": "?1",
|
|
"Priority": "u=1",
|
|
"Pragma": "no-cache",
|
|
"Cache-Control": "no-cache",
|
|
}
|
|
|
|
|
|
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
|
|
|
|
|
|
async def scrape_recipe_ldata(url: str) -> Optional[dict]:
|
|
# Load the requested URL with headers
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.get(url, headers=HEADERS, follow_redirects=True)
|
|
if response.status_code >= 300:
|
|
return None
|
|
|
|
# Extract the recipe ld+json data
|
|
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
|