bobseares-2/seleniumbase-website-scraper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

deploy

Browse Source
This commit is contained in:
a2a-platform
2026-06-27 16:16:11 +00:00
parent 7497ebd603
commit 74d0b664fc
3 changed files with 221 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

199
agent.py
View File

@@ -69,6 +69,11 @@ NOISE_URL_PATTERNS = [
r"cloudflareinsights\.com",
r"sentry\.io",
r"datadog(hq)?\.com",
r"perimeterx|px-captcha|_px",
r"/px/.*/captcha",
r"/akam/",
r"/si/[^/]+/obs",
r"captcha(\.js|/|\?)",
r"/(pixel|beacon|track|pageview|impression)(/|$|\?)",
r"\.(png|jpe?g|gif|svg|webp|ico|woff2?|ttf|eot|otf|css|map|mp4|webm|mp3)(\?|$)",
r"/(sw|service-worker)\.js(\?|$)",
@@ -76,6 +81,9 @@ NOISE_URL_PATTERNS = [
r"/robots\.txt(\?|$)",
r"/favicon\.ico(\?|$)",
]
DEFAULT_CAPTCHA_SOLVE_DOMAINS = ("a2acloud.io", ".a2acloud.io")
CAPTCHA_POLICIES = {"detect_only", "solve_if_allowed", "never"}
BROWSER_MODES = {"seleniumbase_stealth", "playwright"}
class BrowserToApiConfig(BaseModel):
@@ -119,7 +127,7 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]):
"SeleniumBase-style browser agent that turns observed website traffic "
"into OpenAPI specs, coverage reports, samples, and a replay client."
)
version = "0.2.1"
version = "0.2.2"
config_model = BrowserToApiConfig
auth_model = NoAuth
@@ -141,6 +149,7 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]):
],
"noise_filtering": "Drops analytics, static assets, service workers, and obvious bot-defense plumbing by default.",
"schema_inference": "Infers request/response JSON schemas, query params, path params, GraphQL dispatches, and auth-shaped headers.",
"captcha_policy": "SeleniumBase solve_captcha is available only for a2acloud.io and subdomains.",
}
}
@@ -166,9 +175,16 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]):
search_term: str | None = None,
min_samples: int = 1,
redact: list[str] | None = None,
browser_mode: str = "seleniumbase_stealth",
captcha_policy: str = "solve_if_allowed",
allowed_captcha_domains: list[str] | None = None,
) -> dict[str, Any]:
if not url or not str(url).strip().startswith(("http://", "https://")):
return {"error": "url must start with http:// or https://"}
if browser_mode not in BROWSER_MODES:
return {"error": f"browser_mode must be one of: {', '.join(sorted(BROWSER_MODES))}"}
if captcha_policy not in CAPTCHA_POLICIES:
return {"error": f"captcha_policy must be one of: {', '.join(sorted(CAPTCHA_POLICIES))}"}
try:
await ctx.emit_progress("launching browser and capturing network traffic")
@@ -178,6 +194,9 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]):
wait_seconds=max(1, min(int(wait_seconds), 60)),
max_clicks=max(0, min(int(max_clicks), 20)),
search_term=search_term,
browser_mode=browser_mode,
captcha_policy=captcha_policy,
allowed_captcha_domains=allowed_captcha_domains,
)
await ctx.emit_progress(f"captured {len(samples)} network samples; inferring API surface")
@@ -193,7 +212,16 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]):
output_dir = write_bundle(bundle, DEFAULT_OUTPUT_DIR / safe_run_id(url))
await ctx.emit_progress(f"wrote OpenAPI bundle to {output_dir}")
return _result_payload(url=url, samples=samples, bundle=bundle, output_dir=output_dir, capture_log=capture_log)
return _result_payload(
url=url,
samples=samples,
bundle=bundle,
output_dir=output_dir,
capture_log=capture_log,
browser_mode=browser_mode,
captcha_policy=captcha_policy,
allowed_captcha_domains=_effective_captcha_domains(allowed_captcha_domains),
)
except Exception as exc: # noqa: BLE001
return {"error": "browser_to_api_failed", "detail": f"{type(exc).__name__}: {exc}"}
@@ -243,6 +271,9 @@ def capture_browser_traffic(
wait_seconds: int = 8,
max_clicks: int = 4,
search_term: str | None = None,
browser_mode: str = "seleniumbase_stealth",
captcha_policy: str = "solve_if_allowed",
allowed_captcha_domains: list[str] | None = None,
) -> tuple[list[TrafficSample], list[str]]:
"""Capture useful browser traffic with Playwright.
@@ -255,6 +286,15 @@ def capture_browser_traffic(
log: list[str] = []
samples_by_request: dict[int, TrafficSample] = {}
stealth_cookies: list[dict[str, Any]] = []
if browser_mode == "seleniumbase_stealth":
stealth_cookies, stealth_log = seleniumbase_stealth_prep(
url=url,
captcha_policy=captcha_policy,
allowed_captcha_domains=allowed_captcha_domains,
wait_seconds=min(wait_seconds, 12),
)
log.extend(stealth_log)
with sync_playwright() as p:
browser = _launch_chromium(p, log)
@@ -267,6 +307,12 @@ def capture_browser_traffic(
"Chrome/126.0 Safari/537.36"
),
)
if stealth_cookies:
try:
context.add_cookies(stealth_cookies)
log.append(f"browser: transferred {len(stealth_cookies)} SeleniumBase cookies")
except Exception as exc: # noqa: BLE001
log.append(f"browser warning: cookie transfer failed: {type(exc).__name__}: {exc}")
page = context.new_page()
def on_request(request: Any) -> None:
@@ -330,6 +376,123 @@ def capture_browser_traffic(
return list(samples_by_request.values()), log
def seleniumbase_stealth_prep(
*,
url: str,
captcha_policy: str,
allowed_captcha_domains: list[str] | None,
wait_seconds: int,
) -> tuple[list[dict[str, Any]], list[str]]:
log: list[str] = []
cookies: list[dict[str, Any]] = []
allowed_domains = _effective_captcha_domains(allowed_captcha_domains)
captcha_allowed = _host_allowed_for_captcha(url, allowed_domains)
try:
from seleniumbase import SB
except Exception as exc: # noqa: BLE001
return [], [f"seleniumbase warning: import failed: {type(exc).__name__}: {exc}"]
try:
with SB(uc=True, test=True, locale="en-US", use_chromium=True) as sb:
log.append("seleniumbase: started UC/CDP stealth session")
try:
sb.uc_open_with_reconnect(url, reconnect_time=3)
log.append("seleniumbase: uc_open_with_reconnect")
except Exception as exc: # noqa: BLE001
log.append(f"seleniumbase warning: uc reconnect failed: {type(exc).__name__}: {exc}")
try:
sb.open(url)
log.append("seleniumbase: open")
except Exception as open_exc: # noqa: BLE001
log.append(f"seleniumbase warning: open failed: {type(open_exc).__name__}: {open_exc}")
_seleniumbase_sleep(sb, min(max(wait_seconds, 2), 8))
if captcha_policy == "solve_if_allowed":
if captcha_allowed:
try:
sb.solve_captcha()
log.append("seleniumbase: solve_captcha attempted for allowed domain")
_seleniumbase_sleep(sb, 2)
except Exception as exc: # noqa: BLE001
log.append(f"seleniumbase warning: solve_captcha failed: {type(exc).__name__}: {exc}")
else:
host = urlparse(url).netloc
log.append(f"seleniumbase: captcha solve skipped; host not allowlisted: {host}")
elif captcha_policy == "detect_only":
log.append("seleniumbase: captcha policy detect_only")
else:
log.append("seleniumbase: captcha policy never")
try:
raw_cookies = sb.driver.get_cookies()
cookies = _playwright_cookies(raw_cookies, url)
log.append(f"seleniumbase: collected {len(cookies)} cookies")
except Exception as exc: # noqa: BLE001
log.append(f"seleniumbase warning: cookie collection failed: {type(exc).__name__}: {exc}")
except Exception as exc: # noqa: BLE001
log.append(f"seleniumbase warning: stealth prep failed: {type(exc).__name__}: {exc}")
return cookies, log
def _seleniumbase_sleep(sb: Any, seconds: int) -> None:
try:
sb.sleep(seconds)
except Exception:
time.sleep(seconds)
def _effective_captcha_domains(allowed_captcha_domains: list[str] | None) -> list[str]:
domains = [str(item).strip().lower() for item in (allowed_captcha_domains or []) if str(item).strip()]
for domain in DEFAULT_CAPTCHA_SOLVE_DOMAINS:
if domain not in domains:
domains.append(domain)
return domains
def _host_allowed_for_captcha(url: str, allowed_domains: list[str]) -> bool:
host = urlparse(url).hostname or ""
host = host.lower().rstrip(".")
for raw_domain in allowed_domains:
domain = raw_domain.lower().strip().lstrip("*").rstrip(".")
if not domain:
continue
if domain.startswith("."):
suffix = domain[1:]
if host == suffix or host.endswith(domain):
return True
elif host == domain or host.endswith(f".{domain}"):
return True
return False
def _playwright_cookies(raw_cookies: list[dict[str, Any]], url: str) -> list[dict[str, Any]]:
out: list[dict[str, Any]] = []
parsed = urlparse(url)
fallback_domain = parsed.hostname or ""
for item in raw_cookies:
name = item.get("name")
value = item.get("value")
if not name or value is None:
continue
cookie: dict[str, Any] = {
"name": str(name),
"value": str(value),
"domain": str(item.get("domain") or fallback_domain),
"path": str(item.get("path") or "/"),
"secure": bool(item.get("secure", parsed.scheme == "https")),
"httpOnly": bool(item.get("httpOnly", False)),
}
expiry = item.get("expiry") or item.get("expires")
if isinstance(expiry, (int, float)) and expiry > 0:
cookie["expires"] = float(expiry)
same_site = item.get("sameSite")
if same_site in {"Strict", "Lax", "None"}:
cookie["sameSite"] = same_site
out.append(cookie)
return out
def _launch_chromium(p: Any, log: list[str]) -> Any:
launch_kwargs = {
"headless": True,
@@ -872,7 +1035,11 @@ def _result_payload(
output_dir: Path,
url: str | None = None,
capture_log: list[str] | None = None,
browser_mode: str | None = None,
captcha_policy: str | None = None,
allowed_captcha_domains: list[str] | None = None,
) -> dict[str, Any]:
bot_signals = _bot_defense_signals(samples, capture_log or [])
payload: dict[str, Any] = {
"captured_samples": len(samples),
"included_samples": bundle["summary"]["included_samples"],
@@ -886,14 +1053,42 @@ def _result_payload(
"client": str(output_dir / "client.mjs"),
"confidence": str(output_dir / "confidence.json"),
"top_endpoints": bundle["summary"]["top_endpoints"],
"bot_defense_detected": bool(bot_signals),
"bot_defense_signals": bot_signals,
}
if url is not None:
payload["url"] = url
if browser_mode is not None:
payload["browser_mode"] = browser_mode
if captcha_policy is not None:
payload["captcha_policy"] = captcha_policy
if allowed_captcha_domains is not None:
payload["allowed_captcha_domains"] = allowed_captcha_domains
if capture_log is not None:
payload["capture_log"] = capture_log
return payload
def _bot_defense_signals(samples: list[TrafficSample], capture_log: list[str]) -> list[str]:
checks = {
"captcha": (r"captcha",),
"perimeterx": (r"perimeterx", r"/px/", r"_px", r"px-captcha"),
"akamai": (r"/akam/", r"akamai", r"bm_sz", r"_abck"),
"datadome": (r"datadome",),
"cloudflare": (r"cf-chl", r"cloudflare"),
}
haystacks = [sample.url for sample in samples]
for sample in samples:
haystacks.extend(str(key) for key in sample.request_headers)
haystacks.extend(str(key) for key in sample.response_headers)
haystacks.extend(capture_log)
found: list[str] = []
for name, patterns in checks.items():
if any(_regex_search(pattern, haystack) for pattern in patterns for haystack in haystacks):
found.append(name)
return found
def infer_schema(values: list[Any]) -> dict[str, Any]:
values = [value for value in values if value is not None]
if not values: