能否使用Python的websockets库获取直播聊天的二进制消息？

websockets库完全支持处理二进制消息，你的现有代码存在两个核心问题：只接收一次消息就终止程序，且未针对直播聊天的WebSocket协议做必要处理。以下是修正方案和关键说明：

核心修改点 #

• 持续接收消息：用循环保持连接，避免程序接收一次后退出
• 区分消息类型：判断是二进制（bytes）还是文本消息，分别处理
• 适配服务协议：部分直播聊天服务要求客户端先发送初始化/验证消息，才会推送聊天内容

修正后的代码示例 #

import websockets
import asyncio

async def access_chat():
    uri = "wss://chat-af7e4856.afreecatv.com:8001/Websocket/freesias"
    async with websockets.connect(uri, subprotocols=['chat'], ping_interval=None) as websocket:
        # 注意：需根据AFreecaTV的协议发送初始化消息（比如房间验证信息）
        # 可通过浏览器开发者工具抓包获取客户端初始发送的内容，示例：
        # await websocket.send(b'your_initialization_payload')
        
        while True:
            content = await websocket.recv()
            if isinstance(content, bytes):
                print(f"收到二进制消息（长度：{len(content)}）: {content.hex()}")
                # 尝试解码为文本（若部分二进制消息是UTF-8编码的聊天内容）
                try:
                    decoded_text = content.decode('utf-8')
                    print(f"解码后聊天内容：{decoded_text}")
                except UnicodeDecodeError:
                    # 非文本格式的状态检测消息，可根据协议规则过滤
                    pass
            else:
                print(f"收到文本消息: {content}")

asyncio.run(access_chat())

关键注意事项 #

1. 协议解析：AFreecaTV的二进制消息有自定义格式（通常包含消息类型、用户信息、内容长度等字段），需要抓包分析结构，才能准确提取真实聊天内容、过滤状态检测消息。
2. 初始化消息：必须模拟浏览器发送的初始验证消息，否则服务端不会推送聊天数据。可在浏览器开发者工具的「网络」面板中，找到WebSocket连接的「帧」标签，查看客户端发送的首条消息，复制对应字节数据在代码中发送。
3. 心跳处理：若服务端有心跳检测，需定期发送对应消息维持连接（可根据抓包结果实现）。

内容的提问来源于stack exchange，提问作者HyeonPhil Youn