"""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. This module now also exposes an LLM-powered iterative skill that can plan and refine safe scraping runs using the base browser skill as an execution step. It uses a platform-provided LLM grant (ctx.llm) with strict safety rails and never attempts to evade protections. The deep agent proposes selector/URL adjustments limited to the same origin and stops on any challenge. """ from __future__ import annotations import json import re import textwrap import urllib.parse import urllib.request import urllib.robotparser from datetime import datetime, timezone 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/" DEEPAGENTS_RECURSION_LIMIT = 500 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.1.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 LLM use reads 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", "deepagents", "langchain") 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)}, } @skill( name="iterate_scrape_goal", description=( "Iteratively plan and run compliant scraping passes until a goal is met, " "using a platform-provided LLM with strict safety rails. The deep agent may " "refine CSS selectors and pagination hints within the same origin only, and will stop " "on any CAPTCHA/anti-bot/login/paywall/access-control indicator." ), timeout_seconds=900, cost_class="llm-medium", grant_outputs_prefix=OUTPUT_DIR, grant_write_prefixes=(OUTPUT_DIR,), grant_run_timeout_seconds=900, ) async def iterative_scrape_goal( self, ctx: RunContext[NoAuth], goal: str, start_url: str, allowed_domains: list[str] | None = None, seed_selectors: dict[str, str] | None = None, max_steps: int = 2, max_pages: int = 1, min_items: int = 1, max_runtime_seconds: int = 600, 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]: """Use an LLM planner to refine selectors/pagination until the goal is met or a stop condition occurs.""" start_url = str(start_url or "").strip() if not start_url: return {"status": "invalid_input", "items": [], "saved_paths": [], "warnings": ["url is required"], "screenshots": []} # Initial validation (if selectors are provided) selectors = seed_selectors or {"title": "title"} if selectors: v = _validate_inputs( url=start_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 v["errors"]: return {"status": "invalid_input", "items": [], "saved_paths": [], "warnings": v["errors"], "screenshots": []} creds = ctx.llm same_origin_safeguard = _same_origin await ctx.emit_progress("starting deep-agent iterative scrape planner") # Domain guard def _allowed_origin(url: str) -> bool: if not allowed_domains: return same_origin_safeguard(start_url, url) host = urllib.parse.urlparse(url).netloc.lower() return any(host.endswith(dom.lower()) for dom in allowed_domains) # Planner system prompt with strict safety policy system_prompt = ( "You are a cautious scraping planner for a compliant browser scraper. " "Your actions are limited to proposing: (a) CSS selectors to extract fields; " "(b) a same-origin URL (relative or absolute) for the next pass; (c) a reason to stop.\n" "Hard safety rules:\n" "- Never suggest evading CAPTCHAs, anti-bot, or access controls.\n" "- Never suggest login, credential use, or paywall bypass.\n" "- Always respect robots.txt and same-origin restriction with the initial URL's scheme+host.\n" "- Do NOT use stealth, proxies, or automation-evasion.\n" "Output only compact JSON with keys: action ('scrape'|'stop'), reason, selectors (object, optional), url (string, optional), confidence (0..1)." ) decisions: list[dict[str, Any]] = [] step_logs: list[dict[str, Any]] = [] all_items: list[dict[str, Any]] = [] all_pages: list[str] = [] all_htmls: list[str] = [] all_screens: list[str] = [] warnings: list[str] = [] saved_paths: list[str] = [] if not creds.api_key: # Deterministic fallback: run one pass with provided selectors (or trivial title extraction) await ctx.emit_progress("LLM creds unavailable; running deterministic single-pass fallback") sel = selectors or {"title": "title"} one = await self.scrape_website( ctx=ctx, url=start_url, selectors=sel, max_pages=max_pages, wait_seconds=wait_seconds, output_format=output_format, headless=headless, user_agent=user_agent, respect_robots=respect_robots, rate_limit_seconds=rate_limit_seconds, screenshot=screenshot, ) return { "status": one.get("status") or "ok", "items": one.get("items") or [], "saved_paths": one.get("saved_paths") or [], "warnings": one.get("warnings") or [], "screenshots": one.get("screenshots") or [], "pages_visited": one.get("pages_visited") or [], "html_captures": one.get("html_captures") or [], "iterations": 1, "decisions": [], } # Build deep agent planner try: planner = self._build_deep_agent(ctx=ctx, creds=creds, system_prompt=system_prompt) except Exception as exc: # noqa: BLE001 warnings.append("planner_unavailable: " + _friendly_error(exc)) planner = None current_url = start_url current_selectors = selectors or {"title": "title"} hard_cap_steps = max(1, min(int(max_steps), 3)) hard_cap_runtime = max(30, min(int(max_runtime_seconds), 840)) started_at = datetime.now(timezone.utc) for step in range(1, hard_cap_steps + 1): # Runtime budget check if (datetime.now(timezone.utc) - started_at).total_seconds() > hard_cap_runtime: warnings.append("max_runtime_seconds reached; stopping") break # Execute one browser pass plan_note = "initial pass" if step == 1 else "refined pass" run = await self.scrape_website( ctx=ctx, url=current_url, selectors=current_selectors, max_pages=max_pages, wait_seconds=wait_seconds, output_format=output_format, headless=headless, user_agent=user_agent, respect_robots=respect_robots, rate_limit_seconds=rate_limit_seconds, screenshot=screenshot, ) status = str(run.get("status") or "ok") items = list(run.get("items") or []) all_items.extend(items) all_pages.extend(run.get("pages_visited") or []) all_htmls.extend(run.get("html_captures") or []) all_screens.extend(run.get("screenshots") or []) warnings.extend(run.get("warnings") or []) saved_paths.extend(run.get("saved_paths") or []) step_logs.append({ "plan": plan_note, "action": {"url": current_url, "selectors": current_selectors, "max_pages": max_pages}, "observations": {"status": status, "items_this_pass": len(items), "pages": run.get("pages_visited") or []}, "safety": {"robots": bool(respect_robots)}, "decision": None, }) if status in {"challenge_detected", "manual_intervention_required", "blocked", "blocked_by_robots_txt", "browser_runtime_unavailable"}: warnings.append(f"step {step} stopped due to status={status}") step_logs[-1]["decision"] = {"stop": True, "reason": status} break if len(all_items) >= int(min_items): decisions.append({"step": step, "action": "stop", "reason": "min_items reached", "confidence": 1.0}) step_logs[-1]["decision"] = {"stop": True, "reason": "min_items reached"} break # Ask planner what to do next summary = { "step": step, "goal": str(goal or "").strip(), "start_url": start_url, "current_url": current_url, "same_origin": same_origin_safeguard(start_url, current_url), "items_this_pass": len(items), "items_total": len(all_items), "example_item": items[0] if items else None, "current_selectors": current_selectors, "status": status, } prompt = json.dumps( { "instruction": "Propose next step using only same-origin URL and safe CSS selectors, or stop.", "schema": { "action": "scrape|stop", "reason": "string", "selectors": {"field": "CSS"}, "url": "relative or absolute same-origin URL", "confidence": 0.0, }, "context": summary, }, ensure_ascii=False, ) if planner is None: decisions.append({"step": step, "action": "stop", "reason": "planner unavailable", "confidence": 0.0}) break state = await planner.ainvoke( {"messages": [{"role": "user", "content": prompt}]}, config={"recursion_limit": DEEPAGENTS_RECURSION_LIMIT}, ) raw = _last_message_text(state) try: decision = _extract_json_object(raw) except Exception as exc: # noqa: BLE001 warnings.append("planner response was not JSON; stopping: " + _friendly_error(exc)) break action = str(decision.get("action") or "").strip().lower() reason = str(decision.get("reason") or "").strip() next_url = str(decision.get("url") or current_url).strip() or current_url proposed_selectors = decision.get("selectors") or {} confidence = float(decision.get("confidence") or 0.0) # Enforce safety: allowed_domains and same-origin only if not _allowed_origin(next_url): decisions.append({ "step": step, "action": "stop", "reason": "planner suggested URL outside allowed domains; stopping", "confidence": confidence, }) step_logs[-1]["decision"] = {"stop": True, "reason": "outside allowed domains"} break if action == "stop": decisions.append({"step": step, "action": "stop", "reason": reason, "confidence": confidence}) step_logs[-1]["decision"] = {"stop": True, "reason": reason} break if action != "scrape": decisions.append({ "step": step, "action": "stop", "reason": "planner returned unknown action", "confidence": confidence, }) step_logs[-1]["decision"] = {"stop": True, "reason": "unknown action"} break # Adopt next plan current_url = next_url if isinstance(proposed_selectors, dict) and proposed_selectors: safe_map: dict[str, str] = {} for k, v in proposed_selectors.items(): k2 = str(k).strip() v2 = str(v).strip() if k2 and v2: safe_map[k2] = v2 if safe_map: current_selectors = safe_map decisions.append({ "step": step, "action": "scrape", "reason": reason or "refining selectors/pagination", "confidence": confidence, "url": current_url, "selectors": current_selectors, }) step_logs[-1]["decision"] = {"stop": False, "next_url": current_url, "selectors": current_selectors} # Deduplicate items by URL+page_index if present seen_keys: set[tuple[Any, Any]] = set() unique_items: list[dict[str, Any]] = [] for it in all_items: key = (it.get("url"), it.get("page_index")) if key in seen_keys: continue seen_keys.add(key) unique_items.append(it) # Persist a small run manifest for traceability manifest_path = await _write_iter_manifest(ctx, { "goal": goal, "start_url": start_url, "timestamp": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "steps": len(decisions) or min(int(max_steps), 1), "decisions": decisions, "items_total": len(unique_items), }) if manifest_path: saved_paths.append(manifest_path) return { "status": "ok" if unique_items else "no_items", "items": unique_items, "saved_paths": saved_paths, "warnings": warnings, "screenshots": all_screens, "pages_visited": all_pages, "html_captures": all_htmls, "iterations": len(decisions) or 0, "steps": len(decisions) or 0, "decisions": decisions, } def _build_deep_agent(self, *, ctx: RunContext[NoAuth], creds: Any, system_prompt: str): from a2a_pack.deepagents import create_a2a_deep_agent from langchain.agents.middleware import wrap_model_call @wrap_model_call async def safe_model_call(request: Any, handler: Any) -> Any: messages = request.state.get("messages", []) print( "[seleniumbase-website-scraper] planner model_call " f"model={creds.model} source={creds.source} messages={len(messages)}" ) return await handler(request) backend = ctx.workspace_backend() skill_sources = _seed_runtime_skills(backend, ctx) return create_a2a_deep_agent( ctx, creds=creds, backend=backend, skills=skill_sources or None, middleware=[safe_model_call], system_prompt=system_prompt, ) 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(2000000).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(2000000).decode("utf-8", errors="ignore") rp.parse(body.splitlines()) return bool(rp.can_fetch(user_agent, url)), None except Exception as 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() 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("selector {{0}}={{1!r}} failed: {{2}}".format(key, raw_selector, 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("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 "*") if robot_warning: warnings.append(robot_warning) if not allowed: status = "blocked_by_robots_txt" 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"])) 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, "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, "page-{{0}}-challenge.png".format(page_index + 1)) try: sb.save_screenshot(shot_path) screenshots.append(shot_path) except Exception as exc: warnings.append("screenshot failed: " + str(exc)) break 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, "page-{{0}}.png".format(page_index + 1)) try: sb.save_screenshot(shot_path) screenshots.append(shot_path) except Exception as exc: warnings.append("screenshot failed: " + str(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("same-domain safeguard refused pagination URL " + str(next_url)) break current = next_url except Exception as exc: status = "error" 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) 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 def _same_origin(a: str, b: str) -> bool: pa, pb = urllib.parse.urlparse(a), urllib.parse.urlparse(b) return pa.scheme == pb.scheme and pa.netloc.lower() == pb.netloc.lower() 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 def _runtime_skills_root(ctx: RunContext[Any]) -> str: workspace = getattr(ctx, "_workspace", None) prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ()) if not prefixes: outputs_prefix = getattr(workspace, "outputs_prefix", None) prefixes = (outputs_prefix or "outputs/",) prefix = str(prefixes[0]).strip("/") return f"/{prefix}/{RUNTIME_SKILLS_DIR}" if prefix else f"/{RUNTIME_SKILLS_DIR}" def _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]: # Seed any packaged runtime skills if present root = Path(__file__).parent / "skills" if not root.exists(): return [] runtime_root = _runtime_skills_root(ctx) uploads: list[tuple[str, bytes]] = [] for path in root.rglob("*"): if path.is_file(): rel = path.relative_to(root).as_posix() uploads.append((runtime_root + rel, path.read_bytes())) if uploads: backend.upload_files(uploads) return [runtime_root] return [] def _extract_json_object(raw: str) -> dict[str, Any]: text = str(raw or "").strip() if text.startswith("```"): text = re.sub(r"^```(?:json)?\s*", "", text) text = re.sub(r"\s*```$", "", text) try: parsed = json.loads(text) except json.JSONDecodeError: start = text.find("{") end = text.rfind("}") if start < 0 or end <= start: raise parsed = json.loads(text[start : end + 1]) if not isinstance(parsed, dict): raise ValueError("planner result was not a JSON object") return parsed def _last_message_text(state: dict[str, Any]) -> str: messages = state.get("messages") or [] if not messages: return json.dumps(state, default=str) content = getattr(messages[-1], "content", None) if isinstance(content, str): return content if isinstance(content, list): parts: list[str] = [] for item in content: if isinstance(item, dict): text = item.get("text") or item.get("content") if text: parts.append(str(text)) elif item: parts.append(str(item)) return "\n".join(parts) return str(content or messages[-1]) def _friendly_error(exc: Exception) -> str: message = str(exc) or exc.__class__.__name__ message = re.sub(r"(?i)(password|passwd|pwd|secret|token|key)=\S+", r"\1=", message) message = re.sub(r"(?i)(auth(?:entication)? failed).*", r"\1", message) return message[:400] async def _write_iter_manifest(ctx: RunContext[Any], manifest: dict[str, Any]) -> str | "": try: out_dir = OUTPUT_DIR name = f"iter-plan-" + datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + ".json" path = f"{out_dir}/{name}" ctx.workspace.write_bytes(path, json.dumps(manifest, ensure_ascii=False, indent=2).encode("utf-8")) return path except Exception: # noqa: BLE001 return ""