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 23:43:30 +00:00
parent 8888a2cf5d
commit 6042ced143
3 changed files with 247 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

274
agent.py
View File

@@ -22,7 +22,7 @@ import urllib.request
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any
from urllib.parse import parse_qsl, urlparse
from urllib.parse import parse_qsl, urljoin, urlparse
from a2a_pack import A2AAgent, LLMProvisioning, NoAuth, Pricing, RunContext, skill
from pydantic import BaseModel, Field
@@ -129,7 +129,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.9"
version = "0.3.0"
config_model = BrowserToApiConfig
auth_model = NoAuth
@@ -182,7 +182,7 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]):
captcha_policy: str = "solve_if_allowed",
allowed_captcha_domains: list[str] | None = None,
ai_explore: bool = True,
ai_steps: int = 4,
ai_steps: int = 10,
ai_goal: str | None = None,
) -> dict[str, Any]:
if not url or not str(url).strip().startswith(("http://", "https://")):
@@ -204,7 +204,7 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]):
captcha_policy=captcha_policy,
allowed_captcha_domains=allowed_captcha_domains,
ai_explore=ai_explore,
ai_steps=max(0, min(int(ai_steps), 8)),
ai_steps=max(0, min(int(ai_steps), 20)),
ai_goal=ai_goal,
llm_config=_llm_config_from_ctx(ctx),
)
@@ -232,7 +232,7 @@ class BrowserToApiAgent(A2AAgent[BrowserToApiConfig, NoAuth]):
captcha_policy=captcha_policy,
allowed_captcha_domains=_effective_captcha_domains(allowed_captcha_domains),
ai_explore=ai_explore,
ai_steps=max(0, min(int(ai_steps), 8)),
ai_steps=max(0, min(int(ai_steps), 20)),
)
except Exception as exc: # noqa: BLE001
return {"error": "browser_to_api_failed", "detail": f"{type(exc).__name__}: {exc}"}
@@ -287,7 +287,7 @@ def capture_browser_traffic(
captcha_policy: str = "solve_if_allowed",
allowed_captcha_domains: list[str] | None = None,
ai_explore: bool = True,
ai_steps: int = 4,
ai_steps: int = 10,
ai_goal: str | None = None,
llm_config: dict[str, Any] | None = None,
) -> tuple[list[TrafficSample], list[str]]:
@@ -370,8 +370,8 @@ def capture_browser_traffic(
except Exception as exc: # noqa: BLE001
sample.error = f"{type(exc).__name__}: {exc}"
page.on("request", on_request)
page.on("response", on_response)
context.on("request", on_request)
context.on("response", on_response)
try:
page.goto(url, wait_until="domcontentloaded", timeout=45_000)
@@ -391,7 +391,7 @@ def capture_browser_traffic(
_ai_explore_page(
page,
llm_config=llm_config,
steps=min(ai_steps, max_clicks if max_clicks > 0 else ai_steps),
steps=ai_steps,
goal=ai_goal or _default_ai_goal(url),
search_term=search_term,
log=log,
@@ -446,27 +446,63 @@ def _ai_explore_page(
search_term: str | None,
log: list[str],
) -> None:
start_url = page.url
visited_urls = [_normalize_url(start_url)]
history: list[dict[str, Any]] = []
min_steps = min(max(steps, 0), 6)
log.append(f"ai: exploration budget {steps} steps")
for step in range(1, max(steps, 0) + 1):
snapshot = _page_action_snapshot(page)
snapshot = _page_action_snapshot(page, start_url=start_url)
action = _llm_choose_action(
llm_config,
snapshot=snapshot,
goal=goal,
step=step,
total_steps=steps,
search_term=search_term,
history=history,
visited_urls=visited_urls,
log=log,
)
if not action:
log.append(f"ai: step {step}: no valid action returned")
_auto_scroll(page, log)
history.append({"step": step, "action": "fallback_scroll", "url": _safe_page_url(page)})
continue
applied = _apply_ai_action(page, action, snapshot.get("candidates", []), search_term, log, step)
if str(action.get("action") or "").lower().strip() == "stop" and step < min_steps:
log.append(f"ai: step {step}: ignored early stop; continuing exploration")
action = {"action": "scroll", "reason": "continue exploration before stopping"}
applied = _apply_ai_action(
page,
action,
snapshot.get("candidates", []),
search_term,
start_url,
log,
step,
)
if applied and action.get("action") != "stop":
_wait_for_action_settle(page)
_recover_if_external(page, start_url, log, step)
current_url = _normalize_url(_safe_page_url(page))
if current_url and current_url not in visited_urls:
visited_urls.append(current_url)
log.append(f"ai: visited {current_url}")
history.append(
{
"step": step,
"action": str(action.get("action") or ""),
"index": action.get("index"),
"url": current_url,
"reason": str(action.get("reason") or "")[:180],
}
)
if not applied or action.get("action") == "stop":
break
page.wait_for_timeout(1800)
log.append(f"ai: explored {len(visited_urls)} page url(s)")
def _page_action_snapshot(page: Any) -> dict[str, Any]:
def _page_action_snapshot(page: Any, *, start_url: str) -> dict[str, Any]:
try:
candidates = page.evaluate(
"""
@@ -485,33 +521,81 @@ def _page_action_snapshot(page: Any) -> dict[str, Any]:
const tag = node.tagName.toLowerCase();
const text = (node.innerText || node.value || node.getAttribute('aria-label') || node.getAttribute('title') || '').trim();
const href = node.getAttribute('href') || '';
const absoluteHref = href ? new URL(href, window.location.href).href : '';
const placeholder = node.getAttribute('placeholder') || '';
const role = node.getAttribute('role') || '';
const type = node.getAttribute('type') || '';
const disabled = Boolean(node.disabled || node.getAttribute('aria-disabled') === 'true');
const id = `ai-discovery-${index}`;
node.setAttribute('data-ai-discovery-id', id);
out.push({ index, selector: `[data-ai-discovery-id="${id}"]`, tag, text: text.slice(0, 140), href, placeholder, role, type });
out.push({
index,
selector: `[data-ai-discovery-id="${id}"]`,
tag,
text: text.slice(0, 180),
href,
absolute_href: absoluteHref,
placeholder,
role,
type,
disabled
});
index += 1;
if (out.length >= 45) break;
if (out.length >= 90) break;
}
return out;
return {
candidates: out,
scroll: {
x: window.scrollX,
y: window.scrollY,
height: document.documentElement.scrollHeight,
viewportHeight: window.innerHeight
}
};
}
"""
)
except Exception:
candidates = []
candidates = {"candidates": [], "scroll": {}}
raw_candidates = candidates.get("candidates") if isinstance(candidates, dict) else candidates
annotated_candidates = _annotate_action_candidates(raw_candidates, start_url=start_url, current_url=page.url)
try:
visible_text = page.locator("body").inner_text(timeout=1500)[:4000]
visible_text = page.locator("body").inner_text(timeout=1500)[:6000]
except Exception:
visible_text = ""
return {
"url": page.url,
"title": _safe_page_title(page),
"visible_text": visible_text,
"candidates": candidates if isinstance(candidates, list) else [],
"scroll": candidates.get("scroll", {}) if isinstance(candidates, dict) else {},
"candidates": annotated_candidates,
}
def _annotate_action_candidates(raw_candidates: Any, *, start_url: str, current_url: str) -> list[dict[str, Any]]:
if not isinstance(raw_candidates, list):
return []
base_url = current_url or start_url
annotated: list[dict[str, Any]] = []
for item in raw_candidates:
if not isinstance(item, dict):
continue
candidate = dict(item)
href = str(candidate.get("href") or "")
absolute_href = str(candidate.get("absolute_href") or "")
if href and not absolute_href:
absolute_href = urljoin(base_url, href)
candidate["absolute_href"] = absolute_href
if href.startswith(("mailto:", "tel:", "javascript:")):
candidate["same_site"] = False
elif absolute_href:
candidate["same_site"] = _is_same_site_url(absolute_href, start_url)
else:
candidate["same_site"] = True
annotated.append(candidate)
return annotated
def _safe_page_title(page: Any) -> str:
try:
return str(page.title() or "")
@@ -519,37 +603,132 @@ def _safe_page_title(page: Any) -> str:
return ""
def _safe_page_url(page: Any) -> str:
try:
return str(page.url or "")
except Exception:
return ""
def _normalize_url(url: str) -> str:
if not url:
return ""
parsed = urlparse(str(url))
if not parsed.scheme:
return str(url).rstrip("/")
path = (parsed.path or "/").rstrip("/") or "/"
return parsed._replace(path=path, fragment="").geturl()
def _is_same_site_url(candidate_url: str, start_url: str) -> bool:
if not candidate_url:
return True
resolved = urljoin(start_url, candidate_url)
parsed = urlparse(resolved)
if parsed.scheme not in {"http", "https"}:
return False
candidate_site = _site_key(parsed.hostname or "")
start_site = _site_key(urlparse(start_url).hostname or "")
return bool(candidate_site and start_site and candidate_site == start_site)
def _site_key(host: str) -> str:
parts = [part for part in host.lower().rstrip(".").split(".") if part]
if len(parts) <= 2:
return ".".join(parts)
if len(parts[-1]) == 2 and len(parts[-2]) <= 3 and len(parts) >= 3:
return ".".join(parts[-3:])
return ".".join(parts[-2:])
def _recover_if_external(page: Any, start_url: str, log: list[str], step: int) -> None:
current_url = _safe_page_url(page)
if not current_url or _is_same_site_url(current_url, start_url):
return
log.append(f"ai warning: step {step}: landed off-site; returning to start site")
try:
page.go_back(wait_until="domcontentloaded", timeout=6000)
except Exception:
try:
page.goto(start_url, wait_until="domcontentloaded", timeout=10_000)
except Exception as exc: # noqa: BLE001
log.append(f"ai warning: step {step}: recovery failed: {type(exc).__name__}: {exc}")
def _wait_for_action_settle(page: Any) -> None:
page.wait_for_timeout(900)
try:
page.wait_for_load_state("domcontentloaded", timeout=2500)
except Exception:
pass
try:
page.wait_for_load_state("networkidle", timeout=3000)
except Exception:
pass
page.wait_for_timeout(600)
def _llm_choose_action(
llm_config: dict[str, Any],
*,
snapshot: dict[str, Any],
goal: str,
step: int,
total_steps: int,
search_term: str | None,
history: list[dict[str, Any]],
visited_urls: list[str],
log: list[str],
) -> dict[str, Any] | None:
candidates = snapshot.get("candidates", [])
visited_set = {_normalize_url(item) for item in visited_urls if item}
compact_candidates = [
{key: item.get(key) for key in ("index", "tag", "text", "href", "placeholder", "role", "type")}
for item in candidates[:45]
{
**{
key: item.get(key)
for key in (
"index",
"tag",
"text",
"href",
"absolute_href",
"placeholder",
"role",
"type",
"same_site",
"disabled",
)
},
"visited": _normalize_url(str(item.get("absolute_href") or "")) in visited_set,
}
for item in candidates[:90]
if isinstance(item, dict)
]
system = (
"You choose one safe browser action for API discovery. Return only JSON. "
"Allowed actions: click, fill, scroll, wait, stop. Prefer actions likely "
"to trigger XHR/fetch/API traffic. Avoid login, signup, checkout, delete, "
"logout, payment, account settings, destructive operations, and external domains."
"Allowed actions: click, fill, scroll, wait, back, stop. Use the available "
"budget to explore deeply: prefer unvisited same-site links, list/detail "
"pages, search boxes, filters, load-more controls, navigation, docs, feed "
"or content links that are likely to trigger XHR/fetch/API traffic. Avoid "
"external domains, login, signup, checkout, delete, logout, payment, account "
"settings, and destructive operations. Do not stop early unless there are no "
"safe same-site actions left."
)
user = {
"goal": goal,
"step": step,
"total_steps": total_steps,
"remaining_steps": max(total_steps - step, 0),
"current_url": snapshot.get("url"),
"title": snapshot.get("title"),
"visible_text": str(snapshot.get("visible_text") or "")[:2200],
"scroll": snapshot.get("scroll"),
"visited_urls": visited_urls[-14:],
"recent_history": history[-8:],
"visible_text": str(snapshot.get("visible_text") or "")[:3000],
"search_term": search_term,
"candidates": compact_candidates,
"response_schema": {
"action": "click|fill|scroll|wait|stop",
"action": "click|fill|scroll|wait|back|stop",
"index": "candidate index for click/fill, optional",
"text": "text to fill, optional",
"reason": "short reason",
@@ -646,6 +825,7 @@ def _apply_ai_action(
action: dict[str, Any],
candidates: list[dict[str, Any]],
search_term: str | None,
start_url: str,
log: list[str],
step: int,
) -> bool:
@@ -662,18 +842,29 @@ def _apply_ai_action(
log.append(f"ai: step {step}: scroll ({reason})")
_auto_scroll(page, log)
return True
if action_name == "back":
try:
page.go_back(wait_until="domcontentloaded", timeout=6000)
log.append(f"ai: step {step}: back ({reason})")
except Exception as exc: # noqa: BLE001
log.append(f"ai warning: step {step}: back failed: {type(exc).__name__}: {exc}")
return True
index = _int_or_none(action.get("index"))
candidate = _candidate_by_index(candidates, index)
if not candidate:
log.append(f"ai warning: step {step}: missing candidate for action {action_name}")
return False
return True
text_bits = " ".join(str(candidate.get(key) or "") for key in ("text", "href", "placeholder")).lower()
if _looks_destructive(text_bits):
log.append(f"ai: step {step}: skipped destructive-looking candidate {index}")
return True
href = str(candidate.get("absolute_href") or candidate.get("href") or "")
if href and not _is_same_site_url(href, start_url):
log.append(f"ai: step {step}: skipped external candidate {index}")
return True
selector = str(candidate.get("selector") or "")
if not selector:
return False
return True
try:
locator = page.locator(selector).first
if action_name == "fill":
@@ -683,14 +874,19 @@ def _apply_ai_action(
log.append(f"ai: step {step}: fill candidate {index} ({reason})")
return True
if action_name == "click":
try:
locator.evaluate("(node) => node.removeAttribute('target')", timeout=1000)
except Exception:
pass
locator.click(timeout=4000)
log.append(f"ai: step {step}: click candidate {index} ({reason})")
_recover_if_external(page, start_url, log, step)
return True
except Exception as exc: # noqa: BLE001
log.append(f"ai warning: step {step}: {action_name} failed: {type(exc).__name__}: {exc}")
return True
log.append(f"ai warning: step {step}: unsupported action {action_name}")
return False
return True
def _candidate_by_index(candidates: list[dict[str, Any]], index: int | None) -> dict[str, Any] | None:
@@ -967,7 +1163,25 @@ def _click_interesting_elements(page: Any, *, max_clicks: int, log: list[str]) -
def _looks_destructive(text: str) -> bool:
return any(word in text for word in ("delete", "remove", "logout", "sign out", "purchase", "checkout", "pay"))
blocked_words = (
"delete",
"remove",
"logout",
"log out",
"sign out",
"purchase",
"checkout",
"pay",
"billing",
"login",
"log in",
"sign in",
"signup",
"sign up",
"account",
"settings",
)
return any(word in text for word in blocked_words)
def parse_trace_jsonl(*, requests_jsonl: str, responses_jsonl: str = "") -> list[TrafficSample]: