zwitschi
72de657202
Five usable sets added locally (hero daytime façades, the El Capitan Theatre block, night premiere staging, Ovation Hollywood, and the Lee Drug neighbour) — 41 images total, none committed. Records what remains missing: a sidewalk-level frame of the hero entry, the 6806-6822 retail run, and a night frame of 6840 itself rather than its neighbour. commons_fetch.py now survives a throttled download instead of discarding the rest of the batch, and takes --delay.
244 lines
8.7 KiB
Python
Executable File
244 lines
8.7 KiB
Python
Executable File
#!/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/commons_fetch.py --query "Grauman's Chinese Theatre" --slug north-chinese --dry-run
|
|
|
|
# download up to 6 images at 1600 px wide
|
|
python3 Tools/commons_fetch.py --query "Grauman's Chinese Theatre" --slug north-chinese --limit 6
|
|
|
|
# fetch specific files by title
|
|
python3 Tools/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 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("--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:
|
|
ap.error("need --query or --file")
|
|
|
|
titles = list(args.file)
|
|
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,
|
|
"requested_files": args.file,
|
|
"slug": args.slug,
|
|
"thumb_width": args.width,
|
|
"tool": "Tools/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())
|