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

write agent.py

Browse Source
This commit is contained in:
a2a-cloud
2026-06-08 00:01:58 +00:00
parent 68301cae82
commit dabe2dcd95
1 changed files with 13 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
agent.py
View File

@@ -49,7 +49,7 @@ class SeleniumbaseWebsiteScraper(A2AAgent[SeleniumbaseWebsiteScraperConfig, NoAu
"automation with robots.txt checks, same-domain pagination safeguards, "
"screenshots, HTML capture, JSON/CSV output, retries, and polite rate limiting."
)
version = "1.0.2"
version = "1.0.3"
config_model = SeleniumbaseWebsiteScraperConfig
auth_model = NoAuth
@@ -352,7 +352,7 @@ def _render_scraper_script(payload: dict[str, Any]) -> str:
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}}"
return False, "robots.txt could not be checked for pagination URL {{0}}: {{1}}: {{2}}".format(url, type(exc).__name__, exc)
def split_selector(selector):
selector = selector.strip()
@@ -383,7 +383,7 @@ def _render_scraper_script(payload: dict[str, Any]) -> str:
elements = soup.select(css)
except Exception as exc:
item[key] = None
item.setdefault("_selector_warnings", []).append(f"selector {{key}}={{raw_selector!r}} failed: {{exc}}")
item.setdefault("_selector_warnings", []).append("selector {{0}}={{1!r}} failed: {{2}}".format(key, raw_selector, exc))
continue
values = []
for el in elements:
@@ -455,7 +455,7 @@ def _render_scraper_script(payload: dict[str, Any]) -> str:
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}}")
warnings.append("same-domain safeguard stopped pagination to " + str(current))
break
if PAYLOAD.get("respect_robots"):
allowed, robot_warning = robots_allowed(current, PAYLOAD.get("user_agent") or "*")
@@ -463,7 +463,7 @@ def _render_scraper_script(payload: dict[str, Any]) -> str:
warnings.append(robot_warning)
if not allowed:
status = "blocked_by_robots_txt"
warnings.append(f"robots.txt disallows pagination URL {{current}}")
warnings.append("robots.txt disallows pagination URL " + str(current))
break
if page_index > 0 and float(PAYLOAD["rate_limit_seconds"]) > 0:
time.sleep(float(PAYLOAD["rate_limit_seconds"]))
@@ -480,30 +480,30 @@ def _render_scraper_script(payload: dict[str, Any]) -> str:
if challenge_status:
status = challenge_status
warnings.append(challenge_warning)
html_path = os.path.join(out_dir, f"page-{{page_index+1}}-challenge.html")
html_path = os.path.join(out_dir, "page-{{0}}-challenge.html".format(page_index + 1))
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")
shot_path = os.path.join(out_dir, "page-{{0}}-challenge.png".format(page_index + 1))
try:
sb.save_screenshot(shot_path)
screenshots.append(shot_path)
except Exception as exc:
warnings.append(f"screenshot failed: {{exc}}")
warnings.append("screenshot failed: " + str(exc))
break
html_path = os.path.join(out_dir, f"page-{{page_index+1}}.html")
html_path = os.path.join(out_dir, "page-{{0}}.html".format(page_index + 1))
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")
shot_path = os.path.join(out_dir, "page-{{0}}.png".format(page_index + 1))
try:
sb.save_screenshot(shot_path)
screenshots.append(shot_path)
except Exception as exc:
warnings.append(f"screenshot failed: {{exc}}")
warnings.append("screenshot failed: " + str(exc))
item, next_raw = extract_with_bs4(page_source, PAYLOAD["selectors"])
item["url"] = current_url
@@ -516,12 +516,12 @@ def _render_scraper_script(payload: dict[str, Any]) -> str:
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}}")
warnings.append("same-domain safeguard refused pagination URL " + str(next_url))
break
current = next_url
except Exception as exc:
status = "error"
warnings.append(f"browser automation failed: {{type(exc).__name__}}: {{exc}}")
warnings.append("browser automation failed: {{0}}: {{1}}".format(type(exc).__name__, exc))
warnings.append(traceback.format_exc()[-4000:])
saved_paths = write_outputs(items, pages_visited, html_captures, screenshots, warnings, status)