网络爬虫与 CDP 实战（五）：数据一直在变，怎么抓完整？WebSocket 与综合采集实战

网络爬虫与 CDP 实战（五）：数据一直在变，怎么抓完整？WebSocket 与综合采集实战

 实验 19：实时报价 · 实验 20：供应商门户

系列：《网络爬虫与 CDP 实战：拆解网页数据的 20 个实验》 本篇对应 Phase 5，覆盖实验 19—20：实时价格推送与供应商门户综合分析。

前四篇里，我们一直在做一件事：找到一个请求，弄清它需要哪些凭证和参数，再取回一份数据。现在把问题换一下：页面上的价格每隔一段时间都会变化，Network 里却没有对应的新 JSON 请求。

这时继续刷新、复制请求、增加分页，很可能找错了方向。页面可能已经建立了一条长连接，后续数据沿着同一条连接不断到达。

本篇先拆开这种实时数据链路，再进入一个不给取数路线提示的供应商门户。仍然使用前面那批 24 件商品，但目标会从“某个接口返回成功”提高到“从干净起点出发，可重复地取得完整且可信的数据”。

跟着文章动手：在线实验站

这一篇的实时推送和综合门户也有可直接操作的在线页面：我搭建的 网页数据实验场 首页 Phase 5 提供实验 19—20。先打开实时报价页观察消息，再进入供应商门户完成整条采集流程。实时连接最多持续一分钟，到期后可点击“连接”重新开始；门户则适合按本文的方法，从一次全新的访问开始逐步收集证据。

网站使用固定的 24 件虚拟商品，页面负责提供业务操作，操作步骤和分析方法在本文中。下面的请求代码默认使用这个公开站点；如有自己的本地实例，可以设置 CDP_BASE_URL 切换地址。

一、先问业务需要什么，再决定怎样接收数据

设想你正在制作一张商品价格看板。需求可能有三种：每天保存一次价格；每分钟判断价格有没有变化；保留每一次报价变动，方便以后回放。

它们都叫“采集价格”，需要的技术却不同。

SSE 通常适合服务端向浏览器单向发送事件；WebSocket 支持双向通信，客户端可以发送订阅等消息。浏览器分别有对应 API，不能因为它们都保持连接就套用同一套解析代码。1

协议允许实时推送，和业务承诺不漏事件，是两回事。 一个只展示当前价格的网页，未必提供完整的历史变更流。先确定交付的是“最新状态”还是“每次变化”，后面的完整性标准才有意义。

二、实验 19：找到那条不再重复出现的请求

动手入口：打开实时报价实验页。 在 DevTools 的 Network 中选择 WS，观察页面自动建立的连接；修改商品编号并点击订阅，对照文本消息、二进制消息和表格变化。

打开价格推送页面，进入 DevTools 的 Network，筛选 WS。点击连接后，查看消息面板，观察两个方向的内容：页面发送了什么，服务端返回了什么。

建议按这个顺序记录：

公开实验站使用同域连接 wss://web-data-lab.pages.dev/labs/websocket/price/，网页通过 HTTPS 提供。浏览器会根据当前站点生成 wss:// 地址。每次连接持续最多一分钟，之后明确关闭；可以手动重连继续实验。关闭或离开页面会主动断开，避免闲置推送。

不要只截取一条好看的 JSON。至少观察一段包含初始化、订阅和后续变化的过程。否则你可能把欢迎信息当成价格更新，把客户端发送的商品编号当成服务端返回的库存。

2.1 为什么出现 HTTP 101，之后却没有普通响应体？

在本例使用的 HTTP/1.1 握手中，客户端提出协议升级，服务端返回 101 Switching Protocols。握手成功后，后续内容按 WebSocket 帧传输。Sec-WebSocket-Accept 由客户端 key 与固定 GUID 拼接后计算 SHA-1，再进行 Base64 编码。[3]

下面只验证握手中的这一个计算，不实现整个客户端：

import base64
import hashlib
​
GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
​
def compute_accept(key: str) -> str:
   digest = hashlib.sha1((key + GUID).encode("ascii")).digest()
   return base64.b64encode(digest).decode("ascii")
​
assert compute_accept("dGhlIHNhbXBsZSBub25jZQ==") == (
   "s3pPLMBiTxaQ9kYGzzhZRbK+xOo="
)

这个计算用于确认握手协议，不承担业务登录认证。即使握手成功，后续订阅仍可能需要凭证。也不能把这里出现 SHA-1，理解成所有价格数据都经过加密。

2.2 一次 recv() 为什么不等于一条消息？

TCP 提供字节流。应用层消息的边界由上层协议定义：一次接收可能只有半个帧，也可能包含多个帧。WebSocket 还允许一条消息由多个帧组成，并存在独立的控制帧。[3]

为了理解“读够指定字节”这个基础动作，可以先写一个小函数：

def recv_exact(sock, size: int) -> bytes:
   parts = bytearray()
   while len(parts) < size:
       chunk = sock.recv(size – len(parts))
       if not chunk:
           raise EOFError("连接在读取完成前关闭")
       parts.extend(chunk)
   return bytes(parts)

它没有处理 WebSocket 握手、消息分片、控制帧、扩展和关闭流程，也没有替你设置套接字超时。把这个函数和一个简单帧解析器拼在一起，仍然不足以得到可以长期运行的采集服务。

理解帧时，先关注四类字段：结束标记、操作码、负载长度、掩码。文本和二进制负载需要不同解释；客户端向服务端发送的帧需要掩码，但掩码不是加密。Ping、Pong 和 Close 也不能当成商品数据。[3]

实验可以手工拆帧，持续采集应交给具备这些协议处理能力的客户端库。业务代码主要负责订阅、验证、恢复和落库。

2.3 二进制不一定意味着“加密”

本例周期性出现二进制消息，就是为了打破“所有帧都能直接 json.loads()”的习惯。正确第一步是记录类型和长度，再结合页面处理代码确定格式。

可按以下假设逐项验证：它是否是文本的字节表示？是否经过压缩？是否采用某种序列化格式？是否只是与价格无关的控制内容？

不要对任意字节使用 decode(errors="ignore") 然后继续解析。忽略解码错误可能悄悄删除内容，得到一段看似可读却已经损坏的数据。

三、用浏览器先建立消息证据，再移植采集逻辑

第三篇介绍过 CDP 的事件思路。WebSocket 也适合先观察事件：建立连接、发送消息、接收消息、关闭连接。Playwright 提供对应监听接口，收到的负载可能是字符串或字节。[4]

下面是一段完整的有限观察程序。它依赖 playwright 及已安装的 Chromium，默认访问公开实验站；使用本地实例时须先启动对应服务。程序只等待三条可以解析成 JSON 对象的文本消息，不宣称取得全部价格变化。

import os
​
import asyncio
import json
from playwright.async_api import async_playwright
​
BASE_URL = os.environ.get("CDP_BASE_URL", "https://web-data-lab.pages.dev").rstrip("/")
PAGE_URL = BASE_URL + "/labs/websocket/"
​
async def main():
   async with async_playwright() as p:
       browser = await p.chromium.launch()
       try:
           page = await browser.new_page()
           finished = asyncio.Event()
           messages = []
​
           def on_received(payload):
               if finished.is_set():
                   return
               if isinstance(payload, bytes):
                   print("二进制负载长度:", len(payload))
                   return
               try:
                   message = json.loads(payload)
               except json.JSONDecodeError:
                   print("非 JSON 文本消息")
                   return
               if not isinstance(message, dict):
                   return
               messages.append(message)
               print("消息字段:", sorted(message))
               if len(messages) >= 3:
                   finished.set()
​
           def on_socket(ws):
               if "/labs/websocket/price/" in ws.url:
                   ws.on("framereceived", on_received)
​
           page.on("websocket", on_socket)
           await page.goto(PAGE_URL, wait_until="domcontentloaded")
           await asyncio.wait_for(finished.wait(), timeout=20)
           print("观察到的 JSON 对象数:", len(messages))
       finally:
           await browser.close()
​
if __name__ == "__main__":
   asyncio.run(main())

监听必须在打开页面之前注册。否则页面快速建立连接并发送初始化消息时，程序可能错过关键证据。停止条件也应来自你的任务：本例用数量和超时，生产采集则可能按时段、持久化水位或主动取消结束。

得到消息后，再建立字段说明。例如某个 seq 究竟是当前连接内的计数、某个商品的版本号，还是整个服务的全局事件序号？本例序号属于连接中的推送过程，不能直接当成跨重连的永久事件 ID。

字段名是线索，生命周期才决定用途。 对一个新网站，要通过重连、切换订阅、重复观察来判断序号的范围。

四、从“接到消息”到“稳定采集”，还差哪些工作？

以下是面向价格监控系统的设计案例，用来说明工程取舍，并非声称某家真实客户已经采用了这些实现。

4.1 先保存最新状态，还是保存每次变化？

如果看板只需要最新价格，可以维护 product_id → 最新价格与版本。同一商品的多次更新在队列里积压时，是否允许合并，要由业务决定。

如果需要分析价格变化轨迹，则应该保存事件流。至少记录商品 ID、源端事件时间、接收时间、事件标识或版本、原始负载和采集批次。源端时间描述业务发生时刻，接收时间描述采集器看到它的时刻，两者混用会影响延迟分析。

接收时间也不能自动证明源端事件顺序。多条连接、多个订阅和恢复补发都可能改变到达顺序，需要依据源端可用的版本或事件标识建立合并规则。

4.2 重连不是把上一条代码再执行一次

一次断线恢复至少包括：重新建立连接、恢复身份、重新订阅、确认初始化完成、处理断线期间的缺口。

如果服务端支持从序号继续消费，就保存已成功持久化的序号；如果只能重新发送最新快照，就用快照重建状态，并明确这段期间无法恢复全部变更历史。如果既没有补发也没有历史接口，客户端自己加一个重试循环无法补出不存在的数据。

重连通常采用逐步增加的等待时间并加入随机扰动，设置上限和停止条件。认证失败、服务限流、暂时断网的处理应当不同。否则大量采集器同时重连，反而会延长服务恢复时间。

4.3 收到两次，应该删除一次吗？

先检查是否真的是同一业务事件。如果存在稳定事件 ID，可以用唯一约束实现幂等写入；如果只有商品 ID，直接去重会删掉同一商品的正常后续变化。

也不要把 seq=1 当成全系统唯一标识。在连接重建后重新从 1 计数的服务中，这样做会把新连接的大部分事件误判成重复。需要结合序号作用域建立键，或者承认当前协议没有可靠的事件去重依据。

“消息已经交给处理函数”和“消息已经写进数据库”是两个状态。持久化失败后是否重试、是否重复写入，应由事务、唯一键和可恢复的消费位置共同处理。仅凭连接可靠，不能推出业务层恰好处理一次。

4.4 为什么内存会越跑越大？

如果消息到达速度高于处理速度，无界列表和队列会持续增长。浏览器传统 WebSocket API 不提供接收侧背压机制，MDN 明确提示这种情况下的内存与 CPU 风险。[1]

工程上应设置队列容量和可观测指标：每秒收到多少、成功写入多少、积压多少、最老消息等待多久。超出容量后选择合并、暂停上游、断开并重建快照，还是告警停止，都要与业务目标一致。

bufferedAmount 描述待发送字节，并不能直接替你衡量应用收到消息后的处理队列。[5] 用错指标，可能出现“网络发送队列很小，但数据库写入积压已经数十万条”的误判。

五、实验 20：进入一个不给路线提示的供应商门户

动手入口：打开供应商门户实验页。 从一次全新的访问开始，观察公告、刷新库存和翻页操作。这个页面就是本文综合分析的对象，请先保留真实请求证据，再继续往下读。

实时采集训练了我们区分协议层和业务层。最后一个案例回到 HTTP，但把前面的知识组合起来。

打开 /portal/，看到一张可以分页的商品库存表，还有一条公告。目标是取得全部 24 件商品的名称、价格和库存，并说明数据怎样到达页面。

建议先独立分析，再阅读下面的拆解。先写下你的第一步和理由：检查初始 HTML、观察 Network，还是查看脚本？最终判断要由证据决定。

5.1 第一轮观察：排除容易误导你的内容

检查同一次页面导航对应的 Document 响应。商品记录没有直接出现在返回的 HTML 中，这说明当前表格不能仅靠解析这份 HTML 完成采集。

接着观察页面加载和翻页产生的请求。公告响应虽然也是 JSON，却没有商品名称、价格、库存等目标字段。它与目标数据没有对应关系，不应因为“看起来像接口”就成为采集对象。

这里得到的结论要保持精确：这份 Document 响应不包含目标记录，公告接口也不提供目标记录。 不能仅凭这两条证据，断言整个网站没有其他嵌入数据或其他数据来源。

5.2 第二轮观察：找到凭证从哪里来、往哪里去

页面先调用握手接口，取得 ticket；后续库存请求把它放入 X-Portal-Ticket 请求头。握手没有通过 Set-Cookie 下发这个凭证，页面将它保存在 JavaScript 变量里。

这解释了第二篇里一个容易混淆的点：HTTP 客户端保存 Cookie，不等于保存了页面内存中的变量。Client 可以维持连接和 Cookie，但你仍然需要从响应读取 ticket，并显式构造请求头。[6]

不要输出完整 ticket 来证明自己取到了它。分析记录写“来源是握手响应，使用位置是库存请求头”已经足够；运行日志记录凭证取得是否成功即可。

5.3 第三轮观察：为什么复制的请求第二次失败？

本例库存接口依次验证 ticket、时间戳和 nonce。时间戳单位为毫秒，允许窗口为前后两分钟；nonce 是符合约束的随机字符串，已成功使用的值会被记录。

可以做三个小实验，每次只改变一个变量：

接着保留旧 nonce、只更新时间戳，如果仍然返回 409；再同时更换时间戳与 nonce，恢复成功，就可以把失败定位到一次性参数，而非简单地归结为“Python 被识别了”。

这些状态码是本例的约定，不是所有网站的通用规则。真实接口也可能返回 200 加业务错误码，应该同时检查响应结构。

公开实验站把 ticket 与 nonce 记录放在 Durable Objects 中。ticket 签发一小时后失效；nonce 在相关时间窗口结束后清理，并按 ticket 隔离。Worker 实例更换不会清空它们。第四篇讨论过，判断一次性参数时要包含服务端记录的作用域与生命周期。

六、把证据变成一段可重复的采集程序

到这里，路线已经明确：新握手取得 ticket，按页请求库存，每一页生成新的时间戳和 nonce，验证完整性。

下面是面向配套公开实验站静态 24 件商品的完整代码，依赖 httpx。它没有浏览器步骤，也不需要手动复制凭证。真实网站的握手可能还有登录前置条件，不能直接照搬这个接口契约。

import os
import secrets
import time
import httpx
​
BASE_URL = os.environ.get("CDP_BASE_URL", "https://web-data-lab.pages.dev")
​
def collect_inventory(base_url: str) -> list[dict]:
   with httpx.Client(base_url=base_url, timeout=10) as client:
       handshake = client.post("/portal/api/handshake/")
       handshake.raise_for_status()
       ticket = handshake.json()["ticket"]
       if not isinstance(ticket, str) or not ticket:
           raise ValueError("握手没有返回有效 ticket")
​
       collected = {}
       expected_pages = None
       page_no = 1
       while True:
           body = {
               "ts": int(time.time() * 1000),
               "nonce": secrets.token_hex(8),
               "page": page_no,
               "page_size": 10,
          }
           response = client.post(
               "/portal/api/inventory/",
               headers={"X-Portal-Ticket": ticket},
               json=body,
          )
           response.raise_for_status()
           data = response.json()
           if data["count"] != 24:
               raise ValueError("演示数据总数与预期不一致")
           if expected_pages is None:
               expected_pages = data["total_pages"]
               if expected_pages != 3:
                   raise ValueError("演示分页元数据与预期不一致")
           elif data["total_pages"] != expected_pages:
               raise ValueError("分页过程中总页数变化")
​
           for row in data["results"]:
               for key in ("id", "name", "price", "stock"):
                   if key not in row:
                       raise ValueError(f"商品缺少字段: {key}")
               product_id = row["id"]
               if product_id in collected:
                   raise ValueError(f"分页出现重复商品: {product_id}")
               collected[product_id] = row
​
           if page_no == expected_pages:
               break
           page_no += 1
​
       if set(collected) != set(range(1, 25)):
           raise ValueError("演示商品 ID 有缺失或出现意外值")
       return [collected[i] for i in sorted(collected)]
​
if __name__ == "__main__":
   for row in collect_inventory(BASE_URL):
       print(row["id"], row["name"], row["price"], row["stock"])

这里把总数和 ID 范围写成固定断言，是因为教学数据集已知且静态。真实项目需要从业务约定、接口元数据或独立对照来源建立预期，不能把 24 和连续 ID 当成通用完整性规则。

代码故意在发现重复时抛出异常，没有用字典覆盖把重复悄悄消掉。如果分页边界变化，你需要先看到这个异常，才能决定重新采集、改用游标或使用服务端快照。静态样本上“不重不漏”，也不能自动证明动态站点上存在一致性快照。

七、怎样证明综合采集真的完成了？

至少分别回答下面四个问题。

7.1 从干净起点是否能够成功？

关闭用于分析的浏览器，只启动 Python 程序。代码需要自行取得 ticket，自行生成参数。它不应该依赖控制台里残留的变量、手动复制的一组 nonce 或上次运行生成的文件。

连续独立运行两次，两次都建立新的客户端和握手。对于本例静态数据，比较两次商品 ID、名称、价格与库存应当一致。对实时业务则应比较结构、范围和各自采集时间对应的规则，不能把数据变化直接当成失败。

7.2 完整是否只靠“返回 200”判断？

第一页面的 10 条记录也可以全部返回 200。总数、页数、唯一 ID 和必需字段需要单独检查。

这个演示系统还有一条不需要 ticket 的分页接口 /labs/pagination/products/，可以取同一批静态商品做逐项对账。这条对照路径属于本例的便利条件；真实项目未必有，应寻找导出文件、后台统计或其他独立业务证据。

对账时同时比较键集合和字段。只比较“两边都有 24 条”，仍可能放过商品错配、价格单位不同或字段被误映射的问题。

7.3 出错时能否定位到一层？

建议日志至少记录采集批次、阶段、页码、HTTP 状态、业务错误码、耗时、返回条数和累计唯一条数。凭证不入日志。

例如“第 2 页 HTTP 409”可以回查 nonce 的生成与复用；“HTTP 200，但结果字段缺失”应检查响应结构；“累计条数不足但没有报错”应检查分页停止条件。这样的日志比打印整段请求头更有用。

超时后重试也要重新分析。库存查询在业务上只读，但 nonce 可能已经被服务端消费；重新发送旧参数可能被拒绝。对于涉及下单或写入的业务，更不能只换一个 nonce 重发，需要业务幂等设计。

7.4 能否解释为什么选这条路线？

本例选择 HTTPX，是因为商品接口和参数生成已经能够独立复现。如果某个门户依赖页面内存中的复杂计算，第四篇的浏览器辅助路线可能更合适。如果只有渲染后的视图提供目标字段，第三篇的 DOM 路线仍有价值。

选择标准包括可复现性、完整性、身份管理、维护成本和采集规模。用浏览器成功一次，和部署一个持续稳定的数据任务，是不同的验收目标。

八、20 个实验之后，真正可以带走的方法

回顾整个系列，我们并没有为 20 种网站分别背 20 个招数。每个 Phase 都补齐了一种判断能力。

面对一个新网页，可以先写下五句话：我要的业务字段是什么；目前在哪一层找到对应数据；客户端需要哪些状态；哪些参数必须现场生成；怎样证明结果完整。

如果这五句话仍然回答不清，优先补证据。这样，CDP 才会成为拆解浏览器行为的工具，Python 才会成为复现和验证数据链路的工具。

一个可靠的网络爬虫最终要交付两样东西：一份可以继续使用的数据，以及一条能够解释、复现、排查的采集链路。前者让任务有结果，后者让结果可以长期被信任。