""" Adaptive router — per-market compliant-source planner. P014 CORRECTION: this file's previous docstring read "all methods, no compliance restrictions. See b0ts for scraping." That phrasing directly contradicts this product's own compliance stance (CLAUDE.md §1: official API/bulk -> authorized public web -> licensed partner data -> otherwise level 0, never a bypass) and, worse, pointed a future reader at `b0ts` -- a SEPARATE repo that does non-compliant scraping and is out of scope/forbidden to touch from here. Nothing in this file ever called b0ts or any bypass path; the line was simply wrong and is corrected, not extended. What this file actually does: `plan_for_market(market)` returns the ordered list of registered source_ids `pipeline.py._fetch` tries, in priority order, for a given market. Every entry is a name pipeline.py may or may not have an implemented connector for (see sources/connectors.py's CONNECTORS dict) -- an unimplemented source_id is simply skipped by `_fetch`, it is never a license to build a bypass in its place. P014 ADDITION: `record_blocked` / `is_blocked` give this planner a memory of which (market, source_id) pairs recently told the engine to stop (HTTP 401/403/429/451 -- see sources/connectors.py's SourceBlockedError). Before this pass, a source that started rate-limiting or blocking the engine mid-run was retried on every subsequent record with no memory at all -- not a compliance bypass, but not "respecting the block as a stop signal" either (the same principle 0pt-motor's AdaptiveRouter documents for its own, separate method-tier selector). `plan_for_market` now filters a currently- blocked source out of the plan for the rest of the block's cooldown window, so the engine backs off instead of hammering it. This is process-local, in-memory state (matches every other piece of in-memory state this engine documents honestly elsewhere, e.g. orchestrator/main.py's _ORDERS/_JOBS) -- not a persistent, cross-process reputation system. """ from __future__ import annotations import time MARKET_SOURCES: dict[str, list[str]] = { "NO": ["brreg", "brreg_under", "proff", "1881", "gulesider", "flowfirma", "selskapssok", "regnskapsbasen", "okonomi", "visverdi", "gleif"], "GB": ["companies_house", "gleif"], "FI": ["prh", "gleif"], "DK": ["cvr", "krak", "proff_dk", "118", "degulesider", "companydata", "grundigt", "datafordeler", "virk", "gleif"], "DE": ["northdata", "handelsregister", "unternehmensregister", "moneyhouse", "creditreform", "bundesanzeiger", "registercheck", "personensuche", "whitepages", "companyhouse", "gleif"], "SE": ["gleif"], } # How long a (market, source_id) pair is excluded from plan_for_market after # being reported blocked. A fixed cooldown, not exponential backoff -- this # is a planner, not a retry/backoff controller; a source is free to be tried # again on the next run once the window elapses. DEFAULT_BLOCK_COOLDOWN_SECONDS = 300.0 class AdaptiveRouter: """Tracks which (market, source_id) pairs are currently in a block cooldown.""" def __init__(self, cooldown_seconds: float = DEFAULT_BLOCK_COOLDOWN_SECONDS): self._cooldown = cooldown_seconds self._blocked_until: dict[tuple[str, str], float] = {} def record_blocked(self, market: str, source_id: str, *, now: float | None = None) -> None: """Mark (market, source_id) as blocked starting now.""" self._blocked_until[(market, source_id)] = (now if now is not None else time.monotonic()) + self._cooldown def is_blocked(self, market: str, source_id: str, *, now: float | None = None) -> bool: deadline = self._blocked_until.get((market, source_id)) if deadline is None: return False current = now if now is not None else time.monotonic() if current >= deadline: # Cooldown elapsed -- clear it so the map doesn't grow forever. del self._blocked_until[(market, source_id)] return False return True def clear(self) -> None: """Test/process-reset hook -- forgets every recorded block.""" self._blocked_until.clear() # Process-wide singleton, mirroring the in-memory-state convention already # used elsewhere in this engine (e.g. orchestrator/main.py's _ORDERS/_JOBS). # Injectable in `plan_for_market` for tests rather than only reachable via # global mutable state. default_router = AdaptiveRouter() def plan_for_market(market: str, *, router: AdaptiveRouter | None = None) -> list[str]: """ Return the ordered source plan for a market, minus any source_id currently in a block cooldown for that market. `router` defaults to the process-wide `default_router`; tests inject their own instance so they never share state with each other or with production. """ r = router if router is not None else default_router plan = MARKET_SOURCES.get(market, ["gleif"]) return [source_id for source_id in plan if not r.is_blocked(market, source_id)]