Edit on GitHub

agent_search_gateway.providers.web.scrapegraphai

ScrapeGraphAI v2 Search and Scrape adapter.

View Source

 1"""ScrapeGraphAI v2 Search and Scrape adapter."""
 2
 3from ...errors import ExecutionFailure
 4from ...observability import SecretValue
 5from ...providers.contracts import KeywordSearchHit, URLFetchCandidate
 6from ...url_normalization import NormalizedURL
 7from .common import (
 8    JsonRequester,
 9    configured_string,
10    endpoint,
11    failure,
12    non_empty_string,
13    optional_string,
14    require_list,
15    require_object,
16)
17
18
19class ScrapeGraphAIAdapter:
20    def __init__(
21        self,
22        *,
23        name: str,
24        api_url: str,
25        secret: SecretValue,
26        http_executor: JsonRequester,
27    ) -> None:
28        self.name = name
29        self._api_url = configured_string(api_url, "api_url").rstrip("/")
30        self._secret = secret
31        self._http = http_executor
32
33    @property
34    def _headers(self) -> dict[str, str]:
35        return {"SGAI-APIKEY": self._secret.reveal()}
36
37    async def search(self, query: str) -> list[KeywordSearchHit]:
38        payload = await self._http.request_json(
39            "POST",
40            endpoint(self._api_url, "/api/search"),
41            stage="search",
42            headers=self._headers,
43            json_body={"query": query},
44        )
45        root = require_object(payload, self.name, "search", "response")
46        if root.get("error") is not None or root.get("detail") is not None:
47            raise failure(self.name, "search", "provider reported failure")
48        results = require_list(root.get("results"), self.name, "search", "results")
49        hits: list[KeywordSearchHit] = []
50        for item in results:
51            try:
52                result = require_object(item, self.name, "search", "result")
53                content = optional_string(
54                    result.get("content"), self.name, "search", "result.content"
55                )
56                hits.append(
57                    KeywordSearchHit(
58                        url=non_empty_string(result.get("url"), self.name, "search", "result.url"),
59                        title=optional_string(
60                            result.get("title"), self.name, "search", "result.title"
61                        ),
62                        snippet="",
63                        raw_content=content,
64                        content=content,
65                    )
66                )
67            except ExecutionFailure:
68                continue
69        return hits
70
71    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate:
72        payload = await self._http.request_json(
73            "POST",
74            endpoint(self._api_url, "/api/scrape"),
75            stage="fetch",
76            headers=self._headers,
77            json_body={"url": str(url), "formats": [{"type": "markdown"}]},
78        )
79        root = require_object(payload, self.name, "fetch", "response")
80        if root.get("error") is not None or root.get("detail") is not None:
81            raise failure(self.name, "fetch", "provider reported failure")
82        results = require_object(root.get("results"), self.name, "fetch", "results")
83        markdown = require_object(results.get("markdown"), self.name, "fetch", "results.markdown")
84        data = require_list(markdown.get("data"), self.name, "fetch", "results.markdown.data")
85        if not data:
86            raise failure(self.name, "fetch", "page body is empty")
87        text = non_empty_string(data[0], self.name, "fetch", "results.markdown.data[0]")
88        return URLFetchCandidate(text, text)

class ScrapeGraphAIAdapter: View Source

20class ScrapeGraphAIAdapter:
21    def __init__(
22        self,
23        *,
24        name: str,
25        api_url: str,
26        secret: SecretValue,
27        http_executor: JsonRequester,
28    ) -> None:
29        self.name = name
30        self._api_url = configured_string(api_url, "api_url").rstrip("/")
31        self._secret = secret
32        self._http = http_executor
33
34    @property
35    def _headers(self) -> dict[str, str]:
36        return {"SGAI-APIKEY": self._secret.reveal()}
37
38    async def search(self, query: str) -> list[KeywordSearchHit]:
39        payload = await self._http.request_json(
40            "POST",
41            endpoint(self._api_url, "/api/search"),
42            stage="search",
43            headers=self._headers,
44            json_body={"query": query},
45        )
46        root = require_object(payload, self.name, "search", "response")
47        if root.get("error") is not None or root.get("detail") is not None:
48            raise failure(self.name, "search", "provider reported failure")
49        results = require_list(root.get("results"), self.name, "search", "results")
50        hits: list[KeywordSearchHit] = []
51        for item in results:
52            try:
53                result = require_object(item, self.name, "search", "result")
54                content = optional_string(
55                    result.get("content"), self.name, "search", "result.content"
56                )
57                hits.append(
58                    KeywordSearchHit(
59                        url=non_empty_string(result.get("url"), self.name, "search", "result.url"),
60                        title=optional_string(
61                            result.get("title"), self.name, "search", "result.title"
62                        ),
63                        snippet="",
64                        raw_content=content,
65                        content=content,
66                    )
67                )
68            except ExecutionFailure:
69                continue
70        return hits
71
72    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate:
73        payload = await self._http.request_json(
74            "POST",
75            endpoint(self._api_url, "/api/scrape"),
76            stage="fetch",
77            headers=self._headers,
78            json_body={"url": str(url), "formats": [{"type": "markdown"}]},
79        )
80        root = require_object(payload, self.name, "fetch", "response")
81        if root.get("error") is not None or root.get("detail") is not None:
82            raise failure(self.name, "fetch", "provider reported failure")
83        results = require_object(root.get("results"), self.name, "fetch", "results")
84        markdown = require_object(results.get("markdown"), self.name, "fetch", "results.markdown")
85        data = require_list(markdown.get("data"), self.name, "fetch", "results.markdown.data")
86        if not data:
87            raise failure(self.name, "fetch", "page body is empty")
88        text = non_empty_string(data[0], self.name, "fetch", "results.markdown.data[0]")
89        return URLFetchCandidate(text, text)

ScrapeGraphAIAdapter( *, name: str, api_url: str, secret: agent_search_gateway.observability.SecretValue, http_executor: agent_search_gateway.providers.web.common.JsonRequester) View Source

21    def __init__(
22        self,
23        *,
24        name: str,
25        api_url: str,
26        secret: SecretValue,
27        http_executor: JsonRequester,
28    ) -> None:
29        self.name = name
30        self._api_url = configured_string(api_url, "api_url").rstrip("/")
31        self._secret = secret
32        self._http = http_executor

name

async def search( self, query: str) -> list[agent_search_gateway.providers.contracts.KeywordSearchHit]: View Source

38    async def search(self, query: str) -> list[KeywordSearchHit]:
39        payload = await self._http.request_json(
40            "POST",
41            endpoint(self._api_url, "/api/search"),
42            stage="search",
43            headers=self._headers,
44            json_body={"query": query},
45        )
46        root = require_object(payload, self.name, "search", "response")
47        if root.get("error") is not None or root.get("detail") is not None:
48            raise failure(self.name, "search", "provider reported failure")
49        results = require_list(root.get("results"), self.name, "search", "results")
50        hits: list[KeywordSearchHit] = []
51        for item in results:
52            try:
53                result = require_object(item, self.name, "search", "result")
54                content = optional_string(
55                    result.get("content"), self.name, "search", "result.content"
56                )
57                hits.append(
58                    KeywordSearchHit(
59                        url=non_empty_string(result.get("url"), self.name, "search", "result.url"),
60                        title=optional_string(
61                            result.get("title"), self.name, "search", "result.title"
62                        ),
63                        snippet="",
64                        raw_content=content,
65                        content=content,
66                    )
67                )
68            except ExecutionFailure:
69                continue
70        return hits

async def fetch( self, url: agent_search_gateway.url_normalization.NormalizedURL) -> agent_search_gateway.providers.contracts.URLFetchCandidate: View Source

72    async def fetch(self, url: NormalizedURL) -> URLFetchCandidate:
73        payload = await self._http.request_json(
74            "POST",
75            endpoint(self._api_url, "/api/scrape"),
76            stage="fetch",
77            headers=self._headers,
78            json_body={"url": str(url), "formats": [{"type": "markdown"}]},
79        )
80        root = require_object(payload, self.name, "fetch", "response")
81        if root.get("error") is not None or root.get("detail") is not None:
82            raise failure(self.name, "fetch", "provider reported failure")
83        results = require_object(root.get("results"), self.name, "fetch", "results")
84        markdown = require_object(results.get("markdown"), self.name, "fetch", "results.markdown")
85        data = require_list(markdown.get("data"), self.name, "fetch", "results.markdown.data")
86        if not data:
87            raise failure(self.name, "fetch", "page body is empty")
88        text = non_empty_string(data[0], self.name, "fetch", "results.markdown.data[0]")
89        return URLFetchCandidate(text, text)