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('<i', data_length) \
              + struct.pack('<i', data_length) + struct.pack('<i', code)
    s.send(msgHead)
    sent = 0
    while sent < len(msg):
        tn = s.send(msg[sent:])
        sent += tn

def danmu(room_id):
    """主接收线程：登录、入组、循环接收"""
    # 登录请求
    login = 'type@=loginreq/roomid@=%s/\0' % room_id
    login = login.encode('utf-8')
    send_msg(login)

    # 加入分组请求
    joingroup = 'type@=joingroup/rid@=%s/gid@=-9999/\0' % room_id
    joingroup = joingroup.encode('utf-8')
    send_msg(joingroup)

    while True:
        content = s.recv(1024)
        if judge_chatmsg(content):
            nickname = nick_name(content)
            chatmsg = chat_msg(content)
            print('%s : %s' % (nickname, chatmsg))
        else:
            pass

def keep_alive():
    """心跳线程：每 45 秒发送一次"""
    while True:
        msg = 'type@=keeplive/tick@=%s/\0' % str(int(time.time()))
        send_msg(msg)
        time.sleep(45)

def nick_name(content):
    """提取昵称"""
    pattern = re.compile(r'nn@=(.*)/txt@')
    try:
        return pattern.findall(content)[0]
    except IndexError:
        return ''

def chat_msg(content):
    """提取弹幕文本"""
    pattern = re.compile(r'txt@=(.*)/cid@')
    try:
        return pattern.findall(content)[0]
    except IndexError:
        return ''

def judge_chatmsg(content):
    """判断是否为弹幕消息"""
    pattern = re.compile(r'type@=(.*)/rid@')
    data_type = pattern.findall(content)
    try:
        return data_type[0] == 'chatmsg'
    except Exception as e:
        return False

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

注意事项

斗鱼官方文档更新频率不高，实际返回的消息结构可能随版本迭代有所增加。上述代码仅提取了最基础的昵称和文本内容。如果遇到解析失败，建议先打印原始报文查看最新格式，再针对性地调整正则表达式。