jbrechtel
ed70f6e3c2
Downloaded recipe images from source URLs (NYT Cooking, Love & Lemons, Food Network, etc.) via og:image scraping and DuckDuckGo search + page scraping for recipes without direct source URLs.
243 lines
7.9 KiB
Python
243 lines
7.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Final attempt: search DDG text, scrape recipe pages for actual images."""
|
|
|
|
import csv
|
|
import json
|
|
import re
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
from ddgs import DDGS
|
|
|
|
WORKDIR = Path("/home/jbrechtel/recipes/.worktrees/recipe-images")
|
|
HEADERS = {
|
|
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
|
|
}
|
|
SKIP_DOMAINS = {"pinterest", "facebook", "instagram", "twitter", "youtube", "tiktok"}
|
|
MIN_IMAGE_SIZE = 10000 # 10KB minimum for a recipe photo
|
|
|
|
|
|
def search_url(query: str) -> str | None:
|
|
"""Search and return first relevant recipe page URL."""
|
|
try:
|
|
results = list(DDGS().text(query, max_results=8))
|
|
except Exception as e:
|
|
print(f" Search error: {e}", file=sys.stderr)
|
|
return None
|
|
|
|
for r in results:
|
|
url = r.get("href", "")
|
|
domain = url.split("/")[2] if "//" in url else ""
|
|
if any(s in domain for s in SKIP_DOMAINS):
|
|
continue
|
|
# Prefer recipe sites
|
|
if any(kw in domain for kw in ["recipe", "cook", "food", "kitchen", "nytimes", "loveandlemons"]):
|
|
return url
|
|
|
|
# Return first non-wikipedia result
|
|
for r in results:
|
|
url = r.get("href", "")
|
|
if "wikipedia.org" not in url:
|
|
return url
|
|
return None
|
|
|
|
|
|
def find_recipe_images(url: str) -> list[str]:
|
|
"""Fetch a page and find actual recipe/food images."""
|
|
try:
|
|
resp = requests.get(url, headers=HEADERS, timeout=15, allow_redirects=True)
|
|
resp.raise_for_status()
|
|
except requests.RequestException:
|
|
return []
|
|
|
|
soup = BeautifulSoup(resp.text, "html.parser")
|
|
images = []
|
|
|
|
# 1. Check JSON-LD for recipe image
|
|
for script in soup.find_all("script", type="application/ld+json"):
|
|
try:
|
|
data = json.loads(script.string)
|
|
except (json.JSONDecodeError, TypeError):
|
|
continue
|
|
# Handle list or single
|
|
items = data if isinstance(data, list) else [data]
|
|
for item in items:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
if item.get("@type") == "Recipe":
|
|
img = item.get("image")
|
|
if isinstance(img, str):
|
|
images.append(img)
|
|
elif isinstance(img, list):
|
|
for i in img:
|
|
if isinstance(i, str):
|
|
images.append(i)
|
|
elif isinstance(i, dict):
|
|
u = i.get("contentUrl") or i.get("url", "")
|
|
if u:
|
|
images.append(u)
|
|
elif isinstance(img, dict):
|
|
u = img.get("contentUrl") or img.get("url", "")
|
|
if u:
|
|
images.append(u)
|
|
|
|
# 2. Check og:image
|
|
for meta in soup.find_all("meta"):
|
|
content = meta.get("content", "")
|
|
if (meta.get("property") == "og:image" or meta.get("name") == "og:image" or meta.get("name") == "twitter:image"):
|
|
if content and "logo" not in content.lower():
|
|
images.append(content)
|
|
|
|
# 3. Check all img tags for large, relevant images
|
|
for img in soup.find_all("img"):
|
|
src = img.get("src") or img.get("data-src") or ""
|
|
if not src.startswith("http"):
|
|
continue
|
|
src_lower = src.lower()
|
|
if any(kw in src_lower for kw in ["logo", "avatar", "icon", "banner", "pixel", "spacer", "thumb"]):
|
|
continue
|
|
# Check for recipe-related class/id
|
|
parent_classes = ""
|
|
parent = img.parent
|
|
if parent:
|
|
parent_classes = " ".join(parent.get("class", []) or []) + " " + (parent.get("id") or "")
|
|
if any(kw in parent_classes.lower() for kw in ["logo", "nav", "header", "footer", "menu", "icon"]):
|
|
continue
|
|
images.append(src)
|
|
|
|
# Deduplicate
|
|
seen = set()
|
|
unique = []
|
|
for img in images:
|
|
if img not in seen:
|
|
seen.add(img)
|
|
unique.append(img)
|
|
|
|
return unique
|
|
|
|
|
|
def download_image(url: str, dest: Path) -> bool:
|
|
"""Download image, validate it's a real image."""
|
|
try:
|
|
resp = requests.get(url, headers=HEADERS, timeout=30, stream=True)
|
|
resp.raise_for_status()
|
|
ct = resp.headers.get("content-type", "")
|
|
if not ct.startswith("image/"):
|
|
return False
|
|
content = resp.content
|
|
if len(content) < MIN_IMAGE_SIZE:
|
|
return False
|
|
dest.write_bytes(content)
|
|
return True
|
|
except requests.RequestException:
|
|
return False
|
|
|
|
|
|
def get_title(filepath: Path) -> str:
|
|
content = filepath.read_text(encoding="utf-8")
|
|
m = re.search(r"^title:\s*(.+)$", content, re.MULTILINE)
|
|
if m:
|
|
return m.group(1).strip()
|
|
return filepath.stem.replace("_", " ").replace("-", " ")
|
|
|
|
|
|
def add_image_field(filepath: Path, slug: str):
|
|
url = f"https://roux.roo.lol/recipe-images/{slug}.webp"
|
|
content = filepath.read_text(encoding="utf-8")
|
|
if re.search(r"^image:", content, re.MULTILINE):
|
|
content = re.sub(r"^image:.*$", f"image: {url}", count=1, flags=re.MULTILINE, string=content)
|
|
else:
|
|
content = re.sub(r"^---$", f"image: {url}\n---", count=1, flags=re.MULTILINE, string=content)
|
|
filepath.write_text(content, encoding="utf-8")
|
|
print(f" Updated {filepath.name}", file=sys.stderr)
|
|
|
|
|
|
def get_source_url(filepath: Path) -> str | None:
|
|
content = filepath.read_text(encoding="utf-8")
|
|
m = re.search(r"^source:\s*(.+)$", content, re.MULTILINE)
|
|
if m:
|
|
val = m.group(1).strip()
|
|
if val.startswith("http"):
|
|
return val
|
|
return None
|
|
|
|
|
|
def main():
|
|
csv_file = WORKDIR / "docs" / "no-image-has-fm.csv"
|
|
|
|
recipes = []
|
|
with open(csv_file) as f:
|
|
reader = csv.reader(f, delimiter="|")
|
|
for row in reader:
|
|
if len(row) >= 2:
|
|
recipes.append((row[0].strip(), row[1].strip()))
|
|
|
|
success = []
|
|
failed = []
|
|
|
|
for filename, slug in recipes:
|
|
filepath = WORKDIR / filename
|
|
if not filepath.exists():
|
|
failed.append(slug)
|
|
continue
|
|
|
|
dest = WORKDIR / "recipe-images" / f"{slug}.webp"
|
|
|
|
# Skip if already has image AND file exists
|
|
content = filepath.read_text(encoding="utf-8")
|
|
has_field = bool(re.search(r"^image:", content, re.MULTILINE))
|
|
has_file = dest.exists()
|
|
if has_field and has_file and dest.stat().st_size > 10000:
|
|
print(f"[{slug}] Already done, skipping", file=sys.stderr)
|
|
success.append(slug)
|
|
continue
|
|
|
|
title = get_title(filepath)
|
|
query = f"{title} recipe"
|
|
|
|
print(f"[{slug}] Searching: {query}", file=sys.stderr)
|
|
page_url = search_url(query)
|
|
if not page_url:
|
|
print(f" No search results", file=sys.stderr)
|
|
failed.append(slug)
|
|
continue
|
|
|
|
print(f" Page: {page_url}", file=sys.stderr)
|
|
time.sleep(0.3)
|
|
|
|
img_urls = find_recipe_images(page_url)
|
|
if not img_urls:
|
|
print(f" No images found on page", file=sys.stderr)
|
|
failed.append(slug)
|
|
continue
|
|
|
|
downloaded = False
|
|
for img_url in img_urls[:5]:
|
|
print(f" Trying: {img_url[:80]}...", file=sys.stderr)
|
|
if download_image(img_url, dest):
|
|
print(f" Downloaded ({dest.stat().st_size} bytes)", file=sys.stderr)
|
|
add_image_field(filepath, slug)
|
|
downloaded = True
|
|
break
|
|
time.sleep(0.2)
|
|
|
|
if downloaded:
|
|
success.append(slug)
|
|
else:
|
|
failed.append(slug)
|
|
|
|
time.sleep(0.5)
|
|
|
|
print(f"\n=== Results ===", file=sys.stderr)
|
|
print(f"Success: {len(success)}", file=sys.stderr)
|
|
print(f"Failed: {len(failed)}", file=sys.stderr)
|
|
if failed:
|
|
print(f"Failed: {', '.join(failed)}", file=sys.stderr)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|