har-derived-api-client

1#!/usr/bin/env python32"""Capture a HAR from a browser you connect to over CDP (not one you launch).3 4Use this when the browser is owned by someone else and only reachable over the5Chrome DevTools Protocol: Hermes cloud backends (Browserbase, Browser-Use,6Firecrawl), a Camofox session exposing CDP, or anything wired via7`/browser connect <url>` / BROWSER_CDP_URL / browser.cdp_url in config.8 9Why this exists: Playwright's record_har_path only works on a context you10launched locally. connect_over_cdp() attaches to an existing browser, so11record_har is unavailable — we assemble the HAR from CDP Network.* events12ourselves via page.on("request"/"response").13 14Usage:15  python3 har_capture_cdp.py <cdp_url> <output.har> [--wait S] \16      [--goto URL] [--action "fill:SEL:TEXT"] [--action "click:SEL"] ...17 18<cdp_url> is the ws:// or http:// CDP endpoint. For Hermes: run19`/browser connect` to see the active endpoint, or read BROWSER_CDP_URL.20"""21import argparse22import base6423import json24import sys25import time26 27from playwright.sync_api import sync_playwright28 29 30def run_action(page, spec: str) -> None:31    parts = spec.split(":", 2)32    kind = parts[0]33    if kind == "fill":34        page.fill(parts[1], parts[2])35    elif kind == "press":36        page.press(parts[1], parts[2])37    elif kind == "click":38        page.click(parts[1])39    elif kind == "goto":40        page.goto(parts[1] + (":" + parts[2] if len(parts) > 2 else ""))41    elif kind == "sleep":42        time.sleep(float(parts[1]))43    else:44        raise ValueError(f"unknown action: {spec}")45 46 47def _har_entry(req, resp):48    """Build a minimal HAR entry from a Playwright request/response pair."""49    body_text, encoding = "", ""50    if resp is not None:51        try:52            raw = resp.body()53            try:54                body_text = raw.decode("utf-8")55            except UnicodeDecodeError:56                body_text = base64.b64encode(raw).decode("ascii")57                encoding = "base64"58        except Exception:59            pass60    post = req.post_data61    return {62        "_resourceType": req.resource_type,63        "request": {64            "method": req.method,65            "url": req.url,66            "headers": [{"name": k, "value": v} for k, v in req.headers.items()],67            "queryString": [],  # har_to_client.py re-parses the URL, so leave empty68            "postData": {"mimeType": req.headers.get("content-type", ""),69                         "text": post} if post else {},70        },71        "response": {72            "status": resp.status if resp else 0,73            "headers": [{"name": k, "value": v} for k, v in (resp.headers.items() if resp else [])],74            "content": {75                "mimeType": (resp.headers.get("content-type", "") if resp else ""),76                "text": body_text,77                **({"encoding": encoding} if encoding else {}),78            },79        },80    }81 82 83def main() -> int:84    ap = argparse.ArgumentParser()85    ap.add_argument("cdp_url")86    ap.add_argument("har_path")87    ap.add_argument("--goto", default=None, help="URL to navigate to after attaching")88    ap.add_argument("--wait", type=float, default=3.0)89    ap.add_argument("--action", action="append", default=[])90    args = ap.parse_args()91 92    entries = []93    pending = {}  # id(request) -> request94 95    with sync_playwright() as p:96        browser = p.chromium.connect_over_cdp(args.cdp_url)97        context = browser.contexts[0] if browser.contexts else browser.new_context()98        page = context.pages[0] if context.pages else context.new_page()99 100        def on_request(req):101            pending[id(req)] = req102 103        def on_response(resp):104            req = resp.request105            pending.pop(id(req), None)106            entries.append(_har_entry(req, resp))107 108        page.on("request", on_request)109        page.on("response", on_response)110 111        if args.goto:112            page.goto(args.goto, wait_until="domcontentloaded")113        for spec in args.action:114            run_action(page, spec)115            try:116                page.wait_for_load_state("networkidle", timeout=15000)117            except Exception:118                pass119        time.sleep(args.wait)120 121        page.remove_listener("request", on_request)122        page.remove_listener("response", on_response)123        # Do NOT close: we connected to someone else's browser.124 125    har = {"log": {"version": "1.2",126                   "creator": {"name": "har_capture_cdp", "version": "0.1"},127                   "entries": entries}}128    with open(args.har_path, "w", encoding="utf-8") as f:129        json.dump(har, f)130    print(f"HAR written: {args.har_path} ({len(entries)} entries)")131    return 0132 133 134if __name__ == "__main__":135    sys.exit(main())136