119 lines
3.9 KiB
Python
119 lines
3.9 KiB
Python
import logging
|
|
import random
|
|
import threading
|
|
import time
|
|
|
|
import httpx
|
|
from fake_useragent import UserAgent
|
|
|
|
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
|
|
)
|
|
|
|
|
|
def make_client() -> httpx.Client:
|
|
# Route through the Gluetun HTTP proxy (VPN egress) when configured.
|
|
return httpx.Client(
|
|
timeout=30,
|
|
headers={"User-Agent": _ua.random, "Accept": "application/json"},
|
|
follow_redirects=True,
|
|
proxy=GLUETUN_PROXY or None,
|
|
)
|
|
|
|
|
|
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.
|
|
|
|
Returns None on permanent failure. The on_403 argument is kept for
|
|
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()
|
|
if resp.status_code == 403 and on_403:
|
|
log.error("HTTP 403 from %s (forbidden)", url)
|
|
return None
|
|
if resp.status_code in (429, 500, 502, 503, 504):
|
|
delay = RETRY_BASE_DELAY * (2**attempt) + random.uniform(0, 1)
|
|
log.warning(
|
|
"HTTP %d from %s, retry %d/%d in %.1fs",
|
|
resp.status_code,
|
|
url,
|
|
attempt + 1,
|
|
MAX_RETRIES,
|
|
delay,
|
|
)
|
|
shutdown.sleep(delay)
|
|
continue
|
|
log.error("HTTP %d from %s (non-retryable)", resp.status_code, url)
|
|
return None
|
|
except (
|
|
httpx.ConnectError,
|
|
httpx.ReadTimeout,
|
|
httpx.WriteTimeout,
|
|
httpx.PoolTimeout,
|
|
) as e:
|
|
delay = RETRY_BASE_DELAY * (2**attempt) + random.uniform(0, 1)
|
|
log.warning(
|
|
"%s from %s, retry %d/%d in %.1fs",
|
|
type(e).__name__,
|
|
url,
|
|
attempt + 1,
|
|
MAX_RETRIES,
|
|
delay,
|
|
)
|
|
shutdown.sleep(delay)
|
|
log.error("All %d retries exhausted for %s", MAX_RETRIES, url)
|
|
return None
|