andras/perfect-postcode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages 2 Wiki Activity Actions

Hide POIs, show overlay state, remove zoom button, rotate vpn

Browse source
This commit is contained in:
Andras Schmelczer 2026-07-15 21:58:05 +01:00
parent 35276d34fa
commit bce34b73de
32 changed files with 1137 additions and 186 deletions
Download patch file Download diff file Expand all files Collapse all files

6
finder/README.md
View file

@ -187,6 +187,9 @@ Environment variables (override the defaults in `constants.py`):
| `ZOOPLA_OUTCODE_TIMEOUT_SECONDS` | `300` | Per-outcode wall-clock budget for Zoopla. |
| `DETAIL_FETCH_CONCURRENCY` | `8` | Parallel detail fetches (Rightmove/OTM). |
| `REQUESTS_PER_SECOND` | `10` | Global request-rate cap. Lower it if you see `429`/`403`. |
| `BLOCK_403_THRESHOLD` | `5` | Consecutive 403s from one host before the egress IP is rotated. |
| `BLOCK_MAX_ROTATIONS` | `3` | Egress-IP rotations allowed per run. `0` disables rotation. |
| `MAX_ROW_DROP_RATIO` | `0.25` | Reject the run if the merged total falls this far below the previous parquet. |
| `RIGHTMOVE_SKIP_DETAILS_FOR_ACCURATE_PINS` | `1` | Inert today (see note above). |
Non-env code constants worth knowing (`constants.py` / `onthemarket.py`):
@ -216,7 +219,8 @@ uv run --with pytest pytest -q
| `scraper.py` | Orchestration: per-source runners, provider parallelism, cache load/save, merge + write. |
| `rightmove.py` / `onthemarket.py` / `zoopla.py` | Per-portal search + detail scraping and parsing. |
| `transform.py` | Raw listing → output schema; postcode trust rules. |
| `http_client.py` | Shared httpx client, retry/backoff, and the global `RATE_LIMITER`. |
| `http_client.py` | Shared httpx client, retry/backoff, the global `RATE_LIMITER`, and egress-block detection (`BlockedError`). |
| `gluetun.py` | Gluetun control-API client: reads the public IP and rotates the (shared) VPN tunnel. |
| `postcode_cache.py` | Persistent (cross-run) detail-cache load/save. |
| `spatial.py` | Grid spatial index for coordinate → nearest postcode. |
| `storage.py` | Parquet writer (server-ready column names). |

26
finder/constants.py
View file

@ -25,6 +25,32 @@ RETRY_BASE_DELAY = 2.0
# 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"))
# Egress-block detection. A portal that 403s one URL is a quirk; a portal that
# 403s a whole HOST in a row is refusing our egress IP, which no amount of
# per-request retrying fixes. On 2026-07-15 Rightmove's Fastly edge did exactly
# this and every one of the 366 outcodes 403'd on the typeahead call, which the
# scraper silently recorded as "no such outcode" and published a Rightmove-less
# dataset. So: after this many consecutive 403s from one host, rotate the egress
# IP (finder/gluetun.py) and retry; once the rotation budget is spent and the
# host still 403s, raise http_client.BlockedError and fail the run loudly.
#
# Rotation is DISRUPTIVE (see gluetun.py: it restarts the tunnel shared with the
# media stack), so the budget is deliberately small. It is per-run, not per-host.
BLOCK_403_THRESHOLD = int(os.environ.get("BLOCK_403_THRESHOLD", "5"))
BLOCK_MAX_ROTATIONS = int(os.environ.get("BLOCK_MAX_ROTATIONS", "3"))
# Sanity gates on a finished scrape, checked before the parquet is overwritten
# (finder/scraper.py). A selected source yielding nothing, or the merged total
# collapsing against the previous run, means something is broken upstream rather
# than the market having moved; publishing that would quietly gut production.
#
# 10% is loose against real movement and tight against breakage: consecutive
# healthy cycles land within ~0.15% of each other (103,087 / 103,142 / 103,008),
# so a 10% fall is already a ~65x anomaly. It is deliberately BELOW the
# 2026-07-15 incident's own 18.7% drop (103,008 -> 83,785), which a 25% gate
# would have waved through.
MAX_ROW_DROP_RATIO = float(os.environ.get("MAX_ROW_DROP_RATIO", "0.10"))
GRID_CELL_SIZE = 0.01 # degrees for postcode spatial index
MAX_BEDROOMS = 20 # sanity cap: values above this are almost certainly parsing errors

130
finder/gluetun.py Normal file
View file

@ -0,0 +1,130 @@
"""Gluetun VPN control-server client, shared by every scraper.
The scrapers egress through a Gluetun container's network namespace
(docker-compose.yml, network_mode "container:media_gluetun"), so when a portal
starts refusing our egress IP the cheapest unblocker is to make Gluetun
reconnect to a different VPN server. This module wraps Gluetun's HTTP control
API so both callers share one implementation:
* http_client.py rotates when a portal 403s every request (an egress block).
* zoopla.py rotates when Cloudflare Turnstile fires.
Rotation is SHARED and DISRUPTIVE: every container joined to Gluetun's netns
(here also qbittorrent/sonarr/radarr/prowlarr/seerr/jellyfin) loses
connectivity for the few seconds the tunnel takes to come back. It is therefore
serialised behind a module lock, so N blocked worker threads trigger ONE
rotation rather than N, and callers must budget rotations rather than retry
them freely.
"""
import logging
import threading
import time
import httpx
from constants import GLUETUN_API_KEY, GLUETUN_CONTROL_URL
log = logging.getLogger("rightmove")
# Serialises rotation across worker threads: a rotation tears down the shared
# tunnel, so two concurrent ones would fight (and needlessly double the outage).
_ROTATION_LOCK = threading.Lock()
def _base_url() -> str:
return GLUETUN_CONTROL_URL.rstrip("/")
def _client() -> httpx.Client:
# Talks to the control server directly (not through the VPN proxy).
headers = {}
if GLUETUN_API_KEY:
headers["X-API-Key"] = GLUETUN_API_KEY
return httpx.Client(headers=headers)
def public_ip(client: httpx.Client) -> str | None:
try:
resp = client.get(f"{_base_url()}/v1/publicip/ip", timeout=5.0)
if resp.status_code != 200:
return None
data = resp.json()
except (httpx.HTTPError, ValueError):
return None
return data.get("public_ip") or data.get("ip")
def _set_vpn_status(client: httpx.Client, status: str) -> bool:
"""PUT /v1/vpn/status with {'status': status}. Returns True on 2xx."""
try:
resp = client.put(
f"{_base_url()}/v1/vpn/status",
json={"status": status},
timeout=15.0,
)
except httpx.HTTPError as exc:
log.warning("Gluetun vpn/status %s failed: %s", status, exc)
return False
if resp.status_code == 401:
log.warning(
"Gluetun vpn/status %s: 401 Unauthorized. The API key must be "
"authorised for 'PUT /v1/vpn/status' in Gluetun's auth config.toml",
status,
)
return False
if resp.status_code >= 400:
log.warning(
"Gluetun vpn/status %s returned HTTP %d: %s",
status,
resp.status_code,
resp.text[:200],
)
return False
return True
def rotate_ip(wait_seconds: int = 45) -> bool:
"""Restart Gluetun's VPN and wait for the public IP to change.
Returns True if a new IP was observed within ``wait_seconds``. Serialised:
while one thread rotates, others block here and then see the already-rotated
IP, so a 403 storm across many threads costs one tunnel restart. A failed
rotation always attempts to bring the tunnel back up, because leaving it
stopped would strand every container sharing the netns.
"""
with _ROTATION_LOCK, _client() as client:
old_ip = public_ip(client)
log.info("Requesting Gluetun IP rotation (current IP: %s)", old_ip or "unknown")
stop_attempted = False
restart_confirmed = False
try:
stop_attempted = True
if not _set_vpn_status(client, "stopped"):
return False
time.sleep(2)
restart_confirmed = _set_vpn_status(client, "running")
if not restart_confirmed:
return False
deadline = time.monotonic() + wait_seconds
while time.monotonic() < deadline:
time.sleep(2)
new_ip = public_ip(client)
if new_ip and new_ip != old_ip:
log.info("Gluetun rotated IP: %s -> %s", old_ip or "?", new_ip)
return True
finally:
if stop_attempted and not restart_confirmed:
log.warning(
"Gluetun VPN may be stopped after failed rotation; "
"attempting recovery start"
)
if not _set_vpn_status(client, "running"):
log.error(
"Gluetun VPN recovery start failed; manual intervention required"
)
log.warning("Gluetun IP did not change within %ds", wait_seconds)
return False

122
finder/http_client.py
View file

@ -2,12 +2,16 @@ import logging
import random
import threading
import time
from urllib.parse import urlsplit
import httpx
from fake_useragent import UserAgent
import gluetun
import shutdown
from constants import (
BLOCK_403_THRESHOLD,
BLOCK_MAX_ROTATIONS,
GLUETUN_PROXY,
MAX_RETRIES,
REQUESTS_PER_SECOND,
@ -17,6 +21,16 @@ from constants import (
log = logging.getLogger("rightmove")
class BlockedError(RuntimeError):
"""A host is refusing our egress IP and rotating away from it did not help.
Raised rather than returned so a block can never be mistaken for "this query
has no results": that conflation is precisely what turned Rightmove's
2026-07-15 edge block into a silent 0-listing run (see BLOCK_403_THRESHOLD).
Callers should abandon the source, not the outcode.
"""
class RateLimiter:
"""Thread-safe global limiter: spaces request starts by a minimum interval.
@ -51,6 +65,74 @@ class RateLimiter:
# providers). Spacing is global, so politeness is decoupled from concurrency.
RATE_LIMITER = RateLimiter(REQUESTS_PER_SECOND)
class _EgressBlockTracker:
"""Escalates a run of 403s from one host: rotate the egress IP, then give up.
Counts CONSECUTIVE 403s per host, reset by any 200 from that host, so one
forbidden URL never trips it while a refused egress IP does within a couple
of calls. The rotation budget is per-run and shared across hosts and
threads, because the tunnel being rotated is shared too.
"""
def __init__(self, threshold: int, max_rotations: int):
self._threshold = max(threshold, 1)
self._max_rotations = max(max_rotations, 0)
self._lock = threading.Lock()
self._consecutive: dict[str, int] = {}
self._rotations = 0
def record_success(self, host: str) -> None:
with self._lock:
self._consecutive.pop(host, None)
def record_403(self, host: str) -> str:
"""Register a 403 and decide what the caller should do next.
Returns "retry" (below the threshold, back off normally), "rotated"
(the egress IP changed, so the call deserves a fresh retry budget) or
"blocked" (the rotation budget is spent, or rotating failed).
Rotation runs under the lock: concurrent threads caught in the same 403
storm queue behind ONE tunnel restart and then re-read the reset
counters, rather than each spending a slice of the budget.
"""
with self._lock:
count = self._consecutive.get(host, 0) + 1
self._consecutive[host] = count
if count < self._threshold:
return "retry"
if self._rotations >= self._max_rotations:
return "blocked"
self._rotations += 1
log.warning(
"%d consecutive 403s from %s; rotating egress IP (rotation %d/%d)",
count,
host,
self._rotations,
self._max_rotations,
)
rotated = gluetun.rotate_ip()
# Either way the counters restart: on success the next 403s are
# evidence about the NEW IP, and on failure we must not re-enter
# rotation on the very next 403.
self._consecutive.clear()
if not rotated:
log.error("Egress IP rotation failed; treating %s as blocked", host)
return "blocked"
return "rotated"
def reset(self) -> None:
"""Forget all counters and refund the rotation budget (tests only)."""
with self._lock:
self._consecutive.clear()
self._rotations = 0
BLOCK_TRACKER = _EgressBlockTracker(BLOCK_403_THRESHOLD, BLOCK_MAX_ROTATIONS)
_ua = UserAgent(
browsers=["Chrome", "Edge"], os=["Windows", "Mac OS X"], min_version=120.0
)
@ -69,22 +151,48 @@ def make_client() -> httpx.Client:
def fetch_with_retry(
client: httpx.Client, url: str, params: dict | None = None, on_403: bool = True
) -> dict | None:
"""GET JSON with retries on 429/5xx/connection errors.
"""GET JSON with retries on 403/429/5xx/connection errors.
Returns None on permanent failure. The on_403 argument is kept for
compatibility with older callers; 403 is now treated as non-retryable.
Returns None on permanent failure, EXCEPT when the host is refusing our
egress IP, which raises BlockedError instead. A 403 from a shared CDN edge
is usually transient and is retried with backoff; once one host has 403'd
BLOCK_403_THRESHOLD times in a row the egress IP is rotated (and this call
gets a fresh retry budget on the new IP), and once the rotation budget is
spent the block is raised rather than degraded to a None the caller would
read as "no results". ``on_403=False`` opts out for callers where a 403 is
an expected per-URL answer rather than a verdict on our IP.
"""
for attempt in range(MAX_RETRIES):
host = urlsplit(url).netloc
attempt = 0
while attempt < MAX_RETRIES:
if shutdown.stop_requested():
return None
try:
RATE_LIMITER.acquire()
resp = client.get(url, params=params)
if resp.status_code == 200:
BLOCK_TRACKER.record_success(host)
return resp.json()
if resp.status_code == 403 and on_403:
log.error("HTTP 403 from %s (forbidden)", url)
return None
action = BLOCK_TRACKER.record_403(host)
if action == "blocked":
raise BlockedError(
f"HTTP 403 from {url}: {host} is refusing this egress IP "
f"and rotating away from it did not help"
)
delay = RETRY_BASE_DELAY * (2**attempt) + random.uniform(0, 1)
log.warning(
"HTTP 403 from %s, retry %d/%d in %.1fs",
url,
attempt + 1,
MAX_RETRIES,
delay,
)
# A rotation means a different egress IP, so the failures this
# call already accrued say nothing about the new one.
attempt = 0 if action == "rotated" else attempt + 1
shutdown.sleep(delay)
continue
if resp.status_code in (429, 500, 502, 503, 504):
delay = RETRY_BASE_DELAY * (2**attempt) + random.uniform(0, 1)
log.warning(
@ -95,6 +203,7 @@ def fetch_with_retry(
MAX_RETRIES,
delay,
)
attempt += 1
shutdown.sleep(delay)
continue
log.error("HTTP %d from %s (non-retryable)", resp.status_code, url)
@ -114,6 +223,7 @@ def fetch_with_retry(
MAX_RETRIES,
delay,
)
attempt += 1
shutdown.sleep(delay)
log.error("All %d retries exhausted for %s", MAX_RETRIES, url)
return None

7
finder/main.py
View file

@ -218,6 +218,13 @@ def main() -> int:
return 130 # 128 + SIGINT, the conventional Ctrl+C exit code.
log.info("Scrape finished in %.1fs", elapsed)
log.info("Result: %s", result)
if result.get("failures"):
# Nothing was written, so the previous parquet is still the good one.
# Exiting non-zero keeps the caller (scripts/scrape-loop.sh) from
# enriching and publishing on top of a rejected run.
for failure in result["failures"]:
log.error("Scrape rejected: %s", failure)
return 1
if args.test and result.get("errors"):
raise SystemExit("Test scrape failed; see errors in the result above.")
return 0

123
finder/scraper.py
View file

@ -19,6 +19,7 @@ from constants import (
DATA_DIR,
DELAY_BETWEEN_OUTCODES,
LONDON_OUTCODE_PREFIXES,
MAX_ROW_DROP_RATIO,
ZOOPLA_DETAIL_BUDGET_FRACTION,
ZOOPLA_FETCH_DETAILS,
ZOOPLA_FETCHER,
@ -29,7 +30,7 @@ import onthemarket
import rightmove
import shutdown
import zoopla
from http_client import make_client
from http_client import BlockedError, 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
@ -316,6 +317,72 @@ def _record_error(
log.warning(message)
def _previous_row_count(path: Path) -> int | None:
"""Rows in the parquet we are about to replace, or None if there isn't one.
Reads footer metadata only, so it costs nothing on a 30MB file."""
if not path.exists():
return None
try:
return pl.scan_parquet(path).select(pl.len()).collect().item()
except Exception as exc: # noqa: BLE001 - an unreadable baseline must not fail the run
log.warning("Could not read row count from %s: %s", path, exc)
return None
def _sanity_failures(
results: dict[str, list[dict]],
selected_sources: list[str],
merged: list[dict],
output_path: Path,
abandoned: set[str],
) -> list[str]:
"""Reasons this run's data is too broken to publish, empty when it's fine.
Catches upstream breakage that the per-outcode error path cannot see, via
three gates, because no one of them is sufficient:
* ABANDONED. A source that hit an egress block stopped partway, so its
listings cover only the outcodes it reached. This gate is what catches a
block that starts MID-RUN: the source is then non-empty, so the zero-yield
gate below would wave it through on a single banked listing.
* ZERO YIELD. A selected source that returned nothing at all, whatever the
cause (403 storm from the first outcode, markup change, DNS).
* COLLAPSE. The merged total against the previous run, for degradations that
are neither clean-zero nor a raised block.
Cross-source dedup is why the merged total alone cannot be trusted: when
Rightmove vanished on 2026-07-15, OnTheMarket stopped losing dedup ties to
it and backfilled the total from ~9k unique to ~84k, so losing an entire
source showed up as an 18.7% dip.
"""
failures = []
for source in selected_sources:
if source in abandoned:
failures.append(
f"{source} was abandoned mid-run after an egress block, so its "
f"{_source_total(results, source)} listings cover only part of "
f"the outcode list"
)
elif _source_total(results, source) == 0:
failures.append(
f"{source} yielded 0 listings; it was selected for this run, so "
f"treat this as a failure rather than an empty market"
)
if not merged:
failures.append("no listings survived the merge")
previous = _previous_row_count(output_path)
if previous and merged:
drop = (previous - len(merged)) / previous
if drop > MAX_ROW_DROP_RATIO:
failures.append(
f"merged total collapsed from {previous} to {len(merged)} "
f"({drop:.0%} drop, limit {MAX_ROW_DROP_RATIO:.0%})"
)
return failures
def _scrape_rightmove(
outcodes: list[str],
pc_index: PostcodeSpatialIndex,
@ -335,6 +402,12 @@ def _scrape_rightmove(
try:
outcode_id = resolve_outcode_id(client, outcode)
except BlockedError:
# Our egress IP is refused, not this outcode. Grinding through
# the remaining outcodes would just collect 403s, so abandon the
# source and let _run_sources mark it abandoned, which fails the
# run however many outcodes we got through first.
raise
except Exception as exc:
_record_error(errors, "rightmove", outcode, exc)
shutdown.sleep(DELAY_BETWEEN_OUTCODES)
@ -368,6 +441,8 @@ def _scrape_rightmove(
max_properties_per_source,
)
log.info("Rightmove %s: +%d", outcode, added)
except BlockedError:
raise
except Exception as exc:
_record_error(errors, "rightmove", outcode, exc)
@ -396,6 +471,8 @@ def _scrape_rightmove(
max_properties_per_source,
)
log.info("Rightmove %s new-homes: +%d", outcode, added_new)
except BlockedError:
raise
except Exception as exc:
_record_error(errors, "rightmove", outcode, exc)
@ -731,6 +808,7 @@ def _run_sources(
run_zoopla: Callable[[], None] | None,
background_runners: list[tuple[str, Callable[[], None]]],
errors: list[str],
abandoned: set[str],
) -> None:
"""Run the HTTP-based providers concurrently while Zoopla runs inline.
@ -740,17 +818,28 @@ def _run_sources(
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."""
and the remaining sources still finish.
A source that hit an egress block is named in ``abandoned`` as well as
``errors``: it stopped partway, so whatever it did collect covers only the
outcodes it reached, and _sanity_failures must reject the run even though
the source is not empty."""
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 BlockedError as exc:
_record_error(errors, "zoopla", "*", exc)
abandoned.add("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 BlockedError as exc:
_record_error(errors, name, "*", exc)
abandoned.add(name)
except Exception as exc: # noqa: BLE001 - one source must not kill the run
_record_error(errors, name, "*", exc)
@ -775,6 +864,10 @@ def run_scrape(
output_base.mkdir(parents=True, exist_ok=True)
errors: list[str] = []
# Sources that stopped partway because their egress was blocked. Tracked
# separately from `errors`: a partial source is a reason to reject the run,
# not just something to log.
abandoned: set[str] = set()
results = {source: [] for source in SOURCE_ORDER}
started_at = time.time()
@ -834,16 +927,28 @@ def run_scrape(
)
try:
_run_sources(run_zoopla, background_runners, errors)
_run_sources(run_zoopla, background_runners, errors, abandoned)
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:
failures = _sanity_failures(
results, selected_sources, merged, output_path, abandoned
)
if failures:
# Publishing here is worse than publishing nothing: downstream (the
# enrich step, then the server) has no way to tell a gutted dataset from
# a real one, so the previous parquet stays and the run exits non-zero.
# The asking-price history is left alone too, since it is forward-only
# and would bake this run's gaps in permanently.
for failure in failures:
log.error("Sanity check failed: %s", failure)
log.error("Refusing to overwrite %s; keeping the previous data", output_path)
else:
# Accrue the per-listing asking-price history before writing: load the
# persistent store, append this run's price moves, dump it, then embed
# each listing's series in the parquet. Forward-only by nature — the
# each listing's series in the parquet. Forward-only by nature: the
# first run seeds one point per listing and reductions appear over time.
history_path = output_base / "price_history" / "listings.json"
history = load_history(history_path)
@ -853,10 +958,6 @@ def run_scrape(
update_history(history, merged, run_date)
save_history(history_path, history)
write_parquet(merged, output_path, price_history=history)
else:
if output_path.exists():
output_path.unlink()
log.warning("No London-ish properties to write to %s", output_path)
counts = {
"total": len(merged),
@ -881,6 +982,10 @@ def run_scrape(
},
"counts": counts,
"path": str(output_path),
# `errors` are per-outcode and survivable; `failures` mean the run's
# output was rejected and nothing was written. Only the latter is fatal.
"errors": errors,
"failures": failures,
"written": not failures,
"elapsed_seconds": round(time.time() - started_at, 3),
}

321
finder/test_block_detection.py Normal file
View file

@ -0,0 +1,321 @@
"""Egress-block detection: 403 storms must rotate the IP, then fail the run.
Regression cover for 2026-07-15, when Rightmove's edge 403'd every typeahead
call, fetch_with_retry returned None, resolve_outcode_id read that as "no such
outcode" for all 366 outcodes, and the run published a Rightmove-less parquet
with errors: [] and exit 0.
"""
import httpx
import pytest
import http_client
import scraper
from http_client import BlockedError, _EgressBlockTracker
class _StubResponse:
def __init__(self, status_code: int, payload: dict | None = None):
self.status_code = status_code
self._payload = payload or {}
def json(self) -> dict:
return self._payload
class _StubClient:
"""Replays a fixed status sequence, recording the URLs it was asked for."""
def __init__(self, statuses: list[int], payload: dict | None = None):
self._statuses = list(statuses)
self._payload = payload or {"matches": []}
self.calls: list[str] = []
def get(self, url, params=None, headers=None):
self.calls.append(url)
status = self._statuses.pop(0) if self._statuses else 200
return _StubResponse(status, self._payload)
@pytest.fixture(autouse=True)
def _no_sleeping(monkeypatch):
# Backoff delays would otherwise make this suite take minutes.
monkeypatch.setattr(http_client.shutdown, "sleep", lambda _s: None)
monkeypatch.setattr(http_client.RATE_LIMITER, "acquire", lambda: None)
@pytest.fixture(autouse=True)
def _fresh_tracker(monkeypatch):
tracker = _EgressBlockTracker(threshold=3, max_rotations=1)
monkeypatch.setattr(http_client, "BLOCK_TRACKER", tracker)
return tracker
def test_403_is_retried_not_swallowed(monkeypatch):
"""A transient 403 must not end the call: this is what broke last time."""
rotations = []
monkeypatch.setattr(http_client.gluetun, "rotate_ip", lambda: rotations.append(1))
client = _StubClient([403, 200], payload={"matches": ["ok"]})
result = http_client.fetch_with_retry(client, "https://los.example.com/typeahead")
assert result == {"matches": ["ok"]}
assert len(client.calls) == 2, "the 403 should have been retried"
assert rotations == [], "one 403 is not a block; rotating would be disruptive"
def test_sustained_403s_rotate_the_egress_ip(monkeypatch):
"""Threshold consecutive 403s from one host trigger exactly one rotation."""
rotations = []
def _rotate():
rotations.append(1)
return True
monkeypatch.setattr(http_client.gluetun, "rotate_ip", _rotate)
# 3 x 403 hits the threshold and rotates; the retry budget resets and the
# next call succeeds on the "new IP".
client = _StubClient([403, 403, 403, 200], payload={"matches": ["ok"]})
result = http_client.fetch_with_retry(client, "https://los.example.com/typeahead")
assert result == {"matches": ["ok"]}
assert rotations == [1], "expected exactly one rotation"
def test_403s_surviving_rotation_raise_blocked_error(monkeypatch):
"""Once the rotation budget is spent, a block must raise, never return None.
Returning None is what let resolve_outcode_id read a site-wide block as
"this outcode does not exist" and silently produce zero listings.
"""
monkeypatch.setattr(http_client.gluetun, "rotate_ip", lambda: True)
client = _StubClient([403] * 20)
with pytest.raises(BlockedError, match="refusing this egress IP"):
http_client.fetch_with_retry(client, "https://los.example.com/typeahead")
def test_failed_rotation_is_treated_as_blocked(monkeypatch):
"""If we cannot rotate away from a blocked IP, we are blocked. Say so."""
monkeypatch.setattr(http_client.gluetun, "rotate_ip", lambda: False)
client = _StubClient([403] * 10)
with pytest.raises(BlockedError):
http_client.fetch_with_retry(client, "https://los.example.com/typeahead")
def test_success_resets_the_consecutive_count(monkeypatch):
"""Interleaved successes mean the IP is fine, so 403s must not accumulate."""
rotations = []
monkeypatch.setattr(
http_client.gluetun, "rotate_ip", lambda: rotations.append(1) or True
)
# 403,200 repeated: never 3 consecutive, so never a rotation.
for _ in range(5):
client = _StubClient([403, 200], payload={"matches": ["ok"]})
http_client.fetch_with_retry(client, "https://los.example.com/typeahead")
assert rotations == []
def test_on_403_false_opts_out_of_block_detection(monkeypatch):
monkeypatch.setattr(http_client.gluetun, "rotate_ip", lambda: True)
client = _StubClient([403])
assert (
http_client.fetch_with_retry(client, "https://x.example.com/y", on_403=False)
is None
)
def test_403_tracking_is_per_host(monkeypatch):
"""A block on one host must not be charged against another."""
rotations = []
monkeypatch.setattr(
http_client.gluetun, "rotate_ip", lambda: rotations.append(1) or True
)
for host in ("a.example.com", "b.example.com", "c.example.com"):
client = _StubClient([403, 200], payload={"matches": []})
http_client.fetch_with_retry(client, f"https://{host}/typeahead")
assert rotations == [], "one 403 each across three hosts is not a block"
def test_connection_errors_still_retry(monkeypatch):
"""The rewritten loop must not regress non-403 retry behaviour."""
class _FlakyClient:
def __init__(self):
self.calls = 0
def get(self, url, params=None, headers=None):
self.calls += 1
if self.calls == 1:
raise httpx.ConnectError("boom")
return _StubResponse(200, {"matches": ["ok"]})
client = _FlakyClient()
assert http_client.fetch_with_retry(client, "https://x.example.com/y") == {
"matches": ["ok"]
}
assert client.calls == 2
# ---------------------------------------------------------------------------
# Run-level sanity gates
# ---------------------------------------------------------------------------
def test_zero_yield_from_a_selected_source_is_a_failure(tmp_path):
results = {"rightmove": [], "onthemarket": [{"a": 1}], "zoopla": []}
failures = scraper._sanity_failures(
results,
["rightmove", "onthemarket"],
merged=[{"a": 1}],
output_path=tmp_path / "missing.parquet",
abandoned=set(),
)
assert len(failures) == 1
assert "rightmove yielded 0 listings" in failures[0]
def test_healthy_run_has_no_failures(tmp_path):
results = {"rightmove": [{"a": 1}], "onthemarket": [{"b": 2}], "zoopla": []}
assert (
scraper._sanity_failures(
results,
["rightmove", "onthemarket"],
merged=[{"a": 1}, {"b": 2}],
output_path=tmp_path / "missing.parquet",
abandoned=set(),
)
== []
)
def test_unselected_source_yielding_zero_is_fine(tmp_path):
"""zoopla is not scheduled in production; its 0 must not fail the run."""
results = {"rightmove": [{"a": 1}], "onthemarket": [{"b": 2}], "zoopla": []}
assert (
scraper._sanity_failures(
results,
["rightmove", "onthemarket"],
merged=[{"a": 1}],
output_path=tmp_path / "missing.parquet",
abandoned=set(),
)
== []
)
def test_source_abandoned_mid_run_is_a_failure_even_with_listings(tmp_path):
"""A block that starts mid-run must fail even though the source is non-empty.
Without this gate a single banked listing defeats the zero-yield check: the
source stops at outcode 300 of 366, keeps its 60k listings, OnTheMarket
backfills the merged total to within the drop limit, and the partial
dataset publishes with exit 0. That is the original incident with one
outcode's head start.
"""
import polars as pl
path = tmp_path / "online_listings_buy.parquet"
pl.DataFrame({"x": range(103008)}).write_parquet(path)
results = {
"rightmove": [{"a": 1}] * 60000,
"onthemarket": [{"b": 2}] * 83785,
"zoopla": [],
}
failures = scraper._sanity_failures(
results,
["rightmove", "onthemarket"],
merged=[{"a": 1}] * 99000, # only a 4% dip: both other gates pass
output_path=path,
abandoned={"rightmove"},
)
assert len(failures) == 1
assert "abandoned mid-run" in failures[0]
assert "60000 listings cover only part" in failures[0]
def test_the_real_incident_drop_now_trips_the_collapse_gate(tmp_path):
"""103,008 -> 83,785 is an 18.7% drop, which the original 25% limit missed."""
import polars as pl
path = tmp_path / "online_listings_buy.parquet"
pl.DataFrame({"x": range(103008)}).write_parquet(path)
# Pretend rightmove banked listings so only the collapse gate can fire.
results = {
"rightmove": [{"a": 1}] * 100,
"onthemarket": [{"b": 2}] * 83785,
"zoopla": [],
}
failures = scraper._sanity_failures(
results,
["rightmove", "onthemarket"],
merged=[{"b": 2}] * 83785,
output_path=path,
abandoned=set(),
)
assert len(failures) == 1
assert "collapsed from 103008 to 83785" in failures[0]
def test_row_count_collapse_is_a_failure(tmp_path):
"""Backstop for degradation that is neither a clean zero nor a raised block.
Dedup masks a lost source (OnTheMarket backfilled the keys Rightmove used to
win), so a partial block can still look plausible per-source.
"""
import polars as pl
path = tmp_path / "online_listings_buy.parquet"
pl.DataFrame({"x": range(1000)}).write_parquet(path)
results = {"rightmove": [{"a": 1}], "onthemarket": [{"b": 2}], "zoopla": []}
failures = scraper._sanity_failures(
results,
["rightmove", "onthemarket"],
merged=[{"a": 1}] * 700, # 30% drop, over the 10% limit
output_path=path,
abandoned=set(),
)
assert len(failures) == 1
assert "collapsed from 1000 to 700" in failures[0]
def test_normal_market_movement_is_not_a_failure(tmp_path):
import polars as pl
path = tmp_path / "online_listings_buy.parquet"
pl.DataFrame({"x": range(1000)}).write_parquet(path)
results = {"rightmove": [{"a": 1}], "onthemarket": [{"b": 2}], "zoopla": []}
assert (
scraper._sanity_failures(
results,
["rightmove", "onthemarket"],
merged=[{"a": 1}] * 970, # 3% drop, within normal movement
output_path=path,
abandoned=set(),
)
== []
)
def test_first_ever_run_has_no_baseline_to_compare(tmp_path):
results = {"rightmove": [{"a": 1}], "onthemarket": [{"b": 2}], "zoopla": []}
assert (
scraper._sanity_failures(
results,
["rightmove", "onthemarket"],
merged=[{"a": 1}],
output_path=tmp_path / "does-not-exist.parquet",
abandoned=set(),
)
== []
)

8
finder/test_scraper_concurrency.py
View file

@ -31,7 +31,7 @@ def test_run_sources_runs_every_source_and_isolates_failures():
order.append("zoo")
errors: list[str] = []
scraper._run_sources(zoo, [("rightmove", rm), ("onthemarket", otm)], errors)
scraper._run_sources(zoo, [("rightmove", rm), ("onthemarket", otm)], errors, set())
assert set(order) == {"rm", "otm", "zoo"}
# The failing source is recorded but did not stop the others.
@ -45,14 +45,14 @@ def test_run_sources_records_zoopla_failure():
def zoo():
raise ValueError("zoo down")
scraper._run_sources(zoo, [], errors)
scraper._run_sources(zoo, [], errors, set())
assert any("zoopla" in e and "zoo down" in e for e in errors)
def test_run_sources_handles_no_background_runners():
ran = []
errors: list[str] = []
scraper._run_sources(lambda: ran.append("z"), [], errors)
scraper._run_sources(lambda: ran.append("z"), [], errors, set())
assert ran == ["z"]
assert errors == []
@ -60,7 +60,7 @@ def test_run_sources_handles_no_background_runners():
def test_run_sources_handles_zoopla_absent():
ran = []
errors: list[str] = []
scraper._run_sources(None, [("rightmove", lambda: ran.append("rm"))], errors)
scraper._run_sources(None, [("rightmove", lambda: ran.append("rm"))], errors, set())
assert ran == ["rm"]
assert errors == []

109
finder/zoopla.py
View file

@ -27,14 +27,11 @@ import time
from pathlib import Path
from urllib.parse import parse_qsl, urlencode, urljoin, urlparse, urlunparse
import httpx
import gluetun
import shutdown
from constants import (
DATA_DIR,
DELAY_BETWEEN_PAGES,
GLUETUN_API_KEY,
GLUETUN_CONTROL_URL,
GLUETUN_MAX_ROTATIONS,
GLUETUN_PROXY,
MAX_BEDROOMS,
@ -472,110 +469,16 @@ def _challenge_timeout_seconds() -> int:
# ---------------------------------------------------------------------------
#
# When Cloudflare Turnstile fires mid-scrape, the cheapest unblocker is to
# swap the egress IP via Gluetun's HTTP control server. We stop and re-start
# the VPN, poll until the public IP changes, drop the stale cf_clearance
# cookies (bound to the previous IP), then reload and re-check the challenge.
def _gluetun_base_url() -> str:
return GLUETUN_CONTROL_URL.rstrip("/")
def _gluetun_api_key() -> str | None:
return GLUETUN_API_KEY
# swap the egress IP via Gluetun's HTTP control server, then drop the stale
# cf_clearance cookies (bound to the previous IP) and re-check the challenge.
# The control-server plumbing itself lives in gluetun.py, shared with
# http_client.py's egress-block detection.
def _gluetun_max_rotations() -> int:
return max(GLUETUN_MAX_ROTATIONS, 0)
def _gluetun_client() -> httpx.Client:
# Talks to the control server directly (not through the VPN proxy).
headers = {}
api_key = _gluetun_api_key()
if api_key:
headers["X-API-Key"] = api_key
return httpx.Client(headers=headers)
def _gluetun_public_ip(client: httpx.Client) -> str | None:
try:
resp = client.get(f"{_gluetun_base_url()}/v1/publicip/ip", timeout=5.0)
if resp.status_code != 200:
return None
data = resp.json()
except (httpx.HTTPError, ValueError):
return None
return data.get("public_ip") or data.get("ip")
def _gluetun_set_vpn_status(client: httpx.Client, status: str) -> bool:
"""PUT /v1/vpn/status with {'status': status}. Returns True on 2xx."""
try:
resp = client.put(
f"{_gluetun_base_url()}/v1/vpn/status",
json={"status": status},
timeout=15.0,
)
except httpx.HTTPError as exc:
log.warning("Gluetun vpn/status %s failed: %s", status, exc)
return False
if resp.status_code == 401:
log.warning(
"Gluetun vpn/status %s: 401 Unauthorized. The API key must be "
"authorised for 'PUT /v1/vpn/status' in Gluetun's auth config.toml",
status,
)
return False
if resp.status_code >= 400:
log.warning(
"Gluetun vpn/status %s returned HTTP %d: %s",
status, resp.status_code, resp.text[:200],
)
return False
return True
def _rotate_gluetun_ip(wait_seconds: int = 45) -> bool:
"""Restart Gluetun's VPN and wait for the public IP to change.
Returns True if a new IP was observed within wait_seconds."""
with _gluetun_client() as client:
old_ip = _gluetun_public_ip(client)
log.info("Requesting Gluetun IP rotation (current IP: %s)", old_ip or "unknown")
stop_attempted = False
restart_confirmed = False
try:
stop_attempted = True
if not _gluetun_set_vpn_status(client, "stopped"):
return False
time.sleep(2)
restart_confirmed = _gluetun_set_vpn_status(client, "running")
if not restart_confirmed:
return False
deadline = time.monotonic() + wait_seconds
while time.monotonic() < deadline:
time.sleep(2)
new_ip = _gluetun_public_ip(client)
if new_ip and new_ip != old_ip:
log.info("Gluetun rotated IP: %s -> %s", old_ip or "?", new_ip)
return True
finally:
if stop_attempted and not restart_confirmed:
log.warning(
"Gluetun VPN may be stopped after failed rotation; attempting recovery start"
)
if not _gluetun_set_vpn_status(client, "running"):
log.error(
"Gluetun VPN recovery start failed; manual intervention required"
)
log.warning("Gluetun IP did not change within %ds", wait_seconds)
return False
def _clear_cloudflare_cookies(page) -> None:
"""Drop cf_clearance / __cf_bm which are bound to the previous egress IP."""
try:
@ -596,7 +499,7 @@ def _rotate_and_retry_challenge(page, max_rotations: int) -> bool:
"Cloudflare Turnstile challenge, rotating Gluetun IP (attempt %d/%d)",
attempt, max_rotations,
)
if not _rotate_gluetun_ip():
if not gluetun.rotate_ip():
continue
_clear_cloudflare_cookies(page)