PyMySQL==0.9.3websocket_client==0.57.0
斗鱼直播间弹幕最新
我最近在学习python的直播间弹幕爬取，但是由于斗鱼官方把第三方api的接口改变了，必须要注册为开发者才能使用官方提供的方法进行弹幕爬取。所以我通过搜索教程了解到可以使用浏览器自带的爬取功能对弹幕进行爬取。
原理如下：
利用websocket建立wss 连接
wss://:8506/'
8501-8507都可以使用。
发送登录信息
发生入组信息
发送心跳数据，（和b站不一样更高级了，有心跳数据了）。
利用wss必须对发送包进行加密，对接收的数据进行解包，
这些操作官方api 有提供，所以我就不再进行解释了。
就能返回弹幕数据。完整代码如下，将roomid="666743" 改成你需要的id。
以下代码并非我独创，有借鉴别人的。
import websocketimport threadingimport timeclass DyDanmuMsgHandler:# 将字符串数据按照斗鱼协议封装为字节流def dy_encode(self,msg):# 头部8字节，尾部1字节，与字符串长度相加即数据长度# 为什么不加最开头的那个消息长度所占4字节呢？这得问问斗鱼^^data_len = len(msg) + 9# 字符串转化为字节流msg_byte = msg.encode('utf-8')# 将数据长度转化为小端整数字节流len_byte = int.to_bytes(data_len, 4, 'little')# 前两个字节按照小端顺序拼接为0x02b1，转化为十进制即689（《协议》中规定的客户端发送消息类型）# 后两个字节即《协议》中规定的加密字段与保留字段，置0send_byte = bytearray([0xb1, 0x02, 0x00, 0x00])# 尾部以'\0'结束end_byte = bytearray([0x00])# 按顺序拼接在一起data = len_byte + len_byte + send_byte + msg_byte + end_bytereturn datadef __parse_msg(self,raw_msg):'''解析数据:param raw_msg: 原始response数据:return:'''res = {}attrs = raw_msg.split('/')[0:-1]for attr in attrs:attr = attr.replace('@s','/')attr = attr.replace('@A','@')couple = attr.split('@=')res[couple[0]] = couple[1]return resdef dy_decode(self,msg_byte):'''解析斗鱼返回的数据:param msg_byte::return:'''pos = 0msg = []while pos < len(msg_byte):content_length = int.from_bytes(msg_byte[pos: pos + 4], byteorder='little')content = msg_byte[pos + 12: pos + 3 + content_length].decode(encoding='utf-8', errors='ignore')msg.append(content)pos += (4 + content_length)return msgdef get_chat_messages(self,msg_byte):'''从数据获取chatmsg数据:param msg_byte::return:'''decode_msg = self.dy_decode(msg_byte)messages = []for msg in decode_msg:res = self.__parse_msg(msg)if res['type'] !='chatmsg':continuemessages.append(res)return messages class DyDanmuCrawler:def __init__(self,roomid):self.__room_id = roomidself.__heartbeat_thread = Noneself.__client = DyDanmuWebSocketClient(on_open=self.__prepare,on_message=self.__receive_msg,on_close=self.__stop)self.__msg_handler = DyDanmuMsgHandler()self.__keep_HeartBeat = Truedef start(self):'''开启客户端:return:'''self.__client.start()def __stop(self):'''登出停止客户端停止心跳线程:return:'''self.__logout()self.__client.stop()self.__keep_HeartBeat=Falsedef on_error(self, error):print(error)def on_close(self):print('close')# 发送入组消息def join_group(self):'''发送群组消息:return:'''join_group_msg = 'type@=joingroup/rid@=%s/gid@=1/' % (self.__room_id)msg_bytes = self.__msg_handler.dy_encode(join_group_msg)self.__client.send(msg_bytes)# 发送登录请求消息def login(self):'''登陆:return:'''login_msg = 'type@=loginreq/roomid@=%s/dfl@=sn@AA=105@ASss@AA=1/' \'username@=%s/uid@=%s/ver@=0610/aver@=218101901/ct@=0/.'%(self.__room_id,'99047358','99047358')msg_bytes = self.__msg_handler.dy_encode(login_msg)self.__client.send(msg_bytes)def __start_heartbeat(self):self.__heartbeat_thread = threading.Thread(target=self.__heartbeat)self.__heartbeat_thread.start()def __heartbeat(self):heartbeat_msg = 'type@=mrkl/'heartbeat_msg_byte = self.__msg_handler.dy_encode(heartbeat_msg)while True:self.__client.send(heartbeat_msg_byte)for i in range(90):time.sleep(0.5)if not self.__keep_HeartBeat:returndef __prepare(self):self.login()# 登录后发送入组消息self.join_group()self.__start_heartbeat()def __receive_msg(self, msg):'''处理收到的信息:param msg::return:'''chat_messages =self.__msg_handler.get_chat_messages(msg)for message in chat_messages:print(f"{message['nn']}:{message['txt']}")# 将字节流转化为字符串，忽略无法解码的错误（即斗鱼协议中的头部尾部）#print(message.decode(encoding='utf-8', errors='ignore'))class DyDanmuWebSocketClient:def __init__(self,on_open,on_message,on_close):self.__url ='wss://:8506/'self.__websocket = websocket.WebSocketApp(self.__url,on_open=on_open,on_message=on_message,on_error=self.__on_error,on_close=on_close)def start(self):self.__websocket.run_forever()def stop(self):self.__websocket.close()def send(self,msg):self.__websocket.send(msg)def __on_error(self,error):print(error)roomid = "666743"dy_barrage_crawler = DyDanmuCrawler(roomid)dy_barrage_crawler.start()
