0pt/services/api/workers/adaptive_router.py

"""
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)]