【DrissionPage源码-0】了解CDP

前言：

如果你做过爬虫或浏览器自动化，大概率用过 Selenium。它很强大，但也有痛点：启动慢、资源占用高、操作容易被反爬检测。后来 DrissionPage 横空出世，直接用 CDP 协议控制浏览器，性能和灵活性都上了一个台阶。

我用 DrissionPage 做了不少项目，但每次遇到问题去翻源码时，总觉得雾里看花。与其被动地"用"，不如主动地"造"一次。于是打算从头研究一下项目源码，整理一下。同时也缓冲一下之前的旧坑。

什么是cdp

1. CDP 的本质：JSON-RPC over WebSocket

CDP 并不是什么魔法，它本质上就是一堆 JSON 数据包在 WebSocket 上发来发去。

当打开 Chrome 浏览器的F12 开发者工具时，那个“开发者工具窗口”其实就是一个用 HTML/JS 写的前端网页。

• 当点击“Console”选项卡时，工具发送了一条 CDP 命令给浏览器内核。
• 当点击“Network”查看抓包时，浏览器内核通过 CDP 把请求数据推送到工具界面。

通信过程示例：

假设你想让浏览器跳转到百度，你的程序（客户端）会通过 WebSocket 发送这样一段 JSON：

{ "id": 1, "method": "Page.navigate", "params": { "url": "https://www.baidu.com" } }

浏览器（服务端）收到后，执行跳转，然后回传：

{ "id": 1, "result": { "frameId": "A1B2C3..." } } //这个frameid 在

2. CDP 能做什么？（远超简单的 JS）

CDP 把浏览器能力划分为不同的Domains (域)，主要有：

• Runtime 域: 在页面上下文中执行 JS 代码（Runtime.evaluate）。
  • 反检测用途: 可以在页面加载前注入 JS（Page.addScriptToEvaluateOnNewDocument），用来覆盖 navigator.webdriver。
• Network 域: 拦截、修改、阻断网络请求。
  • 反检测用途: 可以拦截浏览器发出的指纹请求，替换掉 Header 或者返回假数据。
• Page 域: 控制页面加载、截图、打印 PDF。
• DOM 域: 直接获取 DOM 树，修改节点（不经过 JS，直接改内核数据）。
• Debugger 域: 设置断点。这就是为什么爬虫可以逆向调试 JS。
• Emulation 域:
  • 关键功能: 它可以模拟移动端、修改 User-Agent、修改时区、修改地理位置。这对于通过环境检测非常有用。

3. CDP 与 Puppeteer / Playwright 的关系

Puppeteer 或 Playwright，它们其实就是CDP 的封装库。

• Chromium: 提供 CDP 接口（服务端）。
• Puppeteer (Node.js): 帮你把 browser.newPage() 这种好写的代码，翻译成底层的 Target.createTarget JSON 命令，并通过 WebSocket 发给 Chromium。

webdriver和cdp

1. WebDriver 与 WebKit 的关系

WebDriver 是什么？（它是“翻译官”）

WebDriver 是一个 W3C 定义的标准接口协议。就像 USB 接口一样，标准是统一的，但插入的设备（浏览器）不同，就需要不同的驱动程序。

• 谷歌 Chrome: 对应 chromedriver
• 微软 Edge: 对应 msedgedriver
• 火狐 Firefox: 对应 geckodriver
• 苹果 Safari: 对应 safaridriver
• https://github.com/LoseNine/AutoWK的 WebKit: 对应 WebDriver.exe (这是一个专门为 WebKit 内核编译的驱动)
  • autowk部分源码

import http.client import subprocess import os import psutil import json def get_bin_file_path(filename): current_dir = os.path.dirname(os.path.abspath(__file__)) exe_path = os.path.join(current_dir, ".", "bin", filename) exe_path = os.path.abspath(exe_path) return exe_path class AutoWKBase: def __init__(self, host, port,webkit_path=None,webdriver_bat=None): self.host = host self.port = port self.headers = {"Content-Type": "application/json"} self.session_id = None self.conn = None if not webkit_path and not webdriver_bat: self.webkit_path = get_bin_file_path("MiniBrowser.exe") self.webdriver_bat = get_bin_file_path("WebDriver.exe") #用於關閉引導頁 self.closePagefile="file:///"+get_bin_file_path("closePage.html") self.minibrowseraddr = f"{self.host}:{self.port + 1}" def launch_webkit(self,x=0,y=0,width=10,height=10,lang="en-US",timezone="America/Chicago", proxyType='',proxyHost='',proxyPort='',proxyUsername='',proxyPassword='', userDataDir='',fpfile='',userAgent='',headless=False,enableListen=False,networkListenPort=0): env = os.environ.copy() env["WEBKIT_INSPECTOR_SERVER"] = self.minibrowseraddr #给进行通信的窗口设置大小，实际上启动完就可以关闭了 args = [ self.webkit_path, f"--x={x}", f"--y={y}", f"--width={width}", f"--height={height}", f"--lang={lang}", f"--timezone={timezone}", f"--url={self.closePagefile}", ] if proxyType and proxyHost and proxyPort: args.append(f"--proxyType={proxyType}") args.append(f"--proxyHost={proxyHost}") args.append(f"--proxyPort={proxyPort}") if proxyUsername and proxyPassword: args.append(f"--proxyUsername={proxyUsername}") args.append(f"--proxyPassword={proxyPassword}") if userDataDir: args.append(f"--userDataDir={userDataDir}") if fpfile: args.append(f"--fpfile={fpfile}") if userAgent: args.append(f"--userAgent={userAgent}") if headless: args.append(f"--headless") if enableListen: args.append(f"--enableListen") if networkListenPort: args.append(f"--networkListenPort={networkListenPort}") self.webkit_process = subprocess.Popen(args, env=env) def launch_webdriver(self): for proc in psutil.process_iter(['name']): try: if proc.info['name'] and 'WebDriver.exe' in proc.info['name']: subprocess.run(["taskkill", "/f", "/im", "WebDriver.exe"], stdout=subprocess.DEVNULL) except (psutil.NoSuchProcess, psutil.AccessDenied): continue args = [ self.webdriver_bat, f"--target={self.minibrowseraddr}", f"--port={str(self.port)}", ] self.webdriver_process = subprocess.Popen(args) def connect(self): self.conn = http.client.HTTPConnection(self.host, self.port) def request(self, method, endpoint, body=None): if body is None or body == {}: body = {"capabilities": {"firstMatch": [{}]}} self.conn.request(method, endpoint, body=json.dumps(body) if body else None, headers=self.headers) return json.loads(self.conn.getresponse().read().decode("utf-8")) def create_session(self): result = self.request("POST", "/session") self.session_id = result["value"]["sessionId"] def delete_session(self): return self.request("DELETE", f"/session/{self.session_id}") def close(self): print("[INFO] Closing connection and shutting down MiniBrowser and WebDriver...") if self.conn: self.conn.close() for proc in psutil.process_iter(['name']): try: if proc.info['name']: if 'MiniBrowser.exe' in proc.info['name']: print(f"[INFO] Terminating process: {proc.info['name']} (PID {proc.pid})") proc.terminate() if 'WebDriver.exe' in proc.info['name']: print(f"[INFO] Terminating process: {proc.info['name']} (PID {proc.pid})") subprocess.run(["taskkill", "/f", "/im", "WebDriver.exe"], stdout=subprocess.DEVNULL) except (psutil.NoSuchProcess, psutil.AccessDenied): continue print("[INFO] autowk processes terminated.")

这些 Driver 的作用就是：接收 Python 发来的统一 HTTP 指令（比如“点击”），翻译成浏览器内部能听懂的私有指令。

WebKit 是什么？（它是“发动机”）

WebKit 是一个浏览器排版引擎（渲染引擎）。它是浏览器的核心，负责把 HTML/CSS 代码变成你屏幕上看到的图像。

• 血缘关系：
  • Safari: 使用纯正的 WebKit 引擎。
  • Chrome: 以前也用 WebKit，后来 Google 觉得不爽，把 WebKit 拿过来改了改，起名叫Blink（现在的 Chrome 内核其实是 WebKit 的一个分支）。
  • AutoWK / MiniBrowser: 这是一个基于纯 WebKit（类似 Safari 内核）编译出来的轻量级浏览器，不是 Chrome。

结论：因为内核不同，所以不能用 chromedriver 去控制 MiniBrowser，必须用配套的 WebDriver.exe。

2. 为什么说 Selenium “庞大”？它依赖了什么？

前面展示的 AutoWK 代码非常“原生”，它只用了 Python 自带的 http.client，不需要安装任何第三方库。

相比之下，Selenium是一个重型框架。当 pip install selenium 时，它不仅仅是下载了代码，还引入了一套复杂的生态：

• 第三方依赖库：
  • urllib3: 处理 HTTP 连接池、重试等（Selenium 不用 Python 自带的 http 库，因为它太弱）。
  • trio / trio-websocket: Selenium 4 为了支持 CDP 和异步，引入了这套庞大的异步 IO 库。
  • certifi: 处理 SSL 证书。
• 对象封装的开销：
  • 在 AutoWK 里，你发个 HTTP 请求就完了。
  • 在 Selenium 里，你获取一个元素 ele = driver.find_element(…)，Selenium 会在内存里创建一个 WebElement 对象，这个对象绑定了 session ID、parent ID 等各种属性。当你有成千上万个元素时，这种封装就是一种负担。
• 启动速度：
  • 加载 selenium 库本身需要解析大量 Python 文件，而 import http.client 几乎是瞬时的。

3. 我能直接用 Python 调用 CDP 操作浏览器吗？

答案是：绝对可以，而且这是目前最高级的玩法。

只要你的 Python 能发 WebSocket 数据包，你就能控制 Chrome/Edge/CEF。

如何实现？

你需要用到 websockets 这个库（比 Selenium 轻量得多，pip install websockets），或者直接用 socket 手撸。

极简代码示例（直接控制 Chrome）：

首先，启动 Chrome 并开启调试端口：

chrome.exe --remote-debugging-port=9222

然后，用 Python 控制它：

import asyncio import websockets import json import requests async def control_chrome(): # 1. 获取 WebSocket 调试地址 # Chrome 会在 http://127.0.0.1:9222/json 暴露当前页面的 WebSocket URL response = requests.get("http://127.0.0.1:9222/json") pages = response.json() # 拿到第一个标签页的 WebSocket 地址 ws_url = pages[0]['webSocketDebuggerUrl'] print(f"Connecting to: {ws_url}") # 2. 建立 WebSocket 连接 async with websockets.connect(ws_url) as ws: # 3. 发送 CDP 命令：跳转到百度 # 每一个命令都有唯一的 id，method 是 CDP 的方法名 command = { "id": 1, "method": "Page.navigate", "params": { "url": "https://www.baidu.com" } } await ws.send(json.dumps(command)) # 4. 接收结果 result = await ws.recv() print(f"Receive: {result}") # 5. 发送 CDP 命令：执行 JS 获取 UserAgent js_command = { "id": 2, "method": "Runtime.evaluate", "params": { "expression": "navigator.userAgent" } } await ws.send(json.dumps(js_command)) result = await ws.recv() print(f"JS Result: {result}") # 运行 asyncio.get_event_loop().run_until_complete(control_chrome())

这种方式的优缺点：

• 优点：
  • 无敌轻量：没有 Selenium，没有 Driver，只有 WebSocket。
  • 权限最高：你可以调用 CDP 的所有隐藏功能（改指纹、拦截网络、模拟地理位置）。
  • 反检测强：因为没有 webdriver 属性注入。
• 缺点：
  • 代码难写：你需要自己管理 JSON 里的 id，自己处理异步回调。
  • 维护累：CDP 协议有时候会变，没有库帮你屏蔽差异。

CDP（Chrome DevTools Protocol）底层架构

1. 启动流程

# 启动 Chrome 并开启远程调试chrome --remote-debugging-port=9222--headless# 输出类似：# DevTools listening on ws://127.0.0.1:9222/devtools/browser/xxx

关键点：

• Chrome 内部启动了一个WebSocket Server
• 端口 9222 监听外部连接
• 不需要安装任何驱动（如 chromedriver），直接和浏览器通信

2. 通信协议（两层架构）

┌──────────────────────────────────────────────────────┐ │ Python / Node.js 客户端 │ │ (DrissionPage, Puppeteer, etc.) │ └───────────────────┬──────────────────────────────────┘ │ │ ① HTTP 获取 Tab 列表 │ GET http://localhost:9222/json │ ▼ ┌──────────────────────────────────────────────────────┐ │ Chrome HTTP 服务器 │ │ 返回所有 Tab 的 WebSocket URL │ └───────────────────┬──────────────────────────────────┘ │ │ ② WebSocket 连接到具体 Tab │ ws://localhost:9222/devtools/page/xxx │ ▼ ┌──────────────────────────────────────────────────────┐ │ Chrome Tab (真实浏览器实例) │ │ - V8 引擎执行 JS │ │ - Blink 渲染引擎处理 DOM │ │ - 真实的 window/document/navigator │ └──────────────────────────────────────────────────────┘

3. HTTP 层：获取 Tab 信息

# 访问 http://localhost:9222/jsoncurlhttp://localhost:9222/json# 返回 JSON（所有打开的 Tab）[{"description":"","devtoolsFrontendUrl":"/devtools/inspector.html?ws=localhost:9222/devtools/page/XXX","id":"E3F9F8C...","title":"百度","type":"page","url":"https://www.baidu.com","webSocketDebuggerUrl":"ws://localhost:9222/devtools/page/XXX"← 关键！}]

所以这个可能会被反检测

但是 获取列表的这个 HTTP 请求 是一个特征。 如果一个网站的 JavaScript 极其狡猾，它可能尝试扫描本地端口（虽然浏览器有 CORS 保护，但在某些特定的网络配置或老版本浏览器下可能泄露）： 网页 JS 尝试请求 http://localhost:9222/json。 如果请求成功并返回了 JSON，说明当前浏览器开启了远程调试端口。 结论：用户是机器人/爬虫。

关键点：

• HTTP 负责总览当前 Tabs 信息
• 每个 Tab 有唯一的webSocketDebuggerUrl
• 客户端通过这个 URL 连接到具体的 Tab

4. WebSocket 层：执行命令

Chrome DevTools 协议主要基于 JSON-RPC：每个命令都是一个带有 id/method 和可选参数的 JavaScript 结构

// 客户端发送命令（通过 WebSocket）{"id":1,// 唯一 ID（用于匹配响应）"method":"Runtime.evaluate",// 命令：执行 JS"params":{"expression":"document.title",// JS 代码"returnByValue":true}}// Chrome 返回响应{"id":1,// 对应请求的 ID"result":{"result":{"type":"string","value":"百度一下，你就知道"}}}

关键点：

• 每个发送到 CDP 的命令必须有唯一的 ‘id’ 参数。消息响应将通过 WebSocket 传递，并具有相同的 ‘id’
• 没有 ‘id’ 参数的传入 WebSocket 消息是协议事件
• JS 在真实浏览器中执行，不是模拟！

5. DrissionPage 的run_script()底层实现

DrissionPage 源码分析（简化版）

# DrissionPage 内部实现classChromiumPage:def__init__(self):# 连接到 Chromeself.ws=websocket.create_connection("ws://localhost:9222/devtools/page/XXX")self.msg_id=0defrun_script(self,js_code):"""执行 JS 代码"""# 构建 CDP 命令msg={"id":self.msg_id,"method":"Runtime.evaluate",# CDP 的 Runtime 域"params":{"expression":js_code,# 你的 JS 代码"returnByValue":True,# 返回值而不是对象引用"awaitPromise":True# 等待 Promise 完成}}# 发送到 Chromeself.ws.send(json.dumps(msg))self.msg_id+=1# 等待响应whileTrue:response=json.loads(self.ws.recv())ifresponse.get("id")==msg["id"]:# 找到对应的响应returnresponse["result"]["result"]["value"]

实际流程

fromDrissionPageimportChromiumPage page=ChromiumPage()page.get('https://www.baidu.com')# 当你执行：result=page.run_script('navigator.userAgent')# 底层发生：# 1. DrissionPage 通过 WebSocket 发送：# {"id": 1, "method": "Runtime.evaluate",# "params": {"expression": "navigator.userAgent"}}## 2. Chrome 在真实浏览器中执行 JS## 3. Chrome 返回：# {"id": 1, "result": {"result": {"value": "Mozilla/5.0..."}}}## 4. DrissionPage 解析并返回结果

6. CDP 的核心域（Domain）

CDP 分为多个功能域，常用的有：

Runtime 域（执行 JS）

# Runtime.evaluate - 执行 JSpage.run_cdp('Runtime.evaluate',expression='1+1')# Runtime.callFunctionOn - 调用对象方法page.run_cdp('Runtime.callFunctionOn',functionDeclaration='function() { return this.title; }',objectId='...')

Page 域（页面控制）

# Page.navigate - 导航到 URLpage.run_cdp('Page.navigate',url='https://example.com')# Page.captureScreenshot - 截图page.run_cdp('Page.captureScreenshot')

Network 域（网络）

# Network.setCookies - 设置 Cookiepage.run_cdp('Network.setCookies',cookies=[...])# Network.setExtraHTTPHeaders - 修改请求头page.run_cdp('Network.setExtraHTTPHeaders',headers={'X-Custom':'value'})

DOM 域（DOM 操作）

# DOM.getDocument - 获取 DOM 树page.run_cdp('DOM.getDocument')# DOM.querySelector - 查询元素page.run_cdp('DOM.querySelector',nodeId=1,selector='#id')

更多文章，敬请关注gzh：零基础爬虫第一天