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 , , 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 - 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