如何在Python爬虫中实现对网页中WebSocket实时数据的抓取？-Python教程-PHP中文网

websocket连接失败主因是http客户端无法处理其协议，必须用专用库（如websocket-client）完成握手、帧解析与保活；需正确配置url、origin头，手动解码二进制数据，并实现带退避的重连逻辑。

如何在python爬虫中实现对网页中websocket实时数据的抓取？

WebSocket连接被拒绝或报400错误

直接用requests发GET请求到ws://地址必然失败——HTTP客户端根本不懂WebSocket协议，服务器会立刻返回400或直接断连。WebSocket需要专用客户端完成握手（Upgrade头）、帧解析和长连接维持。

正确做法是用websocket-client库（非websockets，后者是异步的，同步爬虫里容易阻塞）：

pip install websocket-client

确保URL以ws://或wss://开头，不能是http://
若页面JS中WebSocket地址是相对路径（如/realtime），需拼上当前页面域名和协议，例如wss://example.com/realtime
某些站点校验Origin头，需在header参数中显式传入：{"Origin": "https://www.php.cn/link/b05edd78c294dcf6d960190bf5bde635"}

抓到的数据是二进制帧或加密内容

很多金融、交易类网站把WebSocket消息用ArrayBuffer发送，并做简单XOR或Base64混淆，不是直接可读JSON。浏览器开发者工具的Network → WS → Frames里看到“binary”就说明这点。

用websocket-client时，默认收到的是原始bytes对象，需手动解码：

def on_message(ws, message):
    if isinstance(message, bytes):
        # 尝试UTF-8解码
        try:
            text = message.decode('utf-8')
        except UnicodeDecodeError:
            # 可能是Base64编码的JSON字符串
            import base64
            try:
                decoded = base64.b64decode(message)
                text = decoded.decode('utf-8')
            except Exception:
                print("unknown binary format:", message[:20])
                return
        print(text)
    else:
        print(message)

先检查message类型，别直接json.loads(message)——对bytes会报TypeError: the JSON object must be str, bytes or bytearray
若解码后仍是乱码，打开网页源码搜WebSocket附近JS，看是否有atob()、Uint8Array或自定义解密函数
不要依赖ws.run_forever()无限阻塞——它会让整个爬虫卡住；改用ws.settimeout(5) + 循环ws.recv()

连接自动断开且无法重连

服务端通常30–60秒无心跳就关闭连接，而websocket-client默认不发ping。断开后on_close触发，但不自动重试。

Python数据分析(专业版)

企业级Python数据分析方案，支持机器学习建模、时间序列预测、大数据处理与自动化报表。

下载

必须自己实现保活和重连逻辑：

启用ping机制：ws.run_forever(ping_interval=30, ping_timeout=5)
捕获异常并重试：on_error里记录错误，on_close里调用ws.connect()重新建连（注意别无限递归）
加退避延迟：首次失败等1秒，第二次等2秒，第三次等4秒，避免被频率限制
检查是否被封IP——如果重连总在ConnectionRefusedError或TimeoutError间切换，大概率服务端做了连接数限制

如何从网页JS里准确提取WebSocket地址

地址常藏在动态生成的JS中，比如new WebSocket("wss://" + host + "/stream?token=" + getToken())。靠正则硬匹配容易漏掉变量拼接或混淆。

更可靠的做法是启动一个轻量级无头浏览器（如playwright），执行JS并取值：

from playwright.sync_api import sync_playwright
<p>with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
page = browser.new_page()
page.goto("<a href="https://www.php.cn/link/b05edd78c294dcf6d960190bf5bde635">https://www.php.cn/link/b05edd78c294dcf6d960190bf5bde635</a>")
ws_url = page.evaluate("""
() => {
// 模拟页面中WebSocket初始化逻辑
const host = location.hostname;
const token = document.querySelector('meta[name="token"]')?.content || '';
return <code>wss://${host}/data?token=${token}</code>;
}
""")
print(ws_url)
browser.close()</p>