#!/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()