177 lines
6.6 KiB
Python
177 lines
6.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Local-only reference collector — licence-free, gitignored research material.
|
|
|
|
Downloads images from DuckDuckGo image search into `tmp/research/<slug>/` for
|
|
development reference only. This tool deliberately applies **no licence gate**:
|
|
everything it fetches stays under the gitignored `tmp/` tree and must never be
|
|
committed, shipped, or promoted into `docs/screens/refs/` without passing the
|
|
normal sidecar licence + privacy gate. The owner authorised this local-only
|
|
override on 2026-11-03 (see TODO.md §3D model re-evaluation → A).
|
|
|
|
python3 tools/refcollect.py search "El Capitan Entertainment Centre facade" \
|
|
--slug elcapitan --limit 20
|
|
python3 tools/refcollect.py sheet --slug elcapitan
|
|
|
|
Requires the `duckduckgo_search` package (pip install duckduckgo-search, or the
|
|
renamed `ddgs`). DuckDuckGo rate-limits aggressive queries (HTTP 202): requests
|
|
are spaced and retried with backoff, so a large batch can take a while.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
RESEARCH_DIR = ROOT / "tmp" / "research"
|
|
UA = "Mozilla/5.0 (X11; Linux x86_64; rv:120.0) Gecko/20100101 Firefox/120.0"
|
|
DOWNLOAD_SPACING_S = 2.0
|
|
|
|
|
|
def _rel(path: Path) -> str:
|
|
return str(path.relative_to(ROOT)) if path.is_relative_to(ROOT) else str(path)
|
|
|
|
|
|
def _slug(text: str) -> str:
|
|
return re.sub(r"[^a-z0-9]+", "-", text.strip().lower()).strip("-") or "block"
|
|
|
|
|
|
def _search_images(query: str, limit: int) -> list[dict]:
|
|
"""Return DDG image results as small dicts; retries on 202 rate limits."""
|
|
try:
|
|
from duckduckgo_search import DDGS # legacy import name
|
|
except ImportError:
|
|
try:
|
|
from ddgs import DDGS # current package name
|
|
except ImportError as exc:
|
|
sys.exit("error: install the DuckDuckGo search package first: "
|
|
"pip install duckduckgo-search")
|
|
last_exc = None
|
|
for attempt in range(5):
|
|
try:
|
|
with DDGS() as ddgs:
|
|
rows = ddgs.images(query, max_results=limit)
|
|
return [{
|
|
"title": (r.get("title") or "")[:200],
|
|
"image": r.get("image") or "",
|
|
"thumbnail": r.get("thumbnail") or "",
|
|
"source": r.get("url") or "",
|
|
"width": r.get("width") or 0,
|
|
"height": r.get("height") or 0,
|
|
} for r in rows if r.get("image")]
|
|
except Exception as exc: # noqa: BLE001 - package raises several types
|
|
last_exc = exc
|
|
wait = 8 * (attempt + 1)
|
|
print(f" DDG rate limit or error ({exc.__class__.__name__}); "
|
|
f"retrying in {wait}s ...", file=sys.stderr)
|
|
time.sleep(wait)
|
|
sys.exit(f"error: DuckDuckGo image search kept failing: {last_exc}")
|
|
|
|
|
|
def _download(url: str, path: Path) -> None:
|
|
req = urllib.request.Request(url, headers={"User-Agent": UA})
|
|
with urllib.request.urlopen(req, timeout=60) as resp:
|
|
path.write_bytes(resp.read())
|
|
|
|
|
|
def _manifest_rows(rows: list[dict]) -> list[dict]:
|
|
out = []
|
|
for row in rows:
|
|
out.append({**row, "local_only": True, "licence": "unchecked-local"})
|
|
return out
|
|
|
|
|
|
def cmd_search(args) -> int:
|
|
slug = _slug(args.slug or args.query)
|
|
out_dir = RESEARCH_DIR / slug
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
print(f"searching DDG: {args.query!r} (limit {args.limit})", flush=True)
|
|
rows = _search_images(args.query, args.limit)
|
|
manifest = []
|
|
for index, row in enumerate(rows, start=1):
|
|
stem = f"{index:02d}"
|
|
suffix = Path(row["image"].split("?")[0]).suffix.lower() or ".jpg"
|
|
if suffix not in (".jpg", ".jpeg", ".png", ".webp", ".gif"):
|
|
suffix = ".jpg"
|
|
path = out_dir / f"{stem}{suffix}"
|
|
if path.exists() and not args.force:
|
|
print(f" skip existing {path.name}")
|
|
else:
|
|
try:
|
|
_download(row["image"], path)
|
|
print(f" {path.name} <- {row['image'][:70]}")
|
|
except Exception as exc: # noqa: BLE001 - some hosts block hotlinking
|
|
print(f" FAILED {path.name}: {exc.__class__.__name__}", file=sys.stderr)
|
|
continue
|
|
time.sleep(DOWNLOAD_SPACING_S)
|
|
manifest.append({**row, "file": path.name})
|
|
(out_dir / "manifest.json").write_text(
|
|
json.dumps(_manifest_rows(manifest), indent=2) + "\n")
|
|
print(f"manifest -> {out_dir / 'manifest.json'} ({len(manifest)} entries)")
|
|
return 0
|
|
|
|
|
|
def cmd_sheet(args) -> int:
|
|
"""Compose a local contact sheet (never committed) from a manifest folder."""
|
|
from PIL import Image
|
|
slug = _slug(args.slug)
|
|
out_dir = RESEARCH_DIR / slug
|
|
manifest_path = out_dir / "manifest.json"
|
|
if not manifest_path.is_file():
|
|
sys.exit(f"error: no manifest at {manifest_path}; run `search` first")
|
|
rows = json.loads(manifest_path.read_text())
|
|
images = []
|
|
for row in rows:
|
|
path = out_dir / row["file"]
|
|
if path.is_file():
|
|
with Image.open(path) as img:
|
|
img.load()
|
|
images.append(img.convert("RGB"))
|
|
if not images:
|
|
sys.exit("error: no downloaded frames in this folder")
|
|
cell = 320
|
|
cols = 4
|
|
rows_count = (len(images) + cols - 1) // cols
|
|
sheet = Image.new("RGB", (cols * cell, rows_count * cell), (18, 18, 18))
|
|
for i, img in enumerate(images):
|
|
thumb = img.copy()
|
|
thumb.thumbnail((cell, cell), Image.Resampling.LANCZOS)
|
|
x = (i % cols) * cell
|
|
y = (i // cols) * cell
|
|
sheet.paste(thumb, (x, y))
|
|
out = out_dir / "sheet.jpg"
|
|
sheet.save(out, quality=82)
|
|
print(f"wrote {_rel(out)} {sheet.size[0]}x{sheet.size[1]} "
|
|
f"({len(images)} frames)")
|
|
return 0
|
|
|
|
|
|
def build_parser():
|
|
parser = argparse.ArgumentParser(
|
|
description=__doc__,
|
|
formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
sub = parser.add_subparsers(dest="cmd", required=True)
|
|
search = sub.add_parser("search", help="DDG image search into tmp/research/<slug>/")
|
|
search.add_argument("query")
|
|
search.add_argument("--slug", default="", help="folder slug (default: from query)")
|
|
search.add_argument("--limit", type=int, default=20)
|
|
search.add_argument("--force", action="store_true")
|
|
search.set_defaults(handler=cmd_search)
|
|
sheet = sub.add_parser("sheet", help="compose a local contact sheet")
|
|
sheet.add_argument("--slug", required=True)
|
|
sheet.set_defaults(handler=cmd_sheet)
|
|
return parser
|
|
|
|
|
|
def main(argv=None) -> int:
|
|
args = build_parser().parse_args(argv)
|
|
return args.handler(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|