Add the local-only DuckDuckGo reference collector

tools/refcollect.py

@@ -0,0 +1,176 @@
+#!/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())