七月末的南京简直开启了「微波炉」模式,白天要学车的我,晚上自然选择宅在家看直播,看着狗贼叔叔满屏幕的弹幕,我就想着能不能把弹幕爬下来呢?说干就干
结果的展示:
这里只抓到弹幕内容和发送用户
并输出在终端上,有兴趣的小伙伴
可以在这个基础上接着开发,
搜集弹幕做做数据分析也是很ok的啊!
下面是展示图:
资料的搜集
面向Google编程的我,第一件事当然是键入关键词:「Python 弹幕」
吃惊的是,网上已经有了炒鸡完善的弹幕第三方库:「DanMU」
使用起来也是炒鸡简单,十几行代码就能轻松获取直播间的弹幕了,
有兴趣的同学可以去搜索看看。
本着练手和不折腾会死的态度,我还是想尝试自己写一个版本出来,
然后就找到了 斗鱼居然开放了Api,
这样的话,只要稍微处理一下,就能愉快的获取想要的信息了。
斗鱼Api接口文档和接入协议
仔细观察文档之后,我发现只要自己实现一下协议头,
就能接入弹幕服务器了,
接着构造弹幕请求,
就能实时的获取每一条弹幕了。
请求头的构造
先看文档的要求:
简而言之呢:
请求一共分为三个部分:长度,头部,数据部
分别按照文档的要求构造就行,
需要注意的是,获取和返回的类型是都是 Bytes
代码:
def send_req_msg(msgstr):
'''构造并发送符合斗鱼api的请求'''
msg = msgstr.encode('utf8')
data_length = len(msg) + 8
code = 689
# 构造协议头
msgHead = int.to_bytes(data_length, 4, 'little') \
+ int.to_bytes(data_length, 4, 'little') + \
int.to_bytes(code, 4, 'little')
client.send(msgHead)
sent = 0
while sent < len(msg):
tn = client.send(msg[sent:])
sent = sent + tn
获取弹幕
这里的部分也是按照文档要求写就成
首先 发送登录请求
接着 每隔固定时间发送【心跳请求】防止断线
def DM_start(roomid):
# 构造登录授权请求
msg = 'type@=loginreq/roomid@={}/\0'.format(roomid)
send_req_msg(msg)
# 构造获取弹幕消息请求
msg_more = 'type@=joingroup/rid@={}/gid@=-9999/\0'.format(roomid)
send_req_msg(msg_more)
while True:
# 服务端返回的数据
data = client.recv(1024)
# 通过re模块找发送弹幕的用户名和内容
danmu_username = username_re.findall(data)
danmu_content = danmu_re.findall(data)
if not data:
break
else:
for i in range(0, len(danmu_content)):
try:
# 输出信息
print('[{}]:{}'.format(danmu_username[0].decode(
'utf8'), danmu_content[0].decode(encoding='utf8')))
except:
continue
def keeplive():
'''保持心跳,15秒心跳请求一次'''
while True:
msg = 'type@=keeplive/tick@=' + str(int(time.time())) + '/\0'
send_req_msg(msg)
print('发送心跳包')
time.sleep(15)
tricky 的部分
上面的内容,说起来都不是很难,
但是想要完整的实现需求,
这里需要的知识还是比较多的:socket相关
正则表达式相关
signal相关
多线程、多进程相关
比如我想要实现捕捉「ctrl+c」的信号,
好在我们退出程序的时候,能够正确的处理
这时候就要用到signal相关的知识
说起来,在今天之前,我完全不知道还可以这样用。
总之越是学到后面,
越是会觉得自己的知识储备不足,
Python 作为一门十分强大和容易上手的语言,
能够帮助我们迅速的实现需求,
但是不要认为他单单只能写爬虫哦,
完整的代码
有详细的注释哦:
'''利用斗鱼弹幕 api尝试抓取斗鱼tv指定房间的弹幕'''
import multiprocessing
import socket
import time
import re
import signal
# 构造socket连接,和斗鱼api服务器相连接
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
host = socket.gethostbyname("")
port = 8601
client.connect((host, port))
# 弹幕查询正则表达式
danmu_re = pile(b'txt@=(.+?)/cid@')
username_re = pile(b'nn@=(.+?)/txt@')
def send_req_msg(msgstr):
'''构造并发送符合斗鱼api的请求'''
msg = msgstr.encode('utf-8')
data_length = len(msg) + 8
code = 689
# 构造协议头
msgHead = int.to_bytes(data_length, 4, 'little') \
+ int.to_bytes(data_length, 4, 'little') + \
int.to_bytes(code, 4, 'little')
client.send(msgHead)
sent = 0
while sent < len(msg):
tn = client.send(msg[sent:])
sent = sent + tn
def DM_start(roomid):
# 构造登录授权请求
msg = 'type@=loginreq/roomid@={}/\0'.format(roomid)
send_req_msg(msg)
# 构造获取弹幕消息请求
msg_more = 'type@=joingroup/rid@={}/gid@=-9999/\0'.format(roomid)
send_req_msg(msg_more)
while True:
# 服务端返回的数据
data = client.recv(1024)
# 通过re模块找发送弹幕的用户名和内容
danmu_username = username_re.findall(data)
danmu_content = danmu_re.findall(data)
if not data:
break
else:
for i in range(0, len(danmu_content)):
try:
# 输出信息
print('[{}]:{}'.format(danmu_username[0].decode(
'utf8'), danmu_content[0].decode(encoding='utf8')))
except:
continue
def keeplive():
'''保持心跳,15秒心跳请求一次'''
while True:
msg = 'type@=keeplive/tick@=' + str(int(time.time())) + '/\0'
send_req_msg(msg)
print('发送心跳包')
time.sleep(15)
def logout():
'''与斗鱼服务器断开连接关闭线程'''
msg = 'type@=logout/'
send_req_msg(msg)
print('已经退出服务器')
def signal_handler(signal, frame):
'''捕捉 ctrl+c的信号 即 signal.SIGINT触发hander:登出斗鱼服务器关闭进程'''
p1.terminate()
p2.terminate()
logout()
print('Bye')
if __name__ == '__main__':
#room_id = input('请输入房间ID: ')
# 狗贼的房间号
room_id = 208114
# 开启signal捕捉
signal.signal(signal.SIGINT, signal_handler)
# 开启弹幕和心跳进程
p1 = multiprocessing.Process(target=DM_start, args=(room_id,))
p2 = multiprocessing.Process(target=keeplive)
p1.start()
p2.start()