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.
234 lines
7.6 KiB
Python
234 lines
7.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Search the web for recipe images, download them, and update .cook files.
|
|
|
|
Uses DuckDuckGo search to find recipe pages, then scrapes og:image.
|
|
"""
|
|
|
|
import csv
|
|
import os
|
|
import re
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
from duckduckgo_search 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",
|
|
}
|
|
|
|
IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".webp", ".gif"}
|
|
|
|
|
|
def fetch_og_image(url: str) -> str | None:
|
|
"""Fetch a page and extract og:image URL."""
|
|
try:
|
|
resp = requests.get(url, headers=HEADERS, timeout=15, allow_redirects=True)
|
|
resp.raise_for_status()
|
|
except requests.RequestException:
|
|
return None
|
|
|
|
soup = BeautifulSoup(resp.text, "html.parser")
|
|
|
|
# Try og:image meta tag (property or name)
|
|
for meta in soup.find_all("meta"):
|
|
prop = (meta.get("property") or "").lower()
|
|
name = (meta.get("name") or "").lower()
|
|
content = meta.get("content", "")
|
|
if "og:image" in prop and content:
|
|
return content
|
|
if "og:image" in name and content:
|
|
return content
|
|
if "twitter:image" in name and content:
|
|
return content
|
|
|
|
# Try schema.org JSON-LD
|
|
for script in soup.find_all("script", type="application/ld+json"):
|
|
import json
|
|
try:
|
|
data = json.loads(script.string)
|
|
if isinstance(data, dict):
|
|
img = data.get("image")
|
|
if isinstance(img, str) and img.startswith("http"):
|
|
return img
|
|
if isinstance(img, dict):
|
|
url = img.get("contentUrl") or img.get("url", "")
|
|
if url.startswith("http"):
|
|
return url
|
|
if isinstance(img, list) and img:
|
|
first = img[0]
|
|
if isinstance(first, str):
|
|
return first
|
|
if isinstance(first, dict):
|
|
url = first.get("contentUrl") or first.get("url", "")
|
|
if url.startswith("http"):
|
|
return url
|
|
except (json.JSONDecodeError, AttributeError):
|
|
pass
|
|
|
|
return None
|
|
|
|
|
|
def search_and_find_image(query: str) -> str | None:
|
|
"""Search DuckDuckGo for a recipe and find og:image."""
|
|
try:
|
|
results = list(DDGS().text(query, max_results=5))
|
|
except Exception as e:
|
|
print(f" Search error: {e}", file=sys.stderr)
|
|
return None
|
|
|
|
for r in results:
|
|
url = r.get("href", "")
|
|
if not url or "pinterest" in url or "facebook" in url or "instagram" in url:
|
|
continue
|
|
print(f" Trying: {url}", file=sys.stderr)
|
|
img = fetch_og_image(url)
|
|
if img:
|
|
print(f" Found image: {img}", file=sys.stderr)
|
|
return img
|
|
time.sleep(0.3)
|
|
|
|
return None
|
|
|
|
|
|
def download_image(url: str, dest: Path) -> bool:
|
|
"""Download an image from URL to destination path."""
|
|
try:
|
|
resp = requests.get(url, headers=HEADERS, timeout=30, stream=True)
|
|
resp.raise_for_status()
|
|
dest.write_bytes(resp.content)
|
|
return True
|
|
except requests.RequestException as e:
|
|
print(f" Download failed: {e}", file=sys.stderr)
|
|
return False
|
|
|
|
|
|
def add_image_field(filepath: Path, slug: str):
|
|
"""Add or update the image field in a .cook file's YAML frontmatter."""
|
|
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:
|
|
"""Extract source URL from .cook frontmatter."""
|
|
content = filepath.read_text(encoding="utf-8")
|
|
match = re.search(r"^source:\s*(.+)$", content, re.MULTILINE)
|
|
if match:
|
|
val = match.group(1).strip()
|
|
if val.startswith("http://") or val.startswith("https://"):
|
|
return val
|
|
return None
|
|
|
|
|
|
def process_recipe(filename: str, slug: str) -> bool:
|
|
"""Try to find and download an image for a recipe."""
|
|
filepath = WORKDIR / filename
|
|
dest = WORKDIR / "recipe-images" / f"{slug}.webp"
|
|
|
|
if dest.exists():
|
|
print(f"[{slug}] Already exists", file=sys.stderr)
|
|
add_image_field(filepath, slug)
|
|
return True
|
|
|
|
# Strategy 1: Try source URL if available
|
|
src_url = get_source_url(filepath)
|
|
if src_url:
|
|
print(f"[{slug}] Trying source URL: {src_url}", file=sys.stderr)
|
|
img = fetch_og_image(src_url)
|
|
if img:
|
|
if download_image(img, dest):
|
|
print(f" Downloaded from source", file=sys.stderr)
|
|
add_image_field(filepath, slug)
|
|
return True
|
|
|
|
# Strategy 2: Search by recipe name
|
|
# Get the title from frontmatter
|
|
content = filepath.read_text(encoding="utf-8")
|
|
title_match = re.search(r"^title:\s*(.+)$", content, re.MULTILINE)
|
|
if title_match:
|
|
query = title_match.group(1).strip() + " recipe"
|
|
else:
|
|
query = filename.replace(".cook", "").replace("_", " ").replace("-", " ") + " recipe"
|
|
|
|
print(f"[{slug}] Searching: {query}", file=sys.stderr)
|
|
img = search_and_find_image(query)
|
|
if img and download_image(img, dest):
|
|
print(f" Downloaded from search", file=sys.stderr)
|
|
add_image_field(filepath, slug)
|
|
return True
|
|
|
|
# Strategy 3: Try with "og:image" or "recipe image" variations
|
|
query2 = query + " image"
|
|
print(f"[{slug}] Searching (alt): {query2}", file=sys.stderr)
|
|
img = search_and_find_image(query2)
|
|
if img and download_image(img, dest):
|
|
print(f" Downloaded from alt search", file=sys.stderr)
|
|
add_image_field(filepath, slug)
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
def main():
|
|
# First process the no-image-has-fm list
|
|
csv_file = WORKDIR / "docs" / "no-image-has-fm.csv"
|
|
if not csv_file.exists():
|
|
print(f"CSV not found: {csv_file}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
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"
|
|
if dest.exists() and re.search(r"^image:", filepath.read_text(), re.MULTILINE):
|
|
print(f"[{slug}] Already processed, skipping", file=sys.stderr)
|
|
success.append(slug)
|
|
continue
|
|
|
|
ok = process_recipe(filename, slug)
|
|
if ok:
|
|
success.append(slug)
|
|
else:
|
|
failed.append(slug)
|
|
|
|
# Be polite
|
|
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:
|
|
with open(WORKDIR / "docs" / "failed-images.csv", "w") as f:
|
|
for s in failed:
|
|
f.write(f"{s}\n")
|
|
print(f"Failed: {', '.join(failed)}", file=sys.stderr)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|