使用CDP自动化操作浏览器

chrome.exe --enable-logging --remote-debugging-port=9000

chrome.exe --user-data-dir="%TEMP%\chrome-test" --remote-debugging-port=9000

import json
import time
import requests
import websocket

def get_browser_ws_url(host="127.0.0.1", port=9000):
    # 获取浏览器级 WebSocket 入口
    info = requests.get(f"http://{host}:{port}/json/version", timeout=3).json()
    return info["webSocketDebuggerUrl"]
class CDPClient:
    def __init__(self, ws_url):
        self.ws = websocket.create_connection(ws_url)
        self._id = 0
    def _next_id(self):
        self._id += 1
        return self._id
    def call(self, method, params=None, session_id=None):
        """发送 CDP 命令并等待同 id 的响应；期间会打印遇到的事件。"""
        msg_id = self._next_id()
        payload = {"id": msg_id, "method": method}
        if params:
            payload["params"] = params
        if session_id:
            payload["sessionId"] = session_id
        self.ws.send(json.dumps(payload))
        while True:
            resp = json.loads(self.ws.recv())
            # 命令响应
            if resp.get("id") == msg_id:
                if "error" in resp:
                    raise RuntimeError(resp["error"])
                return resp
            # 事件（可能在响应前到达）
            if "method" in resp:
                print("event:", resp.get("method"))
    def wait_event(self, expect_method, session_id=None, timeout=10):
        """等待指定事件"""
        old_to = self.ws.gettimeout()
        self.ws.settimeout(timeout)
        try:
            while True:
                msg = json.loads(self.ws.recv())
                if msg.get("method") == expect_method:
                    if session_id is None or msg.get("sessionId") == session_id:
                        return msg
        finally:
            self.ws.settimeout(old_to)
    def close(self):
        self.ws.close()
def open_page_via_cdp(url, host="127.0.0.1", port=9000):
    # 1) 连接到浏览器级 WebSocket
    ws_url = get_browser_ws_url(host, port)
    cdp = CDPClient(ws_url)
    # 2) 创建一个新目标页签
    r = cdp.call("Target.createTarget", {"url": "about:blank"})
    target_id = r["result"]["targetId"]
    # 3) 附加到该目标，拿到会话 sessionId（flatten=true 更简单）
    r = cdp.call("Target.attachToTarget", {"targetId": target_id, "flatten": True})
    session_id = r["result"]["sessionId"]
    # 4) 开启需要的域
    cdp.call("Page.enable", session_id=session_id)
    cdp.call("Runtime.enable", session_id=session_id)
    # 5) 导航到目标 URL
    cdp.call("Page.navigate", {"url": url}, session_id=session_id)
    # 6) 等待页面加载完成事件（也可等待 Page.frameStoppedLoading 或 lifecycleEvent）
    try:
        cdp.wait_event("Page.loadEventFired", session_id=session_id, timeout=15)
    except Exception:
        print("loadEventFired 等待超时，继续后续步骤（页面可能已基本可用）")
    # 7) 读取页面标题作为验证
    eval_res = cdp.call(
        "Runtime.evaluate",
        {"expression": "document.title", "returnByValue": True},
        session_id=session_id,
    )
    title = eval_res["result"]["result"].get("value")
    print(f"Title: {title!r}")
    # 8) 可选：关闭该标签页
    # cdp.call("Target.closeTarget", {"targetId": target_id})
    # cdp.close()
if __name__ == "__main__":
    open_page_via_cdp("https://www.baidu.com")
    open_page_via_cdp("https://www.mi.com")