Harden the DDG collector with token-bucket pacing and prefer the ddgs package

tools/refcollect.py

@@ -12,24 +12,221 @@ override on 2026-11-03 (see TODO.md §3D model re-evaluation → A).
         --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.
+Requires the `ddgs` package (`pip install ddgs`); the legacy
+`duckduckgo_search` package still works as a fallback.
+
+Every outbound request goes through a token bucket and a polite retry routine:
+DuckDuckGo and the image hosts rate-limit aggressive clients, so a large batch
+is deliberately slow. Tune the pace with `--rate` (requests per second) and
+`--burst`.
 """
 
+from __future__ import annotations
+
 import argparse
+import datetime
+import email.utils
+import gzip
 import json
+import random
 import re
+import socket
+import ssl
 import sys
+import threading
 import time
 import urllib.error
 import urllib.request
+import zlib
 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
+
+# --- politeness ------------------------------------------------------------
+
+# A current desktop Firefox on Linux. Kept in one place so the search client and
+# the downloader present the same, consistent identity.
+USER_AGENT = ("Mozilla/5.0 (X11; Linux x86_64; rv:159.0) "
+              "Gecko/20100101 Firefox/159.0")
+
+# A coherent Firefox header set. Servers increasingly look past the user agent;
+# a plain UA with no matching Accept/Accept-Language/Sec-Fetch headers is itself
+# a tell, so send the whole set the browser would send.
+#
+# Accept-Encoding deliberately stops at gzip and deflate: urllib cannot decode
+# brotli or zstd, and advertising an encoding we cannot read would corrupt every
+# download. `_decode_body` undoes the two encodings we do advertise.
+BROWSER_HEADERS = {
+    "User-Agent": USER_AGENT,
+    "Accept": ("image/avif,image/webp,image/png,image/svg+xml,image/*;q=0.8,"
+               "*/*;q=0.5"),
+    "Accept-Language": "en-US,en;q=0.9,de-CH;q=0.8,de;q=0.7",
+    "Accept-Encoding": "gzip, deflate",
+    "DNT": "1",
+    "Sec-Fetch-Dest": "image",
+    "Sec-Fetch-Mode": "no-cors",
+    "Sec-Fetch-Site": "cross-site",
+    "Sec-GPC": "1",
+    "Referer": "https://duckduckgo.com/",
+    "Upgrade-Insecure-Requests": "1",
+}
+
+DEFAULT_RATE = 0.5        # requests per second (one every two seconds)
+DEFAULT_BURST = 3.0       # instant burst a cold start may spend
+BACKOFF_BASE_S = 4.0      # first exponential-backoff ceiling
+BACKOFF_CAP_S = 120.0     # never wait longer than two minutes
+MAX_ATTEMPTS = 6
+REQUEST_TIMEOUT_S = 60.0
+RETRY_STATUS = frozenset(
+    {408, 425, 429, 500, 502, 503, 504, 520, 521, 522, 524})
+
+
+class TokenBucket:
+    """Token bucket that paces outbound requests instead of fixed sleeps.
+
+    Tokens accrue at ``rate`` per second up to ``capacity``. Each request spends
+    one token; when the bucket is empty the caller sleeps only as long as it
+    takes for the next token to accrue. A burst allowance lets a cold start fire
+    a few requests back to back, which is what a human opening a page looks
+    like, while the sustained rate stays gentle.
+    """
+
+    def __init__(self, rate: float, capacity: float | None = None) -> None:
+        if rate <= 0:
+            raise ValueError("rate must be positive")
+        self.rate = float(rate)
+        self.capacity = float(
+            capacity if capacity is not None else max(1.0, rate))
+        if self.capacity < 1.0:
+            raise ValueError("capacity must hold at least one token")
+        self._tokens = self.capacity
+        self._stamp = time.monotonic()
+        self._lock = threading.Lock()
+
+    def _refill(self) -> None:
+        now = time.monotonic()
+        self._tokens = min(self.capacity,
+                           self._tokens + (now - self._stamp) * self.rate)
+        self._stamp = now
+
+    @property
+    def tokens(self) -> float:
+        with self._lock:
+            self._refill()
+            return self._tokens
+
+    def acquire(self, tokens: float = 1.0) -> float:
+        """Block until ``tokens`` are available and return the seconds waited."""
+        if tokens > self.capacity:
+            raise ValueError("requested tokens exceed bucket capacity")
+        waited = 0.0
+        while True:
+            with self._lock:
+                self._refill()
+                if self._tokens >= tokens:
+                    self._tokens -= tokens
+                    return waited
+                delay = (tokens - self._tokens) / self.rate
+            time.sleep(delay)
+            waited += delay
+
+
+def _parse_retry_after(value: str | None) -> float | None:
+    """Read a Retry-After header (delta seconds or HTTP date) as seconds."""
+    if not value:
+        return None
+    value = value.strip()
+    if value.isdigit():
+        return float(value)
+    try:
+        when = email.utils.parsedate_to_datetime(value)
+    except (TypeError, ValueError):
+        return None
+    if when is None:
+        return None
+    if when.tzinfo is None:
+        when = when.replace(tzinfo=datetime.timezone.utc)
+    delta = (when - datetime.datetime.now(datetime.timezone.utc)).total_seconds()
+    return max(0.0, delta)
+
+
+def _backoff_delay(attempt: int, retry_after: float | None = None) -> float:
+    """Full-jitter exponential backoff, capped, deferring to Retry-After.
+
+    ``attempt`` starts at 1. When the host told us how long to wait, that wins,
+    so we never come back sooner than asked. Otherwise pick uniformly from the
+    top half of the exponential window: enough spread to avoid a thundering herd
+    of retries, without doubling the total wait time.
+    """
+    if retry_after is not None:
+        return min(BACKOFF_CAP_S, max(1.0, retry_after))
+    ceiling = min(BACKOFF_CAP_S, BACKOFF_BASE_S * 2 ** (attempt - 1))
+    return random.uniform(ceiling / 2.0, ceiling)
+
+
+def _is_permanent_network_error(exc: BaseException) -> bool:
+    """True when retrying cannot help: bad DNS or an untrusted certificate."""
+    reason = getattr(exc, "reason", None) or exc
+    return isinstance(reason, (socket.gaierror, ssl.SSLCertVerificationError))
+
+
+def _decode_body(body: bytes, content_encoding: str | None) -> bytes:
+    """Undo a Content-Encoding we advertised in BROWSER_HEADERS."""
+    encoding = (content_encoding or "").lower().strip()
+    if encoding in ("gzip", "x-gzip"):
+        return gzip.decompress(body)
+    if encoding == "deflate":
+        try:
+            return zlib.decompress(body)
+        except zlib.error:
+            return zlib.decompress(body, -zlib.MAX_WBITS)  # raw deflate
+    return body
+
+
+def polite_get(url: str, *, bucket: TokenBucket, headers: dict | None = None,
+               attempts: int = MAX_ATTEMPTS, timeout: float = REQUEST_TIMEOUT_S,
+               label: str = "request") -> tuple[bytes, object]:
+    """Fetch ``url`` politely and return ``(body, response headers)``.
+
+    Each attempt takes a token from ``bucket`` first. Transient failures are
+    retried with jittered exponential backoff, honouring Retry-After; permanent
+    failures raise immediately so a dead link does not burn the whole budget.
+    """
+    request_headers = dict(BROWSER_HEADERS)
+    if headers:
+        request_headers.update(headers)
+
+    last_exc: BaseException | None = None
+    for attempt in range(1, attempts + 1):
+        bucket.acquire()
+        retry_after = None
+        try:
+            request = urllib.request.Request(url, headers=request_headers)
+            with urllib.request.urlopen(request, timeout=timeout) as response:
+                body = _decode_body(response.read(),
+                                    response.headers.get("Content-Encoding"))
+                return body, response.headers
+        except urllib.error.HTTPError as exc:
+            if exc.code not in RETRY_STATUS:
+                raise
+            last_exc = exc
+            retry_after = _parse_retry_after(exc.headers.get("Retry-After"))
+            reason = f"HTTP {exc.code} {exc.reason or ''}".strip()
+        except (urllib.error.URLError, OSError, ValueError) as exc:
+            if _is_permanent_network_error(exc):
+                raise
+            last_exc = exc
+            reason = exc.__class__.__name__
+        if attempt == attempts:
+            break
+        delay = _backoff_delay(attempt, retry_after)
+        print(f"  {label}: {reason}; politely waiting {delay:.1f}s "
+              f"before attempt {attempt + 1}/{attempts}",
+              file=sys.stderr, flush=True)
+        time.sleep(delay)
+    raise last_exc if last_exc is not None else RuntimeError(
+        f"could not fetch {url}")
 
 
 def _rel(path: Path) -> str:
@@ -40,18 +237,26 @@ 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."""
+def _search_images(query: str, limit: int, bucket: TokenBucket) -> list[dict]:
+    """Return DDG image results as small dicts; polite retry on rate limits."""
     try:
-        from duckduckgo_search import DDGS  # legacy import name
+        from ddgs import DDGS  # current package name
     except ImportError:
+        # Legacy `duckduckgo_search` still works but warns it was renamed to
+        # `ddgs`; silence that warning so it does not leak into the run log.
+        import warnings
         try:
-            from ddgs import DDGS  # current package name
+            with warnings.catch_warnings():
+                warnings.filterwarnings(
+                    "ignore", category=RuntimeWarning,
+                    message=r".*renamed to .ddgs.*")
+                from duckduckgo_search import DDGS  # legacy import name
         except ImportError as exc:
             sys.exit("error: install the DuckDuckGo search package first: "
-                     "pip install duckduckgo-search")
+                     "pip install ddgs")
     last_exc = None
-    for attempt in range(5):
+    for attempt in range(1, MAX_ATTEMPTS + 1):
+        bucket.acquire()
         try:
             with DDGS() as ddgs:
                 rows = ddgs.images(query, max_results=limit)
@@ -65,17 +270,20 @@ def _search_images(query: str, limit: int) -> list[dict]:
             } 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)
+            if attempt == MAX_ATTEMPTS:
+                break
+            delay = _backoff_delay(attempt)
+            print(f"  DDG search failed ({exc.__class__.__name__}); politely "
+                  f"waiting {delay:.1f}s before attempt {attempt + 1}/"
+                  f"{MAX_ATTEMPTS}", file=sys.stderr, flush=True)
+            time.sleep(delay)
     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 _download(url: str, path: Path, bucket: TokenBucket) -> None:
+    """Download one image through the polite, rate-limited fetch routine."""
+    body, _ = polite_get(url, bucket=bucket, label=f"download {path.name}")
+    path.write_bytes(body)
 
 
 def _manifest_rows(rows: list[dict]) -> list[dict]:
@@ -89,8 +297,12 @@ 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)
+    bucket = TokenBucket(args.rate, args.burst)
+    if getattr(args, "user_agent", ""):
+        BROWSER_HEADERS["User-Agent"] = args.user_agent
+    print(f"searching DDG: {args.query!r} (limit {args.limit}, "
+          f"{args.rate:g} req/s, burst {args.burst:g})", flush=True)
+    rows = _search_images(args.query, args.limit, bucket)
     manifest = []
     for index, row in enumerate(rows, start=1):
         stem = f"{index:02d}"
@@ -102,12 +314,12 @@ def cmd_search(args) -> int:
             print(f"  skip existing {path.name}")
         else:
             try:
-                _download(row["image"], path)
+                _download(row["image"], path, bucket)
                 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)
+                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")
@@ -155,11 +367,21 @@ def build_parser():
         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 = 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("--slug", default="",
+                        help="folder slug (default: from query)")
     search.add_argument("--limit", type=int, default=20)
     search.add_argument("--force", action="store_true")
+    search.add_argument("--rate", type=float, default=DEFAULT_RATE,
+                        help="requests per second, enforced by a token bucket "
+                             f"(default {DEFAULT_RATE:g})")
+    search.add_argument("--burst", type=float, default=DEFAULT_BURST,
+                        help="token-bucket capacity, i.e. allowed instant burst "
+                             f"(default {DEFAULT_BURST:g})")
+    search.add_argument("--user-agent", default="",
+                        help="override the default Firefox user agent string")
     search.set_defaults(handler=cmd_search)
     sheet = sub.add_parser("sheet", help="compose a local contact sheet")
     sheet.add_argument("--slug", required=True)

tools/test_refcollect.py

@@ -1,7 +1,12 @@
 """Tests for the local-only reference collector (no network)."""
 
+import datetime
+import email.utils
+import gzip
 import json
+import socket
 import unittest
+import urllib.error
 from pathlib import Path
 from tempfile import TemporaryDirectory
 from unittest import mock
@@ -9,6 +14,37 @@ from unittest import mock
 import refcollect as rc
 
 
+def _fake_clock(start=0.0):
+    """Return (clock, monotonic, sleep) that advance virtual time on sleep."""
+    clock = {"now": start}
+
+    def monotonic():
+        return clock["now"]
+
+    def sleep(seconds):
+        clock["now"] += seconds
+
+    return clock, monotonic, sleep
+
+
+class _FakeResponse:
+    """Minimal stand-in for the urlopen context manager."""
+
+    def __init__(self, body=b"x", content_encoding=None):
+        self._body = body
+        self.headers = ({"Content-Encoding": content_encoding}
+                        if content_encoding else {})
+
+    def read(self):
+        return self._body
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc):
+        return False
+
+
 class SlugTests(unittest.TestCase):
     def test_slug_folds_and_strips(self):
         self.assertEqual(rc._slug("El Capitan Entertainment Centre"),
@@ -39,7 +75,8 @@ class SheetTests(unittest.TestCase):
             slug_dir = root / "block"
             slug_dir.mkdir()
             for i in range(3):
-                Image.new("RGB", (640, 400), (i * 60, 0, 0)).save(slug_dir / f"{i:02d}.jpg")
+                Image.new("RGB", (640, 400), (i * 60, 0, 0)
+                          ).save(slug_dir / f"{i:02d}.jpg")
             (slug_dir / "manifest.json").write_text(json.dumps([
                 {"file": "00.jpg"}, {"file": "01.jpg"}, {"file": "02.jpg"}]))
             with mock.patch.object(rc, "RESEARCH_DIR", root):
@@ -47,5 +84,131 @@ class SheetTests(unittest.TestCase):
             self.assertTrue((slug_dir / "sheet.jpg").is_file())
 
 
+class TokenBucketTests(unittest.TestCase):
+    def test_burst_is_spent_before_sleeping(self):
+        clock, monotonic, sleep = _fake_clock()
+        with mock.patch.object(rc.time, "monotonic", monotonic), \
+                mock.patch.object(rc.time, "sleep", sleep):
+            bucket = rc.TokenBucket(rate=1.0, capacity=2.0)
+            self.assertEqual(bucket.acquire(), 0.0)
+            self.assertEqual(bucket.acquire(), 0.0)
+            self.assertAlmostEqual(bucket.acquire(), 1.0)
+            self.assertAlmostEqual(bucket.acquire(), 1.0)
+        self.assertAlmostEqual(clock["now"], 2.0)
+
+    def test_tokens_refill_up_to_capacity_only(self):
+        clock, monotonic, sleep = _fake_clock()
+        with mock.patch.object(rc.time, "monotonic", monotonic), \
+                mock.patch.object(rc.time, "sleep", sleep):
+            bucket = rc.TokenBucket(rate=1.0, capacity=2.0)
+            bucket.acquire(2.0)
+            self.assertAlmostEqual(bucket.tokens, 0.0)
+            clock["now"] += 100.0
+            self.assertAlmostEqual(bucket.tokens, 2.0)  # capped at capacity
+            clock["now"] += 100.0
+            self.assertAlmostEqual(bucket.tokens, 2.0)
+
+    def test_invalid_configuration_is_rejected(self):
+        with self.assertRaises(ValueError):
+            rc.TokenBucket(rate=0)
+        with self.assertRaises(ValueError):
+            rc.TokenBucket(rate=1.0, capacity=0.5)
+
+
+class BackoffTests(unittest.TestCase):
+    def test_backoff_grows_and_stays_within_its_window(self):
+        for attempt in range(1, 6):
+            ceiling = min(rc.BACKOFF_CAP_S, rc.BACKOFF_BASE_S *
+                          2 ** (attempt - 1))
+            for _ in range(20):
+                delay = rc._backoff_delay(attempt)
+                self.assertGreaterEqual(delay, ceiling / 2)
+                self.assertLessEqual(delay, ceiling)
+
+    def test_backoff_is_capped(self):
+        self.assertLessEqual(rc._backoff_delay(20), rc.BACKOFF_CAP_S)
+
+    def test_retry_after_overrides_the_exponential_guess(self):
+        self.assertEqual(rc._backoff_delay(1, retry_after=30.0), 30.0)
+        self.assertEqual(rc._backoff_delay(
+            1, retry_after=0.0), 1.0)  # never instant
+        self.assertEqual(rc._backoff_delay(
+            1, retry_after=10_000.0), rc.BACKOFF_CAP_S)
+
+    def test_retry_after_parses_seconds_and_http_dates(self):
+        self.assertEqual(rc._parse_retry_after("7"), 7.0)
+        self.assertIsNone(rc._parse_retry_after(None))
+        self.assertIsNone(rc._parse_retry_after("soon"))
+        when = email.utils.format_datetime(
+            datetime.datetime.now(datetime.timezone.utc)
+            + datetime.timedelta(seconds=30))
+        self.assertAlmostEqual(rc._parse_retry_after(when), 30.0, delta=2.0)
+
+
+class PolitenessTests(unittest.TestCase):
+    def test_user_agent_is_the_current_firefox_string_on_linux(self):
+        self.assertEqual(rc.BROWSER_HEADERS["User-Agent"], rc.USER_AGENT)
+        self.assertIn("X11; Linux x86_64", rc.USER_AGENT)
+        self.assertIn("Firefox/", rc.USER_AGENT)
+        # A UA on its own is a tell: the supporting browser headers must match.
+        for header in ("Accept", "Accept-Language", "Sec-Fetch-Dest", "Referer"):
+            self.assertIn(header, rc.BROWSER_HEADERS)
+
+    def test_gzip_bodies_are_decoded(self):
+        payload = gzip.compress(b"jpeg-bytes")
+        self.assertEqual(rc._decode_body(payload, "gzip"), b"jpeg-bytes")
+        self.assertEqual(rc._decode_body(b"plain", None), b"plain")
+
+    def test_polite_get_retries_transient_errors_then_succeeds(self):
+        error = urllib.error.HTTPError("https://x/i.jpg", 429,
+                                       "Too Many Requests",
+                                       {"Retry-After": "3"}, None)
+        bucket = rc.TokenBucket(rate=1000, capacity=1000)
+        with mock.patch.object(rc.urllib.request, "urlopen",
+                               side_effect=[error, _FakeResponse(b"jpeg")]) as opened, \
+                mock.patch.object(rc.time, "sleep") as sleep:
+            body, _ = rc.polite_get("https://x/i.jpg", bucket=bucket)
+        self.assertEqual(body, b"jpeg")
+        self.assertEqual(opened.call_count, 2)
+        # Retry-After is honoured instead of the exponential guess.
+        sleep.assert_called_once_with(3.0)
+        sent = opened.call_args_list[0][0][0]
+        self.assertEqual(sent.get_header("User-agent"), rc.USER_AGENT)
+
+    def test_polite_get_does_not_retry_permanent_errors(self):
+        error = urllib.error.HTTPError(
+            "https://x/i.jpg", 404, "Not Found", {}, None)
+        bucket = rc.TokenBucket(rate=1000, capacity=1000)
+        with mock.patch.object(rc.urllib.request, "urlopen",
+                               side_effect=error) as opened, \
+                mock.patch.object(rc.time, "sleep") as sleep, \
+                self.assertRaises(urllib.error.HTTPError):
+            rc.polite_get("https://x/i.jpg", bucket=bucket)
+        self.assertEqual(opened.call_count, 1)
+        sleep.assert_not_called()
+
+    def test_polite_get_does_not_retry_dns_failures(self):
+        error = urllib.error.URLError(socket.gaierror(-2, "Name not known"))
+        bucket = rc.TokenBucket(rate=1000, capacity=1000)
+        with mock.patch.object(rc.urllib.request, "urlopen",
+                               side_effect=error) as opened, \
+                mock.patch.object(rc.time, "sleep") as sleep, \
+                self.assertRaises(urllib.error.URLError):
+            rc.polite_get("https://nope.invalid/i.jpg", bucket=bucket)
+        self.assertEqual(opened.call_count, 1)
+        sleep.assert_not_called()
+
+    def test_polite_get_gives_up_after_the_attempt_budget(self):
+        error = urllib.error.HTTPError(
+            "https://x/i.jpg", 503, "Unavailable", {}, None)
+        bucket = rc.TokenBucket(rate=1000, capacity=1000)
+        with mock.patch.object(rc.urllib.request, "urlopen",
+                               side_effect=error) as opened, \
+                mock.patch.object(rc.time, "sleep"), \
+                self.assertRaises(urllib.error.HTTPError):
+            rc.polite_get("https://x/i.jpg", bucket=bucket, attempts=3)
+        self.assertEqual(opened.call_count, 3)
+
+
 if __name__ == "__main__":
     unittest.main()