services.py

"""Backend orchestration, separated from HTTP wiring.

The policy in here - which backend to try, in what order, when to fall back
to another, what counts as evidence a backend is unhealthy, and when a
patent is genuinely absent rather than merely unreachable - is domain
policy. It would be equally true behind a CLI or a queue consumer, and it
was previously written inline in FastAPI route handlers, which meant
exercising any of it cost an ASGI round-trip.

The stateful collaborators (HTTP client, browser, circuit breaker, OPS
credentials) are constructor arguments rather than module globals reached
into at call time. That is what lets a test build a service with its own
circuit breaker instead of resetting a process-wide singleton between
tests, and hand in a stand-in for the OPS credentials instead of assigning
to private attributes on the real token manager.

The stateless backend functions (`query_arxiv` and friends) stay
module-level imports: they hold no state, so a test that wants to replace
one can monkeypatch it here without any of the problems shared mutable
state causes.

Services raise domain errors - `PatentNotFound`, `UpstreamUnavailable` -
rather than `HTTPException`. Mapping those onto status codes is the HTTP
layer's job, and lives in app.py.
"""

import asyncio
import logging
from typing import Callable, Optional

import httpx
from httpx import AsyncClient, HTTPStatusError
from playwright.async_api import Browser

from circuit_breaker import CircuitBreaker, CircuitOpenError
from ops import (OPSBulkResponse, ops_scrap_patent, ops_scrap_patent_bulk,
                 ops_search)
from scrap import (PatentScrapBulkResponse, PatentScrapResult,
                   scrap_patent_async, scrap_patent_bulk_async)
from serp import (EmptyResultsError, SerpQuery, SerpResults, query_arxiv,
                  query_bing_search, query_brave_search, query_ddg_search,
                  query_google_patents, query_google_scholar)
from utils import log_gathered_exceptions

logger = logging.getLogger(__name__)


# ================================ domain errors ================================


class PatentNotFound(Exception):
    """Every backend consulted positively said the document is absent.

    Distinct from UpstreamUnavailable on purpose: MCP_INSTRUCTIONS tells an
    agent that a not-found patent is genuinely absent everywhere and to
    move on without retrying, so this must never stand in for "we couldn't
    reach anyone".
    """


class UpstreamUnavailable(Exception):
    """A backend failed to answer. Says nothing about whether the document
    exists."""

    def __init__(self, message: str, *, timeout: bool = False):
        super().__init__(message)
        self.timeout = timeout


class OPSUnconfigured(Exception):
    """An OPS-only operation was requested without credentials configured."""


def _is_not_found(exc: BaseException) -> bool:
    """True only when a backend positively said the document is absent."""
    return isinstance(exc, HTTPStatusError) and exc.response.status_code == 404


def _as_upstream(exc: BaseException, message: str) -> UpstreamUnavailable:
    return UpstreamUnavailable(message, timeout=isinstance(exc, httpx.TimeoutException))


# ================================ result shaping ================================


def shape_serp_results(results: list) -> SerpResults:
    """Flatten a list of per-query result-lists (or Exceptions, from a
    `return_exceptions=True` gather) into one SerpResults, surfacing the
    last error only when every query failed.
    """
    if not results:
        # The request models bound the query list, so this is unreachable
        # from the endpoints; keep the helper total anyway rather than
        # indexing into an empty list, since every search path shares it.
        return SerpResults(results=[], error="No queries were provided.")

    filtered_results = [r for r in results if not isinstance(r, Exception)]
    flattened_results = [
        item for sublist in filtered_results for item in sublist]

    if len(filtered_results) == 0:
        return SerpResults(results=[], error=str(results[-1]))

    return SerpResults(results=flattened_results, error=None)


# ================================= search service =================================


class SearchService:
    """Runs SERP queries across the available backends."""

    def __init__(self, *, http_client: AsyncClient,
                 browser_provider: Callable[[], Optional[Browser]],
                 circuit_breaker: CircuitBreaker, ops_tokens):
        self._http = http_client
        # A provider rather than the browser itself: the browser is started
        # by the app's lifespan, after this service may already exist, and
        # is None when Playwright failed to start.
        self._browser_provider = browser_provider
        self._breaker = circuit_breaker
        self._ops_tokens = ops_tokens

    @property
    def browser(self) -> Optional[Browser]:
        return self._browser_provider()

    async def _run_queries(self, fn, params: SerpQuery, context: str) -> SerpResults:
        """Run `fn(query, n_results)` concurrently for every query, log any
        failures, and shape the result the way every simple search path
        expects.
        """
        results = await asyncio.gather(
            *[fn(q, params.n_results) for q in params.queries], return_exceptions=True)
        log_gathered_exceptions(results, context, params.queries)
        return shape_serp_results(results)

    # ------------------------------ single backend ------------------------------

    async def google_scholar(self, params: SerpQuery) -> SerpResults:
        return await self._run_queries(
            lambda q, n: query_google_scholar(self.browser, q, n), params,
            "google scholar search")

    async def arxiv(self, params: SerpQuery) -> SerpResults:
        return await self._run_queries(
            lambda q, n: query_arxiv(self._http, q, n), params, "arxiv search")

    async def brave(self, params: SerpQuery) -> SerpResults:
        return await self._run_queries(
            lambda q, n: query_brave_search(self.browser, q, n), params, "brave search")

    async def bing(self, params: SerpQuery) -> SerpResults:
        return await self._run_queries(
            lambda q, n: query_bing_search(self.browser, q, n), params, "bing search")

    async def duckduckgo(self, params: SerpQuery) -> SerpResults:
        return await self._run_queries(
            lambda q, n: query_ddg_search(q, n), params, "duckduckgo search")

    async def ops_keyword_search(self, params: SerpQuery) -> SerpResults:
        return await self._run_queries(
            lambda q, n: ops_search(self._http, q, n), params, "OPS search")

    # -------------------------------- patents --------------------------------

    async def patents(self, params: SerpQuery) -> SerpResults:
        """Google Patents, falling back to EPO OPS for any query it returns
        nothing for (when OPS credentials are configured)."""
        results = await asyncio.gather(
            *[query_google_patents(self.browser, q, params.n_results)
              for q in params.queries],
            return_exceptions=True)
        log_gathered_exceptions(results, "google patent search", params.queries)

        # Gathered rather than awaited one at a time: this path exists for
        # when Google Patents is unavailable, so it is exactly when a
        # request can least afford N sequential round-trips to the slower
        # backend.
        if self._ops_tokens.configured:
            needs_fallback = [
                i for i, res in enumerate(results)
                if isinstance(res, Exception) or not res]
            if needs_fallback:
                logger.info(
                    f"Google Patents empty for {len(needs_fallback)} quer(y/ies), trying OPS.")
                fallback = await asyncio.gather(
                    *[ops_search(self._http, params.queries[i], params.n_results)
                      for i in needs_fallback],
                    return_exceptions=True)
                # strict: gather returns exactly one result per index, so a
                # length mismatch here would be a bug worth surfacing.
                for i, res in zip(needs_fallback, fallback, strict=True):
                    if isinstance(res, Exception):
                        logger.warning(f"OPS fallback failed for `{params.queries[i]}`: {res}")
                    else:
                        results[i] = res

        return shape_serp_results(results)

    # ----------------------------- all backends -----------------------------

    async def _search_one(self, q: str, n_results: int) -> tuple[str, list[dict], Optional[str]]:
        """Try DDG, then Brave, then Bing for a single query; stop at the
        first backend that actually returns something.

        A backend that answers with an empty list has not answered the
        question - Brave does that whenever its block page lacks the word
        "suspicious", and Bing whenever no result item yields a title and
        href - so the chain continues rather than reporting success with
        zero results.

        When nothing produces results, the error names what each backend
        did. "Failed", "returned nothing" and "never asked" are different
        problems that lead somewhere different, and collapsing them into
        one guess is how this previously reported rate limiting it had no
        evidence for.
        """
        backends = [
            ("DuckDuckGo", lambda: query_ddg_search(q, n_results)),
            ("Brave Search", lambda: query_brave_search(self.browser, q, n_results)),
            ("Bing", lambda: query_bing_search(self.browser, q, n_results)),
        ]
        outcomes: list[str] = []
        for name, call in backends:
            try:
                self._breaker.before_call(name)
            except CircuitOpenError as e:
                logger.info(f"Skipping {name} for query `{q}`: {e}")
                outcomes.append(f"{name}: skipped ({e})")
                continue

            try:
                logger.info(f"Querying {name} with query: `{q}`")
                result = await call()
            except EmptyResultsError:
                # Zero results is ambiguous - no hits, or a soft block that
                # parsed to nothing. Move on, but don't hold it against
                # this backend: the breaker is shared across every request,
                # so counting it would let a few unusual-but-legitimate
                # queries disable the primary backend for everyone. A hard
                # block normally raises a real error, handled below.
                logger.info(f"{name} returned no results for query `{q}`")
                outcomes.append(f"{name}: no results")
                continue
            except Exception as e:
                logger.error(f"Failed to query {name} with query `{q}`: {e}")
                self._breaker.record_failure(name)
                outcomes.append(f"{name}: failed ({e})")
                continue

            if result:
                self._breaker.record_success(name)
                return q, result, None

            # Answered, but with nothing. Same ambiguity as the exception
            # above, so it counts neither for nor against the backend -
            # recording it as a success would reset one part-way to
            # tripping its breaker, which is how a soft-blocking backend
            # stays in rotation forever.
            logger.info(f"{name} returned an empty result set for query `{q}`")
            outcomes.append(f"{name}: no results")

        return q, [], f"No results for query '{q}' ({'; '.join(outcomes)})"

    async def search(self, params: SerpQuery) -> SerpResults:
        """Search every query across the full backend fallback chain."""
        outcomes = await asyncio.gather(
            *[self._search_one(q, params.n_results) for q in params.queries])

        results: list[dict] = []
        errors: list[str] = []
        for _q, res, err in outcomes:
            results.extend(res)
            if err:
                errors.append(err)

        if len(results) == 0:
            # _search_one always explains itself when it produces nothing,
            # so `errors` is populated here; the fallback text is a plain
            # statement of fact rather than the invented rate-limit
            # diagnosis it replaced.
            return SerpResults(
                results=[],
                error="; ".join(errors) if errors else "No results found.")

        return SerpResults(results=results, error="; ".join(errors) if errors else None)


# ================================= patent service =================================


class PatentService:
    """Retrieves patent full text, from Google Patents or EPO OPS."""

    def __init__(self, *, http_client: AsyncClient, ops_tokens):
        self._http = http_client
        self._ops_tokens = ops_tokens

    async def scrap(self, patent_id: str) -> PatentScrapResult:
        """Scrape from Google Patents, falling back to EPO OPS.

        Raises PatentNotFound only when every backend consulted said the
        document is absent; anything else raises UpstreamUnavailable.
        """
        google_error: BaseException
        try:
            return await scrap_patent_async(
                self._http, f"https://patents.google.com/patent/{patent_id}/en")
        except HTTPStatusError as e:
            google_error = e
            logger.warning(
                f"Google Patents returned {e.response.status_code} for {patent_id}.")
        except Exception as e:
            google_error = e
            logger.warning(f"Failed to scrap patent {patent_id}: {e}")

        if not self._ops_tokens.configured:
            if _is_not_found(google_error):
                raise PatentNotFound(
                    f"Patent '{patent_id}' not found on Google Patents "
                    "(EPO OPS fallback not configured).")
            raise _as_upstream(
                google_error,
                f"Google Patents request failed for '{patent_id}' and the EPO OPS "
                f"fallback is not configured: {google_error}")

        try:
            logger.info(f"Trying OPS for patent {patent_id}.")
            return await ops_scrap_patent(self._http, patent_id)
        except Exception as e:
            logger.warning(f"OPS fallback failed for {patent_id}: {e}")
            # Only claim the patent doesn't exist when *both* backends said so.
            if _is_not_found(google_error) and _is_not_found(e):
                raise PatentNotFound(
                    f"Patent '{patent_id}' not found on Google Patents or EPO OPS.") from e
            if isinstance(e, HTTPStatusError):
                raise _as_upstream(
                    e, f"EPO OPS returned {e.response.status_code} for '{patent_id}'.") from e
            raise _as_upstream(e, f"EPO OPS request failed for '{patent_id}': {e}") from e

    async def scrap_bulk(self, patent_ids: list[str]) -> PatentScrapBulkResponse:
        return await scrap_patent_bulk_async(self._http, patent_ids)

    async def ops_scrap(self, patent_id: str) -> PatentScrapResult:
        """Retrieve via EPO OPS only, with no Google Patents fallback."""
        if not self._ops_tokens.configured:
            raise OPSUnconfigured(
                "EPO OPS is not configured (OPS_CONSUMER_KEY / OPS_CONSUMER_SECRET missing).")
        try:
            return await ops_scrap_patent(self._http, patent_id)
        except HTTPStatusError as e:
            if e.response.status_code == 404:
                # OPS is the only backend asked here, so its 404 is the
                # whole answer.
                raise PatentNotFound(
                    f"Patent '{patent_id}' not found in EPO OPS.") from e
            raise _as_upstream(
                e, f"EPO OPS returned {e.response.status_code} for '{patent_id}'.") from e
        except Exception as e:
            logger.warning(f"Failed to retrieve patent {patent_id} from OPS: {e}")
            raise _as_upstream(e, f"EPO OPS request failed for '{patent_id}': {e}") from e

    async def ops_scrap_bulk(self, patent_ids: list[str]) -> OPSBulkResponse:
        return await ops_scrap_patent_bulk(self._http, patent_ids)