为什么是CDP而不是API
这个问题我被人问过不下二十遍。回答之前需要先厘清一个事实：绝大多数AI桌面客户端，底层就是一个Electron壳子包了一层Web应用。你看到的聊天界面、图片生成面板、历史记录列表——全是跑在Chromium渲染引擎上的HTML、CSS和JavaScript。
这意味着什么？意味着这些桌面端天然具备Chrome DevTools Protocol（CDP）的调试能力，只不过默认被关闭了。
有句话说，所有能被浏览器渲染的东西，都能被浏览器协议驱动。API当然是最优雅的路径，但当平台没有开放API、API收费过高、API功能受限、或者你需要绕过某些API层面不方便做的事情时，CDP就成了一条"侧门"。
| 方案 |
优势 |
劣势 |
| 官方API |
稳定、合规、有文档 |
可能不存在、有调用限制、收费 |
| CDP自动化 |
无需API Key、所见即所得、能操作用户视角的一切功能 |
依赖UI结构、有反自动化风险 |
| 逆向抓包 |
最灵活、性能最好 |
接口随时变、加密签名难搞、维护成本高 |
| 屏幕OCR+鼠标模拟 |
平台无关 |
极慢、极不稳定、分辨率依赖 |
本文聚焦第二条路径，以某主流AI桌面客户端的图片生成功能为例，完整走一遍从开启调试端口、建立协议会话、注入登录态到批量出图的全流程。不是玩具demo，是真正跑过几百张图之后的工程总结。
前置知识：CDP协议的核心概念
在动手之前，有必要花几分钟理解CDP协议的几个基础概念，否则后面的代码你只能照抄而不能调试。
CDP本质上是一套基于WebSocket的JSON-RPC协议。浏览器（或Electron应用）作为Server端，你的自动化脚本作为Client端，双方通过WebSocket双向通信。每一次交互都是一个"方法调用+响应"的模式：
- 你发送
{"id": 1, "method": "Page.navigate", "params": {"url": "..."}}
- 浏览器返回
{"id": 1, "result": {"frameId": "xxx"}}
除了请求-响应模式，CDP还有事件推送机制。浏览器会主动给你发消息，比如页面加载完成（Page.loadEventFired）、控制台输出（Runtime.consoleAPICalled）、网络请求（Network.requestWillBeSent）。这些事件需要先通过XXX.enable方法来激活。
CDP的方法被组织成若干"域"（Domain），我们最常用的有：
- Page：页面导航、加载、截图
- Runtime：在页面上下文中执行JavaScript
- DOM：查询和操作DOM树
- Network：拦截/修改网络请求、管理Cookie
- Input：模拟鼠标点击、键盘输入、触摸
- Target：管理浏览器中的所有可调试目标（页面、iframe、Service Worker等）
理解这些域的职责分工，后面遇到问题时你就知道该去哪个域里找答案。
找到Electron的调试入口
启动参数注入
Electron应用启动时接受一组Chromium原生命令行参数。其中与我们最相关的是：
1
2
|
--remote-debugging-port=9222
--remote-debugging-address=127.0.0.1
|
在Windows上，你可以通过修改快捷方式的目标字段来追加这些参数。在macOS上，用终端直接拉起：
1
2
3
|
/Applications/SomeAI.app/Contents/MacOS/SomeAI \
--remote-debugging-port=9222 \
--remote-debugging-address=127.0.0.1
|
Linux环境下更直接，改.desktop文件或者写个wrapper脚本即可。
启动成功后，用浏览器访问http://127.0.0.1:9222/json，你应该能看到一个JSON数组，里面列出了所有可调试的页面（target）。每个target有webSocketDebuggerUrl字段——这就是CDP会话的入口。典型的返回结构长这样：
1
2
3
4
5
6
7
8
9
10
11
|
[
{
"description": "",
"devtoolsFrontendUrl": "/devtools/inspector.html?ws=...",
"id": "A1B2C3D4...",
"title": "AI Chat",
"type": "page",
"url": "file:///app.asar/index.html",
"webSocketDebuggerUrl": "ws://127.0.0.1:9222/devtools/page/A1B2C3D4..."
}
]
|
注意type字段——Electron应用通常有多个target，包括主窗口页面、隐藏的preload页面、Service Worker等。你需要找type: "page"且url指向主应用的那个。
找不到调试端口？常见防护与破解
有些应用做了防护，常见手段有三种：
第一种：启动时检测参数。 应用代码中有类似if (app.commandLine.hasSwitch('remote-debugging-port')) process.exit()的逻辑。应对方法是解包asar文件（Electron的资源打包格式），找到检测逻辑并patch掉：
1
2
3
4
5
|
# 解包asar
npx asar extract app.asar app_extracted/
# 在app_extracted中搜索remote-debugging-port关键字
# 修改检测逻辑后重新打包
npx asar pack app_extracted/ app.asar
|
第二种：端口扫描反制。 应用启动后主动占用或检测9222端口，发现被占用就退出。换一个不常见的端口号（比如19876或37421）通常就能绕开。如果它检测的是一个范围，那就找一个更偏僻的端口。
第三种：多进程隔离。 Electron的BrowserWindow可能跑在独立的渲染进程中，调试端口暴露的是主进程的DevTools，你需要通过Target.setDiscoverTargets来发现子target。这种情况在采用了多窗口架构的复杂Electron应用中比较常见。
踩坑提醒：部分应用使用了定制版Electron，其Chromium内核版本可能较老，CDP协议的部分方法签名与最新版Chrome不同。建议先查清楚目标应用的Electron版本（通常在About页面或通过CDP的Browser.getVersion方法获取），再对照对应版本的CDP文档。我有一次在某个使用Electron 18的应用上调Fetch.enable（CDP的Fetch域），死活报错，后来才发现那个版本的Chromium根本不支持Fetch域，只能退回到Network.setRequestInterception。
建立CDP会话：从握手到控制
拿到webSocketDebuggerUrl之后，我们需要通过WebSocket与目标页面建立双向通信。Python生态中比较好用的库是pychrome和websockets裸连方案，JavaScript侧则有chrome-remote-interface和Puppeteer的connect模式。
Python裸连方案
我喜欢裸连，因为它让你对协议有完全的掌控，出了问题也知道卡在哪一步：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
|
import asyncio
import websockets
import json
async def connect_cdp(ws_url):
ws = await websockets.connect(ws_url, max_size=10*1024*1024)
msg_id = 0
pending = {}
async def send(method, params=None, timeout=30):
nonlocal msg_id
msg_id += 1
current_id = msg_id
payload = {"id": current_id, "method": method}
if params:
payload["params"] = params
future = asyncio.get_event_loop().create_future()
pending[current_id] = future
await ws.send(json.dumps(payload))
return await asyncio.wait_for(future, timeout=timeout)
async def receiver():
"""后台任务：持续接收消息并分发"""
async for raw in ws:
msg = json.loads(raw)
if "id" in msg and msg["id"] in pending:
pending[msg["id"]].set_result(msg)
del pending[msg["id"]]
# 事件消息可以在这里通过回调处理
asyncio.ensure_future(receiver())
return send, ws
|
这段代码比前面那个简洁版多了异步future分发机制——每个请求发出后注册一个future，接收循环根据id匹配响应并resolve对应的future。这样就支持并发发送多个CDP命令而不会互相阻塞。
实际使用时务必加上重连逻辑。WebSocket连接在Electron应用中不算特别稳定，尤其是当渲染进程因为内存压力被系统杀掉再重启时，你的连接会断掉。一个健壮的批量系统必须能处理断连-重连-恢复状态这个循环。
Puppeteer的connect模式
如果你更习惯JavaScript生态，或者想快速验证某个操作流程，Puppeteer提供了开箱即用的远程连接：
1
2
3
4
5
6
7
8
9
10
11
12
13
|
const puppeteer = require('puppeteer-core');
const browser = await puppeteer.connect({
browserURL: 'http://127.0.0.1:9222',
defaultViewport: null
});
const pages = await browser.pages();
const targetPage = pages.find(p => p.url().includes('chat'));
// 现在可以像操作普通页面一样操作AI客户端
await targetPage.type('[class*="input"] textarea', '画一只猫');
await targetPage.click('[data-testid="send-button"]');
|
puppeteer-core不包含Chromium二进制文件，体积小很多，专门用于连接已有浏览器实例的场景。它的API封装了CDP的底层细节，写起来快，但当你要做一些CDP原生不支持的操作时（比如直接操作Network域的底层拦截），就需要用page._client()拿到原始的CDP Session来发送自定义命令。
会话注入：让自动化脚本"登录"
连接上CDP只是第一步。大多数AI客户端需要登录态才能使用图片生成功能。手动登录后再连接CDP当然可以，但批量任务不可能每次都手动操作。我们需要的是会话注入——把有效的登录凭证通过CDP写入浏览器环境。
Cookie注入
登录态最常见的载体是Cookie。通过Network.setCookies可以一次性写入：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
await send("Network.enable")
await send("Network.setCookies", {
"cookies": [
{
"name": "session_id",
"value": "abc123def456...",
"domain": ".example.com",
"path": "/",
"secure": True,
"httpOnly": True,
"sameSite": "Lax"
},
{
"name": "_token",
"value": "xyz789...",
"domain": ".example.com",
"path": "/",
"secure": True
}
]
})
|
Cookie从哪来？两种途径：一是手动登录后从DevTools的Application面板逐个导出；二是写一个辅助脚本，在首次手动登录成功后自动抓取所有Cookie并保存到本地文件，后续批量任务直接加载这个文件。
LocalStorage与Token注入
有些应用不用Cookie存登录态，而是把JWT或access token放在localStorage或sessionStorage里。这时候需要通过Runtime.evaluate执行一段JS来完成注入：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
await send("Runtime.evaluate", {
"expression": """
localStorage.setItem('access_token', 'eyJhbGciOiJIUzI1NiIs...');
localStorage.setItem('refresh_token', 'dGhpcyBpcyBhIHJlZnJl...');
localStorage.setItem('user_info', JSON.stringify({
id: '12345',
name: 'testuser',
avatar: 'https://cdn.example.com/avatar.jpg'
}));
// 有些应用还会在 sessionStorage 里存一份
sessionStorage.setItem('auth_state', 'authenticated');
""",
"returnByValue": True
})
|
注入之后需要刷新页面让应用重新读取凭证。通过Page.reload即可：
1
2
3
4
|
await send("Page.enable")
await send("Page.reload", {"ignoreCache": True})
# 等待页面加载完成
await send("Page.loadEventFired") # 这是个事件等待，需要先注册监听
|
踩坑提醒：部分应用做了token绑定校验——比如把token和设备指纹（Canvas指纹、WebGL渲染器信息、时区等）绑定在一起，注入的token如果来自另一台设备，可能直接被服务端拒绝并强制退出登录。这种情况下你需要在同一台机器上完成登录和token提取，不能跨设备搬运。还有一个更隐蔽的坑：某些应用在localStorage里存的不是token本身，而是一个加密后的blob，解密密钥在IndexedDB或者内存里。这种情况下单纯注入localStorage是不够的，你还需要把IndexedDB的内容一起迁移。
登录状态验证
注入凭证后，不要急着开始批量任务。先跑一个轻量的验证步骤，确认登录确实生效了：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
verify_result = await send("Runtime.evaluate", {
"expression": """
(function() {
// 检查页面上是否有已登录用户的标识
const avatar = document.querySelector('.user-avatar');
const loginBtn = document.querySelector('.login-button');
return {
hasAvatar: !!avatar,
hasLoginBtn: !!loginBtn,
isLoggedIn: !!avatar && !loginBtn
};
})()
""",
"returnByValue": True
})
if not verify_result["result"]["value"]["isLoggedIn"]:
raise RuntimeError("Session injection failed - not logged in")
|
定位图片生成功能的DOM结构
会话建立后，下一步是找到"输入提示词→点击生成→等待结果"这条路径在DOM中的对应元素。
打开DevTools（直接访问http://127.0.0.1:9222可以在浏览器中打开目标页面的完整DevTools面板），用Elements面板检查输入框、发送按钮和图片容器。
定位策略的选择
常见的定位策略及其稳定性对比：
| 选择器类型 |
示例 |
稳定性 |
说明 |
| data-testid |
[data-testid="send"] |
高 |
专门为测试保留，不易变 |
| aria-label |
[aria-label="发送消息"] |
高 |
无障碍属性，产品不太会改 |
| role属性 |
[role="textbox"] |
中高 |
语义化，但可能被复用 |
| class片段 |
[class*="chat-input"] |
中 |
CSS modules可能加hash |
| 文本内容 |
按钮文本=“生成” |
中 |
多语言场景会失效 |
| XPath路径 |
/html/body/div[3]/... |
极低 |
DOM结构一变就废 |
实际操作中，最可靠的方式是组合多种策略：优先用data-testid或aria-label，找不到再退到class片段匹配，最后才用文本内容。
AI客户端输入框的特殊性
AI聊天类应用的输入框有一个共同特点：它们几乎都不是原生的<textarea>或<input>元素。为了支持富文本、@提及、文件拖入等功能，开发者通常会用contenteditable的div来模拟输入框。这意味着你用传统的element.value = "xxx"方式赋值是无效的。
正确的做法是模拟真实的输入事件序列：
1
2
3
4
5
6
7
8
9
10
|
async def type_into_editor(send, selector, text):
# 先聚焦到编辑器
await send("Runtime.evaluate", {
"expression": f"document.querySelector('{selector}').focus()",
"returnByValue": True
})
# 使用Input.insertText来模拟真实输入
# 这个CDP方法会在聚焦的元素上触发完整的输入事件链
await send("Input.insertText", {"text": text})
|
Input.insertText是CDP提供的一个高级方法，它会触发beforeinput、input、compositionstart、compositionend等完整的输入事件序列，基本上等同于用户在键盘上打字。对于React/Vue这类框架管理的受控组件，这是最可靠的文本输入方式。
等待图片生成：异步轮询的艺术
点击发送按钮之后，AI模型需要时间来生成图片。这个等待时间从几秒到几十秒不等，取决于模型负载、prompt复杂度和排队情况。
轮询检测策略
最朴素的方式是每隔N秒检查一次DOM中是否出现了图片元素。但间隔太短浪费资源，间隔太长又浪费时间。我推荐指数退避加抖动：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
|
async def wait_for_image(send, timeout=120):
"""等待图片生成完毕，返回图片元素的src"""
start = time.time()
interval = 1.0 # 初始间隔1秒
while time.time() - start < timeout:
result = await send("Runtime.evaluate", {
"expression": """
(function() {
// 检查是否有生成完毕的图片
const imgs = document.querySelectorAll('.message-assistant img');
const lastImg = imgs[imgs.length - 1];
if (lastImg && lastImg.naturalWidth > 0) {
return {ready: true, src: lastImg.src};
}
// 检查是否还在生成中（loading状态）
const loading = document.querySelector('.generating-indicator');
if (loading) {
return {ready: false, status: 'generating'};
}
// 检查是否出错
const error = document.querySelector('.generation-error');
if (error) {
return {ready: false, status: 'error',
message: error.textContent};
}
return {ready: false, status: 'unknown'};
})()
""",
"returnByValue": True
})
value = result["result"]["value"]
if value.get("ready"):
return value["src"]
if value.get("status") == "error":
raise RuntimeError(f"Generation failed: {value['message']}")
# 指数退避：1s → 2s → 4s → 8s → 10s（上限）
await asyncio.sleep(min(interval, 10))
interval *= 1.5
raise TimeoutError(f"Image not ready after {timeout}s")
|
这里有一个细节：检查lastImg.naturalWidth > 0。仅凭<img>元素存在不能判断图片已经加载完毕——有些应用会先渲染img标签再异步加载src。naturalWidth为0说明图片还没真正加载，这时候提取出来的是一个空白图。
处理blob URL
AI客户端生成的图片经常以blob: URL的形式展示（形如blob:http://localhost:9222/xxxx-xxxx）。这种URL只在当前页面上下文中有效，你不能直接拿它去下载。需要在页面内通过JavaScript把blob转成base64：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
|
extract_script = """
(async function() {
const imgs = document.querySelectorAll('.message-assistant img');
const lastImg = imgs[imgs.length - 1];
if (!lastImg) return null;
// blob URL 需要通过 fetch 获取实际数据
const response = await fetch(lastImg.src);
const blob = await response.blob();
return new Promise(resolve => {
const reader = new FileReader();
reader.onload = () => resolve({
data: reader.result, // data:image/png;base64,...
width: lastImg.naturalWidth,
height: lastImg.naturalHeight
});
reader.readAsDataURL(blob);
});
})()
"""
result = await send("Runtime.evaluate", {
"expression": extract_script,
"returnByValue": True,
"awaitPromise": True # 关键：等待Promise resolve
})
image_data = result["result"]["value"]
# 保存为文件
import base64
raw = base64.b64decode(image_data["data"].split(",")[1])
with open(f"output_{task_id}.png", "wb") as f:
f.write(raw)
|
注意awaitPromise: True这个参数。当你的JS表达式返回一个Promise时，CDP默认会直接返回Promise对象本身（你拿到的是{type: "object", className: "Promise"}）。加上这个参数后，CDP会等待Promise resolve再返回实际的值。
批量任务编排：从单张到流水线
单个prompt生成一张图片，整个流程大概是：定位输入框→填入prompt→点击发送→轮询等待图片出现→提取图片URL→下载保存。这个流程跑一次大概需要30秒到2分钟。
批量场景下，你需要考虑以下工程问题。
任务队列与状态机
最简单的方式是用一个数组加索引遍历。但在生产环境中，我推荐引入一个显式的状态机来管理每个任务的生命周期：
1
2
3
|
PENDING → INJECTING → TYPING → SENDING → WAITING → EXTRACTING → SAVING → DONE
↓ ↓ ↓ ↓ ↓
FAILED ←←←←←←←←←←←←←←←←←←←←←←←←← RETRY(最多3次)
|
每个状态对应一个明确的操作步骤和超时阈值。状态机的好处是：当你需要从中断处恢复时（比如应用崩溃了），你知道每个任务停在哪一步，可以从那一步重新开始而不是从头来。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
class TaskState:
PENDING = "pending"
TYPING = "typing"
SENDING = "sending"
WAITING = "waiting"
EXTRACTING = "extracting"
DONE = "done"
FAILED = "failed"
class Task:
def __init__(self, prompt, task_id):
self.prompt = prompt
self.task_id = task_id
self.state = TaskState.PENDING
self.retries = 0
self.max_retries = 3
self.result = None
self.error = None
self.started_at = None
self.finished_at = None
|
并发控制策略
CDP是单WebSocket连接、串行通信的协议。你不能同时在一个页面上操作两个不同的输入框。但你有几种方式来提高吞吐量：
方案一：多窗口并发。 通过Target.createTarget创建多个BrowserWindow，每个窗口跑一个独立的生成任务。注意Electron应用可能对同时存在的会话数有限制，开太多窗口可能导致内存暴涨或被应用自身的安全机制拦截。
1
2
3
4
5
6
7
8
|
# 创建新窗口target
new_target = await send("Target.createTarget", {
"url": "file:///app.asar/index.html",
"newWindow": True
})
# 连接到新窗口的CDP session
new_ws_url = new_target["result"]["targetId"]
# ... 建立新的WebSocket连接
|
方案二：串行加流水线。 一个窗口串行操作，但在等待图片生成的空窗期（通常10到30秒），提前准备好下一个prompt、执行文件写入等不涉及CDP的操作。这不会减少单任务的延迟，但能提高整体吞吐。
方案三：多实例并行。 启动多个应用实例，每个实例用不同的调试端口。这是最暴力但最可靠的方案：
1
2
3
4
5
6
7
8
9
10
11
12
|
import subprocess
ports = [9222, 9223, 9224, 9225]
processes = []
for port in ports:
proc = subprocess.Popen([
app_path,
f"--remote-debugging-port={port}",
f"--user-data-dir=/tmp/ai_instance_{port}",
"--no-sandbox"
])
processes.append(proc)
|
注意--user-data-dir参数——每个实例必须有独立的用户数据目录，否则多个进程争抢同一个配置目录会导致各种诡异问题（Cookie互相覆盖、IndexedDB锁冲突等）。
错误恢复与断点续跑
批量任务最让人崩溃的场景是：跑了200个prompt，第187个的时候应用崩了。如果没有断点续跑机制，你得从头开始。
我的做法是在每个任务状态变更时写入一个JSON Lines日志文件：
1
2
3
4
5
6
7
8
9
10
11
12
13
|
import json
def log_task_state(task):
record = {
"task_id": task.task_id,
"prompt": task.prompt,
"state": task.state,
"retries": task.retries,
"error": task.error,
"timestamp": time.time()
}
with open("task_log.jsonl", "a") as f:
f.write(json.dumps(record, ensure_ascii=False) + "\n")
|
重新启动时，读取这个日志文件，找到每个任务的最后状态，跳过已经DONE的任务，从FAILED或PENDING的任务继续。
反自动化对策与稳定性工程
跑过几十次之后，你大概率会遇到以下问题。
会话过期
Cookie或Token有有效期，批量任务跑一半过期了。解决方案是在任务循环中定期检查登录状态，过期时暂停任务、重新注入凭证、然后从断点继续：
1
2
3
4
5
6
7
8
9
10
11
12
|
async def check_session_health(send):
"""每次任务前检查会话是否还有效"""
result = await send("Runtime.evaluate", {
"expression": """
fetch('/api/user/profile', {credentials: 'include'})
.then(r => ({ok: r.ok, status: r.status}))
.catch(e => ({ok: false, error: e.message}))
""",
"returnByValue": True,
"awaitPromise": True
})
return result["result"]["value"].get("ok", False)
|
行为检测绕过
部分应用在JS层做了自动化检测。常见的检测点和对应的绕过方式：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
// 1. 覆盖 webdriver 属性（Puppeteer/Playwright会设为true）
Object.defineProperty(navigator, 'webdriver', {get: () => undefined});
// 2. 伪造 plugins 数组（自动化环境通常为空）
Object.defineProperty(navigator, 'plugins', {
get: () => Object.create(PluginArray.prototype, {
length: {value: 5}
})
});
// 3. 伪造 languages
Object.defineProperty(navigator, 'languages', {
get: () => ['zh-CN', 'zh', 'en-US', 'en']
});
// 4. 覆盖 chrome.runtime（正常Electron应用有这个对象）
window.chrome = window.chrome || {};
window.chrome.runtime = {
connect: function(){},
sendMessage: function(){}
};
|
这些注入要在页面加载后、应用逻辑执行前完成。可以通过Page.addScriptToEvaluateOnNewDocument来确保每次页面加载时自动注入：
1
2
3
|
await send("Page.addScriptToEvaluateOnNewDocument", {
"source": stealth_js_code # 上面那些绕过代码的集合
})
|
页面崩溃与重连
Electron的渲染进程偶尔会因为内存不足而崩溃，尤其是在连续生成大量图片的情况下（每张图片的blob数据都会占用内存）。你需要：
- 监听WebSocket的
close事件，检测到断连后启动重连
- 通过
Target.setDiscoverTargets监听target的创建和销毁
- 定期调用
Page.getResourceTree检查页面是否还活着
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
async def robust_cdp_session(ws_url, on_disconnect=None):
"""带自动重连的CDP会话管理"""
max_retries = 5
for attempt in range(max_retries):
try:
send, ws = await connect_cdp(ws_url)
# 注入stealth脚本和登录凭证
await setup_session(send)
return send, ws
except (websockets.exceptions.ConnectionClosed,
ConnectionRefusedError):
if on_disconnect:
on_disconnect(attempt)
await asyncio.sleep(2 ** attempt) # 指数退避重连
raise RuntimeError("Failed to establish CDP session after retries")
|
DOM结构变更的防御
应用更新后，你依赖的CSS选择器可能全部失效。建议在定位元素时优先使用语义化选择器，并在任务启动时跑一次"探针"——验证所有关键元素是否可定位：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
|
PROBE_SELECTORS = {
"input_box": "[role='textbox'], [class*='chat-input'] textarea, "
"[class*='editor'] [contenteditable='true']",
"send_button": "[data-testid='send'], [aria-label='发送'], "
"button[class*='send']",
"image_container": "[class*='generated-image'], "
"[class*='image-result'] img",
}
async def probe_elements(send):
"""启动前探测所有关键元素是否存在"""
for name, selector in PROBE_SELECTORS.items():
result = await send("Runtime.evaluate", {
"expression": f"!!document.querySelector(`{selector}`)",
"returnByValue": True
})
if not result["result"]["value"]:
raise RuntimeError(
f"Element probe failed: '{name}' not found. "
f"The app UI may have been updated."
)
|
完整编排框架的设计
把上面的模块串起来，最终的批量系统大致是这样的结构：
1
2
3
4
5
6
7
8
9
|
config.yaml ← prompt列表路径、并发数、超时参数、重试策略
↓
session_manager.py ← CDP连接池管理、会话注入、健康检查、自动重连
↓
task_runner.py ← 任务状态机、并发调度、重试逻辑、结果收集
↓
image_extractor.py ← blob→base64转换、文件保存、MD5去重、缩略图生成
↓
report.json ← 每条prompt的生成结果、耗时、失败原因、截图路径
|
config.yaml的一个例子：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
app:
path: /Applications/SomeAI.app/Contents/MacOS/SomeAI
debug_port: 9222
user_data_dir: /tmp/ai_automation
session:
cookies_file: ./cookies.json
stealth_enabled: true
health_check_interval: 10
tasks:
prompts_file: ./prompts.txt
concurrency: 2
timeout_per_task: 120
retry_limit: 3
interval_range: [3, 8]
checkpoint_file: ./task_log.jsonl
output:
dir: ./generated_images/
format: png
dedup_by_content: true
generate_thumbnails: true
|
任务完成后输出的report.json：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
|
{
"summary": {
"total": 50,
"success": 47,
"failed": 3,
"avg_duration_sec": 22.5,
"total_duration_sec": 1125
},
"tasks": [
{
"prompt": "一只橘猫坐在窗台上看雨，水彩风格",
"status": "ok",
"image_path": "./generated_images/001.png",
"duration_sec": 18.3,
"dimensions": [1024, 1024]
},
{
"prompt": "赛博朋克城市夜景，霓虹灯倒映在积水中",
"status": "failed",
"error": "timeout after 120s",
"retries": 3,
"duration_sec": 126.1
}
]
}
|
合规边界与使用建议
有句话说，工具本身没有善恶，但使用者需要清楚边界。CDP自动化驱动AI客户端在技术上完全可行，但你需要注意几点：
- 服务条款：大多数AI平台的用户协议明确禁止自动化访问和爬取。如果你用于商业用途或大规模滥用，可能面临账号封禁甚至法律风险。
- 资源消耗：批量生成图片意味着消耗平台的GPU算力资源。如果是免费额度的滥用，本质上是在薅羊毛，平台有权封号。
- 个人学习场景：如果你只是想用技术手段提升自己的工作效率——比如批量生成一组配图的草稿、测试不同prompt的效果差异——在自己的电脑上操作自己的账号、在合理使用额度内，这属于合理的个人使用范畴。
技术文章的价值在于让你理解原理和可能性，具体用在哪里、怎么用，需要你自己做判断。
回顾与延伸
CDP协议打开了一扇门——它让你看到，那些看起来"原生"的桌面应用，底层不过是一个跑在Chromium上的Web程序。理解了这一点，很多看似复杂的自动化需求，其实都有章可循。
关键不在协议本身，而在工程上的细节处理：超时控制的粒度、状态恢复的可靠性、选择器的鲁棒性、异常情况的优雅降级。这些东西没有捷径，只有反复调试和踩坑才能积累出来。
如果你对这个方向感兴趣，可以继续探索几个延伸话题：用CDP的Network域拦截和修改AI客户端的API请求（比如替换模型参数来解锁隐藏功能）；用Performance域分析图片生成的端到端延迟瓶颈；或者结合WebSocket代理在CDP之上构建一个统一的自动化控制面。协议是死的，玩法是活的。
