Python 实时爬取斗鱼弹幕

Python 实时爬取斗鱼弹幕

实现目标

输入斗鱼房间号，通过 TCP 协议实时获取并解析弹幕信息。

核心逻辑梳理

斗鱼开放了弹幕 API，直接对接服务器即可。整体流程其实并不复杂，主要涉及两个线程的协作：一个负责建立连接并接收数据，另一个负责定时发送心跳保持长连接。

1. 建立连接与握手

首先通过 TCP 连接到弹幕服务器（IP: openbarrage.douyutv.com, Port: 8601）。连接成功后需要完成两步握手：

• 登录请求：发送格式为 type@=loginreq/roomid@=房间号/\0 的消息。这里不需要账号密码，只要知道房间号即可。
• 加入分组：登录成功后，服务器会返回确认信息，此时需继续发送进入弹幕分组请求，格式为 type@=joingroup/rid@=房间号/gid@=-9999/\0。其中 gid 设为 -9999 代表海量弹幕模式。

此后服务器会持续推送消息流，除了普通弹幕外，还可能包含礼物、进场通知等，可以通过解析消息头中的 type 字段来过滤所需内容。

2. 心跳保活

TCP 长连接并非永久有效，为了防止服务器因超时断开，需要定期发送心跳包。心跳格式为 type@=keeplive/tick@=当前秒级时间戳/\0，建议每隔 45 秒左右发送一次。

代码实现

下面是一个完整的 Python 3 示例。为了保持代码健壮性，我将其拆分为连接、发送、接收和解析几个模块。注意，实际运行时请根据最新的服务器返回报文调整正则匹配规则。

# -*- coding:utf-8 -*-
import socket
import re
import time
import struct
import threading

# 全局 socket 对象
s = None

def connect():
    """建立 TCP 连接"""
    global s
    print('-----*-----DouYu_Spider-----*-----\n')
    host = socket.gethostbyname("openbarrage.douyutv.com")
    port = 8601
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.connect((host, port))

def send_msg(msg):
    """封装发送逻辑，处理长度头"""
    data_length = len(msg) + 8
    code = 689
    # 构造头部：长度 + 长度 + 类型码
    msgHead = struct.pack(, data_length) \
              + struct.pack(, data_length) + struct.pack(, code)
    s.send(msgHead)
    sent = 
     sent < (msg):
        tn = s.send(msg[sent:])
        sent += tn

 ():
    
    
    login =  % room_id
    login = login.encode()
    send_msg(login)

    
    joingroup =  % room_id
    joingroup = joingroup.encode()
    send_msg(joingroup)

     :
        content = s.recv()
         judge_chatmsg(content):
            nickname = nick_name(content)
            chatmsg = chat_msg(content)
            ( % (nickname, chatmsg))
        :
            

 ():
    
     :
        msg =  % ((time.time()))
        send_msg(msg)
        time.sleep()

 ():
    
    pattern = re.()
    :
         pattern.findall(content)[]
     IndexError:
         

 ():
    
    pattern = re.()
    :
         pattern.findall(content)[]
     IndexError:
         

 ():
    
    pattern = re.()
    data_type = pattern.findall(content)
    :
         data_type[] == 
     Exception  e:
         

 __name__ == :
    connect()
    t1 = threading.Thread(target=danmu, args=(,))
    t2 = threading.Thread(target=keep_alive)
    t1.start()
    t2.start()