Python脚本爬虫WebSocket如何抓取

本文目录导读：

1. 基础工具安装
2. 方法一：使用websocket-client库（同步）
3. 方法二：使用websockets库（异步）
4. 方法三：抓取加密交易所WebSocket示例
5. 方法四：使用Selenium模拟浏览器WebSocket
6. 高级特性：自动重连与数据持久化
7. 注意事项

我来详细说明如何使用Python爬虫抓取WebSocket数据。

基础工具安装

# 安装websocket客户端库
pip install websocket-client
pip install websockets  # 异步版本
# 可选工具
pip install aiohttp  # 异步HTTP客户端，支持WebSocket
pip install ws4py    # 另一种WebSocket客户端

使用websocket-client库（同步）

import websocket
import json
import threading
def on_message(ws, message):
    """收到消息时的回调"""
    data = json.loads(message)
    print(f"收到数据: {data}")
    # 这里处理你的数据，如保存到文件或数据库
def on_error(ws, error):
    """错误处理"""
    print(f"错误: {error}")
def on_close(ws, close_status_code, close_msg):
    """连接关闭处理"""
    print(f"连接关闭: {close_status_code} - {close_msg}")
def on_open(ws):
    """连接成功建立"""
    print("WebSocket连接已建立")
    # 发送订阅消息
    subscribe_msg = {
        "event": "subscribe",
        "channel": "ticker",
        "symbol": "BTC/USDT"
    }
    ws.send(json.dumps(subscribe_msg))
def connect_websocket():
    """建立WebSocket连接"""
    # 禁用SSL验证（如果需要）
    websocket.enableTrace(False)
    # 连接WebSocket
    ws = websocket.WebSocketApp(
        "wss://your-websocket-url.com/ws",  # 替换为实际URL
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )
    # 运行连接（会阻塞）
    ws.run_forever()
# 启动连接
connect_websocket()

使用websockets库（异步）

import asyncio
import websockets
import json
async def subscribe_market_data():
    """订阅市场数据"""
    uri = "wss://your-websocket-url.com/ws"  # 替换为实际URL
    async with websockets.connect(uri) as websocket:
        # 发送订阅请求
        subscribe_msg = {
            "event": "subscribe",
            "channel": "ticker",
            "symbol": "BTC/USDT"
        }
        await websocket.send(json.dumps(subscribe_msg))
        # 持续接收数据
        while True:
            try:
                response = await websocket.recv()
                data = json.loads(response)
                print(f"收到数据: {data}")
                # 数据处理逻辑
                # save_to_database(data)
            except websockets.exceptions.ConnectionClosed:
                print("连接已关闭，尝试重连...")
                break
            except Exception as e:
                print(f"错误: {e}")
                break
async def main():
    """主函数"""
    # 创建重连机制
    while True:
        try:
            await subscribe_market_data()
        except Exception as e:
            print(f"连接失败: {e}")
            await asyncio.sleep(5)  # 等待5秒重试
# 运行
asyncio.run(main())

抓取加密交易所WebSocket示例

import websocket
import json
import threading
import time
from datetime import datetime
class CryptoWebSocket:
    """加密货币WebSocket抓取器"""
    def __init__(self, url, channels=None):
        self.url = url
        self.channels = channels or []
        self.ws = None
        self.running = False
        self.reconnect_count = 0
    def on_message(self, ws, message):
        """处理接收到的消息"""
        try:
            data = json.loads(message)
            # 根据数据类型处理
            if 'channel' in data:
                channel = data['channel']
                if channel == 'ticker':
                    self.handle_ticker(data)
                elif channel == 'trade':
                    self.handle_trade(data)
                elif channel == 'orderbook':
                    self.handle_orderbook(data)
        except json.JSONDecodeError:
            print(f"无法解析的JSON数据: {message[:100]}")
    def handle_ticker(self, data):
        """处理ticker数据"""
        ticker_data = {
            'timestamp': datetime.now().isoformat(),
            'symbol': data.get('symbol'),
            'price': data.get('price'),
            'volume': data.get('volume'),
            'change': data.get('change')
        }
        print(f"Ticker: {ticker_data}")
        # 保存到文件或数据库
    def handle_trade(self, data):
        """处理交易数据"""
        print(f"Trade: {json.dumps(data, indent=2)}")
    def handle_orderbook(self, data):
        """处理订单簿数据"""
        print(f"Orderbook update received")
    def on_error(self, ws, error):
        """处理错误"""
        print(f"WebSocket错误: {error}")
    def on_close(self, ws, close_status_code, close_msg):
        """处理连接关闭"""
        print(f"WebSocket连接关闭: {close_status_code}")
        if self.running:
            self.reconnect()
    def on_open(self, ws):
        """处理连接建立"""
        print("WebSocket连接已建立")
        self.reconnect_count = 0
        # 发送订阅消息
        for channel in self.channels:
            subscribe_msg = {
                "event": "subscribe",
                "channel": channel
            }
            ws.send(json.dumps(subscribe_msg))
            print(f"已订阅: {channel}")
    def connect(self):
        """建立连接"""
        websocket.enableTrace(False)
        self.ws = websocket.WebSocketApp(
            self.url,
            on_open=self.on_open,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close
        )
        self.running = True
        # 在另一个线程中运行
        wst = threading.Thread(target=self.ws.run_forever)
        wst.daemon = True
        wst.start()
    def reconnect(self):
        """重连机制"""
        if self.reconnect_count < 10:  # 最多重连10次
            wait_time = min(30, 2 ** self.reconnect_count)
            print(f"将在 {wait_time} 秒后重连...")
            time.sleep(wait_time)
            self.reconnect_count += 1
            self.connect()
    def disconnect(self):
        """断开连接"""
        self.running = False
        if self.ws:
            self.ws.close()
    def send_message(self, message):
        """发送消息"""
        if self.ws and self.running:
            self.ws.send(json.dumps(message))
# 使用示例
def main():
    # 配置WebSocket连接
    ws_client = CryptoWebSocket(
        url="wss://stream.binance.com:9443/ws",
        channels=["btcusdt@ticker", "ethusdt@ticker"]
    )
    # 启动连接
    ws_client.connect()
    try:
        # 保持主线程运行
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("正在关闭连接...")
        ws_client.disconnect()
if __name__ == "__main__":
    main()

使用Selenium模拟浏览器WebSocket

from selenium import webdriver
from selenium.webdriver.common.by import By
import json
import time
def capture_websocket_via_selenium():
    """通过Selenium捕获WebSocket通信"""
    options = webdriver.ChromeOptions()
    options.add_argument('--disable-gpu')
    options.add_argument('--headless')  # 无头模式
    # 启用性能日志
    options.set_capability('goog:loggingPrefs', {'performance': 'ALL'})
    driver = webdriver.Chrome(options=options)
    try:
        # 访问目标网站
        driver.get("https://example.com")  # 替换为实际URL
        # 等待页面加载
        time.sleep(5)
        # 获取性能日志
        logs = driver.get_log('performance')
        websocket_messages = []
        for entry in logs:
            log_data = json.loads(entry['message'])
            message = log_data.get('message', {})
            # 过滤WebSocket消息
            if message.get('method') == 'Network.webSocketFrameReceived':
                params = message.get('params', {})
                response = params.get('response', {})
                payload = response.get('payloadData', '')
                try:
                    data = json.loads(payload)
                    websocket_messages.append(data)
                    print(f"WebSocket消息: {data}")
                except:
                    pass
        return websocket_messages
    finally:
        driver.quit()

高级特性：自动重连与数据持久化

import asyncio
import websockets
import json
import sqlite3
from datetime import datetime
import logging
logging.basicConfig(level=logging.INFO)
class WebSocketCrawler:
    """高级WebSocket爬虫"""
    def __init__(self, url, db_path='websocket_data.db'):
        self.url = url
        self.db_path = db_path
        self.setup_database()
    def setup_database(self):
        """初始化数据库"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        # 创建数据表
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS websocket_data (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp TEXT,
                channel TEXT,
                data TEXT,
                received_at TEXT
            )
        ''')
        conn.commit()
        conn.close()
    def save_to_database(self, channel, data):
        """保存数据到数据库"""
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            cursor.execute('''
                INSERT INTO websocket_data (timestamp, channel, data, received_at)
                VALUES (?, ?, ?, ?)
            ''', (
                datetime.now().isoformat(),
                channel,
                json.dumps(data),
                datetime.now().isoformat()
            ))
            conn.commit()
            conn.close()
            return True
        except Exception as e:
            logging.error(f"保存数据失败: {e}")
            return False
    async def connect_with_retry(self, max_retries=5):
        """带重试的连接"""
        for attempt in range(max_retries):
            try:
                async with websockets.connect(
                    self.url,
                    ping_interval=30,
                    ping_timeout=10
                ) as websocket:
                    logging.info(f"连接成功 (尝试 {attempt + 1})")
                    # 订阅数据
                    await websocket.send(json.dumps({
                        "event": "subscribe",
                        "channel": "ticker"
                    }))
                    # 持续接收数据
                    async for message in websocket:
                        try:
                            data = json.loads(message)
                            channel = data.get('channel', 'unknown')
                            # 保存到数据库
                            self.save_to_database(channel, data)
                            # 可选：发送到消息队列
                            # await self.send_to_queue(data)
                        except json.JSONDecodeError:
                            continue
            except websockets.exceptions.ConnectionClosed:
                logging.warning(f"连接断开，准备重试 ({attempt + 1}/{max_retries})")
                await asyncio.sleep(2 ** attempt)  # 指数退避
            except Exception as e:
                logging.error(f"连接错误: {e}")
                await asyncio.sleep(5)
        logging.error("达到最大重试次数，放弃连接")
# 使用示例
async def main():
    crawler = WebSocketCrawler("wss://your-websocket-url.com/ws")
    await crawler.connect_with_retry()
if __name__ == "__main__":
    asyncio.run(main())

注意事项

1. 法律合规：确保抓取行为符合网站服务条款和当地法律法规
2. 频率限制：遵守WebSocket的速率限制，避免过度请求
3. 数据解析：不同平台的数据格式可能不同，需要适配解析
4. 错误处理：实现完善的错误处理和重连机制
5. 资源管理：及时关闭连接，释放资源
6. 安全性：避免在代码中硬编码敏感信息

选择哪种方法取决于你的具体需求：同步库适合简单场景，异步库适合高并发，Selenium适合抓取动态加载的WebSocket数据。