agent_search_gateway.providers.web.scraperapi
ScraperAPI synchronous Google SERP and page scrape adapter.
1"""ScraperAPI synchronous Google SERP and page scrape adapter.""" 2 3from urllib.parse import urlencode 4 5from ...errors import ExecutionFailure 6from ...observability import SecretValue 7from ...providers.contracts import KeywordSearchHit, URLFetchCandidate 8from ...url_normalization import NormalizedURL 9from .common import ( 10 HttpRequester, 11 configured_string, 12 endpoint, 13 failure, 14 non_empty_string, 15 optional_string, 16 require_list, 17 require_object, 18) 19 20 21class ScraperAPIAdapter: 22 def __init__( 23 self, 24 *, 25 name: str, 26 api_url: str, 27 secret: SecretValue, 28 http_executor: HttpRequester, 29 ) -> None: 30 self.name = name 31 self._api_url = configured_string(api_url, "api_url").rstrip("/") 32 self._secret = secret 33 self._http = http_executor 34 35 async def search(self, query: str) -> list[KeywordSearchHit]: 36 request_url = ( 37 f"{endpoint(self._api_url, '/structured/google/search')}?" 38 f"{urlencode({'api_key': self._secret.reveal(), 'query': query})}" 39 ) 40 payload = await self._http.request_json("GET", request_url, stage="search") 41 root = require_object(payload, self.name, "search", "response") 42 if root.get("error") is not None: 43 raise failure(self.name, "search", "provider reported failure") 44 results = require_list(root.get("organic_results"), self.name, "search", "organic_results") 45 hits: list[KeywordSearchHit] = [] 46 for item in results: 47 try: 48 result = require_object(item, self.name, "search", "result") 49 hits.append( 50 KeywordSearchHit( 51 url=non_empty_string( 52 result.get("link"), self.name, "search", "result.link" 53 ), 54 title=optional_string( 55 result.get("title"), self.name, "search", "result.title" 56 ), 57 snippet=optional_string( 58 result.get("snippet"), self.name, "search", "result.snippet" 59 ), 60 ) 61 ) 62 except ExecutionFailure: 63 continue 64 return hits 65 66 async def fetch(self, url: NormalizedURL) -> URLFetchCandidate: 67 params = urlencode({"api_key": self._secret.reveal(), "url": str(url)}) 68 request_url = f"{self._api_url}?{params}" 69 text = await self._http.request_text("GET", request_url, stage="fetch") 70 if not text.strip(): 71 raise failure(self.name, "fetch", "page body is empty") 72 return URLFetchCandidate(text, text)
class
ScraperAPIAdapter:
22class ScraperAPIAdapter: 23 def __init__( 24 self, 25 *, 26 name: str, 27 api_url: str, 28 secret: SecretValue, 29 http_executor: HttpRequester, 30 ) -> None: 31 self.name = name 32 self._api_url = configured_string(api_url, "api_url").rstrip("/") 33 self._secret = secret 34 self._http = http_executor 35 36 async def search(self, query: str) -> list[KeywordSearchHit]: 37 request_url = ( 38 f"{endpoint(self._api_url, '/structured/google/search')}?" 39 f"{urlencode({'api_key': self._secret.reveal(), 'query': query})}" 40 ) 41 payload = await self._http.request_json("GET", request_url, stage="search") 42 root = require_object(payload, self.name, "search", "response") 43 if root.get("error") is not None: 44 raise failure(self.name, "search", "provider reported failure") 45 results = require_list(root.get("organic_results"), self.name, "search", "organic_results") 46 hits: list[KeywordSearchHit] = [] 47 for item in results: 48 try: 49 result = require_object(item, self.name, "search", "result") 50 hits.append( 51 KeywordSearchHit( 52 url=non_empty_string( 53 result.get("link"), self.name, "search", "result.link" 54 ), 55 title=optional_string( 56 result.get("title"), self.name, "search", "result.title" 57 ), 58 snippet=optional_string( 59 result.get("snippet"), self.name, "search", "result.snippet" 60 ), 61 ) 62 ) 63 except ExecutionFailure: 64 continue 65 return hits 66 67 async def fetch(self, url: NormalizedURL) -> URLFetchCandidate: 68 params = urlencode({"api_key": self._secret.reveal(), "url": str(url)}) 69 request_url = f"{self._api_url}?{params}" 70 text = await self._http.request_text("GET", request_url, stage="fetch") 71 if not text.strip(): 72 raise failure(self.name, "fetch", "page body is empty") 73 return URLFetchCandidate(text, text)
ScraperAPIAdapter( *, name: str, api_url: str, secret: agent_search_gateway.observability.SecretValue, http_executor: agent_search_gateway.providers.web.common.HttpRequester)
async def
search( self, query: str) -> list[agent_search_gateway.providers.contracts.KeywordSearchHit]:
36 async def search(self, query: str) -> list[KeywordSearchHit]: 37 request_url = ( 38 f"{endpoint(self._api_url, '/structured/google/search')}?" 39 f"{urlencode({'api_key': self._secret.reveal(), 'query': query})}" 40 ) 41 payload = await self._http.request_json("GET", request_url, stage="search") 42 root = require_object(payload, self.name, "search", "response") 43 if root.get("error") is not None: 44 raise failure(self.name, "search", "provider reported failure") 45 results = require_list(root.get("organic_results"), self.name, "search", "organic_results") 46 hits: list[KeywordSearchHit] = [] 47 for item in results: 48 try: 49 result = require_object(item, self.name, "search", "result") 50 hits.append( 51 KeywordSearchHit( 52 url=non_empty_string( 53 result.get("link"), self.name, "search", "result.link" 54 ), 55 title=optional_string( 56 result.get("title"), self.name, "search", "result.title" 57 ), 58 snippet=optional_string( 59 result.get("snippet"), self.name, "search", "result.snippet" 60 ), 61 ) 62 ) 63 except ExecutionFailure: 64 continue 65 return hits
async def
fetch( self, url: agent_search_gateway.url_normalization.NormalizedURL) -> agent_search_gateway.providers.contracts.URLFetchCandidate:
67 async def fetch(self, url: NormalizedURL) -> URLFetchCandidate: 68 params = urlencode({"api_key": self._secret.reveal(), "url": str(url)}) 69 request_url = f"{self._api_url}?{params}" 70 text = await self._http.request_text("GET", request_url, stage="fetch") 71 if not text.strip(): 72 raise failure(self.name, "fetch", "page body is empty") 73 return URLFetchCandidate(text, text)