agent_search_gateway.providers.web.scrape_do
Scrape.do Google Search plugin and Markdown scrape adapter.
1"""Scrape.do Google Search plugin and Markdown 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 ScrapeDoAdapter: 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 params = urlencode({"token": self._secret.reveal(), "q": query}) 37 request_url = f"{endpoint(self._api_url, '/plugin/google/search')}?{params}" 38 payload = await self._http.request_json("GET", request_url, stage="search") 39 root = require_object(payload, self.name, "search", "response") 40 if root.get("error") is not None: 41 raise failure(self.name, "search", "provider reported failure") 42 results = require_list(root.get("organic_results"), self.name, "search", "organic_results") 43 hits: list[KeywordSearchHit] = [] 44 for item in results: 45 try: 46 result = require_object(item, self.name, "search", "result") 47 hits.append( 48 KeywordSearchHit( 49 url=non_empty_string( 50 result.get("link"), self.name, "search", "result.link" 51 ), 52 title=optional_string( 53 result.get("title"), self.name, "search", "result.title" 54 ), 55 snippet=optional_string( 56 result.get("snippet"), self.name, "search", "result.snippet" 57 ), 58 ) 59 ) 60 except ExecutionFailure: 61 continue 62 return hits 63 64 async def fetch(self, url: NormalizedURL) -> URLFetchCandidate: 65 params = urlencode({"token": self._secret.reveal(), "url": str(url), "output": "markdown"}) 66 request_url = f"{self._api_url}?{params}" 67 text = await self._http.request_text("GET", request_url, stage="fetch") 68 if not text.strip(): 69 raise failure(self.name, "fetch", "page body is empty") 70 return URLFetchCandidate(text, text)
class
ScrapeDoAdapter:
22class ScrapeDoAdapter: 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 params = urlencode({"token": self._secret.reveal(), "q": query}) 38 request_url = f"{endpoint(self._api_url, '/plugin/google/search')}?{params}" 39 payload = await self._http.request_json("GET", request_url, stage="search") 40 root = require_object(payload, self.name, "search", "response") 41 if root.get("error") is not None: 42 raise failure(self.name, "search", "provider reported failure") 43 results = require_list(root.get("organic_results"), self.name, "search", "organic_results") 44 hits: list[KeywordSearchHit] = [] 45 for item in results: 46 try: 47 result = require_object(item, self.name, "search", "result") 48 hits.append( 49 KeywordSearchHit( 50 url=non_empty_string( 51 result.get("link"), self.name, "search", "result.link" 52 ), 53 title=optional_string( 54 result.get("title"), self.name, "search", "result.title" 55 ), 56 snippet=optional_string( 57 result.get("snippet"), self.name, "search", "result.snippet" 58 ), 59 ) 60 ) 61 except ExecutionFailure: 62 continue 63 return hits 64 65 async def fetch(self, url: NormalizedURL) -> URLFetchCandidate: 66 params = urlencode({"token": self._secret.reveal(), "url": str(url), "output": "markdown"}) 67 request_url = f"{self._api_url}?{params}" 68 text = await self._http.request_text("GET", request_url, stage="fetch") 69 if not text.strip(): 70 raise failure(self.name, "fetch", "page body is empty") 71 return URLFetchCandidate(text, text)
ScrapeDoAdapter( *, 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 params = urlencode({"token": self._secret.reveal(), "q": query}) 38 request_url = f"{endpoint(self._api_url, '/plugin/google/search')}?{params}" 39 payload = await self._http.request_json("GET", request_url, stage="search") 40 root = require_object(payload, self.name, "search", "response") 41 if root.get("error") is not None: 42 raise failure(self.name, "search", "provider reported failure") 43 results = require_list(root.get("organic_results"), self.name, "search", "organic_results") 44 hits: list[KeywordSearchHit] = [] 45 for item in results: 46 try: 47 result = require_object(item, self.name, "search", "result") 48 hits.append( 49 KeywordSearchHit( 50 url=non_empty_string( 51 result.get("link"), self.name, "search", "result.link" 52 ), 53 title=optional_string( 54 result.get("title"), self.name, "search", "result.title" 55 ), 56 snippet=optional_string( 57 result.get("snippet"), self.name, "search", "result.snippet" 58 ), 59 ) 60 ) 61 except ExecutionFailure: 62 continue 63 return hits
async def
fetch( self, url: agent_search_gateway.url_normalization.NormalizedURL) -> agent_search_gateway.providers.contracts.URLFetchCandidate:
65 async def fetch(self, url: NormalizedURL) -> URLFetchCandidate: 66 params = urlencode({"token": self._secret.reveal(), "url": str(url), "output": "markdown"}) 67 request_url = f"{self._api_url}?{params}" 68 text = await self._http.request_text("GET", request_url, stage="fetch") 69 if not text.strip(): 70 raise failure(self.name, "fetch", "page body is empty") 71 return URLFetchCandidate(text, text)