Cache postcodes

finder/constants.py

@@ -15,6 +15,16 @@ DELAY_BETWEEN_PAGES = 0.3
 DELAY_BETWEEN_OUTCODES = 0.5
 MAX_RETRIES = 3
 RETRY_BASE_DELAY = 2.0
+
+# Concurrency + global rate limiting for the HTTP-based scrapers (Rightmove and
+# OnTheMarket). Detail-page fetches dominate runtime and are independent,
+# idempotent GETs, so they are fetched concurrently rather than one-at-a-time.
+# A single shared token-bucket limiter (http_client.RATE_LIMITER) caps the
+# COMBINED request rate across every worker thread and both providers, so the
+# VPN egress IP stays polite no matter how high the concurrency is. Tune both
+# down if the portals start returning 429/403.
+DETAIL_FETCH_CONCURRENCY = int(os.environ.get("DETAIL_FETCH_CONCURRENCY", "8"))
+REQUESTS_PER_SECOND = float(os.environ.get("REQUESTS_PER_SECOND", "10"))
 GRID_CELL_SIZE = 0.01  # degrees for postcode spatial index
 MAX_BEDROOMS = 20  # sanity cap — values above this are almost certainly parsing errors
 
@@ -35,6 +45,21 @@ RIGHTMOVE_DETAIL_URL = "https://www.rightmove.co.uk/properties/{id}"
 RIGHTMOVE_FETCH_DETAILS = True  # fetch detail pages for true per-listing postcodes
 RIGHTMOVE_MAX_DETAILS_PER_OUTCODE = 4000  # max detail-page fetches per outcode
 
+# Skip the detail fetch when the search result already pins the property
+# precisely. Each search-result `location` carries a `pinType`: for
+# "ACCURATE_POINT" listings the coordinates are rooftop-exact, so the
+# coordinate-nearest postcode already has the right outcode (and almost always
+# the right unit) and the detail page adds little. The detail fetch earns its
+# keep on APPROXIMATE pins (new-builds/developments) where Rightmove
+# deliberately fuzzes the coordinates. Degrades safely: when `pinType` is absent
+# from the search payload, nothing is skipped (behaviour is unchanged), so this
+# is only a speed-up to the extent the field is present — verify against a live
+# search response before relying on the saving.
+RIGHTMOVE_SKIP_DETAILS_FOR_ACCURATE_PINS = (
+    os.environ.get("RIGHTMOVE_SKIP_DETAILS_FOR_ACCURATE_PINS", "1") != "0"
+)
+RIGHTMOVE_ACCURATE_PIN_TYPE = "ACCURATE_POINT"
+
 # OnTheMarket
 ONTHEMARKET_BASE = "https://www.onthemarket.com"
 

finder/http_client.py

@@ -1,14 +1,56 @@
 import logging
 import random
+import threading
 import time
 
 import httpx
 from fake_useragent import UserAgent
 
-from constants import GLUETUN_PROXY, MAX_RETRIES, RETRY_BASE_DELAY
+import shutdown
+from constants import (
+    GLUETUN_PROXY,
+    MAX_RETRIES,
+    REQUESTS_PER_SECOND,
+    RETRY_BASE_DELAY,
+)
 
 log = logging.getLogger("rightmove")
 
+
+class RateLimiter:
+    """Thread-safe global limiter: spaces request starts by a minimum interval.
+
+    Detail-page fetches run concurrently across many worker threads (and across
+    providers), but a single shared limiter caps their COMBINED rate so the VPN
+    egress IP stays polite. Each ``acquire()`` reserves the next free time slot
+    under a lock, then sleeps (outside the lock) until that slot — so N threads
+    calling concurrently are spaced ``1/rate_per_second`` apart rather than all
+    firing at once. ``rate_per_second <= 0`` disables limiting."""
+
+    def __init__(self, rate_per_second: float):
+        self._interval = 1.0 / rate_per_second if rate_per_second > 0 else 0.0
+        self._lock = threading.Lock()
+        self._next = 0.0
+
+    def acquire(self) -> None:
+        if self._interval <= 0:
+            return
+        with self._lock:
+            now = time.monotonic()
+            if now >= self._next:
+                self._next = now + self._interval
+                wait = 0.0
+            else:
+                wait = self._next - now
+                self._next += self._interval
+        if wait > 0:
+            shutdown.sleep(wait)
+
+
+# Shared by every HTTP-based fetch (search pages and detail pages, all
+# providers). Spacing is global, so politeness is decoupled from concurrency.
+RATE_LIMITER = RateLimiter(REQUESTS_PER_SECOND)
+
 _ua = UserAgent(
     browsers=["Chrome", "Edge"], os=["Windows", "Mac OS X"], min_version=120.0
 )
@@ -33,7 +75,10 @@ def fetch_with_retry(
     compatibility with older callers; 403 is now treated as non-retryable.
     """
     for attempt in range(MAX_RETRIES):
+        if shutdown.stop_requested():
+            return None
         try:
+            RATE_LIMITER.acquire()
             resp = client.get(url, params=params)
             if resp.status_code == 200:
                 return resp.json()
@@ -50,7 +95,7 @@ def fetch_with_retry(
                     MAX_RETRIES,
                     delay,
                 )
-                time.sleep(delay)
+                shutdown.sleep(delay)
                 continue
             log.error("HTTP %d from %s (non-retryable)", resp.status_code, url)
             return None
@@ -69,6 +114,6 @@ def fetch_with_retry(
                 MAX_RETRIES,
                 delay,
             )
-            time.sleep(delay)
+            shutdown.sleep(delay)
     log.error("All %d retries exhausted for %s", MAX_RETRIES, url)
     return None

finder/main.py

@@ -5,10 +5,12 @@ import tempfile
 import time
 from pathlib import Path
 
+import shutdown
 from constants import DATA_DIR, REPO_DIR
 
 
-SOURCE_CHOICES = ("rightmove", "onthemarket", "zoopla", "all")
+SOURCES = ("rightmove", "onthemarket", "zoopla")
+SOURCE_CHOICES = (*SOURCES, "all")
 TEST_MAX_PROPERTIES_PER_SOURCE = 100
 TEST_OUTCODES = (
     "E1",
@@ -47,9 +49,13 @@ def parse_args() -> argparse.Namespace:
     )
     parser.add_argument(
         "--source",
-        choices=SOURCE_CHOICES,
         default="all",
-        help="Portal to scrape. 'all' runs Rightmove, OnTheMarket, and Zoopla.",
+        metavar="SOURCES",
+        help=(
+            "Comma-separated portal(s) to scrape: any of "
+            f"{', '.join(SOURCES)}, or 'all' (default). "
+            "E.g. --source rightmove,onthemarket."
+        ),
     )
     parser.add_argument(
         "--output-dir",
@@ -100,15 +106,35 @@ def configure_logging() -> None:
 
 
 def selected_sources(source: str) -> list[str]:
-    if source == "all":
+    """Resolve --source: one or more comma-separated portals, or 'all'.
-        return ["rightmove", "onthemarket", "zoopla"]
+
-    return [source]
+    Accepts e.g. ``rightmove,onthemarket`` (whitespace and case tolerant).
+    Unknown values are rejected; the result is deduplicated and returned in the
+    canonical ``SOURCES`` order so downstream merge/dedup stays deterministic."""
+    requested = [part.strip().lower() for part in source.split(",")]
+    requested = [part for part in requested if part]
+    if not requested:
+        raise SystemExit("--source was empty")
+
+    unknown = sorted(set(requested) - set(SOURCE_CHOICES))
+    if unknown:
+        raise SystemExit(
+            f"Unknown --source value(s): {', '.join(unknown)}. "
+            f"Choose from {', '.join(SOURCE_CHOICES)} (comma-separated)."
+        )
+    if "all" in requested:
+        return list(SOURCES)
+    return [portal for portal in SOURCES if portal in requested]
 
 
 def main() -> int:
     args = parse_args()
     configure_standalone_runtime()
     configure_logging()
+    # Ctrl+C (and SIGTERM, e.g. `docker stop`) asks the scrapers to wind down
+    # gracefully — each source stops at its next outcode boundary and the run
+    # still persists detail caches and writes the listings collected so far.
+    shutdown.install_signal_handlers()
 
     if args.limit_outcodes is not None and args.limit_outcodes < 1:
         raise SystemExit("--limit-outcodes must be greater than zero")
@@ -182,6 +208,14 @@ def main() -> int:
     )
 
     elapsed = time.monotonic() - started
+    if shutdown.stop_requested():
+        log.warning(
+            "Scrape interrupted after %.1fs; partial results written to %s",
+            elapsed,
+            result.get("path"),
+        )
+        log.info("Result: %s", result)
+        return 130  # 128 + SIGINT, the conventional Ctrl+C exit code.
     log.info("Scrape finished in %.1fs", elapsed)
     log.info("Result: %s", result)
     if args.test and result.get("errors"):

finder/onthemarket.py

@@ -40,17 +40,19 @@ import json
 import logging
 import random
 import re
-import time
+from concurrent.futures import ThreadPoolExecutor
 
 import httpx
 
+import shutdown
 from constants import (
-    DELAY_BETWEEN_PAGES,
+    DETAIL_FETCH_CONCURRENCY,
     MAX_BEDROOMS,
     MAX_RETRIES,
     ONTHEMARKET_BASE,
     RETRY_BASE_DELAY,
 )
+from http_client import RATE_LIMITER
 from spatial import PostcodeSpatialIndex
 from transform import (
     clean_listing_address,
@@ -89,10 +91,24 @@ _HTML_HEADERS = {
 
 # listingId -> recovered full postcode (or None). Failures are cached too so a
 # broken or postcode-less detail page is not re-fetched within a run (the same
-# listing can reappear across overlapping outcode searches).
+# listing can reappear across overlapping outcode searches). Seeded from /
+# dumped to a persistent on-disk cache by the orchestrator (see
+# postcode_cache.py) so a recurring scrape only re-fetches newly-listed
+# properties rather than every listing every run.
 _detail_postcode_cache: dict[str, str | None] = {}
 
 
+def seed_detail_cache(mapping: dict) -> None:
+    """Pre-populate the in-memory detail cache from a persisted snapshot."""
+    if mapping:
+        _detail_postcode_cache.update(mapping)
+
+
+def detail_cache_snapshot() -> dict:
+    """Return a JSON-serialisable copy of the in-memory detail cache."""
+    return dict(_detail_postcode_cache)
+
+
 def _fetch_page_json(client: httpx.Client, outcode: str, page_num: int) -> dict | None:
     """GET one search-results page and return the embedded __NEXT_DATA__ JSON.
 
@@ -103,7 +119,10 @@ def _fetch_page_json(client: httpx.Client, outcode: str, page_num: int) -> dict
     params = {"page": str(page_num)} if page_num > 1 else None
 
     for attempt in range(MAX_RETRIES):
+        if shutdown.stop_requested():
+            return None
         try:
+            RATE_LIMITER.acquire()
             resp = client.get(
                 url,
                 params=params,
@@ -121,7 +140,7 @@ def _fetch_page_json(client: httpx.Client, outcode: str, page_num: int) -> dict
                 "%s from %s, retry %d/%d in %.1fs",
                 type(exc).__name__, url, attempt + 1, MAX_RETRIES, delay,
             )
-            time.sleep(delay)
+            shutdown.sleep(delay)
             continue
 
         if 300 <= resp.status_code < 400:
@@ -151,7 +170,7 @@ def _fetch_page_json(client: httpx.Client, outcode: str, page_num: int) -> dict
                 "HTTP %d from %s, retry %d/%d in %.1fs",
                 resp.status_code, url, attempt + 1, MAX_RETRIES, delay,
             )
-            time.sleep(delay)
+            shutdown.sleep(delay)
             continue
         log.error("HTTP %d from %s (non-retryable)", resp.status_code, url)
         return None
@@ -218,7 +237,8 @@ def _fetch_detail_postcode(
     reappears across overlapping outcode searches is fetched at most once. Plain
     HTTPS GET — OnTheMarket detail pages have no Cloudflare challenge. Network /
     parse errors degrade gracefully to None so the caller falls back to the
-    coordinate-nearest postcode.
+    coordinate-nearest postcode. Safe to call concurrently: distinct listing ids
+    write distinct cache keys, and the shared RATE_LIMITER spaces the GETs.
     """
     if listing_id in _detail_postcode_cache:
         return _detail_postcode_cache[listing_id]
@@ -231,7 +251,10 @@ def _fetch_detail_postcode(
     result: str | None = None
     if full_url:
         for attempt in range(MAX_RETRIES):
+            if shutdown.stop_requested():
+                break
             try:
+                RATE_LIMITER.acquire()
                 resp = client.get(
                     full_url, headers=_HTML_HEADERS, follow_redirects=True
                 )
@@ -246,7 +269,7 @@ def _fetch_detail_postcode(
                     "%s from %s, retry %d/%d in %.1fs",
                     type(exc).__name__, full_url, attempt + 1, MAX_RETRIES, delay,
                 )
-                time.sleep(delay)
+                shutdown.sleep(delay)
                 continue
 
             if resp.status_code == 200:
@@ -258,7 +281,7 @@ def _fetch_detail_postcode(
                     "HTTP %d from %s, retry %d/%d in %.1fs",
                     resp.status_code, full_url, attempt + 1, MAX_RETRIES, delay,
                 )
-                time.sleep(delay)
+                shutdown.sleep(delay)
                 continue
             log.debug(
                 "OnTheMarket detail %s returned HTTP %d (no postcode)",
@@ -422,6 +445,47 @@ def transform_property(
     }
 
 
+def _prime_detail_postcodes(
+    client: httpx.Client,
+    raw_listings: list[dict],
+    detail_cap: int,
+) -> None:
+    """Fill ``_detail_postcode_cache`` for the listings that need a detail page.
+
+    Picks the fresh (uncached) listings — up to ``detail_cap`` per outcode — then
+    fetches their detail pages CONCURRENTLY, bounded by
+    ``DETAIL_FETCH_CONCURRENCY`` (the shared RATE_LIMITER keeps the combined
+    request rate polite). Cached listings cost neither a slot nor a GET. The
+    worklist is deduplicated, so distinct ids write distinct cache keys and the
+    concurrent fetches never race on the same key."""
+    if not OTM_FETCH_DETAILS or detail_cap <= 0:
+        return
+
+    worklist: list[tuple[str, str]] = []  # (details_url, listing_id)
+    seen: set[str] = set()
+    for raw in raw_listings:
+        listing_id = str(raw.get("id") or "")
+        if not listing_id or listing_id in seen:
+            continue
+        seen.add(listing_id)
+        if listing_id in _detail_postcode_cache:
+            continue
+        worklist.append((raw.get("details-url") or "", listing_id))
+        if len(worklist) >= detail_cap:
+            break
+
+    if not worklist:
+        return
+
+    workers = min(DETAIL_FETCH_CONCURRENCY, len(worklist))
+    with ThreadPoolExecutor(max_workers=workers) as pool:
+        # Drain the iterator so every fetch runs and populates the cache here.
+        for _ in pool.map(
+            lambda item: _fetch_detail_postcode(client, item[0], item[1]), worklist
+        ):
+            pass
+
+
 def search_outcode(
     client: httpx.Client,
     outcode: str,
@@ -430,17 +494,19 @@ def search_outcode(
 ) -> list[dict]:
     """Paginate through OnTheMarket sale results for one outcode.
 
-    When ``OTM_FETCH_DETAILS`` is enabled, up to
+    Search pages are paginated serially (the RATE_LIMITER spaces the GETs); then,
-    ``OTM_MAX_DETAILS_PER_OUTCODE`` listings per outcode have their detail page
+    when ``OTM_FETCH_DETAILS`` is enabled, up to ``OTM_MAX_DETAILS_PER_OUTCODE``
-    fetched for the property's own postcode (see ``_fetch_detail_postcode``);
+    listings per outcode have their detail page fetched CONCURRENTLY for the
-    the rest fall back to the coordinate-nearest postcode.
+    property's own postcode (see ``_fetch_detail_postcode``); the rest fall back
+    to the coordinate-nearest postcode.
     """
-    properties: list[dict] = []
+    raw_listings: list[dict] = []
     seen_ids: set[str] = set()
     page_num = 1
-    details_fetched = 0
 
     while True:
+        if shutdown.stop_requested():
+            break
         data = _fetch_page_json(client, outcode, page_num)
         if data is None:
             break
@@ -453,47 +519,45 @@ def search_outcode(
             )
             break
 
-        raw_listings = state.get("list") or []
+        page_listings = state.get("list") or []
-        if not raw_listings:
+        if not page_listings:
             break
 
-        for raw in raw_listings:
+        for raw in page_listings:
             listing_id = str(raw.get("id") or "")
             if listing_id and listing_id in seen_ids:
                 continue
             seen_ids.add(listing_id)
+            raw_listings.append(raw)
 
-            detail_postcode = None
+        if max_properties is not None and len(raw_listings) >= max_properties:
-            if OTM_FETCH_DETAILS and listing_id:
+            break
-                # Cached lookups are free; only fresh GETs count toward the cap
-                # and incur the inter-request delay.
-                cached = listing_id in _detail_postcode_cache
-                if cached or details_fetched < OTM_MAX_DETAILS_PER_OUTCODE:
-                    detail_postcode = _fetch_detail_postcode(
-                        client, raw.get("details-url") or "", listing_id
-                    )
-                    if not cached:
-                        details_fetched += 1
-                        time.sleep(DELAY_BETWEEN_PAGES)
-
-            try:
-                transformed = transform_property(raw, pc_index, detail_postcode)
-            except Exception as exc:
-                log.warning(
-                    "OnTheMarket %s property %s failed to transform: %s",
-                    outcode, listing_id or "?", exc,
-                )
-                continue
-            if transformed:
-                properties.append(transformed)
-                if max_properties is not None and len(properties) >= max_properties:
-                    return properties
 
         pagination = state.get("paginationControls") or {}
         if not pagination.get("next"):
             break
 
         page_num += 1
-        time.sleep(DELAY_BETWEEN_PAGES)
+
+    _prime_detail_postcodes(client, raw_listings, OTM_MAX_DETAILS_PER_OUTCODE)
+
+    properties: list[dict] = []
+    for raw in raw_listings:
+        listing_id = str(raw.get("id") or "")
+        detail_postcode = (
+            _detail_postcode_cache.get(listing_id) if OTM_FETCH_DETAILS else None
+        )
+        try:
+            transformed = transform_property(raw, pc_index, detail_postcode)
+        except Exception as exc:
+            log.warning(
+                "OnTheMarket %s property %s failed to transform: %s",
+                outcode, listing_id or "?", exc,
+            )
+            continue
+        if transformed:
+            properties.append(transformed)
+            if max_properties is not None and len(properties) >= max_properties:
+                break
 
     return properties

finder/rightmove.py

@@ -1,20 +1,23 @@
 import json
 import logging
 import re
-import time
+from concurrent.futures import ThreadPoolExecutor
 
 import httpx
 
+import shutdown
 from constants import (
+    DETAIL_FETCH_CONCURRENCY,
     PAGE_SIZE,
-    DELAY_BETWEEN_PAGES,
+    RIGHTMOVE_ACCURATE_PIN_TYPE,
     RIGHTMOVE_DETAIL_URL,
     RIGHTMOVE_FETCH_DETAILS,
     RIGHTMOVE_MAX_DETAILS_PER_OUTCODE,
+    RIGHTMOVE_SKIP_DETAILS_FOR_ACCURATE_PINS,
     SEARCH_URL,
     TYPEAHEAD_URL,
 )
-from http_client import fetch_with_retry
+from http_client import RATE_LIMITER, fetch_with_retry
 from spatial import PostcodeSpatialIndex
 from transform import extract_full_postcode, normalize_postcode, transform_property
 
@@ -168,16 +171,32 @@ def parse_detail_postcode(html: str) -> str | None:
 
 # listingId -> true full postcode (or None when unavailable). Failures are
 # cached too, so a broken/duplicate listing is fetched at most once per run (the
-# same listing can reappear across overlapping outcode searches).
+# same listing can reappear across overlapping outcode searches). Seeded from /
+# dumped to a persistent on-disk cache by the orchestrator (see
+# postcode_cache.py) so a recurring scrape only re-fetches newly-listed
+# properties rather than every listing every run.
 _detail_postcode_cache: dict[str, str | None] = {}
 
 
+def seed_detail_cache(mapping: dict) -> None:
+    """Pre-populate the in-memory detail cache from a persisted snapshot."""
+    if mapping:
+        _detail_postcode_cache.update(mapping)
+
+
+def detail_cache_snapshot() -> dict:
+    """Return a JSON-serialisable copy of the in-memory detail cache."""
+    return dict(_detail_postcode_cache)
+
+
 def _fetch_detail_postcode(client: httpx.Client, property_id: str) -> str | None:
     """GET a listing detail page and return its true full postcode (or None).
 
     Results (including failures) are cached by listing id. The detail page is a
     plain HTML GET — no Cloudflare, unlike Zoopla — so a single httpx call
-    suffices; any error degrades gracefully to the coordinate fallback."""
+    suffices; any error degrades gracefully to the coordinate fallback. Safe to
+    call concurrently: distinct listing ids write distinct cache keys, and the
+    shared RATE_LIMITER spaces the GETs."""
     if not property_id:
         return None
     if property_id in _detail_postcode_cache:
@@ -186,6 +205,7 @@ def _fetch_detail_postcode(client: httpx.Client, property_id: str) -> str | None
     postcode: str | None = None
     url = RIGHTMOVE_DETAIL_URL.format(id=property_id)
     try:
+        RATE_LIMITER.acquire()
         resp = client.get(url, headers={"Accept": "text/html"})
         if resp.status_code == 200:
             postcode = parse_detail_postcode(resp.text)
@@ -219,53 +239,83 @@ def resolve_outcode_id(client: httpx.Client, outcode: str) -> str | None:
     return None
 
 
-def _detail_postcode_for(
+def _needs_detail_fetch(prop: dict) -> bool:
+    """Whether a search result still needs a detail-page fetch for its postcode.
+
+    Skips listings the search already pins precisely: an "ACCURATE_POINT"
+    ``pinType`` means rooftop-exact coordinates, so the coordinate-nearest
+    postcode is trustworthy and the detail page would only confirm it. Listings
+    with an approximate pin — or no ``pinType`` field at all — still get fetched,
+    so this degrades safely to the previous behaviour when the search payload
+    omits ``pinType``."""
+    if not RIGHTMOVE_SKIP_DETAILS_FOR_ACCURATE_PINS:
+        return True
+    loc = prop.get("location") or {}
+    return loc.get("pinType") != RIGHTMOVE_ACCURATE_PIN_TYPE
+
+
+def _prime_detail_postcodes(
     client: httpx.Client,
-    prop: dict,
+    props: list[dict],
     fetch_details: bool,
-    detail_budget: dict,
+    detail_cap: int,
-) -> str | None:
+) -> None:
-    """Look up a listing's true postcode, honouring the per-outcode fetch cap.
+    """Fill ``_detail_postcode_cache`` for the listings that need a detail page.
 
-    Cached listings are always served (they cost neither a cap slot nor a GET);
+    Picks the fresh (uncached, not-skipped) listings — up to ``detail_cap`` per
-    a fresh fetch is made only while ``detail_budget['remaining'] > 0``."""
+    outcode — then fetches their detail pages CONCURRENTLY, bounded by
-    if not fetch_details:
+    ``DETAIL_FETCH_CONCURRENCY`` (the shared RATE_LIMITER keeps the combined
-        return None
+    request rate polite). Cached listings cost neither a slot nor a GET. The
-    property_id = str(prop.get("id") or "")
+    worklist is deduplicated, so distinct ids write distinct cache keys and the
-    if not property_id:
+    concurrent fetches never race on the same key."""
-        return None
+    if not fetch_details or detail_cap <= 0:
-    if property_id in _detail_postcode_cache:
+        return
-        return _detail_postcode_cache[property_id]
+
-    if detail_budget["remaining"] <= 0:
+    worklist: list[str] = []
-        return None
+    seen: set[str] = set()
-    detail_budget["remaining"] -= 1
+    for prop in props:
-    postcode = _fetch_detail_postcode(client, property_id)
+        property_id = str(prop.get("id") or "")
-    time.sleep(DELAY_BETWEEN_PAGES)
+        if not property_id or property_id in seen:
-    return postcode
+            continue
+        seen.add(property_id)
+        if property_id in _detail_postcode_cache:
+            continue
+        if not _needs_detail_fetch(prop):
+            continue
+        worklist.append(property_id)
+        if len(worklist) >= detail_cap:
+            break
+
+    if not worklist:
+        return
+
+    workers = min(DETAIL_FETCH_CONCURRENCY, len(worklist))
+    with ThreadPoolExecutor(max_workers=workers) as pool:
+        # Drain the iterator so every fetch runs and populates the cache here.
+        for _ in pool.map(lambda pid: _fetch_detail_postcode(client, pid), worklist):
+            pass
 
 
-def _paginate(
+def _collect_search_props(
     client: httpx.Client,
     outcode_id: str,
     outcode: str,
     channel_cfg: dict,
-    pc_index: PostcodeSpatialIndex,
     max_properties: int | None = None,
-    fetch_details: bool = False,
-    detail_cap: int = 0,
 ) -> tuple[list[dict], int]:
-    """Paginate through search results. Returns (properties, result_count).
+    """Paginate the search API for one outcode+channel, collecting raw results.
 
-    When ``fetch_details`` is set, up to ``detail_cap`` listings per outcode have
+    Returns ``(raw_props, result_count)``. Pagination stays serial — each page
-    their detail page fetched for the property's TRUE full postcode (see
+    reveals the next — but is cheap relative to detail fetching, and the
-    ``parse_detail_postcode``); the rest fall back to coordinate-derived
+    RATE_LIMITER spaces the page GETs. Collection stops at ``max_properties`` raw
-    postcodes."""
+    listings, the end of results, or Rightmove's ``_MAX_INDEX`` page cap."""
-    properties = []
+    raw_props: list[dict] = []
     index = 0
     result_count = 0
-    detail_budget = {"remaining": detail_cap}
 
     while True:
+        if shutdown.stop_requested():
+            break
         params = {
             "useLocationIdentifier": "true",
             "locationIdentifier": f"OUTCODE^{outcode_id}",
@@ -284,37 +334,17 @@ def _paginate(
             )
             break
 
-        raw_props = data.get("properties", [])
+        page_props = data.get("properties", [])
-        if not raw_props:
+        if not page_props:
             break
+        raw_props.extend(page_props)
 
-        for prop in raw_props:
-            try:
-                detail_postcode = _detail_postcode_for(
-                    client, prop, fetch_details, detail_budget
-                )
-                transformed = transform_property(
-                    prop, outcode, pc_index, detail_postcode=detail_postcode
-                )
-            except Exception as exc:
-                log.warning(
-                    "Rightmove %s/%s property %s failed to transform: %s",
-                    outcode,
-                    channel_cfg["channel"],
-                    prop.get("id", "?"),
-                    exc,
-                )
-                continue
-            if transformed:
-                properties.append(transformed)
-                if max_properties is not None and len(properties) >= max_properties:
-                    return properties, result_count
-
-        # Check if there are more pages
         result_count_str = data.get("resultCount", "0")
         result_count = int(result_count_str.replace(",", ""))
-        index += PAGE_SIZE
 
+        if max_properties is not None and len(raw_props) >= max_properties:
+            break
+        index += PAGE_SIZE
         if index >= result_count:
             break
         if index >= _MAX_INDEX:
@@ -327,7 +357,55 @@ def _paginate(
             )
             break
 
-        time.sleep(DELAY_BETWEEN_PAGES)
+    return raw_props, result_count
+
+
+def _paginate(
+    client: httpx.Client,
+    outcode_id: str,
+    outcode: str,
+    channel_cfg: dict,
+    pc_index: PostcodeSpatialIndex,
+    max_properties: int | None = None,
+    fetch_details: bool = False,
+    detail_cap: int = 0,
+) -> tuple[list[dict], int]:
+    """Collect search results, recover true postcodes, and transform them.
+
+    Search pages are paginated serially; then — when ``fetch_details`` is set —
+    up to ``detail_cap`` listings per outcode have their detail page fetched
+    CONCURRENTLY for the property's TRUE full postcode (see
+    ``parse_detail_postcode``), with listings the search already pins precisely
+    skipped (see ``_needs_detail_fetch``). The rest fall back to
+    coordinate-derived postcodes. Returns ``(properties, result_count)``."""
+    raw_props, result_count = _collect_search_props(
+        client, outcode_id, outcode, channel_cfg, max_properties
+    )
+    _prime_detail_postcodes(client, raw_props, fetch_details, detail_cap)
+
+    properties: list[dict] = []
+    for prop in raw_props:
+        property_id = str(prop.get("id") or "")
+        detail_postcode = (
+            _detail_postcode_cache.get(property_id) if fetch_details else None
+        )
+        try:
+            transformed = transform_property(
+                prop, outcode, pc_index, detail_postcode=detail_postcode
+            )
+        except Exception as exc:
+            log.warning(
+                "Rightmove %s/%s property %s failed to transform: %s",
+                outcode,
+                channel_cfg["channel"],
+                prop.get("id", "?"),
+                exc,
+            )
+            continue
+        if transformed:
+            properties.append(transformed)
+            if max_properties is not None and len(properties) >= max_properties:
+                break
 
     return properties, result_count
 

finder/scraper.py

@@ -3,9 +3,11 @@ import os
 import re
 import signal
 import time
+from concurrent.futures import ThreadPoolExecutor
 from contextlib import contextmanager
+from functools import partial
 from pathlib import Path
-from typing import Iterable
+from typing import Callable, Iterable
 
 import polars as pl
 
@@ -21,8 +23,13 @@ from constants import (
     ZOOPLA_MAX_DETAILS_PER_OUTCODE,
 )
 
+import onthemarket
+import rightmove
+import shutdown
+import zoopla
 from http_client import make_client
 from onthemarket import search_outcode as onthemarket_search_outcode
+from postcode_cache import load_cache, save_cache
 from rightmove import resolve_outcode_id
 from rightmove import search_outcode as rightmove_search_outcode
 from spatial import PostcodeSpatialIndex
@@ -38,6 +45,16 @@ SALE_CHANNEL = CHANNELS[0]
 LONDON_AREAS = sorted({prefix.upper() for prefix in LONDON_OUTCODE_PREFIXES})
 OUTCODE_RE = re.compile(r"^([A-Z]{1,2}\d[A-Z0-9]?)")
 
+# Per-source modules exposing the persistent detail-cache hooks
+# (seed_detail_cache / detail_cache_snapshot). A listing's recovered postcode
+# never changes, so the cache is loaded before a run and dumped after it, letting
+# a recurring scrape skip the detail fetch for every listing it has already seen.
+_CACHE_MODULES = {
+    "rightmove": rightmove,
+    "onthemarket": onthemarket,
+    "zoopla": zoopla,
+}
+
 
 def _arcgis_columns() -> tuple[str, str]:
     """Return postcode and country column names for supported ARCGIS schemas."""
@@ -306,6 +323,9 @@ def _scrape_rightmove(
     client = make_client()
     try:
         for outcode in outcodes:
+            if shutdown.stop_requested():
+                log.info("Rightmove: shutdown requested, stopping early")
+                return
             if _source_remaining(results, "rightmove", max_properties_per_source) == 0:
                 log.info("Rightmove cap reached")
                 return
@@ -314,12 +334,12 @@ def _scrape_rightmove(
                 outcode_id = resolve_outcode_id(client, outcode)
             except Exception as exc:
                 _record_error(errors, "rightmove", outcode, exc)
-                time.sleep(DELAY_BETWEEN_OUTCODES)
+                shutdown.sleep(DELAY_BETWEEN_OUTCODES)
                 continue
 
             if not outcode_id:
                 log.debug("No Rightmove outcode ID for %s", outcode)
-                time.sleep(DELAY_BETWEEN_OUTCODES)
+                shutdown.sleep(DELAY_BETWEEN_OUTCODES)
                 continue
 
             remaining = _source_remaining(
@@ -348,7 +368,7 @@ def _scrape_rightmove(
             except Exception as exc:
                 _record_error(errors, "rightmove", outcode, exc)
 
-            time.sleep(DELAY_BETWEEN_OUTCODES)
+            shutdown.sleep(DELAY_BETWEEN_OUTCODES)
     finally:
         client.close()
 
@@ -494,6 +514,9 @@ def _scrape_zoopla_flaresolverr(
 
     try:
         for outcode in outcodes:
+            if shutdown.stop_requested():
+                log.info("Zoopla: shutdown requested, stopping early")
+                return
             remaining = _source_remaining(results, "zoopla", max_properties_per_source)
             if remaining == 0:
                 log.info("Zoopla cap reached")
@@ -511,7 +534,7 @@ def _scrape_zoopla_flaresolverr(
                 log.info("Zoopla %s: +%d", outcode, added)
             except Exception as exc:  # noqa: BLE001 - one outcode must not kill the run
                 _record_error(errors, "zoopla", outcode, exc)
-            time.sleep(DELAY_BETWEEN_OUTCODES)
+            shutdown.sleep(DELAY_BETWEEN_OUTCODES)
     finally:
         session.__exit__(None, None, None)
 
@@ -547,6 +570,9 @@ def _scrape_zoopla(
 
     try:
         for outcode in outcodes:
+            if shutdown.stop_requested():
+                log.info("Zoopla: shutdown requested, stopping early")
+                return
             if _source_remaining(results, "zoopla", max_properties_per_source) == 0:
                 log.info("Zoopla cap reached")
                 return
@@ -596,7 +622,7 @@ def _scrape_zoopla(
                         _record_error(errors, "zoopla", outcode, relaunch_exc)
                         return
 
-            time.sleep(DELAY_BETWEEN_OUTCODES)
+            shutdown.sleep(DELAY_BETWEEN_OUTCODES)
     finally:
         if detail_page is not None:
             try:
@@ -616,6 +642,9 @@ def _scrape_onthemarket(
     client = make_client()
     try:
         for outcode in outcodes:
+            if shutdown.stop_requested():
+                log.info("OnTheMarket: shutdown requested, stopping early")
+                return
             if (
                 _source_remaining(results, "onthemarket", max_properties_per_source)
                 == 0
@@ -644,11 +673,57 @@ def _scrape_onthemarket(
             except Exception as exc:
                 _record_error(errors, "onthemarket", outcode, exc)
 
-            time.sleep(DELAY_BETWEEN_OUTCODES)
+            shutdown.sleep(DELAY_BETWEEN_OUTCODES)
     finally:
         client.close()
 
 
+def _seed_detail_caches(sources: list[str], cache_dir: Path) -> None:
+    """Load each selected source's persisted detail cache into memory."""
+    for source in sources:
+        module = _CACHE_MODULES.get(source)
+        if module is None:
+            continue
+        module.seed_detail_cache(load_cache(cache_dir / f"{source}.json"))
+
+
+def _save_detail_caches(sources: list[str], cache_dir: Path) -> None:
+    """Persist each selected source's in-memory detail cache to disk."""
+    for source in sources:
+        module = _CACHE_MODULES.get(source)
+        if module is None:
+            continue
+        save_cache(cache_dir / f"{source}.json", module.detail_cache_snapshot())
+
+
+def _run_sources(
+    run_zoopla: Callable[[], None] | None,
+    background_runners: list[tuple[str, Callable[[], None]]],
+    errors: list[str],
+) -> None:
+    """Run the HTTP-based providers concurrently while Zoopla runs inline.
+
+    Rightmove and OnTheMarket are plain-httpx and thread-safe, so each runs in a
+    worker thread. Zoopla must stay on the MAIN thread because its per-outcode
+    wall-clock guard uses SIGALRM, which Python only delivers there. Each source
+    writes only its own ``results[source]`` list and appends to the shared
+    ``errors`` list (atomic under the GIL), so there is no cross-source data
+    race. One source raising never kills the others: each failure is recorded
+    and the remaining sources still finish."""
+    with ThreadPoolExecutor(max_workers=max(1, len(background_runners))) as pool:
+        futures = {pool.submit(fn): name for name, fn in background_runners}
+        if run_zoopla is not None:
+            try:
+                run_zoopla()
+            except Exception as exc:  # noqa: BLE001 - one source must not kill the run
+                _record_error(errors, "zoopla", "*", exc)
+        for future, name in futures.items():
+            try:
+                future.result()
+            except Exception as exc:  # noqa: BLE001 - one source must not kill the run
+                _record_error(errors, name, "*", exc)
+
+
 def run_scrape(
     outcodes: list[str],
     pc_index: PostcodeSpatialIndex,
@@ -679,28 +754,46 @@ def run_scrape(
         max_properties_per_source,
     )
 
+    # Persistent detail caches: load before, dump after. A listing's recovered
+    # postcode never changes, so a recurring scrape only fetches new listings.
+    cache_dir = output_base / "detail_cache"
+    _seed_detail_caches(selected_sources, cache_dir)
+
+    # Rightmove and OnTheMarket are plain-httpx and run in worker threads;
+    # Zoopla runs on the main thread (its SIGALRM timeout requires it). They
+    # share one RATE_LIMITER, so running them at once stays polite.
+    background_runners: list[tuple[str, Callable[[], None]]] = []
     if "rightmove" in selected_sources:
-        _scrape_rightmove(
+        background_runners.append((
-            selected_outcodes,
+            "rightmove",
-            pc_index,
+            partial(
-            results,
+                _scrape_rightmove,
-            errors,
+                selected_outcodes,
-            max_properties_per_source,
+                pc_index,
-        )
+                results,
-
+                errors,
+                max_properties_per_source,
+            ),
+        ))
     if "onthemarket" in selected_sources:
-        _scrape_onthemarket(
+        background_runners.append((
-            selected_outcodes,
+            "onthemarket",
-            pc_index,
+            partial(
-            results,
+                _scrape_onthemarket,
-            errors,
+                selected_outcodes,
-            max_properties_per_source,
+                pc_index,
-        )
+                results,
+                errors,
+                max_properties_per_source,
+            ),
+        ))
 
+    run_zoopla: Callable[[], None] | None = None
     if "zoopla" in selected_sources:
         if pc_coords is None:
             pc_coords = build_postcode_coords()
-        _scrape_zoopla(
+        run_zoopla = partial(
+            _scrape_zoopla,
             selected_outcodes,
             pc_index,
             pc_coords,
@@ -709,6 +802,11 @@ def run_scrape(
             max_properties_per_source,
         )
 
+    try:
+        _run_sources(run_zoopla, background_runners, errors)
+    finally:
+        _save_detail_caches(selected_sources, cache_dir)
+
     merged, source_counts, deduped = _merge_properties(results)
     output_path = output_base / "online_listings_buy.parquet"
     if merged:

finder/zoopla.py

@@ -29,6 +29,7 @@ from urllib.parse import parse_qsl, urlencode, urljoin, urlparse, urlunparse
 
 import httpx
 
+import shutdown
 from constants import (
     DATA_DIR,
     DELAY_BETWEEN_PAGES,
@@ -973,7 +974,7 @@ def _paginate(
         listing["_detail"] = fetch_detail(url)
         if not cached:
             detail_state["fetched"] += 1
-            time.sleep(DELAY_BETWEEN_PAGES)
+            shutdown.sleep(DELAY_BETWEEN_PAGES)
 
     all_listings = _extract_listings(page)
     for listing in all_listings:
@@ -988,13 +989,15 @@ def _paginate(
     page_num = 2
 
     while True:
+        if shutdown.stop_requested():
+            break
         next_url = _find_next_page_url(page)
         if not next_url:
             if total_results > 0 and len(all_listings) >= total_results:
                 break
             next_url = _url_with_page(page.url, page_num)
 
-        time.sleep(DELAY_BETWEEN_PAGES)
+        shutdown.sleep(DELAY_BETWEEN_PAGES)
 
         try:
             page.goto(next_url, wait_until="domcontentloaded", timeout=30000)
@@ -1119,9 +1122,23 @@ def _extract_outcode(text: str) -> str | None:
 
 # listingId -> parsed detail dict (or None). Failures are cached too, so a
 # broken listing is not re-fetched within a run (the same listing reappears
-# across overlapping outcode searches).
+# across overlapping outcode searches). Seeded from / dumped to a persistent
+# on-disk cache by the orchestrator (see postcode_cache.py) so a recurring
+# scrape only re-fetches newly-listed properties — the biggest saving for
+# Zoopla, whose detail fetch drives a real browser tab.
 _detail_cache: dict[str, dict | None] = {}
 
+
+def seed_detail_cache(mapping: dict) -> None:
+    """Pre-populate the in-memory detail cache from a persisted snapshot."""
+    if mapping:
+        _detail_cache.update(mapping)
+
+
+def detail_cache_snapshot() -> dict:
+    """Return a JSON-serialisable copy of the in-memory detail cache."""
+    return dict(_detail_cache)
+
 _LISTING_ID_RE = re.compile(r"/details/(\d+)/?")
 
 # The property's own location is carried by a `"location":{...}` wrapper and a