gig-line/Tools/gen/commons_fetch.py

#!/usr/bin/env python3
"""Fetch reference photographs from Wikimedia Commons.

Local, private development reference only. Images fetched here are **not**
committed: the output directory is gitignored. The tool records provenance
(author, licence, source URL) in a JSON sidecar next to the images so a local
working copy stays traceable.

    # dry run: list what the search returns, download nothing
    python3 Tools/gen/commons_fetch.py --query "Grauman's Chinese Theatre" --slug north-chinese --dry-run

    # download up to 6 images at 1600 px wide
    python3 Tools/gen/commons_fetch.py --query "Grauman's Chinese Theatre" --slug north-chinese --limit 6

    # fetch specific files by title
    python3 Tools/gen/commons_fetch.py --file "File:Grauman's Chinese Theatre, Hollywood.JPG" --slug north-chinese

Defaults: out dir `Art/reference/<slug>/`, 1600 px wide thumbnails, one request
per second, JPEG/PNG only, at least 800 px wide on the original.

Requires only the standard library.
"""

from __future__ import annotations

import argparse
import datetime
import hashlib
import json
import os
import re
import sys
import time
import urllib.parse
import urllib.request

API = "https://commons.wikimedia.org/w/api.php"
UA = "gig-line-references/0.1 (local development reference fetch)"
ALLOWED_MIME = {"image/jpeg", "image/png"}
MIN_WIDTH = 800


def api_get(params: dict, retries: int = 3) -> dict:
    query = urllib.parse.urlencode({**params, "format": "json", "formatversion": "2"})
    url = f"{API}?{query}"
    for attempt in range(retries):
        req = urllib.request.Request(url, headers={"User-Agent": UA})
        try:
            with urllib.request.urlopen(req, timeout=45) as resp:
                return json.loads(resp.read().decode("utf-8"))
        except Exception as exc:  # noqa: BLE001 - report and retry, then fail
            if attempt == retries - 1:
                raise SystemExit(f"Commons API failed: {exc}")
            time.sleep(2 * (attempt + 1))
    return {}


def strip_html(value: str) -> str:
    text = re.sub(r"<[^>]+>", " ", value or "")
    return re.sub(r"\s+", " ", text).strip()


def search_titles(query: str, limit: int) -> list[str]:
    data = api_get(
        {
            "action": "query",
            "generator": "search",
            "gsrsearch": query,
            "gsrnamespace": "6",
            "gsrlimit": str(limit),
            "prop": "imageinfo",
            "iiprop": "url|size|mime|extmetadata",
        }
    )
    pages = data.get("query", {}).get("pages", [])
    # generator search returns unordered; keep relevance order from the index list
    order = [p.get("pageid") for p in pages]
    titles = []
    for page in pages:
        info = (page.get("imageinfo") or [{}])[0]
        if info.get("mime") not in ALLOWED_MIME:
            continue
        if info.get("width", 0) < MIN_WIDTH:
            continue
        titles.append(page["title"])
    return titles, order


def category_titles(category: str, limit: int) -> list[str]:
    data = api_get(
        {
            "action": "query",
            "list": "categorymembers",
            "cmtitle": category,
            "cmtype": "file",
            "cmlimit": str(max(limit, 50)),
        }
    )
    members = data.get("query", {}).get("categorymembers", [])
    return [m["title"] for m in members][:limit]


def image_info(titles: list[str], thumb_width: int) -> list[dict]:
    data = api_get(
        {
            "action": "query",
            "titles": "|".join(titles),
            "prop": "imageinfo",
            "iiprop": "url|size|mime|extmetadata|sha1|user",
            "iiurlwidth": str(thumb_width),
        }
    )
    out = []
    for page in data.get("query", {}).get("pages", []):
        info = (page.get("imageinfo") or [{}])[0]
        if not info or info.get("mime") not in ALLOWED_MIME:
            continue
        meta = info.get("extmetadata", {})
        out.append(
            {
                "title": page["title"],
                "page_url": info.get("descriptionurl"),
                "file_url": info.get("url"),
                "download_url": info.get("thumburl") or info.get("url"),
                "width": info.get("width"),
                "height": info.get("height"),
                "mime": info.get("mime"),
                "uploader": info.get("user"),
                "sha1": info.get("sha1"),
                "artist": strip_html(meta.get("Artist", {}).get("value", "")),
                "credit": strip_html(meta.get("Credit", {}).get("value", "")),
                "licence": strip_html(
                    meta.get("LicenseShortName", {}).get("value", "")
                    or meta.get("UsageTerms", {}).get("value", "")
                ),
                "licence_url": strip_html(meta.get("LicenseUrl", {}).get("value", "")),
                "attribution_required": strip_html(
                    meta.get("AttributionRequired", {}).get("value", "")
                ),
                "date": strip_html(
                    meta.get("DateTimeOriginal", {}).get("value", "")
                    or meta.get("DateTime", {}).get("value", "")
                ),
                "description": strip_html(meta.get("ImageDescription", {}).get("value", ""))[:400],
            }
        )
    return out


def safe_name(title: str, index: int) -> str:
    base = title.split(":", 1)[-1]
    base = re.sub(r"[^A-Za-z0-9._-]+", "-", base).strip("-").lower()
    ext = os.path.splitext(base)[1] or ".jpg"
    stem = os.path.splitext(base)[0][:60]
    return f"{index:02d}-{stem}{ext}"


def download(url: str, path: str, retries: int = 3) -> str:
    for attempt in range(retries):
        req = urllib.request.Request(url, headers={"User-Agent": UA})
        try:
            with urllib.request.urlopen(req, timeout=90) as resp:
                payload = resp.read()
            with open(path, "wb") as handle:
                handle.write(payload)
            return hashlib.sha256(payload).hexdigest()
        except Exception as exc:  # noqa: BLE001
            if attempt == retries - 1:
                raise SystemExit(f"download failed for {url}: {exc}")
            time.sleep(2 * (attempt + 1))
    return ""


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--query", help="Commons search string")
    ap.add_argument("--category", help='Commons category, e.g. "Category:Hollywood Masonic Temple"')
    ap.add_argument("--file", action="append", default=[], help="exact File: title, repeatable")
    ap.add_argument("--slug", required=True, help="output folder name under the out dir")
    ap.add_argument("--out", default="Art/reference", help="base output directory")
    ap.add_argument("--limit", type=int, default=6)
    ap.add_argument("--width", type=int, default=1600, help="thumbnail width in px")
    ap.add_argument("--dry-run", action="store_true")
    ap.add_argument("--delay", type=float, default=2.0, help="seconds between downloads")
    args = ap.parse_args()

    if not args.query and not args.file and not args.category:
        ap.error("need --query, --category, or --file")

    titles = list(args.file)
    if args.category:
        found = category_titles(args.category, args.limit * 4)
        print(f"{len(found)} file(s) in {args.category}")
        titles += [t for t in found if t not in titles]
    if args.query:
        found, _ = search_titles(args.query, args.limit * 3)
        titles += [t for t in found if t not in titles]

    if not titles:
        print("no candidate files matched")
        return 1

    print(f"{len(titles)} candidate file(s)")
    if args.dry_run:
        for title in titles[: args.limit]:
            print(f"  {title}")
        return 0

    out_dir = os.path.join(args.out, args.slug)
    os.makedirs(out_dir, exist_ok=True)

    selected = image_info(titles[: args.limit], args.width)
    if not selected:
        print("no downloadable candidates")
        return 1

    records = []
    for index, item in enumerate(selected, start=1):
        name = safe_name(item["title"], index)
        path = os.path.join(out_dir, name)
        print(f"  [{index}/{len(selected)}] {item['title']}")
        try:
            digest = download(item["download_url"], path)
        except SystemExit as exc:
            # one throttled file must not discard the ones already fetched
            print(f"    skipped: {exc}", flush=True)
            time.sleep(args.delay * 5)
            continue
        item.update(
            {
                "local_file": name,
                "sha256": digest,
                "bytes": os.path.getsize(path),
            }
        )
        records.append(item)
        time.sleep(args.delay)

    sidecar = {
        "fetched_utc": datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds"),
        "source": "Wikimedia Commons API",
        "api": API,
        "query": args.query,
        "category": args.category,
        "requested_files": args.file,
        "slug": args.slug,
        "thumb_width": args.width,
        "tool": "Tools/gen/commons_fetch.py",
        "note": "local development reference only; not committed to the repository",
        "images": records,
    }
    sidecar_path = os.path.join(out_dir, "sidecar.json")
    with open(sidecar_path, "w", encoding="utf-8") as handle:
        json.dump(sidecar, handle, indent=2, ensure_ascii=False)

    print(f"\nwrote {len(records)} image(s) + sidecar.json to {out_dir}")
    for item in records:
        print(f"  {item['local_file']}  {item['licence'] or 'licence unknown'}  {item['artist'][:40]}")
    return 0


if __name__ == "__main__":
    sys.exit(main())