精华内容
下载资源
问答
  • [Python] [Bilibili] B站历史弹幕爬虫

    千次阅读 2020-02-26 16:39:57
    b站历史弹幕爬虫,以b站弹幕最多的天鹅臂视频为例。

    b站弹幕最多的天鹅臂视频为例,可视化效果见哪位艺人的老婆们最有毅力?b站弹幕爬虫及可视化

    1. 最新弹幕

    最新弹幕可以通过NetWork中的list.so...获取,具体操作可参考历史弹幕的爬取。
    在这里插入图片描述
    但是最新弹幕只能最多获取3000条,如果视频的弹幕超过3000条,则需要通过别的方式获取。
    在这里插入图片描述

    2. 历史弹幕

    2.1 探索

    观察弹幕列表可以发现,b站提供了历史弹幕的查看途径。
    在这里插入图片描述
    点击不同的日期,可以发现不断有以history?开头的资源加载进来。
    在这里插入图片描述
    从预览中可以看到,其实就是对应日期最新的3000条弹幕。观察一下url,有关的参数是typeoiddateoid就是视频对应的id,date则是日期,type暂时不知道干哈的。

    https://api.bilibili.com/x/v2/dm/history?type=1&oid=5627945&date=2020-03-02
    

    再把这个网址放到没有登录b站账号的浏览器打开,显示账号未登录,也就是需要登录才能获取数据,可以考虑用模拟登录解决。
    在这里插入图片描述

    2.2 数据获取

    2.2.1 模拟登录

    用的是selenium搭配谷歌浏览器,需要先配置一下和浏览器版本匹配的驱动,可自行百度或参考这里的第五点

    # 导入需要的包
    from bs4 import BeautifulSoup as bs
    import pandas as pd
    import re
    from selenium import webdriver
    import time, datetime
    
    # 模拟登录
    ## 登录界面链接
    url = 'https://passport.bilibili.com/login'
    ## 模拟登录启动
    driver = webdriver.Chrome()
    driver.get(url)
    ## 接下来就是在打开的谷歌浏览器里手动登录一下自己的账号,
    ## 	* 因为b站登录需要验证,这一步自动化的代价比较大,所以手动了。
    

    2.2.2 获取每一天的数据

    探索部分,看看即可:

    # 以2019-01-01为例
    date = '2019-01-01'
    url = 'https://api.bilibili.com/x/v2/dm/history?type=1&oid=5627945&date=%s'%date
    
    # 转入历史弹幕的页面
    driver.get(url)
    
    # 获取数据部分的html
    html = driver.find_element_by_tag_name('i').get_attribute('innerHTML')
    
    # 解析
    soup = bs(html, 'lxml')
    
    # 先康康第一条
    print(soup.find('d'))
    # 弹幕内容
    print(soup.find('d').text)
    # 弹幕其他信息
    print(soup.find('d').get('p'))
    
    # 经过一番探索,以逗号分割,第一个参数是视频时间(秒),第五个参数是时间戳,第七个参数是用户id,其他反正暂时不重要,不探索了
    # 整理成DataFrame
    data = pd.DataFrame([i.get('p').split(',') + [i.text] for i in soup.findAll('d')], \
                        columns=['second', 'b', 'c', 'd', 'timestamp', 'f', 'user', 'h', 'text'])
    data.head(2)
    

    在这里插入图片描述
    封装成函数后:

    def get_per_day(date):
        # url
        url = 'https://api.bilibili.com/x/v2/dm/history?type=1&oid=5627945&date=%s'%date
        # 转入历史弹幕的页面
        driver.get(url)
        # 获取并解析
        soup = bs(driver.find_element_by_tag_name('i').get_attribute('innerHTML'), 'lxml')
        # 整理成DataFrame
        data = pd.DataFrame([i.get('p').split(',') + [i.text] for i in soup.findAll('d')], \
                        columns=['second', 'b', 'c', 'd', 'timestamp', 'f', 'user', 'h', 'text'])
        # 只取有用的列
        data = data[['timestamp', 'text']]
        return data
    

    2.2.3 获取一段时间的数据

    探索部分,看看即可:

    # 设定开始和结束日期
    start = '2019-01-01'
    end = '2019-02-01'
    
    # 创建空DataFrame方便后面放数据
    data = pd.DataFrame()
    
    # 转成日期形式方便加天数
    start = pd.datetime.strptime(start, '%Y-%m-%d')
    end = pd.datetime.strptime(end, '%Y-%m-%d')
    
    # 循环
    while start <= end:
        date = pd.datetime.strftime(start, '%Y-%m-%d')
        print(date,end=', ')
        data = data.append(get_per_day(date))
        start = start + datetime.timedelta(days=1)
    

    封装成函数后:

    def get_period(start, end):
        # 创建空DataFrame方便后面放数据
        data = pd.DataFrame()
    
        # 转成日期形式方便加天数
        start = pd.datetime.strptime(start, '%Y-%m-%d')
        end = pd.datetime.strptime(end, '%Y-%m-%d')
    
        # 循环
        while start <= end:
            date = pd.datetime.strftime(start, '%Y-%m-%d')
            print(date,end=', ')
            data = data.append(get_per_day(date))
            # 去重(因为如果当天弹幕量不到3000会获取到前一天的弹幕,导致数据大量重复)
            data.drop_duplicates(inplace=True)
            start = start + datetime.timedelta(days=1)
            
        return data
    

    于是就可以通过设定开始和结束的时间爬取一段时间的弹幕。

    data = get_period('2019-01-01', '2019-01-31')
    

    2.2.4 代码整理

    如果不想看上面的部分,直接复制下面这段。

    def get_per_day(date, driver):
    	'''
    	获取一天的数据
    	date: 	str, %Y-%m-%d
    	'''
        # url
        url = 'https://api.bilibili.com/x/v2/dm/history?type=1&oid=5627945&date=%s'%date
        # 转入历史弹幕的页面
        driver.get(url)
        # 获取并解析
        soup = bs(driver.find_element_by_tag_name('i').get_attribute('innerHTML'), 'lxml')
        # 整理成DataFrame
        data = pd.DataFrame([i.get('p').split(',') + [i.text] for i in soup.findAll('d')], \
                        columns=['second', 'b', 'c', 'd', 'timestamp', 'f', 'user', 'h', 'text'])
        # 只取有用的列
        data = data[['timestamp', 'text']]
        return data
    
    def get_period(start, end):
    	'''
    	先登录,然后获取数据。
    	start: 	str, %Y-%m-%d
    	end:	str, %Y-%m-%d
    	'''
    	# 登录
    	url = 'https://passport.bilibili.com/login'
    	driver = webdriver.Chrome()
    	driver.get(url)
    	input('请在浏览器登录账号后回车')
    	
        # 创建空DataFrame方便后面放数据
        data = pd.DataFrame()
    
        # 转成日期形式方便加天数
        start = pd.datetime.strptime(start, '%Y-%m-%d')
        end = pd.datetime.strptime(end, '%Y-%m-%d')
    
        # 循环
        while start <= end:
            date = pd.datetime.strftime(start, '%Y-%m-%d')
            # 时不时打印一下方便追踪进度
            if start.day == 1:
            	print(date,end=', ')
           	# 拼接数据
            data = data.append(get_per_day(date, driver))
            # 去重
            data.drop_duplicates(inplace=True)
            # 下一天
            start = start + datetime.timedelta(days=1)
            
        return data
    
    data = get_period('2019-01-01', '2019-12-31')
    

    2.3 数据处理

    2.3.1 计算弹幕发送时间

    把获取的时间由时间戳专成字符串或整型。

    # 转成字符串
    #data['time'] = data['timestamp'].apply(lambda x: time.strftime("%Y%m%d", time.localtime(int(x))))
    # 转成整型方便后面计算
    data['time'] = data['timestamp'].apply(lambda x: int(time.strftime("%Y%m%d", time.localtime(int(x)))))
    

    2.3.2 从弹幕中提取各种粉丝类型和明星名字

    有很多种方法,规则的学习成本最低所以用规则,具体不讲了,主要就是正则匹配。

    def clean(text):
        for i in ['前面的', '前面', \
                  '前边儿的', '前边儿', '前边', '你的', '讨厌', '感觉'\
                  '我是', '小姐姐', '挥着翅膀', '这位', '这个动作', \
                  '我以后养', '大姨妈', '来姨妈', '姨妈', '第四天', '第五天', '第三天', '第二天', \
                  '坠机', '哈哈', '广场舞大妈', '做到这里', '已经', '安利', '我的妈', '一半', \
                  '没有', '要做', '撞妈', '心疼', '鸡翼', '要成为', '好多', '未来']:
            text = text.replace(i, '')
        
        pattern = r'([^,.,。 !!??()\t0-9吗呜了啊的🟢着呦他说我你是·]{2,})'
        return text, pattern
    
    def find_relation(text, names):
        text, pattern = clean(text)
        for i in names:
            pa = re.compile(pattern+i)
            if len(pa.findall(text)) != 0:
                return pa.findall(text)[0]
        return None
    
    wife = ['的大老婆', '大老婆', \
            '的小老婆', '小老婆', \
            '的老婆', '老婆', \
            '的真老婆', '真老婆', \
            '的脑婆', '脑婆', \
            '的夫人', '夫人', \
            '的妻子', '妻子']
    girlfriend  = ['的女朋友', '女朋友', \
                   '的女票', '女票', \
                   '的对象', '脑婆', \
                   '的女盆友', '女盆友', \
                   '的女友', '女友']
    mon = ['的麻麻', '麻麻', \
          '的妈妈', '妈妈', \
          '的妈', '妈', \
          '的亲妈', '亲妈', \
          '的亲麻', '亲麻', \
          '的母亲', '母亲', \
          '的马麻', '马麻', \
          '的妈粉', '妈粉']
    women = ['的女人', '女人']
    girl = ['的女孩', '女孩']
    daughter = ['的女儿', '女儿']
    sister = ['的姐姐', '姐姐', \
             '的妹妹', '妹妹']
    
    relation_dic = {'wife': '老婆', 'girlfriend': '女朋友', 'mon': '妈妈', \
                   'women': '女人', 'girl': '女孩', 'daughter': '女儿', \
                   'sister': '姐妹'}
    
    data['wife'] = data['text'].apply(lambda x: find_relation(x, wife))
    data['girlfriend'] = data['text'].apply(lambda x: find_relation(x, girlfriend))
    data['mon'] = data['text'].apply(lambda x: find_relation(x, mon))
    data['women'] = data['text'].apply(lambda x: find_relation(x, women))
    data['girl'] = data['text'].apply(lambda x: find_relation(x, girl))
    data['daughter'] = data['text'].apply(lambda x: find_relation(x, daughter))
    data['sister'] = data['text'].apply(lambda x: find_relation(x, sister))
    

    在这里插入图片描述

    2.3.3 汇总每一天

    relation_count = pd.DataFrame()
    for k,v in relation_dic.items():
        df = pd.DataFrame(data.groupby(['time', k]).size())
        df.reset_index(inplace=True)
        df.columns = ['date', 'name', 'value']
        df['type'] = v
        relation_count = relation_count.append(df)
    

    在这里插入图片描述

    2.3.4 计算累计值

    dates = data['time'].unique()
    relation_sum = pd.DataFrame()
    for i in dates:
        df = pd.DataFrame(relation_count[relation_count['date'] <= i].groupby(['name', 'type']).sum()['value'])
        df.reset_index(inplace=True)
        df['date'] = '%s-%s-%s'%(str(i)[:4], str(i)[4:6], str(i)[6:])
        df.sort_values('value', ascending=False, inplace=True)
        relation_sum = relation_sum.append(df.iloc[:30])
    relation_sum['name'] = relation_sum.apply(lambda x: str(x['name'])+'的'+x['type'], axis=1)
    

    在这里插入图片描述

    展开全文
  • 提供python爬取B站历史弹幕详细思路,全部弹幕一览入怀,小伙真的做到了。

    嘻嘻嘻,今天我们来爬小B站的弹幕。


    前言

    在写这篇博客前,我也在csdn上搜索了关于爬取B站弹幕的内容,怎么说呢,关于历史弹幕的爬取实际上是少的。

    这个也好理解,因为有些B站视频它从那个发布到爬取的时候弹幕内容都不到3000条(我印象里弹幕列表里面是显示的最近的3000条),自然也没有爬取历史弹幕的必要了,其实实际上是可以得到的,我不仅要得到,也需要得到实际上不重复的,这也算补了csdn这块爬虫的空缺。


    以下是本篇文章正文内容。

    一、爬取历史弹幕的思路讲解

    1.如何找到弹幕的爬取位置

    这个也是爬取弹幕信息的第一步,我开始觉得是十分简单,但是实际操作起来还是有难度的。我们用B站视频寻味顺德为例。
    在这里插入图片描述
    进入谷歌开发者工具,点击Network,展开弹幕列表,多点击几次查看历史历史弹幕,会出现很多xhr文件来,如果后面内容比较多可以点左上角的禁止符号,面板上的内容就会清理掉。

    在这里插入图片描述

    其实实际上一直点查看历史弹幕,是无法爬取到历史弹幕的,或者说连默认的弹幕列表里面的也爬不到,我们需要的是一个开头是history的文件,该文件的网址后缀是年月日形式的。

    在这里插入图片描述
    找到这个文件只需要先点击查看历史弹幕,再在页面上选择当日的(这个的是默认的最近多少条也是大部分的博主选择的爬取网址,这个网址是没有日期后缀的)或者其他日子的就可以在Network出现相应内容。

    在这里插入图片描述
    进入网址,就会发现这实际上就是一个又一个弹幕,其实到这里如果是爬取默认的弹幕的话,这块网站的部分就可以结束了,直接复制网址,爬取就欧克,不过历史弹幕的才是刚刚开始。

    在这里插入图片描述

    2.如何找到历史弹幕的位置

    我们依旧是在查看历史弹幕这个选项里面下心思,那么网址的后缀是可变的那么直接编写网址的后缀不就好了,实际是方案是可行的,不过有些繁琐了。

    我在网址分析的时候发现B站会把有新弹幕的日期存在一个xhr文件里面,找到这个文件也很简单,只需要在查看历史弹幕那块进行月份的切换,边切边在Network找文件前缀是index,网址的后缀是年月形式的。

    在这里插入图片描述
    访问的结果是这样的,其实在这里大致的思路已经出来了,编写月份的网址,循环遍历月份的网址,提取出来含有新弹幕的日期,再通过这些日期编写当天的历史弹幕的网址,最后循环遍历将弹幕信息全部爬出来,爬取结束。

    在这里插入图片描述

    二、代码讲解部分

    1.引入库

    代码如下:

    import re,requests,time,pandas as pd,json
    

    json库是为了在月份里提取有效日期所使用的,其他的都好理解。

    2.编写自己的headers

    参数还是谷歌开发者工具里找就可以找到,字典形式用键值对连接起来,这里就不打完整代码了,大家自己都用自己的就好,我一般用的是cookie和user-agent这两个参数,当然爬B站弹幕这两个也是必备的。

    h={'cookie':'自己的cookie',
       'user-agent':'自己的user-agent'}
    

    3.编写视频的oid与爬取的开始年份与结束年份

    这个oid实际上在历史月份与历史日期的网址都是有的,如果仅仅是爬一次的话,那略过这里也是可以的,后期把循环编写的部分改掉就好,年份那块也是一样,这只是笔者的习惯罢了。

    oid='7582034'
    begin=2019###在这里输入你需要爬取的最初年份
    end=2020###这里是结束年份
    

    4.编写历史月份的网址后缀

    y_m=[]###这个用来存年月份数据
    for sz1 in range(begin,end+1):
        for sz2 in range(1,13):
            y_m.append(str(sz1)+'-0'+str(sz2))
    

    在这里插入图片描述
    实际上这样编写,会出现如上图所示的情况,我把这样的后缀加到基出网址上访问,得到的不是我们需要的内容,我们需要对这样的进行替换,使它变成有效的后缀。

    for s1 in range(len(y_m)):
        y_m[s1]=re.sub('-010','-10',y_m[s1])
        y_m[s1]=re.sub('-011','-11',y_m[s1])
        y_m[s1]=re.sub('-012','-12',y_m[s1])
    

    上面我编写的代码,出来的效果是非常成功的。
    在这里插入图片描述

    5.爬取有效的历史弹幕网址的日期后缀

    前面的月份的后缀在这里也就用到了https://api.bilibili.com/x/v2/dm/history/index?type=1&oid='+oid+'&month='+月份后缀
    这里需要看一下月份的网址,我在这截了一段。

    {“code”:0,“message”:“0”,“ttl”:1,“data”:[“2020-08-01”,“2020-08-02”,“2020-08-03”,“2020-08-04”,“2020-08-05”,“2020-08-06”,“2020-08-07”,“2020-08-08”,“2020-08-09”,“2020-08-10”,“2020-08-11”,“2020-08-12”,“2020-08-13”,“2020-08-14”,“2020-08-15”,“2020-08-16”,“2020-08-17”,“2020-08-18”,“2020-08-19”,“2020-08-20”,“2020-08-21”,“2020-08-22”,“2020-08-23”,“2020-08-24”,“2020-08-25”,“2020-08-26”,“2020-08-27”,“2020-08-28”,“2020-08-29”,“2020-08-30”,“2020-08-31”]}

    很明显的用json就一部到位json.loads(''.join(文本))['data'],那么这里的代码也迎刃而解了。

    Month_day=[]#这里用来存有效的日期
    for _y_m in y_m:
        Q=requests.get('https://api.bilibili.com/x/v2/dm/history/index?type=1&oid='+oid+'&month='+_y_m,headers=h)
        if re.findall('("data":null)',Q.text)==['"data":null']:
            print(oid+'的'+_y_m+'无效')
        else:
            Month_day=Month_day+json.loads(''.join(Q.text))['data']
            print(oid+'的'+_y_m+'已到位')
        time.sleep(20)
    

    这里的时间休眠设了20秒,实际上不需要那么长,大概来个5秒就好,我设置的长是因为我都是晚上爬,时间多(实际是害怕被B站关小黑屋)。

    这里也做了一个判断,是因为前面编写月份后缀的时候,没有办法排错的,在这里进行了排错,下面的是未存在的月份的页面反馈。

    在这里插入图片描述
    这里也是直接用正则做的判断,看看运行中打印的效果,还是蛮不错的。
    在这里插入图片描述

    5.访问每一个历史弹幕网址,爬取有效信息

    这块就进行到爬取历史弹幕的网址了,看看网站吧。在这里插入图片描述
    有弹幕的部分,这个好理解,可是前面的数字都代表什么呢,这些要去问问度娘。

    在这里插入图片描述
    那自己需要什么呢,自己自行判断,博主这里选择了3个内容进行爬取,都是用的正则,看自己习惯吧,自己喜欢用那个就好。

    dm=[]#弹幕主体
    sjend=[]#弹幕发布时间
    spsjend=[]#弹幕在视频中出现的位置
    for _m_d in Month_day:
        q=requests.get('https://api.bilibili.com/x/v2/dm/history?type=1&oid='+oid+'&date='+_m_d,headers=h)
        txt=q.content.decode("utf-8")
        dm=dm+re.findall('<d p=.*?>(.*?)</d>',txt)
        sj=re.findall('<d p=".*?,.*?,.*?,.*?,(.*?),.*?">.*?</d>',txt)
        for zy in sj:
            zy=time.strftime("%Y-%m-%d %H:%M:%S",time.gmtime(int(zy)))
            sjend.append(zy)
        spsj=re.findall('<d p="(.*?),.*?,.*?,.*?,.*?,.*?">.*?</d>',txt)
        for zy1 in spsj:
            zy1=time.strftime('%M:%S', time.gmtime(int(re.sub('\.\d+','',zy1))))
            spsjend.append(zy1)
        print(oid+'的'+_m_d+"已完成")
        time.sleep(20)
    

    在中间使用到了time.strftime()这个方法,用来做时间的转换,要求都是int格式的,所以做了一些操作。

    6.写字典,构建数据框

    zd={'弹幕发布日期':sjend,'弹幕内容':dm,'视频中弹幕的何时发布':spsjend}
    bg=pd.DataFrame(zd)
    

    第五步运行完,爬取的内容就算结束了,这里是运行了10来页爬到的内容,是十分不错的,下面就是要做是处理掉三项都重复的数据。
    在这里插入图片描述

    6.去重数据

    bgz=bg.drop_duplicates(subset=['弹幕发布日期','弹幕内容','视频中弹幕的何时发布'])
    

    这里采取的是按照多列去重,看看去重完的数据。
    在这里插入图片描述
    25w去重变6000多,但的的确确是这样的,不过这只是一个来月的数据,并且不是热播时期,这样的结果已经很满意了。

    在这里插入图片描述
    用excel去重的效果也是一样,证明代码可行,弹幕积少成多嘛。

    总结

    这种对历史弹幕的爬取还是十分容易理解的,不过最后的筛选我考虑的还是少了一些,会不会出现这种情况,在同一秒里同一时刻不同的人发布了相同的弹幕内容。

    实际上,这样的情况我在写的时候是没有考虑到的,解决这种的方法还是有的,就是把去重时所考虑的参数增多(增加其他参数),或者再爬取时不用改为标准时间格式,去重后改为标准时间。这样也可以更加精准一点。

    因为B站历史弹幕的特殊变换,爬取是十分暴力的,不过也得到了可观的效果,思路比较直接。

    第一次编辑时间:2020/09/09

    本次爬取仅作为学习研究使用

    展开全文
  • 前言关于这个小项目的由来。最开始是想要利用b站弹幕进行...开码一丶利用 POST 方式获取 B 直播弹幕参考:【python】b站直播弹幕获取首先,随便打开一个b站的直播页面,按F12打开控制台,点进“网络(Network...

    前言

    关于这个小项目的由来。

    最开始是想要利用b站的弹幕进行一些互动之类的。原本也有想过可以利用现有的弹幕姬做个插件来解决的,但无奈不会C#,所以只能自己研究b站的弹幕协议。

    后来有写过一个C++版本的,不过有一些小问题,这在后文中会提到。

    开码

    一丶利用 POST 方式获取 B 站直播弹幕

    参考:【python】b站直播弹幕获取

    首先,随便打开一个b站的直播页面,按F12打开控制台,点进“网络(Network)”标签,刷新一下,然后审计一下里面的内容,可以找到“gethistory”这个文件里面就是我们要的弹幕了。

    90f045b106cfcebd57ed618d28d232aa

    实际上,仔细观察便不难发现,请求 gethistory 的时候返回的是请求时最近的10条历史弹幕,不过根据这些就可以写出来一个简易的弹幕姬了。具体做法就是每隔一定的时间请求一次,然后与上次的请求做对比。不同的部分就是这段时间新发的弹幕了,这样就可以对弹幕进行一些操作了。

    我们点进“headers”标签:

    9af260662a1e4cf14be28539047b2b3b

    有了这些我们就可以开写一个弹幕姬了。

    虽然headers很乱,不过实际上我们在请求弹幕的时候并不需要这么多headers,具体哪些headers是必要的可以用实验试出来,不过具体过程和结果我就直接略去了。最后的代码可以参考:

    B站直播弹幕爬取

    或我自己写的C++版本:

    【笔记/学习】c++实现b站弹幕姬

    (代码有点长而且不是本文的重点这里就不放了)

    注:之前的时候获取弹幕的URL是:https://api.live.bilibili.com/ajax/msg,不过我写这篇文再去复现的时候发现这个URL已经没了,经过观察发现变成了 https://api.live.bilibili.com/xlive/web-room/v1/dM/gethistory 截至写文时二者都能用。不过这些都不是重点了。

    二丶利用 WebSocket 获取 B 站弹幕

    前文利用 POST 的方式获取B站弹幕。这种方法虽然简单,但也有不方便的地方。我设的时间间隔为3s,但如果3s内发送的弹幕数量超过了10条,这种方法就会丢失一部分的弹幕,而如果简单的减小时间间隔,不仅会占用更多的网络资源,如果太过频繁的话还可能会被封IP。

    而 HTTP 请求的这种缺陷也正好就是 WebSocket 的出现所为了解决的问题。事实上,我们在看B站直播的时候正是通过 WebSocket 的方式与服务器通信的。

    让我们继续打开 F12 :

    94ebb8d91fa935532d34b8ba50d5eb56

    这个sub 就是与弹幕服务器通信的 WebSocket 啦。

    点进 Message 标签,会看见一大堆东西。不过我们并不需要自己去研究这个通信协议,在 Github 上已经有了B站的API可以直接使用。

    弹幕WS协议

    API文档使用 JavaScript 写的,不过这并不妨碍我们移植一个 Python 版本的。

    由 API 可知,我们与服务器进行通信所发送的数据大多是 json 的数据,偶尔还会有 zlib 数据。所以我们自然需要导入这两个包。我们与服务器使用 WebSocket 进行通信,但原生 Python 并不能直接发送 WebSocket,我们自然也不可能使用 socket 去造轮子。不过好在已经有很多好用的 WebSocket 的库可供我们使用了。

    目前常用的 WebSocket 库有: websocket-client, websockets, aiowebsocket 三个。其中 websocket-client 是同步的,因为我们在收弹幕的同时还得要发送心跳包才能不被服务器断开连接,使用异步io会方便一些。所以不用他。另外两个我也都试过。感觉上 aiowebsocket 更稳定一些,所以这里我们使用这个库。

    安装:

    pip install aiowebsocket

    除了aiowebsocket 要安装外,其他的库都是 python 自带的,直接导入就行了。自然不要忘了用了异步操作要加上 asyncio 库哦。

    import asyncio

    import zlib

    from aiowebsocket.converses import AioWebSocket

    import json

    之后我们写好入口函数:

    if __name__ == '__main__':

    remote = 'wss://broadcastlv.chat.bilibili.com:2245/sub'

    try:

    asyncio.get_event_loop().run_until_complete(startup(remote))

    except KeyboardInterrupt as exc:

    pring('Quit.')

    remote 自然就是API中弹幕服务器的地址了。然后是startup()

    roomid = '5322'

    data_raw='000000{headerLen}0010000100000007000000017b22726f6f6d6964223a{roomid}7d'

    data_raw=data_raw.format(headerLen=hex(27+len(roomid))[2:], roomid=''.join(map(lambda x:hex(ord(x))[2:],list(roomid))))

    async def startup(url):

    async with AioWebSocket(url) as aws:

    converse = aws.manipulator

    await converse.send(bytes.fromhex(data_raw))

    tasks=[receDM(converse), sendHeartBeat(converse)]

    await asyncio.wait(tasks)

    在连接到弹幕服务器后必须先发一个数据包写出进入的房间,否则连接会被断开。这里我是直接从浏览器抄的。数据包中的包长度必须要正确,所以这里要计算一下包长度。

    然后这里先把接受弹幕的receDM() 和发送心跳包的 sendHeartBeat 先写好。接下来是这两个函数:

    hb = '00000010001000010000000200000001'

    async def sendHeartBeat(websocket):

    while True:

    await asyncio.sleep(30)

    await websocket.send(bytes.fromhex(hb))

    print('[Notice] Sent HeartBeat.')

    async def receDM(websocket):

    while True:

    recv_text = await websocket.receive()

    printDM(recv_text)

    B站的弹幕服务器是如果70秒没有心跳就断开连接,这里是30s发送一次。因为只需要发一个没有内容的数据包就行了,所以这里也是直接从浏览器抄的。。

    对于接收到的数据包的处理比较复杂,这里我们单独写一个函数来处理它。

    首先,由API我们可以看到每个数据包的头部是怎样的:

    位置

    0-3

    4-5

    6-7

    8-11

    12-15

    16-

    说明

    数据包长度

    数据包头部长度

    协议版本

    操作类型

    数据包头部长度

    数据包内容

    不过这些内容我们并不都需要用到。

    下面上代码,具体说明在注释中写了:

    # 将数据包传入:

    def printDM(data):

    # 获取数据包的长度,版本和操作类型

    packetLen = int(data[:4].hex(),16)

    ver = int(data[6:8].hex(),16)

    op = int(data[8:12].hex(),16)

    # 有的时候可能会两个数据包连在一起发过来,所以利用前面的数据包长度判断,

    if(len(data)>packetLen):

    printDM(data[packetLen:])

    data=data[:packetLen]

    # 有时会发送过来 zlib 压缩的数据包,这个时候要去解压。

    if(ver == 2):

    data = zlib.decompress(data)

    printDM(data)

    print('db3')

    return

    # ver 为1的时候为进入房间后或心跳包服务器的回应。op 为3的时候为房间的人气值。

    if(ver == 1):

    if(op == 3):

    print('[RENQI] {}'.format(int(data[16:].hex(),16)))

    return

    # ver 不为2也不为1目前就只能是0了,也就是普通的 json 数据。

    # op 为5意味着这是通知消息,cmd 基本就那几个了。

    if(op==5):

    try:

    jd = json.loads(data[16:].decode('utf-8', errors='ignore'))

    if(jd['cmd']=='DANMU_MSG'):

    print('[DANMU] ', jd['info'][2][1], ': ', jd['info'][1])

    elif(jd['cmd']=='SEND_GIFT'):

    print('[GITT]',jd['data']['uname'], ' ', jd['data']['action'], ' ', jd['data']['num'], 'x', jd['data']['giftName'])

    elif(jd['cmd']=='LIVE'):

    print('[Notice] LIVE Start!')

    elif(jd['cmd']=='PREPARING'):

    print('[Notice] LIVE Ended!')

    else:

    print('[OTHER] ', jd['cmd'])

    except Exception as e:

    pass

    这样,一个简单的弹幕姬就完成了!

    三丶试试搞一些其他的事情吧!

    至此,我们已经有了一个可以在控制台输出弹幕内容的弹幕姬了。不过,这并不是结束,有了这个我们就可以利用弹幕搞事情了(

    先放个简单的功能:

    把弹幕保存至本地

    先到如下时间:

    import time

    然后我们把前面的print()函数改掉:

    def log(typ, body):

    with open('D:/danmu.txt','a') as fd:

    fd.write(time.strftime("[%H:%M:%S] ", time.localtime()))

    fd.write(body+'\n')

    这样就可以爬取弹幕到本地了。

    利用SAPI朗读弹幕

    利用 SAPI 朗读需要导入相应的包:

    import win32com.client

    然后改写log函数:

    def log(typ, body):

    speak = win32com.client.Dispatch("SAPI.SpVoice")

    #创建发声对象

    speak.Speak(body)

    #使用发生对象读取文字

    with open('D:/danmu.txt','a') as fd:

    fd.write(time.strftime("[%H:%M:%S] ", time.localtime()))

    fd.write(body+'\n')

    来使python自动朗读弹幕。

    一般 Windows 都可以直接使用,不能用的话再上网查吧。。

    利用聊天机器人实现自动聊天

    首先打开b站一个直播间,发条弹幕截下包:

    ca6c37911a73df7ac7d7515fc4f93038

    照着参考,可以大致写出一份发送弹幕的python脚本

    import requests

    import timeform_data = {

    'color': '65532',

    'fontsize': '25',

    'mode': '1',

    'msg': 'test',

    'rnd': int(time.time()),

    'roomid': '1136753',

    'csrf_token': 'cce335cbfa5bfd292a049b813175bd12',

    'csrf': 'cce335cbfa5bfd292a049b813175bd12'

    }

    # 设置cookie值帮助我们在发送弹幕的时候,服务器识别我们的身份

    cookie = { '你的cookie(上图红色部分)' }

    res = requests.post('https://api.live.bilibili.com/msg/send', cookies=cookie, data=form_data)

    print (res.status_code)

    上面的csrf 和 csrf_token 在使用的时候最好也换成自己的。

    然后可以去注册图灵机器人/思知机器人等API(不推荐图灵,之前还好,后来一去看感觉有点贵),申请到 appid。

    具体可以参考我以前写的这篇文章

    最后代码差不多是这样:

    def talk(msg):

    form_data = {

    'color': '65532',

    'fontsize': '25',

    'mode': '1',

    'msg': 'test',

    'rnd': int(time.time()),

    'roomid': roomid,

    'csrf_token': 'cce335cbfa5bfd292a049b813175bd12',

    'csrf': 'cce335cbfa5bfd292a049b813175bd12'

    }

    cookie = { '你的cookie(上图红色部分)' }

    payload = {

    'appid': '你的appid',

    'userid': '1234',

    }

    payload['spoken'] = msg

    res1 = requests.get("https://api.ownthink.com/bot", params=payload)

    form_data['msg'] = res1.json()['data']['info']['text']

    res2 = requests.post('https://api.live.bilibili.com/msg/send', cookies=cookie, data=form_data)

    好啦,去直播间发条弹幕看看效果:

    7109cd7d0ac754a8d02cf4bab28f2277

    注意使用的时候加上限制不要对机器人的回复再去回复就行了

    。。。

    后记

    嗯。。。暂时就先写这么多吧,还有什么要补充的以后再说吧。。

    源码什么的之后再传吧。。

    点赞

    收藏

    分享

    文章举报

    d2c3be6a770e0ea15cedcad39a918348

    e1089cbef1b3fe9d76f1d8e0af658b17.png

    Panedioic

    发布了2 篇原创文章 · 获赞 0 · 访问量 79

    私信

    关注

    展开全文
  • python爬取B站视频历史弹幕演示 演示 1.运行程序,输入Bvid和爬取日期。 2.程序运行完成后会在当前文件夹下生成一个csv格式文件。

    python爬取B站视频历史弹幕

    演示

    1.运行程序,输入Bvid和爬取日期。
    程序运行输入

    2.程序运行完成后会在当前文件夹下生成一个csv格式文件。
    程序运行结果

    展开全文
  • 需要准备的环境:一个B站账号,需要先登录,否则不能查看历史弹幕记录联网的电脑和顺手的浏览器,我用的ChromePython3环境以及request模块,安装使用命令,换源比较快:pip3 install request -i ... 登录后打开需要爬取的视频...
  • 需要准备的环境:一个B站账号,需要先登录,否则不能查看历史弹幕记录联网的电脑和顺手的浏览器,我用的ChromePython3环境以及request模块,安装使用命令,换源比较快:pip3 install request -i ... 登录后打开需要爬取的视频...
  • 原本也有想过可以利用现有的弹幕姬做个插件来解决的,但无奈不会C#,所以只能自己研究b站弹幕协议。 后来有写过一个C++版本的,不过有一些小问题,这在后文中会提到。 开码 一丶利用 POST 方式获取 B 直播弹幕 ...
  • 分析网页视频地址:https://www.bilibili.com/video/BV19E411W7BE本身博主同时也是一名up主,虽然已经断更好久了,但是不妨碍我爬取弹幕信息来分析呀。这次我选取的是自己唯一的爆款视频。就是下面这个。希望大家...
  • 需要先登录,否则不能查看历史记录点击历史弹幕,有两个请求 包括弹幕id,返回的是所有的有弹幕的日期,格式为json 将timestamp提取出来拼接弹幕url,点击指定日期,显示弹幕的时候有dmroll请求,获取弹幕信息 返回xml格式...
  • 前几天我做了B站《后浪》视频的弹幕分析,感兴趣的朋友可以去B站看一下现在我将在制作这个视频背后用到的文本分析相关的关键技术点做一个总结,因为涉及知识点太多,有些点会一笔带过,着重讲一些主要的点。...

空空如也

空空如也

1 2 3 4 5 ... 20
收藏数 2,040
精华内容 816
关键字:

b站历史弹幕