"""Small, isolated Patchright/Camoufox browser comparison.""" import argparse import json import os from pathlib import Path import subprocess import sys import time TARGETS = { "wikipedia": { "url": "https://en.wikipedia.org/wiki/Web_scraping", "required": ["web scraping"], "min_text": 500, }, "amazon": { "url": "https://www.amazon.com/s?k=laptop", "required": ["laptop"], "min_text": 500, }, "g2": { "url": "https://www.g2.com/products/asana/reviews", "required": ["asana", "reviews"], "min_text": 500, }, } BLOCK_MARKERS = ( "just a moment", "verify you are human", "robot check", "pardon our interruption", "access denied", "checking your browser", "unusual traffic", "press and hold", ) def classify(target, status, text, title, final_url): low = (title + "\n" + text).lower() if status != 200: return "http_error" if status is not None else "no_response" if any(marker in low for marker in BLOCK_MARKERS): return "block_or_challenge" if len(text) < target["min_text"]: return "empty_or_short_page" if any(marker not in low for marker in target["required"]): return "required_content_missing" if "/captcha" in final_url.lower() or "/challenge" in final_url.lower(): return "challenge_redirect" return "likely_usable" def worker(args): target = TARGETS[args.target] output = Path(args.output) output.mkdir(parents=True, exist_ok=True) result = { "tool": args.tool, "target": args.target, "url": target["url"], "mode": args.mode, "proxy": "none", "headless": True, "started_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), } started = time.monotonic() try: if args.tool == "patchright": from patchright.sync_api import sync_playwright with sync_playwright() as playwright: browser = playwright.chromium.launch(headless=True) try: inspect_page(browser, target, output, result) result["browser_version"] = browser.version finally: browser.close() else: from camoufox.sync_api import Camoufox with Camoufox(headless=True, os="linux") as browser: inspect_page(browser, target, output, result) result["browser_version"] = browser.version except Exception as exc: result["error"] = f"{type(exc).__name__}: {str(exc)[:400]}" result.setdefault("classification", "transport_or_browser_error") finally: result["wall_seconds"] = round(time.monotonic() - started, 3) result["finished_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) (output / "result.json").write_text(json.dumps(result, indent=2) + "\n") def inspect_page(browser, target, output, result): page = browser.new_page() try: response = page.goto(target["url"], wait_until="domcontentloaded", timeout=30000) result["status"] = response.status if response else None if result["status"] == 200: try: page.wait_for_function( "minimum => document.body && document.body.innerText.length >= minimum", arg=target["min_text"], timeout=8000, ) except Exception: pass page.wait_for_timeout(500) result["final_url"] = page.url result["title"] = page.title() page.screenshot(path=str(output / "screenshot.png"), full_page=False, timeout=10000) html = page.content() text = page.locator("body").inner_text(timeout=5000) (output / "response.html").write_text(html) (output / "visible.txt").write_text(text) result["html_bytes"] = len(html.encode()) result["visible_chars"] = len(text) result["classification"] = classify( target, result["status"], text, result["title"], result["final_url"] ) finally: page.close() def run_group(tool, mode, targets, output_root, comparison_dir): import psutil launched = [] for target in targets: output = output_root / mode / tool / target worker_python = comparison_dir / tool / "bin" / "python" env = os.environ.copy() env["PLAYWRIGHT_BROWSERS_PATH"] = str(comparison_dir / "browsers") env["XDG_CACHE_HOME"] = str(comparison_dir / "cache") proc = subprocess.Popen( [str(worker_python), __file__, "worker", tool, mode, target, str(output)], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, env=env, ) launched.append((proc, target, output)) start = time.monotonic() peak_pss = 0 peak_rss = 0 max_cpu_by_pid = {} max_processes = 0 while any(proc.poll() is None for proc, _, _ in launched): pss, rss, count = 0, 0, 0 for proc, _, _ in launched: try: parent = psutil.Process(proc.pid) members = [parent, *parent.children(recursive=True)] except psutil.NoSuchProcess: continue for member in members: try: info = member.memory_full_info() pss += getattr(info, "pss", info.rss) rss += info.rss times = member.cpu_times() max_cpu_by_pid[member.pid] = max( max_cpu_by_pid.get(member.pid, 0), times.user + times.system ) count += 1 except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): pass peak_pss = max(peak_pss, pss) peak_rss = max(peak_rss, rss) max_processes = max(max_processes, count) time.sleep(0.2) results = [] for proc, target, output in launched: _, stderr = proc.communicate() result_path = output / "result.json" result = json.loads(result_path.read_text()) if result_path.exists() else { "tool": tool, "target": target, "error": stderr.decode(errors="replace")[-500:] } result["exit_code"] = proc.returncode results.append(result) summary = { "tool": tool, "mode": mode, "targets": targets, "group_wall_seconds": round(time.monotonic() - start, 3), "peak_pss_mb": round(peak_pss / 1048576, 1), "peak_rss_sum_mb": round(peak_rss / 1048576, 1), "sampled_cpu_seconds": round(sum(max_cpu_by_pid.values()), 2), "max_processes": max_processes, "results": results, } path = output_root / mode / f"{tool}.json" path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(summary, indent=2) + "\n") print(json.dumps({ "tool": tool, "mode": mode, "group_wall_seconds": summary["group_wall_seconds"], "peak_pss_mb": summary["peak_pss_mb"], "sampled_cpu_seconds": summary["sampled_cpu_seconds"], "results": [{"target": r["target"], "classification": r.get("classification"), "status": r.get("status"), "error": r.get("error")} for r in results], }), flush=True) def main(): parser = argparse.ArgumentParser() sub = parser.add_subparsers(dest="command", required=True) worker_parser = sub.add_parser("worker") worker_parser.add_argument("tool", choices=["patchright", "camoufox"]) worker_parser.add_argument("mode") worker_parser.add_argument("target", choices=TARGETS) worker_parser.add_argument("output") driver_parser = sub.add_parser("driver") driver_parser.add_argument("comparison_dir", type=Path) args = parser.parse_args() if args.command == "worker": worker(args) else: output = args.comparison_dir / "results-v2" for target in TARGETS: for tool in ("patchright", "camoufox"): run_group(tool, f"single-{target}", [target], output, args.comparison_dir) for tool in ("patchright", "camoufox"): run_group(tool, "concurrent-3", list(TARGETS), output, args.comparison_dir) if __name__ == "__main__": main()