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

remove agent.py

Browse Source
This commit is contained in:
a2a-cloud
2026-06-07 23:13:42 +00:00
parent ee13879136
commit dec9158f08
1 changed files with 0 additions and 578 deletions
Download Patch File Download Diff File Expand all files Collapse all files

578
agent.py
View File

@@ -1,578 +0,0 @@
"""Private compliant website scraper using SeleniumBase/browser automation.
The public skill is intentionally narrow: it accepts one starting URL, a CSS
selector mapping, pagination limits, browser options, robots.txt behavior, and
output format. It never attempts to bypass CAPTCHAs, anti-bot challenges,
access controls, paywalls, login gates, robots.txt restrictions, or terms of
service restrictions.
"""
from __future__ import annotations
import json
import re
import textwrap
import time
import urllib.parse
import urllib.request
import urllib.robotparser
from pathlib import Path
from typing import Any, Literal
from pydantic import BaseModel, Field
from a2a_pack import (
A2AAgent,
EgressPolicy,
LLMProvisioning,
NoAuth,
Pricing,
Resources,
RunContext,
WorkspaceAccess,
WorkspaceMode,
skill,
)
OUTPUT_DIR = "outputs/seleniumbase-website-scraper"
RUNTIME_SKILLS_DIR = "seleniumbase-website-scraper/.deepagents/skills/"
class SeleniumbaseWebsiteScraperConfig(BaseModel):
max_pages_limit: int = Field(default=50, ge=1, le=200)
default_user_agent: str = "A2A seleniumbase-website-scraper/1.0 (+authorized public/user-owned scraping only)"
class SeleniumbaseWebsiteScraper(A2AAgent[SeleniumbaseWebsiteScraperConfig, NoAuth]):
name = "seleniumbase-website-scraper"
description = (
"Private compliant website scraping agent using SeleniumBase/browser "
"automation with robots.txt checks, same-domain pagination safeguards, "
"screenshots, HTML capture, JSON/CSV output, retries, and polite rate limiting."
)
version = "1.0.0"
config_model = SeleniumbaseWebsiteScraperConfig
auth_model = NoAuth
llm_provisioning = LLMProvisioning.PLATFORM
pricing = Pricing(
price_per_call_usd=0.0,
caller_pays_llm=True,
notes=(
"Browser scraping is deterministic; any future LLM use must read the "
"caller-provided ctx.llm credential. The caller pays any forwarded LLM cost."
),
)
resources = Resources(cpu="2", memory="2Gi", max_runtime_seconds=900)
egress = EgressPolicy(deny_internet_by_default=False)
tools_used = ("seleniumbase", "selenium", "chromium", "browser-automation")
workspace_access = WorkspaceAccess.dynamic(
max_files=256,
allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
require_reason=False,
max_total_size_bytes=250 * 1024 * 1024,
)
@skill(
description=(
"Authorized website scraping with SeleniumBase/browser automation, "
"robots.txt checks, selector extraction, same-domain pagination, "
"screenshots, and JSON/CSV output. Stops on CAPTCHA, anti-bot, login, "
"paywall, or access-control challenges."
),
timeout_seconds=900,
cost_class="browser-heavy",
grant_outputs_prefix=OUTPUT_DIR,
grant_write_prefixes=(OUTPUT_DIR,),
grant_run_timeout_seconds=900,
)
async def scrape_website(
self,
ctx: RunContext[NoAuth],
url: str,
selectors: dict[str, str],
max_pages: int = 1,
wait_seconds: float = 3.0,
output_format: Literal["json", "csv"] = "json",
headless: bool = True,
user_agent: str = "",
respect_robots: bool = True,
rate_limit_seconds: float = 1.0,
screenshot: bool = False,
) -> dict[str, Any]:
"""Scrape authorized public/user-owned pages and save structured output."""
creds = ctx.llm # Read platform-forwarded LLM metadata; this skill does not construct a model.
await ctx.emit_progress(f"starting compliant scrape; llm credential source={creds.source}")
validation = _validate_inputs(
url=url,
selectors=selectors,
max_pages=max_pages,
max_pages_limit=self.config.max_pages_limit,
wait_seconds=wait_seconds,
output_format=output_format,
user_agent=user_agent,
rate_limit_seconds=rate_limit_seconds,
)
if validation["errors"]:
return {
"status": "invalid_input",
"items": [],
"saved_paths": [],
"warnings": validation["errors"],
"screenshots": [],
}
start_url = validation["url"]
effective_user_agent = user_agent.strip() or self.config.default_user_agent
warnings: list[str] = list(validation["warnings"])
if respect_robots:
robots = _robots_allowed(start_url, effective_user_agent)
warnings.extend(robots.get("warnings", []))
if not robots.get("allowed", False):
return {
"status": "blocked_by_robots_txt",
"items": [],
"saved_paths": [],
"warnings": warnings + ["robots.txt disallows fetching the requested URL for this user agent."],
"screenshots": [],
"robots": robots,
}
script_payload = {
"url": start_url,
"selectors": selectors,
"max_pages": max_pages,
"wait_seconds": wait_seconds,
"output_format": output_format,
"headless": headless,
"user_agent": effective_user_agent,
"respect_robots": respect_robots,
"rate_limit_seconds": rate_limit_seconds,
"screenshot": screenshot,
"output_dir": f"/workspace/{OUTPUT_DIR}",
}
script = _render_scraper_script(script_payload)
try:
result = await ctx.workspace_python(
script,
image="seleniumbase/seleniumbase:latest",
timeout_seconds=900,
memory_mib=2048,
cpus=2,
)
except Exception as exc: # noqa: BLE001
await ctx.emit_error(str(exc), code="browser_runtime_unavailable")
return {
"status": "browser_runtime_unavailable",
"items": [],
"saved_paths": [],
"warnings": warnings
+ [
"Browser sandbox could not be started. The platform must allow the SeleniumBase image, "
"or rerun where browser automation is available.",
f"runtime error: {type(exc).__name__}: {exc}",
],
"screenshots": [],
}
stdout = (result.stdout or "").strip()
stderr = (result.stderr or "").strip()
parsed = _parse_last_json(stdout)
if not parsed:
return {
"status": "error",
"items": [],
"saved_paths": [],
"warnings": warnings
+ [
"Browser script did not return a parseable JSON result.",
f"returncode={getattr(result, 'exit_code', None)}",
f"stdout_tail={stdout[-2000:]}",
f"stderr_tail={stderr[-2000:]}",
],
"screenshots": [],
}
parsed_warnings = list(parsed.get("warnings") or [])
if getattr(result, "exit_code", 0) not in (0, None):
parsed_warnings.append(
f"browser command returned nonzero exit code {getattr(result, 'exit_code', None)}; usable emitted files were preserved when present"
)
if stderr:
parsed_warnings.append(f"stderr_tail={stderr[-2000:]}")
saved_paths = [str(path).replace("/workspace/", "") for path in (parsed.get("saved_paths") or [])]
screenshot_paths = [str(path).replace("/workspace/", "") for path in (parsed.get("screenshots") or [])]
for path in saved_paths[:5] + screenshot_paths[:5]:
await _emit_workspace_artifact_if_available(ctx, path)
status = str(parsed.get("status") or "ok")
if status in {"challenge_detected", "manual_intervention_required", "blocked"}:
await ctx.emit_error("Scrape stopped because a CAPTCHA, anti-bot, login, paywall, or access-control challenge was detected.", code=status)
else:
await ctx.emit_progress(f"scrape finished with status={status}; items={len(parsed.get('items') or [])}")
return {
"status": status,
"items": parsed.get("items") or [],
"saved_paths": saved_paths,
"warnings": warnings + parsed_warnings,
"screenshots": screenshot_paths,
"pages_visited": parsed.get("pages_visited") or [],
"html_captures": [str(path).replace("/workspace/", "") for path in (parsed.get("html_captures") or [])],
"robots": {"checked": bool(respect_robots)},
}
def _validate_inputs(
*,
url: str,
selectors: dict[str, str],
max_pages: int,
max_pages_limit: int,
wait_seconds: float,
output_format: str,
user_agent: str,
rate_limit_seconds: float,
) -> dict[str, Any]:
errors: list[str] = []
warnings: list[str] = []
clean_url = str(url or "").strip()
parsed = urllib.parse.urlparse(clean_url)
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
errors.append("url must be an absolute http(s) URL.")
if parsed.username or parsed.password:
errors.append("url must not contain embedded credentials.")
if not selectors:
errors.append("selectors mapping must contain at least one CSS selector.")
for key, value in selectors.items():
if not isinstance(key, str) or not key.strip():
errors.append("selector field names must be non-empty strings.")
if not isinstance(value, str) or not value.strip():
errors.append(f"selector for {key!r} must be a non-empty string.")
if re.search(r"password|token|secret|credential", key, re.I):
warnings.append(f"selector field {key!r} looks sensitive; do not scrape secrets or credentials.")
if max_pages < 1 or max_pages > max_pages_limit:
errors.append(f"max_pages must be between 1 and {max_pages_limit}.")
if wait_seconds < 0 or wait_seconds > 60:
errors.append("wait_seconds must be between 0 and 60.")
if output_format not in {"json", "csv"}:
errors.append("output_format must be 'json' or 'csv'.")
if "\n" in user_agent or "\r" in user_agent:
errors.append("user_agent must be a single-line string.")
if rate_limit_seconds < 0 or rate_limit_seconds > 120:
errors.append("rate_limit_seconds must be between 0 and 120.")
return {"errors": errors, "warnings": warnings, "url": clean_url}
def _robots_allowed(url: str, user_agent: str) -> dict[str, Any]:
parsed = urllib.parse.urlparse(url)
robots_url = urllib.parse.urlunparse((parsed.scheme, parsed.netloc, "/robots.txt", "", "", ""))
rp = urllib.robotparser.RobotFileParser()
rp.set_url(robots_url)
try:
with urllib.request.urlopen(
urllib.request.Request(robots_url, headers={"User-Agent": user_agent}),
timeout=10,
) as response:
body = response.read(2_000_000).decode("utf-8", errors="ignore")
rp.parse(body.splitlines())
return {"allowed": bool(rp.can_fetch(user_agent, url)), "robots_url": robots_url, "warnings": []}
except urllib.error.HTTPError as exc:
if exc.code in {401, 403}:
return {
"allowed": False,
"robots_url": robots_url,
"warnings": [f"robots.txt returned HTTP {exc.code}; treating as disallow for safety."],
}
if exc.code == 404:
return {"allowed": True, "robots_url": robots_url, "warnings": ["robots.txt not found; proceeding because no robots rules were published."]}
return {
"allowed": False,
"robots_url": robots_url,
"warnings": [f"robots.txt check failed with HTTP {exc.code}; treating as disallow for safety."],
}
except Exception as exc: # noqa: BLE001
return {
"allowed": False,
"robots_url": robots_url,
"warnings": [f"robots.txt could not be checked ({type(exc).__name__}: {exc}); treating as disallow for safety."],
}
def _render_scraper_script(payload: dict[str, Any]) -> str:
payload_json = json.dumps(payload, ensure_ascii=False)
return textwrap.dedent(
f"""
import csv
import json
import os
import re
import sys
import time
import traceback
import urllib.parse
import urllib.request
import urllib.robotparser
from datetime import datetime, timezone
PAYLOAD = json.loads({payload_json!r})
CHALLENGE_PATTERNS = [
r"captcha", r"recaptcha", r"hcaptcha", r"cf-challenge", r"cloudflare", r"turnstile",
r"are you human", r"verify you are human", r"bot detection", r"automated traffic",
r"access denied", r"temporarily blocked", r"unusual traffic", r"login required",
r"sign in to continue", r"subscribe to continue", r"paywall", r"forbidden",
]
PAGINATION_KEYS = {{"next", "_next", "__next__", "pagination_next", "next_page"}}
def emit(payload):
print(json.dumps(payload, ensure_ascii=False))
def safe_name(value):
value = re.sub(r"[^a-zA-Z0-9._-]+", "-", value).strip("-._")
return value[:80] or "page"
def same_origin(a, b):
pa, pb = urllib.parse.urlparse(a), urllib.parse.urlparse(b)
return pa.scheme == pb.scheme and pa.netloc.lower() == pb.netloc.lower()
def robots_allowed(url, user_agent):
parsed = urllib.parse.urlparse(url)
robots_url = urllib.parse.urlunparse((parsed.scheme, parsed.netloc, "/robots.txt", "", "", ""))
rp = urllib.robotparser.RobotFileParser()
rp.set_url(robots_url)
try:
req = urllib.request.Request(robots_url, headers={{"User-Agent": user_agent}})
with urllib.request.urlopen(req, timeout=10) as response:
body = response.read(2_000_000).decode("utf-8", errors="ignore")
rp.parse(body.splitlines())
return bool(rp.can_fetch(user_agent, url)), None
except Exception as exc:
return False, f"robots.txt could not be checked for pagination URL {{url}}: {{type(exc).__name__}}: {{exc}}"
def split_selector(selector):
selector = selector.strip()
m = re.search(r"::attr\\(([^)]+)\\)\\s*$", selector)
if m:
return selector[:m.start()].strip(), "attr", m.group(1).strip()
if selector.endswith("::text"):
return selector[:-6].strip(), "text", None
if selector.endswith("::html"):
return selector[:-6].strip(), "html", None
return selector, "text", None
def detect_challenge(page_source, title, current_url):
haystack = "\\n".join([title or "", current_url or "", page_source[:50000] or ""]).lower()
matched = [pat for pat in CHALLENGE_PATTERNS if re.search(pat, haystack, re.I)]
if matched:
return "manual_intervention_required", "challenge/access-control indicator detected: " + ", ".join(sorted(set(matched))[:8])
return None, None
def extract_with_bs4(page_source, selectors):
from bs4 import BeautifulSoup
soup = BeautifulSoup(page_source, "html.parser")
item = {{}}
next_url = None
for key, raw_selector in selectors.items():
css, mode, attr = split_selector(str(raw_selector))
try:
elements = soup.select(css)
except Exception as exc:
item[key] = None
item.setdefault("_selector_warnings", []).append(f"selector {{key}}={{raw_selector!r}} failed: {{exc}}")
continue
values = []
for el in elements:
if mode == "attr":
values.append(el.get(attr))
elif mode == "html":
values.append(el.decode_contents())
else:
values.append(el.get_text(" ", strip=True))
values = [v for v in values if v is not None]
if key in PAGINATION_KEYS:
next_url = values[0] if values else None
else:
item[key] = values if len(values) != 1 else values[0] if values else None
return item, next_url
def write_outputs(items, pages_visited, html_captures, screenshots, warnings, status):
out_dir = PAYLOAD["output_dir"]
os.makedirs(out_dir, exist_ok=True)
stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
base = safe_name(urllib.parse.urlparse(PAYLOAD["url"]).netloc) + "-" + stamp
metadata = {{
"status": status,
"items": items,
"pages_visited": pages_visited,
"html_captures": html_captures,
"screenshots": screenshots,
"warnings": warnings,
"safety": "Authorized public/user-owned scraping only. No CAPTCHA, anti-bot, login, paywall, robots.txt, ToS, or access-control circumvention attempted.",
}}
json_path = os.path.join(out_dir, base + ".json")
with open(json_path, "w", encoding="utf-8") as f:
json.dump(metadata, f, ensure_ascii=False, indent=2)
saved_paths = [json_path]
if PAYLOAD["output_format"] == "csv":
csv_path = os.path.join(out_dir, base + ".csv")
fieldnames = sorted({{k for item in items for k in item.keys() if k != "_selector_warnings"}})
with open(csv_path, "w", newline="", encoding="utf-8") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames or ["url"])
writer.writeheader()
for item in items:
row = {{k: (json.dumps(v, ensure_ascii=False) if isinstance(v, (list, dict)) else v) for k, v in item.items() if k in (fieldnames or ["url"])}}
writer.writerow(row)
saved_paths.append(csv_path)
return saved_paths
def run():
warnings = []
items = []
pages_visited = []
html_captures = []
screenshots = []
status = "ok"
out_dir = PAYLOAD["output_dir"]
os.makedirs(out_dir, exist_ok=True)
current = PAYLOAD["url"]
start = current
try:
from seleniumbase import SB
browser_kwargs = {{
"headless": bool(PAYLOAD["headless"]),
"uc": False,
"test": False,
"locale_code": "en",
}}
if PAYLOAD.get("user_agent"):
browser_kwargs["agent"] = PAYLOAD["user_agent"]
with SB(**browser_kwargs) as sb:
for page_index in range(int(PAYLOAD["max_pages"])):
if not same_origin(start, current):
warnings.append(f"same-domain safeguard stopped pagination to {{current}}")
break
if PAYLOAD.get("respect_robots"):
allowed, robot_warning = robots_allowed(current, PAYLOAD.get("user_agent") or "*")
if robot_warning:
warnings.append(robot_warning)
if not allowed:
status = "blocked_by_robots_txt"
warnings.append(f"robots.txt disallows pagination URL {{current}}")
break
if page_index > 0 and float(PAYLOAD["rate_limit_seconds"]) > 0:
time.sleep(float(PAYLOAD["rate_limit_seconds"]))
sb.open(current)
if float(PAYLOAD["wait_seconds"]) > 0:
sb.sleep(float(PAYLOAD["wait_seconds"]))
current_url = sb.get_current_url()
title = sb.get_title()
page_source = sb.get_page_source()
pages_visited.append(current_url)
challenge_status, challenge_warning = detect_challenge(page_source, title, current_url)
if challenge_status:
status = challenge_status
warnings.append(challenge_warning)
html_path = os.path.join(out_dir, f"page-{{page_index+1}}-challenge.html")
with open(html_path, "w", encoding="utf-8") as f:
f.write(page_source)
html_captures.append(html_path)
if PAYLOAD.get("screenshot"):
shot_path = os.path.join(out_dir, f"page-{{page_index+1}}-challenge.png")
try:
sb.save_screenshot(shot_path)
screenshots.append(shot_path)
except Exception as exc:
warnings.append(f"screenshot failed: {{exc}}")
break
html_path = os.path.join(out_dir, f"page-{{page_index+1}}.html")
with open(html_path, "w", encoding="utf-8") as f:
f.write(page_source)
html_captures.append(html_path)
if PAYLOAD.get("screenshot"):
shot_path = os.path.join(out_dir, f"page-{{page_index+1}}.png")
try:
sb.save_screenshot(shot_path)
screenshots.append(shot_path)
except Exception as exc:
warnings.append(f"screenshot failed: {{exc}}")
item, next_raw = extract_with_bs4(page_source, PAYLOAD["selectors"])
item["url"] = current_url
item["page_index"] = page_index + 1
if item.get("_selector_warnings"):
warnings.extend(item.pop("_selector_warnings"))
items.append(item)
if page_index >= int(PAYLOAD["max_pages"]) - 1 or not next_raw:
break
next_url = urllib.parse.urljoin(current_url, str(next_raw))
if not same_origin(start, next_url):
warnings.append(f"same-domain safeguard refused pagination URL {{next_url}}")
break
current = next_url
except Exception as exc:
status = "error"
warnings.append(f"browser automation failed: {{type(exc).__name__}}: {{exc}}")
warnings.append(traceback.format_exc()[-4000:])
saved_paths = write_outputs(items, pages_visited, html_captures, screenshots, warnings, status)
emit({{
"status": status,
"items": items,
"saved_paths": saved_paths,
"screenshots": screenshots,
"html_captures": html_captures,
"warnings": warnings,
"pages_visited": pages_visited,
}})
return 0 if status in {{"ok", "blocked_by_robots_txt", "manual_intervention_required"}} else 1
if __name__ == "__main__":
sys.exit(run())
"""
)
def _parse_last_json(stdout: str) -> dict[str, Any] | None:
for line in reversed(stdout.splitlines()):
line = line.strip()
if not line.startswith("{"):
continue
try:
value = json.loads(line)
except json.JSONDecodeError:
continue
if isinstance(value, dict):
return value
return None
async def _emit_workspace_artifact_if_available(ctx: RunContext[Any], path: str) -> None:
try:
reader = getattr(ctx.workspace, "read_bytes", None)
if reader is None:
return
data = reader(path)
if not isinstance(data, (bytes, bytearray)):
return
mime = "application/json"
if path.endswith(".csv"):
mime = "text/csv"
elif path.endswith(".png"):
mime = "image/png"
elif path.endswith(".html"):
mime = "text/html"
ref = await ctx.write_artifact(Path(path).name, bytes(data), mime)
await ctx.emit_artifact(ref)
except Exception: # noqa: BLE001
return