From 8e4c56bb0d71e29c20802dacf64e80daa0650010 Mon Sep 17 00:00:00 2001 From: Andras Schmelczer Date: Sun, 14 Jun 2026 14:50:38 +0100 Subject: [PATCH] Cache postcodes --- finder/constants.py | 25 ++++++ finder/http_client.py | 51 ++++++++++- finder/main.py | 46 ++++++++-- finder/onthemarket.py | 150 ++++++++++++++++++++++--------- finder/rightmove.py | 204 +++++++++++++++++++++++++++++------------- finder/scraper.py | 144 ++++++++++++++++++++++++----- finder/zoopla.py | 23 ++++- 7 files changed, 502 insertions(+), 141 deletions(-) diff --git a/finder/constants.py b/finder/constants.py index 17e6938..66cc419 100644 --- a/finder/constants.py +++ b/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" diff --git a/finder/http_client.py b/finder/http_client.py index 3f7aaa0..635371f 100644 --- a/finder/http_client.py +++ b/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 diff --git a/finder/main.py b/finder/main.py index 15b6811..f7773fb 100644 --- a/finder/main.py +++ b/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": - return ["rightmove", "onthemarket", "zoopla"] - return [source] + """Resolve --source: one or more comma-separated portals, or 'all'. + + 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"): diff --git a/finder/onthemarket.py b/finder/onthemarket.py index 00b90c4..8ccaf40 100644 --- a/finder/onthemarket.py +++ b/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 - ``OTM_MAX_DETAILS_PER_OUTCODE`` listings per outcode have their detail page - fetched for the property's own postcode (see ``_fetch_detail_postcode``); - the rest fall back to the coordinate-nearest postcode. + Search pages are paginated serially (the RATE_LIMITER spaces the GETs); then, + when ``OTM_FETCH_DETAILS`` is enabled, up to ``OTM_MAX_DETAILS_PER_OUTCODE`` + listings per outcode have their detail page fetched CONCURRENTLY for the + 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 [] - if not raw_listings: + page_listings = state.get("list") or [] + 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 OTM_FETCH_DETAILS and listing_id: - # 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 + if max_properties is not None and len(raw_listings) >= max_properties: + break 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 diff --git a/finder/rightmove.py b/finder/rightmove.py index 956b73d..cabfdd9 100644 --- a/finder/rightmove.py +++ b/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, -) -> str | None: - """Look up a listing's true postcode, honouring the per-outcode fetch cap. + detail_cap: int, +) -> None: + """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); - a fresh fetch is made only while ``detail_budget['remaining'] > 0``.""" - if not fetch_details: - return None - property_id = str(prop.get("id") or "") - if not property_id: - return None - if property_id in _detail_postcode_cache: - return _detail_postcode_cache[property_id] - if detail_budget["remaining"] <= 0: - return None - detail_budget["remaining"] -= 1 - postcode = _fetch_detail_postcode(client, property_id) - time.sleep(DELAY_BETWEEN_PAGES) - return postcode + Picks the fresh (uncached, not-skipped) 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 fetch_details or detail_cap <= 0: + return + + worklist: list[str] = [] + seen: set[str] = set() + for prop in props: + property_id = str(prop.get("id") or "") + if not property_id or property_id in seen: + 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 - their detail page fetched for the property's TRUE full postcode (see - ``parse_detail_postcode``); the rest fall back to coordinate-derived - postcodes.""" - properties = [] + Returns ``(raw_props, result_count)``. Pagination stays serial — each page + reveals the next — but is cheap relative to detail fetching, and the + RATE_LIMITER spaces the page GETs. Collection stops at ``max_properties`` raw + listings, the end of results, or Rightmove's ``_MAX_INDEX`` page cap.""" + 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", []) - if not raw_props: + page_props = data.get("properties", []) + 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 diff --git a/finder/scraper.py b/finder/scraper.py index 25ddf90..7e4e358 100644 --- a/finder/scraper.py +++ b/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( - selected_outcodes, - pc_index, - results, - errors, - max_properties_per_source, - ) - + background_runners.append(( + "rightmove", + partial( + _scrape_rightmove, + selected_outcodes, + pc_index, + results, + errors, + max_properties_per_source, + ), + )) if "onthemarket" in selected_sources: - _scrape_onthemarket( - selected_outcodes, - pc_index, - results, - errors, - max_properties_per_source, - ) + background_runners.append(( + "onthemarket", + partial( + _scrape_onthemarket, + selected_outcodes, + 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: diff --git a/finder/zoopla.py b/finder/zoopla.py index 6cffe5f..538d1ea 100644 --- a/finder/zoopla.py +++ b/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