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