123 lines
4.2 KiB
Python
123 lines
4.2 KiB
Python
|
"""
|
||
|
|
Snapshot recipe pages locally for offline inspection and future parsing tests.
|
||
|
|
|
||
|
|
Run explicitly (not part of CI by default):
|
||
|
|
|
||
|
|
python -m scripts.save_recipe_pages
|
||
|
|
|
||
|
|
This will download each URL in scripts/recipe_urls.py and save into:
|
||
|
|
tests/sample_files/recipes/
|
||
|
|
|
||
|
|
For each URL, it writes:
|
||
|
|
- <slug>.html The raw page HTML
|
||
|
|
- <slug>.meta.json JSON metadata (url, fetched_at, status, final_url, error)
|
||
|
|
|
||
|
|
Notes:
|
||
|
|
- Uses the same browser-like headers as our scraper.
|
||
|
|
- Respects redirects; stores final_url.
|
||
|
|
- Skips writing HTML on hard errors but still writes a .meta.json with the error.
|
||
|
|
- Creates parent folders as needed.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import re
|
||
|
|
from datetime import datetime, timezone
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Dict, Tuple
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
|
||
|
|
from scripts.recipe_urls import URLS
|
||
|
|
from recipes.scraping import DEFAULT_HEADERS, HEADER_PROFILES, _fallback_urls, fetch_first_2xx_html, looks_like_jsonld
|
||
|
|
|
||
|
|
|
||
|
|
OUT_DIR = Path("tests/sample_files/recipes")
|
||
|
|
|
||
|
|
|
||
|
|
def _safe_slug_from_url(url: str) -> str:
|
||
|
|
# Keep domain and last path segment(s) for readability; replace non-word with '-'
|
||
|
|
from urllib.parse import urlparse
|
||
|
|
p = urlparse(url)
|
||
|
|
host = p.netloc.replace(":", "-")
|
||
|
|
path = p.path.rstrip("/")
|
||
|
|
if not path:
|
||
|
|
seg = "index"
|
||
|
|
else:
|
||
|
|
# use last two segments if available to avoid collisions like /recipe/ vs /recipe-2/
|
||
|
|
parts = [s for s in path.split("/") if s]
|
||
|
|
seg = "-".join(parts[-2:]) if len(parts) >= 2 else parts[-1]
|
||
|
|
raw = f"{host}-{seg}".lower()
|
||
|
|
slug = re.sub(r"[^a-z0-9._-]+", "-", raw).strip("-")
|
||
|
|
return slug or "page"
|
||
|
|
|
||
|
|
|
||
|
|
async def _fetch_best_html(client: httpx.AsyncClient, url: str) -> Tuple[str, Dict]:
|
||
|
|
meta: Dict = {
|
||
|
|
"url": url,
|
||
|
|
"fetched_at": datetime.now(timezone.utc).isoformat(),
|
||
|
|
"status": None,
|
||
|
|
"final_url": None,
|
||
|
|
"error": None,
|
||
|
|
"profile": None,
|
||
|
|
"content_type": None,
|
||
|
|
"content_encoding": None,
|
||
|
|
"candidate": None,
|
||
|
|
}
|
||
|
|
visited: set[str] = set()
|
||
|
|
candidates = list(_fallback_urls(url))
|
||
|
|
for cand in candidates:
|
||
|
|
if cand in visited:
|
||
|
|
continue
|
||
|
|
visited.add(cand)
|
||
|
|
html, prof_meta = await fetch_first_2xx_html(client, cand, safe_accept_encoding=True)
|
||
|
|
# Merge selected metadata
|
||
|
|
meta.update({
|
||
|
|
"status": int(prof_meta.get("status") or 0) or None,
|
||
|
|
"final_url": prof_meta.get("final_url"),
|
||
|
|
"content_type": prof_meta.get("content_type"),
|
||
|
|
"content_encoding": prof_meta.get("content_encoding"),
|
||
|
|
"profile": prof_meta.get("profile"),
|
||
|
|
"request_headers": prof_meta.get("request_headers"),
|
||
|
|
"candidate": cand,
|
||
|
|
"error": prof_meta.get("error"),
|
||
|
|
})
|
||
|
|
if html and looks_like_jsonld(html):
|
||
|
|
return html, meta
|
||
|
|
# If html is present, keep as a fallback in case later candidates fail
|
||
|
|
if html:
|
||
|
|
fallback_html = html
|
||
|
|
fallback_meta = dict(meta)
|
||
|
|
continue
|
||
|
|
# Nothing succeeded; return empty payload with last meta
|
||
|
|
return "", meta
|
||
|
|
|
||
|
|
|
||
|
|
async def main() -> None:
|
||
|
|
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
|
async with httpx.AsyncClient(http2=True) as client:
|
||
|
|
sem = asyncio.Semaphore(6)
|
||
|
|
|
||
|
|
async def worker(u: str):
|
||
|
|
async with sem:
|
||
|
|
html, meta = await _fetch_best_html(client, u)
|
||
|
|
slug = _safe_slug_from_url(u)
|
||
|
|
html_path = OUT_DIR / f"{slug}.html"
|
||
|
|
meta_path = OUT_DIR / f"{slug}.meta.json"
|
||
|
|
try:
|
||
|
|
# Always write metadata
|
||
|
|
meta_path.write_text(json.dumps(meta, indent=2))
|
||
|
|
# Only persist HTML on successful fetch
|
||
|
|
if meta.get("status") and 200 <= int(meta["status"]) < 300 and html:
|
||
|
|
html_path.write_text(html)
|
||
|
|
print(f"saved: {u} -> {html_path.name} (status={meta.get('status')}, error={meta.get('error')})")
|
||
|
|
except Exception as e:
|
||
|
|
print(f"failed to save for {u}: {type(e).__name__}: {e}")
|
||
|
|
|
||
|
|
await asyncio.gather(*(worker(u) for u in URLS))
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
asyncio.run(main())
|