InboundVerify/docs/安能门户CDP连接指南.md

安能全网门户 CDP 连接与驱动指南

目标：连接到一个已经以调试模式启动的「安能全网门户」Electron 应用，像 Playwright 一样查看和操作界面元素。

实测环境：Windows 11、Electron 19 / Chrome 102、Python 3.13、playwright 1.60、websocket-client 1.9。

---

1. 启动应用

用 Electron/Chromium 的远程调试参数启动可执行文件：

"安能全网门户.exe" --remote-debugging-port=9222

启动后，应用会在 http://localhost:9222 暴露一套 Chrome DevTools Protocol (CDP) 的 HTTP + WebSocket 接口。

快速自检（任意浏览器或 curl）：

curl http://localhost:9222/json/version   # 确认端口在线、看 Electron/Chrome 版本
curl http://localhost:9222/json           # 列出所有「页面目标」

/json/version 里 User-Agent 会带 Electron/19.0.4 Chrome/102.0.5005.63，据此判断 Chromium 内核版本（影响后续兼容性判断）。

---

2. 技术路线选型（重要）

❌ 路线 A：Playwright connect_over_cdp —— 不可用

直觉上最省事，但实测在这个 Electron 构建上握手后立刻断开：

playwright._impl._errors.Error: BrowserType.connect_over_cdp:
Protocol error (Browser.setDownloadBehavior): Browser context management is not supported.

原因：Playwright 在 CDP 连接建立后，会调用浏览器域的 Browser.setDownloadBehavior 来初始化默认下载上下文；而这个 Electron 19 / Chrome 102 构建的 CDP 只部分实现了浏览器域， 对“浏览器上下文管理”一类命令直接返回 not supported，于是连接被关闭。

为什么不能靠降级 Playwright 绕过：该行为由来已久，需要降到 2022 年的 Playwright 才可能规避； 但本机 venv 是 Python 3.13，而 playwright < 1.48 没有提供 3.13 的 wheel，装不上。

✅ 路线 B：页面级 CDP（本项目采用）

关键观察：Electron 的浏览器域受限，但页面域（Page / DOM / Runtime）完全正常。

因此绕开 Playwright 的高层连接，直接：

1. GET http://localhost:9222/json 拿到所有页面目标的 webSocketDebuggerUrl；
2. 用 WebSocket 直连某个页面目标（不是 browser 目标）；
3. 在该 WebSocket 上收发 CDP 消息（Runtime.evaluate / DOM.* / Input.* …）。

在上面再封装一层 Playwright 风格的薄封装（Page.eval / Page.query_all / Page.click …）， 用起来接近 Playwright，又不受 Electron 浏览器域限制。

备选：也可用 pychrome 这类 CDP 客户端，它同样是页面级、不做上下文管理， 同样不会踩这个坑。本项目为减少依赖，直接用已安装的 websocket-client 手写。

---

3. 连接原理

3.1 发现目标

GET /json 返回一个数组，每个元素形如：

{
  "id": "4E4584692AF9A17D12E70E024F6DC938",
  "type": "page",
  "title": "安能全网门户",
  "url": "file:///.../app.asar/website/index.html#/ai-button",
  "webSocketDebuggerUrl": "ws://localhost:9222/devtools/page/4E458469..."
}

• type == "page" 的才是普通页面（另有 background_page、service_worker、webview 等，按需过滤）。
• id / webSocketDebuggerUrl 每次启动都会变，所以一定要动态发现，绝不能硬编码（早期 site_anneng.py 硬编码 WS URL，重启就失效）。
• 用 title 或 url 来挑选你真正要操作的那一页。

3.2 CDP 消息往返

CDP 是 JSON-RPC 风格：客户端发 {id, method, params}，服务端回 {id, result} 或 {id, error}； 服务端也会主动推送事件 {method, params}（没有 id）。

所以收消息时要过滤掉事件，只取 id 与本次请求匹配的那条回复：

while True:
    msg = json.loads(ws.recv())
    if msg.get("id") == my_id:
        return msg.get("result", {})
    # 否则是事件，忽略

3.3 取值：Runtime.evaluate + returnByValue

直接执行 JS，让浏览器把结果序列化成 JSON 带回来：

result = call("Runtime.evaluate",
              expression="document.title",
              returnByValue=True, awaitPromise=True)
title = result["result"]["value"]

• returnByValue=True：把 JS 返回值按值序列化为 JSON（适合结构化数据）。
• awaitPromise=True：表达式中若返回 Promise 会被自动 await（写普通表达式也无副作用）。

绝大多数“查询/读取”需求都能用一段 JS + Runtime.evaluate 解决，比走 DOM.* 更直观。

---

4. 可运行的最小封装

下面是项目里 _probe_anneng.py 的核心结构（精简版），可直接复用：

"""页面级 CDP：连接 安能全网门户（:9222），Playwright 风格的薄封装。"""
import json
import sys
import urllib.request

import websocket

sys.stdout.reconfigure(encoding="utf-8")  # Windows 控制台是 GBK，否则打印中文/emoji 会崩

CDP_URL = "http://localhost:9222"


def list_pages():
    with urllib.request.urlopen(f"{CDP_URL}/json") as resp:
        data = json.load(resp)
    return [p for p in data if p.get("type") == "page"]


class CDP:
    """绑定到单个页面目标的同步 CDP 客户端。"""

    def __init__(self, ws_url):
        self.ws = websocket.create_connection(ws_url)
        self._id = 0

    def call(self, method, **params):
        self._id += 1
        self.ws.send(json.dumps({"id": self._id, "method": method, "params": params}))
        while True:
            msg = json.loads(self.ws.recv())
            if msg.get("id") == self._id:
                if "error" in msg:
                    raise RuntimeError(f"{method} failed: {msg['error']}")
                return msg.get("result", {})

    def eval(self, expression):
        res = self.call("Runtime.evaluate", expression=expression,
                         returnByValue=True, awaitPromise=True)
        return res.get("result", {}).get("value")

    def close(self):
        self.ws.close()


class Page:
    def __init__(self, cdp):
        self.cdp = cdp
        cdp.call("Page.enable")
        cdp.call("Runtime.enable")

    @property
    def title(self):
        return self.cdp.eval("document.title")

    def query_all(self, selector, limit=20):
        return self.cdp.eval(
            "(() => [...document.querySelectorAll(%r)].slice(0,%d).map(e => ({"
            "tag:e.tagName.toLowerCase(), text:(e.innerText||'').trim().slice(0,40),"
            "id:e.id||'', cls:(e.className||'').toString().slice(0,40),"
            "title:e.getAttribute('title')||'' })))()" % (selector, limit)
        ) or []

用法：

pages = list_pages()
# 挑你要的那一页（示例：取标题里含“鱼洞”的那页，即应用外壳）
target = next(p for p in pages if "鱼洞" in p["title"])
page = Page(CDP(target["webSocketDebuggerUrl"]))
print(page.title)
print(page.query_all("div.rc-menu-submenu-title"))

---

5. 关键发现：界面结构与定位

实测两个页面目标：

页面 title	url 片段	说明
安能全网门户	…/index.html#/ai-button	仅 AI 按钮落地页，几乎是空的，不是主界面
重庆鱼洞镇	…/index.html	真正的应用外壳，左侧整棵导航菜单都在这里

主界面（“重庆鱼洞镇”页）的左侧一级菜单（实测抓取）：

基础数据 / 客服 / 运单管理 / 运营管理 / 扫描操作 / 物料管理 /
网点金融 / 自营客户管理 / 综合查询 / 新闻与问卷 / 问题反馈

定位特征（React + styled-components + rc-menu 技术栈）：

• 一级菜单项：div.rc-menu-submenu-title[title='运营管理']
• 站点信息（顶部）：div[title='02330019400049']
• 菜单搜索框：input[placeholder*='菜单搜索']
• 折叠按钮：div[title='折叠菜单']

这是 React 应用，class 名是 styled-components 生成的散列（如 sc-dntSTA ginXyJ）， 不要用这些散列 class 做定位——换一次构建就变。优先用 [title=…]、[role=…]、 稳定的 rc-menu-* 类名或文本内容来定位。

---

6. 独立 webContents：tab 页（独立网页）的元素操作

应用里有些功能（如「导出下载」）点击导航菜单后，会在右侧打开一个 tab 页。 这类 tab 的内容是独立的 webContents（一个远程网页），而不是主页面里的普通 iframe。

判据（出现以下现象即说明是独立 webContents）：

• 用主窗口 DevTools（Ctrl+Shift+I）的元素拾取，能抓导航栏，却抓不到 tab 内的元素；
• 在 tab 内点右键只有「检查」一项，点开后会另开一个 DevTools 窗口，在那里才能拾取 tab 元素。

关键结论：这种 tab 会作为 /json 里一个独立的 page 目标出现（url 是远程 https 地址）。 所以不需要任何特殊 API —— 把它当成“又一个页面”：按 URL 从 /json 里挑出来、直连它的 webSocketDebuggerUrl，用同一套 Runtime.evaluate / DOM 操作即可，和主页面毫无区别。

打开并定位 tab 目标的通用做法（已打开就按 URL 复用，没打开就点菜单触发后轮询新目标）：

import time

def find_or_open_tab(url_hint, menu_text=None, timeout=20):
    """按 url_hint 找到 tab 目标；没打开就点 menu_text 菜单触发，再轮询新目标。"""
    # 1) 已打开则直接复用（不必重复点击）
    for p in list_pages():
        if url_hint in p.get("url", ""):
            return CDP(p["webSocketDebuggerUrl"])
    # 2) 未打开 → 点导航菜单触发，再轮询新出现的目标
    assert menu_text, "tab 未打开且未提供 menu_text"
    main = next(CDP(p["webSocketDebuggerUrl"]) for p in list_pages()
                if p.get("title") and "鱼洞" in p["title"])   # 主应用外壳页
    before = {p["id"] for p in list_pages()}
    main.eval(
        "[...document.querySelectorAll('li.rc-menu-item')]"
        f".find(e => e.textContent.trim() === {json.dumps(menu_text)})?.click()"
    )
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        for p in list_pages():
            if url_hint in p.get("url", "") and p["id"] not in before:
                return CDP(p["webSocketDebuggerUrl"])
        time.sleep(0.5)
    raise TimeoutError("tab 目标未出现")

实测：「导出下载」tab 的 URL 含 exportAllRecords，打开后即是一个远程 https 页面目标， 连上后读表格、点按钮等都与主页面完全一样。

---

7. 实战：免对话框下载导出文件

目标：触发导出下载并自动存到默认下载目录，不弹 Windows 保存对话框。

核心经验：下载不要靠点页面按钮，直接 GET 下载接口即可。 （点按钮会走 Electron 主进程的下载流程并弹原生保存对话框，CDP 无法从外部抑制； 而直接发 HTTP 请求取回文件字节，根本不经过那条流程，自然无对话框。）

下载接口与鉴权（实测自「导出下载」tab）：

项	值
接口	GET https://uep.ane56.com/uep/foreign/api/fileDownloadRecord/download
参数	fileId=<文件路径>、appId=LB、aneFile=false、index=<同 fileId>
鉴权头	x-auth = sessionStorage["x-auth"]（base64 JWT）
cookie	TGC（httpOnly，用 Network.getCookies 读取；document.cookie 取不到）
响应	application/vnd.openxmlformats-officedocument.spreadsheetml.sheet（xlsx）

文件名：用表格「文件名」列的可读名（如 交接单明细-20260621_153527.xlsx）； 服务端 Content-Disposition 里给的是 UUID 名，不要用。

完整步骤（CDP 取鉴权 + Python urllib 下载）：

import urllib.parse
import urllib.request

# 1) 连到「导出下载」tab 目标，取出鉴权信息
cdp = find_or_open_tab("exportAllRecords", menu_text="导出下载")
x_auth  = cdp.eval('sessionStorage.getItem("x-auth")')
ua      = cdp.eval("navigator.userAgent")
referer = cdp.eval("location.href")
tgc = next(c["value"] for c in
           cdp.call("Network.getCookies", urls=["https://uep.ane56.com"])["cookies"]
           if c["name"] == "TGC")

# 2) 直接 GET 下载接口（不点按钮 → 不弹对话框）
params = {"fileId": file_id, "appId": "LB", "aneFile": "false", "index": file_id}
url = ("https://uep.ane56.com/uep/foreign/api/fileDownloadRecord/download?"
       + urllib.parse.urlencode(params))
req = urllib.request.Request(url, headers={
    "x-auth": x_auth, "Cookie": f"TGC={tgc}",
    "Referer": referer, "User-Agent": ua, "Accept": "*/*",
})
data = urllib.request.urlopen(req, timeout=30).read()
assert data[:2] == b"PK", "非 xlsx"                 # ZIP 魔数校验
open(r"C:\Users\pengq\Downloads\交接单明细-xxx.xlsx", "wb").write(data)

实测：GET 返回 200 / 21508 字节 / PK 头合法，全程不点按钮、不弹对话框、不留 <uuid>.tmp。 完整可运行脚本见项目 download_clean.py。

关于 fileId：下载按钮的 <a> 没有 href（是 JS 点击），fileId 不在可见 DOM 里。 download_clean.py 当前用抓包所得的 fileId 验证通过；要批量下载多行，需从表格行的 React record 里取每行的 fileId（路径模式 /task/YYYYMMDD/<uuid>.xlsx）。

---

8. 常见坑

1. 页面 ID 会变：每次重启应用，/json 里的 id / webSocketDebuggerUrl 都不同 → 必须运行时动态发现。
2. Windows 控制台编码：默认 GBK，打印中文/emoji 抛 UnicodeEncodeError → 脚本里加 sys.stdout.reconfigure(encoding="utf-8")，或设环境变量 PYTHONUTF8=1 运行。
3. 必须过滤 CDP 事件：recv 循环里要跳过没有匹配 id 的事件消息，否则会把事件当成回复解析出错。
4. 直连 page 目标，不是 browser 目标：/json/version 里的 webSocketDebuggerUrl 是 browser 级 WS， 连它同样会触发上下文管理类命令的坑；要用 /json 里各 page 的 WS。
5. 不要关掉用户的进程：我们只是“附加”到外部进程，ws.close() 只断开自己的连接，不会杀掉 Electron。

---

9. 扩展：点击与输入

9.1 点击（JS 触发，最简单）

def click(self, selector):
    self.cdp.eval(
        "document.querySelector(%r)?.click() || false" % selector
    )

适合菜单展开、按钮提交这类纯逻辑点击。

9.2 点击（真实鼠标事件，需要坐标时）

用 Input.dispatchMouseEvent 模拟真实移动/按下/抬起，坐标来自元素包围盒：

def click_real(self, selector):
    box = self.cdp.eval(
        "(() => {const r=document.querySelector(%r).getBoundingClientRect();"
        "return {x:r.x+r.width/2, y:r.y+r.height/2};})()" % selector
    )
    for t in ("mouseMoved", "mousePressed", "mouseReleased"):
        self.cdp.call("Input.dispatchMouseEvent", type=t,
                      x=box["x"], y=box["y"], button="left", clickCount=1)

9.3 文本输入（React 应用特别注意）

React 受控组件不会响应直接赋的 el.value = '...'。要触发它认得的 input 事件：

def fill(self, selector, text):
    self.cdp.eval(
        "(() => {const el=document.querySelector(%r);"
        "const s=Object.getOwnPropertyDescriptor(el.constructor.prototype,'value').set;"
        "s.call(el, %r);"
        "el.dispatchEvent(new Event('input',{bubbles:true}));"
        "})()" % (selector, text)
    )

9.4 等待元素出现

CDP 没有现成的 wait_for_selector，轮询最稳：

import time
def wait_for(self, selector, timeout=10, interval=0.3):
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if self.cdp.eval(f"!!document.querySelector({selector!r})"):
            return True
        time.sleep(interval)
    raise TimeoutError(selector)

---

10. 运行环境约定

• 所有 Python 脚本在项目 .venv 内运行（python -m venv .venv，激活后安装依赖）。
• 依赖：websocket-client（已装）、black（格式化，已装）、可选 pychrome。
• 相关脚本：_probe_anneng.py（探查）、site_anneng.py（菜单导航）、download_clean.py（免对话框下载）。
• 运行：.venv/Scripts/python.exe download_clean.py

---

11. 速查清单

任务	做法
确认应用在线	curl http://localhost:9222/json/version
列出页面	curl http://localhost:9222/json（或 list_pages()）
选目标页	按 title / url 过滤，别用 id
连接	直连该页的 webSocketDebuggerUrl
读取元素	Runtime.evaluate + 一段 querySelector JS
点击	JS .click() 或 Input.dispatchMouseEvent
输入（React）	走原型 setter + dispatchEvent('input')
操作 tab 页（独立网页）	它是 /json 里独立 page 目标，按 URL 取出直连即可
免对话框下载	别点按钮；用 x-auth(sessionStorage)+TGC(Network.getCookies) 直接 GET 下载接口
格式化代码	black
终端乱码	sys.stdout.reconfigure(encoding="utf-8")