Python Selenium简单爬取虎牙直播弹幕（仅学习）

2024-07-03 04:30| 来源: 网络整理| 查看: 265

目录前言一、爬虫思路二、使用步骤1.引入库2.通过selenium进入直播间3.分析HTML4.实现伪监听总结

前言

学习python的第二周，在此记录学习进程：

这次有了需求，由于要做主播的视频剪辑，但是虎牙提供的弹幕热力流在长时间轴中极不敏感，很难快速在6个小时的录播中找到热点，因此打算直接使用爬虫爬取直播弹幕。

本文内容仅用于学习，请勿商用

一、爬虫思路

直播界面的弹幕和礼物都不需要登录态，因此不需要借助cookie；但是直播本身是使用了socket，而且初步研究之后觉得使用socket破解加密实在超出水平太多，因此打算使用神器selenium。

二、使用步骤 1.引入库

代码如下（示例）：

这次的库除了基本库套餐+selenium外，加入了signal，用于监听中断信号。

import csv,time,sys,signal from selenium import webdriver from selenium.webdriver.chrome.options import Options 2.通过selenium进入直播间

安装webdriver的具体事宜有很多文章可参考，这里就不赘述了；

代码如下（示例）：

一句话打开浏览器：

driver = webdriver.Chrome()

组装URL，打开网页：

url ='https://www.huya.com/'+self.target driver.get(url) 3.分析HTML

到了直播间页面，首先找到我们想要的东西：弹幕窗口在这里插入图片描述然后F12，分析HTML结构：显然，弹幕（聊天）窗口在一个id=‘chat-room__list’的ul标签中，里面有li标签装起来的每条聊天（class=‘J_msg’）。

chatRoomList = driver.find_element_by_id("chat-room__list") chatMsgs = chatRoomList.find_elements_by_class_name("J_msg")

J_msg有很多条，先观察一下大致的长相：

礼物

消息

其他的都不需要收集，忽略就好。

for chatMsg in chatMsgs: try: content = {} #初始化弹幕内容字典 #尝试是否为礼物弹幕 try: hSend = chatMsg.find_element_by_class_name("tit-h-send") content['username'] = hSend.find_elements_by_class_name("cont-item")[0].text content['gift'] = hSend.find_element_by_class_name("send-gift").find_element_by_tag_name("img").get_attribute("alt") content['num'] = hSend.find_elements_by_class_name("cont-item")[3].text except: pass #尝试是否为消息弹幕 try: mSend = chatMsg.find_element_by_class_name("msg-normal") content["username"] = mSend.find_element_by_class_name("J_userMenu").text content["msg"] = mSend.find_element_by_class_name("msg").text except: pass #存入弹幕列表 #... except: continue

刚学try-except，所以写了好几个想要的都找到了。跑起来看看效果：

在这里插入图片描述效果不错，空字典是因为存储的方法没有去空，导致进入直播间的J_msg也被当成消息存了。稍微修改一下存储方法：

def SaveToBarrageList(self,content):#弹幕列表存储 if not content: #去空 pass else: content['time'] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())

再加了一条时间，这样可以大致知道这些弹幕出现的时间（不精确）

4.实现伪监听

由于直播的内容都是通过socket实时传输的，而我们没办法通过socket去监听响应，只能高频获取HTML来实现伪监听了——这种形式并没有发送请求，因此也不会因为高频访问被封号或禁止。

while True: #无限循环，伪监听 time.sleep(1) #等待1秒加载 #...

用一个非常简单的无限循环将刚刚的selenium部分框进来，这样脚本就会不停地去重复刚刚的动作，采集HTML里的J_msg，正好配合J_msg的更新。

但是这样会导致另一个问题，也就是J_msg的重复读取：因此我们再观察J_msg的标签，可以看到J_msg的标签带有属性data-id 显然这个id不重复，因此可以将data-id作为弹幕的id来导入：

dataId = chatMsg.get_attribute('data-id') #每条弹幕都有独立data-id self.SaveToBarrageList(dataId,content)

把data-id作为key存入字典中，利用字典去重：

if dataId in self.barrageList or not content: #去重、去空 pass

跑起来看看：在这里插入图片描述

这样就可以得到一个完美的弹幕字典了！这里再加一个监听ctrl+C信号的方法，这样录完弹幕就可以直接ctrl+C退出了：

def QuitAndSave(signum, frame):#监听退出信号 print ('catched singal: %d' % signum) hyObj.SaveToCSV('test',['username','time','msg','gift','num'],hyObj.barrageList.values()) sys.exit(0)

看看最终导出到csv的结果：在这里插入图片描述大功告成！

总结

以下为完整代码：

#!/usr/bin/env python3 # coding=utf-8 # author:sakuyo #---------------------------------- import csv,time,sys,signal from selenium import webdriver from selenium.webdriver.chrome.options import Options class Huya(object): def SaveToCSV(self,fileName,headers,contents): titles = headers data = contents #csv用utf-8-sig来保存 with open(fileName+'.csv','a',newline='',encoding='utf-8-sig') as f: writer = csv.DictWriter(f,fieldnames=titles) writer.writeheader() writer.writerows(data) print('写入完成！') class HuyaLive(Huya): def __init__(self,target):#初始化输入值target为直播间ID self.target = target self.barrageList = {} def Connect(self):#连接直播间 chrome_options = Options() # 使用headless无界面浏览器模式 chrome_options.add_argument('--headless') #增加无界面选项 chrome_options.add_experimental_option("detach", True) driver = webdriver.Chrome(options=chrome_options) url ='https://www.huya.com/'+self.target driver.get(url) time.sleep(5) while True: #无限循环，伪监听 time.sleep(1) #等待1秒加载 chatRoomList = driver.find_element_by_id("chat-room__list") chatMsgs = chatRoomList.find_elements_by_class_name("J_msg") #定位弹幕div，逐条解析 for chatMsg in chatMsgs: try: dataId = chatMsg.get_attribute('data-id') #每条弹幕都有独立data-id content = {} #初始化弹幕内容字典 #尝试是否为礼物弹幕 try: hSend = chatMsg.find_element_by_class_name("tit-h-send") content['username'] = hSend.find_elements_by_class_name("cont-item")[0].text content['gift'] = hSend.find_element_by_class_name("send-gift").find_element_by_tag_name("img").get_attribute("alt") content['num'] = hSend.find_elements_by_class_name("cont-item")[3].text except: pass #尝试是否为消息弹幕 try: mSend = chatMsg.find_element_by_class_name("msg-normal") content["username"] = mSend.find_element_by_class_name("J_userMenu").text content["msg"] = mSend.find_element_by_class_name("msg").text except: pass #存入弹幕列表 self.SaveToBarrageList(dataId,content) except: continue def SaveToBarrageList(self,dataId,content):#弹幕列表存储 if dataId in self.barrageList or not content: #去重、去空 pass else: content['time'] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) self.barrageList[dataId] = content print(dataId,content) def QuitAndSave(signum, frame):#监听退出信号 print ('catched singal: %d' % signum) hyObj.SaveToCSV('test',['username','time','msg','gift','num'],hyObj.barrageList.values()) sys.exit(0) if __name__ == '__main__':#执行层 #信号监听 signal.signal(signal.SIGTERM, QuitAndSave) signal.signal(signal.SIGINT, QuitAndSave) hyObj = HuyaLive('chuhe') hyObj.Connect()

【本文地址】

公司简介

联系我们