Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 21 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -151,6 +151,27 @@ se = ws.SearchEngine(
)
```

##### API method (no browser, no scraping)

`method="serpbase"` fetches results from the [SerpBase](https://serpbase.dev)
Google Search Results API instead of driving a browser or scraping HTML, so it
keeps working on headless hosts and avoids Google's `/sorry/` CAPTCHA blocks.
Results are rendered into a minimal SERP document and flow through the same
parser, so the output schema is unchanged.

```python
import WebSearcher as ws

# Reads SERPBASE_API_KEY from the environment (or pass serpbase_config).
se = ws.SearchEngine(method="serpbase", serpbase_config={"api_key": "..."})
se.search("election news", num_results=10)
se.parse_serp()
se.parsed.results[0]
```

Without an API key the backend logs a warning and returns an empty response, so
a crawl configured for `serpbase` degrades gracefully instead of failing.

#### 2. Conduct a Search

Logs are emitted as JSON Lines -- one structured object per line, with only the
Expand Down
2 changes: 1 addition & 1 deletion WebSearcher/demos/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ def _add_engine_args(p: argparse.ArgumentParser) -> None:
"method",
nargs="?",
default="patchright",
choices=["requests", "patchright"],
choices=["requests", "patchright", "serpbase"],
help="Search method",
)
p.add_argument("--data-dir", default=None, help="Directory to save outputs")
Expand Down
13 changes: 13 additions & 0 deletions WebSearcher/models/configs.py
Original file line number Diff line number Diff line change
Expand Up @@ -58,9 +58,21 @@ def sesh(self) -> requests.Session:
return sesh


class SerpBaseConfig(BaseConfig):
"""Configuration for the SerpBase REST API backend."""

api_key: str = Field(
default="",
description="SerpBase API key (falls back to the SERPBASE_API_KEY env var)",
)
base_url: str = "https://api.serpbase.dev"
timeout: int = 15


class SearchMethod(Enum):
REQUESTS = "requests"
PATCHRIGHT = "patchright"
SERPBASE = "serpbase"

@classmethod
def create(cls, method=None):
Expand All @@ -85,3 +97,4 @@ class SearchConfig(BaseConfig):
log: LogConfig = Field(default_factory=LogConfig)
requests: RequestsConfig = Field(default_factory=RequestsConfig)
patchright: PatchrightConfig = Field(default_factory=PatchrightConfig)
serpbase: SerpBaseConfig = Field(default_factory=SerpBaseConfig)
14 changes: 11 additions & 3 deletions WebSearcher/searchers/searchers.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,12 +8,14 @@
RequestsConfig,
SearchConfig,
SearchMethod,
SerpBaseConfig,
)
from ..models.data import BaseSERP, ParsedSERP
from ..models.searches import SearchParams
from ..parsers.parse_serp import parse_serp
from .patchright_searcher import PatchrightSearcher
from .requests_searcher import RequestsSearcher
from .serpbase_searcher import SerpBaseSearcher

WS_VERSION = metadata.version("WebSearcher")

Expand All @@ -27,17 +29,20 @@ def __init__(
log_config: dict | LogConfig = {},
requests_config: dict | RequestsConfig = {},
patchright_config: dict | PatchrightConfig = {},
serpbase_config: dict | SerpBaseConfig = {},
crawl_id: str = "",
) -> None:
"""Initialize the search engine

Args:
method: The method to use for searching: 'patchright' (a headed Chrome
via the patchright stealth fork) or 'requests' (pure HTTP, no
browser). Defaults to SearchMethod.PATCHRIGHT.
via the patchright stealth fork), 'requests' (pure HTTP, no
browser), or 'serpbase' (SerpBase REST API, no browser, no
scraping). Defaults to SearchMethod.PATCHRIGHT.
log_config: Common search configuration. Defaults to {}.
requests_config: Requests-specific configuration. Defaults to {}.
patchright_config: Patchright-specific configuration. Defaults to {}.
serpbase_config: SerpBase-specific configuration. Defaults to {}.
crawl_id: A unique identifier for the crawl. Defaults to ''.
"""

Expand All @@ -49,6 +54,7 @@ def __init__(
"log": LogConfig.create(log_config),
"requests": RequestsConfig.create(requests_config),
"patchright": PatchrightConfig.create(patchright_config),
"serpbase": SerpBaseConfig.create(serpbase_config),
}
)
# Name the logger after the subpackage, not __name__ (which doubles to
Expand All @@ -61,12 +67,14 @@ def __init__(
}

# Initialize searcher based on method
self.searcher: RequestsSearcher | PatchrightSearcher
self.searcher: RequestsSearcher | PatchrightSearcher | SerpBaseSearcher
if self.config.method == SearchMethod.REQUESTS:
self.searcher = RequestsSearcher(config=self.config.requests, logger=self.log)
elif self.config.method == SearchMethod.PATCHRIGHT:
self.searcher = PatchrightSearcher(config=self.config.patchright, logger=self.log)
self.searcher.init_driver()
elif self.config.method == SearchMethod.SERPBASE:
self.searcher = SerpBaseSearcher(config=self.config.serpbase, logger=self.log)

# Initialize search params and output
self.search_params = SearchParams.create()
Expand Down
122 changes: 122 additions & 0 deletions WebSearcher/searchers/serpbase_searcher.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,122 @@
"""SerpBase REST API searcher backend.

A no-browser, no-scraping alternative to the ``requests`` method: results are
fetched from the SerpBase Google Search Results API as JSON and rendered into a
minimal Google-style SERP document, so the existing parser pipeline
(``#rso`` > ``div.g`` > ``yuRUbf``/``VwiC3b``) produces the usual ``general``
results with the standard schema.

Graceful degradation: when no API key is configured the searcher logs a warning
and returns an empty ``ResponseOutput`` (``response_code`` 0) instead of
raising, so a crawl run without ``SERPBASE_API_KEY`` leaves the other backends
unaffected.
"""

import html
import os
from datetime import UTC, datetime

import requests

from ..models.configs import SerpBaseConfig
from ..models.data import ResponseOutput
from ..models.searches import SearchParams


class SerpBaseSearcher:
"""Handle SerpBase REST API-based web interactions for search engines"""

def __init__(self, config: SerpBaseConfig, logger):
"""Initialize a SerpBase searcher with the given configuration

Args:
config: SerpBaseConfig instance
logger: Logger instance
"""
self.config = config
self.log = logger
self.sesh = requests.Session()
self.sesh.headers.update({"Accept": "application/json"})

def cleanup(self) -> bool:
"""Close the requests session (uniform interface with the other backends)."""
try:
self.sesh.close()
return True
except Exception as e:
self.log.debug(f"Failed to close session: {e}", extra={"event": "cleanup"})
return False

def send_request(self, search_params: SearchParams) -> ResponseOutput:
"""Send a request to the SerpBase API and return a parseable response.

Args:
search_params: SearchParams instance

Returns:
ResponseOutput with the SERP rendered as minimal Google-style HTML.
When no API key is available, returns an empty ResponseOutput
(response_code 0) so the caller's pipeline degrades gracefully.
"""
api_key = self.config.api_key or os.environ.get("SERPBASE_API_KEY", "")
ts = datetime.now(UTC).replace(tzinfo=None).isoformat()
url = f"{self.config.base_url}/google/search"
user_agent = "SerpBaseSearcher/1.0"

if not api_key:
self.log.warning(
"SERPBASE_API_KEY not set - skipping SerpBase request. Get a key at https://serpbase.dev",
extra={"event": "fetch"},
)
return ResponseOutput(url=url, user_agent=user_agent, timestamp=ts)

params = {"q": search_params.qry, "api_key": api_key}
if search_params.num_results:
params["num"] = search_params.num_results
if search_params.lang:
params["hl"] = search_params.lang

response_output = ResponseOutput(url=url, user_agent=user_agent, timestamp=ts)
try:
response = self.sesh.get(url, params=params, timeout=self.config.timeout)
response_output.url = response.url
response_output.response_code = response.status_code
if response.status_code == 200:
response_output.html = self._json_to_html(response.json())
else:
self.log.warning(
f"SerpBase API returned {response.status_code}",
extra={"event": "fetch"},
)
except requests.exceptions.RequestException:
self.log.exception("SerpBase | Request error", extra={"event": "fetch"})
except ValueError:
self.log.exception("SerpBase | Invalid JSON response", extra={"event": "fetch"})

return response_output

@staticmethod
def _json_to_html(payload: dict) -> str:
"""Render SerpBase JSON results as a minimal Google-style SERP document.

The synthesized markup targets the classic result structure the parser
already handles: ``div#rso`` > ``div.g`` > ``div.yuRUbf`` (``h3``/``a``)
with a ``div.VwiC3b`` snippet and ``cite`` URL.
"""
blocks = []
for result in payload.get("organic_results", []):
title = html.escape(str(result.get("title", "")))
link = html.escape(str(result.get("link", "")))
snippet = html.escape(str(result.get("snippet", "")))
blocks.append(
'<div class="g">'
f'<div class="yuRUbf"><a href="{link}"><h3>{title}</h3></a></div>'
f'<div class="VwiC3b">{snippet}</div>'
f"<cite>{link}</cite>"
"</div>"
)
body = "".join(blocks)
return (
"<!DOCTYPE html><html><head><title>SerpBase results</title></head>"
f'<body><div id="rcnt"><div id="rso">{body}</div></div></body></html>'
)
76 changes: 76 additions & 0 deletions tests/test_searchers.py
Original file line number Diff line number Diff line change
Expand Up @@ -130,3 +130,79 @@ def test_save_record_writes_metadata_only_line_when_unparsed(tmp_path):
assert loaded["serp_id"] == "abc123"
assert loaded["features"] == {}
assert loaded["results"] == []


# SerpBase searcher -----------------------------------------------------------


def test_serpbase_searcher_no_key_graceful():
# Missing API key -> empty ResponseOutput (response_code 0), no exception,
# so a crawl without SERPBASE_API_KEY leaves the other methods unaffected.
from WebSearcher.models.configs import SerpBaseConfig
from WebSearcher.models.searches import SearchParams
from WebSearcher.searchers.serpbase_searcher import SerpBaseSearcher

searcher = SerpBaseSearcher(
config=SerpBaseConfig(api_key=""), logger=logging.getLogger("test_searchers")
)
out = searcher.send_request(SearchParams(qry="pizza"))
assert out.response_code == 0
assert out.html == ""


def test_serpbase_searcher_renders_parseable_serp(monkeypatch):
# With a key (here via env var) the JSON response renders into minimal
# Google-style HTML that the standard parser turns into `general` results.
from WebSearcher.models.configs import SerpBaseConfig
from WebSearcher.models.searches import SearchParams
from WebSearcher.parsers.parse_serp import parse_serp
from WebSearcher.searchers.serpbase_searcher import SerpBaseSearcher

monkeypatch.setenv("SERPBASE_API_KEY", "test-key")

class FakeResponse:
status_code = 200
url = "https://api.serpbase.dev/google/search?q=pizza&api_key=test-key"

def json(self):
return {
"organic_results": [
{
"title": "Best Pizza in Town",
"link": "https://example.com/pizza",
"snippet": "A tiny pizzeria serving wood-fired margherita since 1987.",
},
{
"title": "Pizza Wiki",
"link": "https://example.org/wiki/pizza",
"snippet": "Pizza is a traditional Italian dish consisting of a flat base.",
},
]
}

class FakeSession:
headers = {}

def get(self, url, params=None, timeout=None):
assert params["q"] == "pizza"
assert params["api_key"] == "test-key"
return FakeResponse()

def close(self):
pass

searcher = SerpBaseSearcher(
config=SerpBaseConfig(api_key=""), logger=logging.getLogger("test_searchers")
)
searcher.sesh = FakeSession()
out = searcher.send_request(SearchParams(qry="pizza"))
assert out.response_code == 200
assert '<div id="rso">' in out.html
assert '<div class="g">' in out.html

parsed = parse_serp(out.html, url=out.url)
general = [r for r in parsed["results"] if r["type"] == "general"]
assert len(general) == 2
assert general[0]["title"] == "Best Pizza in Town"
assert general[0]["url"] == "https://example.com/pizza"
assert "wood-fired margherita" in general[0]["text"]