当前位置:首页 » 《关注互联网》 » 正文

Python:抓取 Bilibili(B站)评论、弹幕、字幕等

5 人参与  2024年10月19日 08:41  分类 : 《关注互联网》  评论

点击全文阅读


        个人学习需求,需要获取一些 UGCuser generated content),包括 UP 的内容、弹幕、评论等。于是从 哔哩哔哩 (゜-゜)つロ 干杯~-bilibili 抓取了一些数据,以下内容仅供学习参考。

目录

1. Python 包:bilibili-api

1.1 bilibili-api

1.1.1 安装

1.1.2 示例

1.1.3 Credential

1.1.4 config.ini

2 自定义 Video 类

2.1 bvid

2.2 代码

2.2.1 get_video()

2.2.2 get_info()

2.2.3 get_comments()

2.2.4 get_sub_comments()

2.2.5 get_danmakus()

2.2.6 get_subtitle()

3 示例


1. Python 包:bilibili-api

        API 文档:bilibili-api 开发文档

        B站讲解视频,UP主:w海底捞不动w,【bilibili-api】爬取某个视频的所有评论 | python开发b站常用功能

1.1 bilibili-api

        引用开发文档中的简介。

这是一个用 Python 写的调用 Bilibili 各种 API 的库, 范围涵盖视频、音频、直播、动态、专栏、用户、番剧等。

        这里简单说明一些重要的内容,详细的内容可查看开发文档。

1.1.1 安装
pip3 install bilibili-api-python
1.1.2 示例

        粘贴开发文档中提供的代码。

import asynciofrom pprint import pprintfrom bilibili_api import videoasync def main() -> None:    # 实例化 Video 类    v = video.Video(bvid="BV1uv411q7Mv")    # 获取信息    info = await v.get_info()    # 打印信息    pprint(info)if __name__ == "__main__":    asyncio.get_event_loop().run_until_complete(main())

        注:这里修改 print pprint,目的是更优雅的查看结果。结果如下,部分省略。

DeprecationWarning: There is no current event loop  asyncio.get_event_loop().run_until_complete(main()){'aid': 243922477, 'argue_info': {'argue_link': '', 'argue_msg': '', 'argue_type': 0}, 'bvid': 'BV1uv411q7Mv', 'cid': 214334689, 'copyright': 1, 'ctime': 1595168654, 'desc': '相关游戏:\n'         '----Minecraft、しゅがてん!-sugarfull tempering\n'         '====================\n'         '制作名单:\n'         '----Minecraft 游戏内建筑:-落忆-\n'         '----程序:-落忆-\n'         '----游戏内摄影:-落忆-、Passkou\n'         '----视频后期:Passkou\n'         '----音乐编曲:Passkou\n'...and more

        注意,这里报 DeprecationWarning,可能是开发文档长期未维护的原因吧。正确(不报异常)的代码如下:

# coding=utf-8# @Author: Fulai Cui (cuifulai@mail.hfut.edu.cn)# @Time: 2024/9/15 15:22from pprint import pprintfrom bilibili_api import video, syncdef main() -> None:    # 实例化 Video 类    v = video.Video(bvid="BV1uv411q7Mv")    # 获取信息    info = sync(v.get_info())    # 打印信息    pprint(info)if __name__ == "__main__":    main()
1.1.3 Credential

        正如开发文档所讲,

如何给这个视频点赞?我们需要登录自己的账号。

这里设计是传入一个 Credential 类,获取所需的信息参照:获取 Credential 类所需信息

        bilibili-api 提供了一个 Credential 类,用于处理用户的“专属信息”。

        主要的内容就是开发文档的第二章:获取 Credential 类所需信息。具体内容:

SESSDATA

        用于一般在获取对应用户信息时提供,通常是 GET 操作下提供,此类操作一般不会进行操作,仅读取信息。如获取个人简介、获取个人空间信息等情况下需要提供。

BILI_JCT

        用于进行操作用户数据时提供,通常是 POST 操作下提供,此类操作会修改用户数据。如发送评论、点赞三连、上传视频等等情况下需要提供。

BUVID3 / BUVID4我只找到BUVID3

        设备验证码。通常不需要提供,但如放映室内部分接口需要提供,同时与风控有关。

DEDEUSERID

        通常为用户 UID几乎不需要提供。

AC_TIME_VALUE这个我没找到,没有对我来说没有影响

        在登录时获取,登录状态过期后用于刷新 Cookies,没有此值则只能重新登录,如不需要凭据刷新则不需要提供。

1.1.4 config.ini

        将上诉关键信息保存在 config.ini,以免意外泄露个人账号相关信息。以下为 config.ini 内容。

[Credential]SESSDATA = XXXBILI_JCT = XXXBUVID3 = XXXDEDEUSERID = XXX

        读取 config 可以使用 import configparser 包来完成。        

import configparserdef get_configs():    parser = configparser.RawConfigParser()    parser.read("../config.ini")    configs = {        'SESSDATA': parser.get('Credential', 'SESSDATA'),        'BILI_JCT': parser.get('Credential', 'BILI_JCT'),        'BUVID3': parser.get('Credential', 'BUVID3'),        'DEDEUSERID': parser.get('Credential', 'DEDEUSERID')    }    return configs

        然后, bilibili_api.Credential 类的对象可以通过检索 configs 的值来定义。

configs = get_configs()credential = Credential(    sessdata=configs['SESSDATA'],    bili_jct=configs['BILI_JCT'],    buvid3=configs['BUVID3'],    dedeuserid=configs['DEDEUSERID'])

         需要注意的是,这里的 credential 在后续频繁被使用。

2 自定义 Video 类

2.1 bvid

        BV 号(bvid),比如 URL https://www.bilibili.com/video/BV1Bx4y1s7n3 中,bvid 是连同 BV 在内的 BV1Bx4y1s7n3。获取某视频相关的内容都需要该 BV 号。

2.2 代码

       这里先直接 post 我自定义的 Video 类的 全部代码,后续再详细挑重点进行解释。

# coding=utf-8# @Author: Fulai Cui (cuifulai@mail.hfut.edu.cn)# @Time: 2024/7/12 15:14import jsonfrom time import sleepfrom random import randintfrom tqdm import trange, tqdmfrom bilibili_api import commentfrom bilibili_api import Credentialfrom bilibili_api import videofrom bilibili_api import syncfrom bilibili_api import Danmakufrom bilibili_api import assimport requestsHEADERS = {    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36',    "Referer": "https://www.bilibili.com",}class Video:    def __init__(self, credential: Credential, bv_id: str):        self.credential = credential        self.bv_id = bv_id        self.video: video.Video = self.get_video()        self.aid = self.video.get_aid()        self.info = self.get_info()    def get_video(self):        return video.Video(bvid=self.bv_id, credential=self.credential)    def get_info(self):        return sync(self.video.get_info())    def get_comments(self, page_index: int = 1):        num = 0        comments: dict = sync(            comment.get_comments(                oid=self.aid,                type_=comment.CommentResourceType.VIDEO,                page_index=page_index,                order=comment.OrderType.TIME,                credential=self.credential,            )        )        all_comments = comments        while True:            sleep(randint(1, 3))            page_index += 1            num += comments['page']['size']            if num >= comments['page']['count']:                break            comments: dict = sync(                comment.get_comments(                    oid=self.aid,                    type_=comment.CommentResourceType.VIDEO,                    page_index=page_index,                    order=comment.OrderType.TIME,                    credential=self.credential,                )            )            all_comments['replies'].extend(comments['replies'])        return all_comments    def get_upper(self, page_index: int = 1):        comments: dict = sync(            comment.get_comments(                oid=self.aid,                type_=comment.CommentResourceType.VIDEO,                page_index=page_index,                order=comment.OrderType.TIME,                credential=self.credential,            )        )        upper = comments['upper']        if self.check_ad(upper):            replies = self.get_sub_comments(upper)            upper['replies'] = replies            return upper        else:            return None    def get_sub_comments(self, upper):        replies = []        for pn in trange((upper['top']['rcount'] // 10) + 1, desc=f'Get sub comments of upper 【{self.bv_id}】'):            url = ''.join([                'https://api.bilibili.com/x/v2/reply/reply'                f'?oid={self.aid}'                '&type=1'                f'&root={upper['top']['rpid']}'                '&ps=10'                f'&pn={pn + 1}'                '&web_location=333.788'            ])            response = requests.get(url, headers=HEADERS).text            response = json.loads(response)            sleep(randint(1, 3))            replies.extend(response['data']['replies'])        return replies    def get_danmakus(self):        danmakus: list[Danmaku] = sync(            self.video.get_danmakus(                page_index=0,                date=None,                cid=None,                from_seg=0,                to_seg=self.info['duration'] // 360            )        )        def to_json(danmaku: Danmaku):            txt = danmaku.text.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")            return {                "dm_time": danmaku.dm_time,                "send_time": danmaku.send_time,                "crc32_id": danmaku.crc32_id,                "color": danmaku.color,                "weight": danmaku.weight,                "id_": danmaku.id_,                "id_str": danmaku.id_str,                "action": danmaku.action,                "mode": danmaku.mode,                "font_size": danmaku.font_size,                "is_sub": danmaku.is_sub,                "pool": danmaku.pool,                "attr": danmaku.attr,                "content": txt            }        xmls = {'danmakus': []}        for _ in tqdm(danmakus, desc=f'Get danmakus【{self.bv_id}】'):            xmls['danmakus'].append(to_json(_))        return xmls    def get_subtitle(self, root_dir):        sync(            ass.make_ass_file_subtitle(                obj=self.video,                page_index=0,                cid=self.info['cid'],                out=f"{root_dir}/{self.bv_id}.ass",                lan_name="中文(自动生成)",                lan_code="ai-zh",                credential=self.credential            )        )    @staticmethod    def check_ad(upper):        if upper['top']:            if 'https://b23.tv/mall' in str(upper['top']['content']['jump_url']):                return True        else:            return False
2.2.1 get_video()

        调用 bilibili_api.video.Video 类来获取 video 对象。

def get_video(self):    return video.Video(bvid=self.bv_id, credential=self.credential)

bilibili_api.video.Video
def __init__(self,
             bvid: str | None = None,
             aid: int | None = None,
             credential: Credential | None = None) -> Any
 
Args:

bvid (str | None, optional) : BV 号. bvid 和 aid 必须提供其中之一。  

aid (int | None, optional) : AV 号. bvid 和 aid 必须提供其中之一。  

credential (Credential | None, optional): Credential 类. Defaults to None.

2.2.2 get_info()

        获取视频信息。

def get_info(self):    return sync(self.video.get_info())

        需要注意到,bilibili_api.sync 多次使用,其目的是“同步执行异步函数”。

2.2.3 get_comments()

        获取评论信息。

def get_comments(self, page_index: int = 1):    num = 0    comments: dict = sync(        comment.get_comments(            oid=self.aid,            type_=comment.CommentResourceType.VIDEO,            page_index=page_index,            order=comment.OrderType.TIME,            credential=self.credential,        )    )    all_comments = comments...more code

bilibili_api.comment
async def get_comments(oid: int,
                       type_: CommentResourceType,
                       page_index: int = 1,
                       order: OrderType = OrderType. TIME,
                       credential: Credential | None = None) -> Coroutine[Any, Any, dict]
 
获取资源评论列表。  

第二页以及往后需要提供 `credential` 参数。  

Args:

oid (int) : 资源 ID。  

type_ (CommentsResourceType) : 资源类枚举。  

page_index (int, optional) : 页码. Defaults to 1.  

order (OrderType, optional) : 排序方式枚举. Defaults to OrderType. TIME.  

credential (Credential, optional): 凭据。Defaults to None.  

Returns:

dict: 调用 API 返回的结果

         为了获取全部评论,自然需要递增 page_index,直到全部获取。也就自然而然的有了接着的代码。

def get_comments(self, page_index: int = 1):    num = 0    comments: dict = sync(        comment.get_comments(            oid=self.aid,            type_=comment.CommentResourceType.VIDEO,            page_index=page_index,            order=comment.OrderType.TIME,            credential=self.credential,        )    )    all_comments = comments    while True:        sleep(randint(1, 3))        page_index += 1        num += comments['page']['size']        if num >= comments['page']['count']:            break        comments: dict = sync(            comment.get_comments(                oid=self.aid,                type_=comment.CommentResourceType.VIDEO,                page_index=page_index,                order=comment.OrderType.TIME,                credential=self.credential,            )        )        all_comments['replies'].extend(comments['replies'])    return all_comments
2.2.4 get_sub_comments()

        我这里额外的关注置顶评论的子评论,但是发现 bilibili_api 并没有提供这样的 API 接口。由此定义了一个获取评论的子评论的方法。

        这里采用的是 requests.get() 的方法。

def get_sub_comments(self, upper):    replies = []    for pn in trange((upper['top']['rcount'] // 10) + 1, desc=f'Get sub comments of upper 【{self.bv_id}】'):        url = ''.join([            'https://api.bilibili.com/x/v2/reply/reply'            f'?oid={self.aid}'            '&type=1'            f'&root={upper['top']['rpid']}'            '&ps=10'            f'&pn={pn + 1}'            '&web_location=333.788'        ])        response = requests.get(url, headers=HEADERS).text        response = json.loads(response)        sleep(randint(1, 3))        replies.extend(response['data']['replies'])    return replies

         这样能获取的原因是 Bilibili 对该 api 的反爬比较“松”。

2.2.5 get_danmakus()

        获取弹幕信息。

def get_danmakus(self):    danmakus: list[Danmaku] = sync(        self.video.get_danmakus(            page_index=0,            date=None,            cid=None,            from_seg=0,            to_seg=self.info['duration'] // 360        )    )...more code

bilibili_api.video.Video
async def get_danmakus(self,
                       page_index: int = 0,
                       date: date | None = None,
                       cid: int | None = None,
                       from_seg: int | None = None,
                       to_seg: int | None = None) -> Coroutine[Any, Any, list[Danmaku]]
 
获取弹幕。  

Args:

page_index (int, optional): 分 P 号,从 0 开始。Defaults to None  

date (datetime. Date | None, optional): 指定日期后为获取历史弹幕,精确到年月日。Defaults to None.  

cid (int | None, optional): 分 P 的 ID。Defaults to None  

from_seg (int, optional): 从第几段开始(0 开始编号,None 为从第一段开始,一段 6 分钟). Defaults to None.  

to_seg (int, optional): 到第几段结束(0 开始编号,None 为到最后一段,包含编号的段,一段 6 分钟). Defaults to None.  

注意

- 1. 段数可以使用 `get_danmaku_view()["dm_seg"]["total"]` 查询。

- 2. `from_seg` 和 `to_seg` 仅对 `date == None` 的时候有效果。

- 3. 例:取前 `12` 分钟的弹幕:`from_seg=0, to_seg=1`  

Returns:

List[Danmaku]: Danmaku 类的列表

2.2.6 get_subtitle()

        获取字幕信息。

def get_subtitle(self, root_dir):    sync(        ass.make_ass_file_subtitle(            obj=self.video,            page_index=0,            cid=self.info['cid'],            out=f"{root_dir}/{self.bv_id}.ass",            lan_name="中文(自动生成)",            lan_code="ai-zh",            credential=self.credential        )    )

bilibili_api.ass
async def make_ass_file_subtitle(obj: Video | Episode,
                                 page_index: int | None = 0,
                                 cid: int | None = None,
                                 out: str | None = "test. ass",
                                 lan_name: str | None = "中文(自动生成)",
                                 lan_code: str | None = "ai-zh",
                                 credential: Credential = Credential()) -> Coroutine[Any, Any, None]
 
生成视频字幕文件  

Args:

obj (Union[Video,Episode]): 对象  

page_index (int, optional) : 分 P 索引  

cid (int, optional) : cid  

out (str, optional) : 输出位置. Defaults to "test. ass".  

lan_name (str, optional) : 字幕名,如”中文(自动生成)“,是简介的 subtitle 项的'list'项中的弹幕的'lan_doc'属性。Defaults to "中文(自动生成)".  

lan_code (str, optional) : 字幕语言代码,如 ”中文(自动翻译)” 和 ”中文(自动生成)“ 为 "ai-zh"  

credential (Credential) : Credential 类. 必须在此处或传入的视频 obj 中传入凭据,两者均存在则优先此处

3 示例

        使用代码如下。

# coding=utf-8# @Author: Fulai Cui (cuifulai@mail.hfut.edu.cn)# @Time: 2024/7/12 15:19import configparserfrom bilibili_api import Credentialfrom video import Videoimport jsondef get_configs():    parser = configparser.RawConfigParser()    parser.read("../config.ini")    configs = {        'SESSDATA': parser.get('Credential', 'SESSDATA'),        'BILI_JCT': parser.get('Credential', 'BILI_JCT'),        'BUVID3': parser.get('Credential', 'BUVID3'),        'DEDEUSERID': parser.get('Credential', 'DEDEUSERID')    }    return configsdef json_write(json_filename: str, json_dict: dict):    if json_dict:        with open(json_filename, 'w', encoding='utf-8') as f:            json.dump(json_dict, f, indent=4, ensure_ascii=False)def main():    bv_id = 'BV1iC4y177Sb'    root_dir = '../Data'    configs = get_configs()    credential = Credential(        sessdata=configs['SESSDATA'],        bili_jct=configs['BILI_JCT'],        buvid3=configs['BUVID3'],        dedeuserid=configs['DEDEUSERID']    )    video = Video(credential=credential, bv_id=bv_id)        # comments = video.get_comments()    upper = video.get_upper()    if upper:        danmakus = video.get_danmakus()        json_write(f'{root_dir}/Info/{bv_id}.json', video.info)        # json_write(f'{root_dir}/Comment/{bv_id}.json', comments)        json_write(f'{root_dir}/Upper/{bv_id}.json', upper)        json_write(f'{root_dir}/Danmaku/{bv_id}.json', danmakus)if __name__ == '__main__':    main()


点击全文阅读


本文链接:http://m.zhangshiyu.com/post/173941.html

<< 上一篇 下一篇 >>

  • 评论(0)
  • 赞助本站

◎欢迎参与讨论,请在这里发表您的看法、交流您的观点。

关于我们 | 我要投稿 | 免责申明

Copyright © 2020-2022 ZhangShiYu.com Rights Reserved.豫ICP备2022013469号-1