""" netutil.py — shared network safety: error classification + an adaptive, thread-safe throttle / circuit breaker used by scraper.py and discover.py. Goal: NEVER hammer a store into blocking our IP. Being slow is fine; getting blocked is not. So the moment we see a rate-limit signal (or a run of unexplained errors), every worker pauses for a cooldown that grows on repeat and relaxes again once requests succeed. """ import logging import re import threading import time log = logging.getLogger("net") # HTTP statuses that mean "you're going too fast / not allowed" -> back off hard. RATE_LIMIT_CODES = {403, 429} # Temporary server hiccups -> short retry. TRANSIENT_CODES = {408, 500, 502, 503, 504} def http_status(exc: Exception): """Best-effort extract an HTTP status code from any exception.""" resp = getattr(exc, "response", None) code = getattr(resp, "status_code", None) if isinstance(code, int): return code m = re.search(r"\b(\d{3})\b", str(exc)) # libs often embed the code in the message return int(m.group(1)) if m else None def is_not_found(exc: Exception) -> bool: if http_status(exc) == 404: return True return "notfound" in type(exc).__name__.lower() def is_rate_limited(exc: Exception) -> bool: if http_status(exc) in RATE_LIMIT_CODES: return True msg = str(exc).lower() return ( "429" in msg or "too many requests" in msg or ("rate" in msg and "limit" in msg) or "quota" in msg or "throttl" in msg ) def is_transient(exc: Exception) -> bool: if http_status(exc) in TRANSIENT_CODES: return True name = type(exc).__name__.lower() return any(w in name for w in ("timeout", "connection", "ssl", "chunked")) class Throttle: """ Adaptive, shared-across-threads throttle + circuit breaker. - wait() : call BEFORE each request; blocks while a cooldown is active. - record_success() : call after a good response; relaxes the throttle. - record_fail(rate) : call after a failure; trips the cooldown if it's a rate-limit OR enough consecutive failures pile up. """ def __init__(self, cooldown_start: float = 30.0, cooldown_max: float = 300.0, fail_threshold: int = 5): self.lock = threading.Lock() self.cooldown_start = cooldown_start self.cooldown_max = cooldown_max self.fail_threshold = fail_threshold self.current_cooldown = cooldown_start self.pause_until = 0.0 # monotonic time until which everyone waits self.extra_delay = 0.0 # added spacing that lingers after a trip self.consecutive_fail = 0 def wait(self) -> None: while True: with self.lock: remaining = self.pause_until - time.monotonic() extra = self.extra_delay if remaining > 0: time.sleep(min(remaining, 5.0)) continue break if extra: time.sleep(extra) def record_success(self) -> None: with self.lock: self.consecutive_fail = 0 if self.extra_delay > 0: self.extra_delay = max(0.0, self.extra_delay - 0.25) if self.current_cooldown > self.cooldown_start: self.current_cooldown = max( self.cooldown_start, self.current_cooldown * 0.9 ) def record_fail(self, rate_limited: bool) -> float: """Return the cooldown (seconds) if the breaker tripped, else 0.""" with self.lock: self.consecutive_fail += 1 tripped = rate_limited or self.consecutive_fail >= self.fail_threshold if not tripped: return 0.0 cd = self.current_cooldown self.pause_until = max(self.pause_until, time.monotonic() + cd) self.current_cooldown = min(self.current_cooldown * 2, self.cooldown_max) self.extra_delay = min(self.extra_delay + 1.0, 15.0) self.consecutive_fail = 0 return cd def retry_call(fn, *args, retries: int = 6, base: float = 2.0, max_wait: float = 120.0, throttle: "Throttle | None" = None): """ Sequential call-with-retry. Waits out rate limits (slow but safe). Re-raises not-found immediately (don't retry a healthy 404) and re-raises after the retry budget is exhausted. """ attempt = 0 while True: if throttle: throttle.wait() try: result = fn(*args) if throttle: throttle.record_success() return result except Exception as exc: if is_not_found(exc): if throttle: throttle.record_success() raise rate = is_rate_limited(exc) if throttle: throttle.record_fail(rate) attempt += 1 if attempt > retries: raise wait = min(base * (2 ** (attempt - 1)), max_wait) if rate: wait = max(wait, 30.0) # rate limits: wait it out, never rush back log.warning("retry %d/%d in %.0fs (%s)", attempt, retries, wait, exc) time.sleep(wait)