Python：抓取 Bilibili（B站）评论、弹幕、字幕等

        个人学习需求，需要获取一些 UGC（user 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 asyncio
from 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 loopasyncio.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 ([e-mail])
# @Time: 2024/9/15 15:22
from 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 = XXX
BILI_JCT = XXX
BUVID3 = XXX
DEDEUSERID = 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 ([e-mail])
# @Time: 2024/7/12 15:14
import jsonfrom time import sleep
from random import randint
from tqdm import trange, tqdmfrom bilibili_api import comment
from bilibili_api import Credential
from bilibili_api import video
from bilibili_api import sync
from bilibili_api import Danmaku
from 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 = credentialself.bv_id = bv_idself.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 = 0comments: 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 = commentswhile True:sleep(randint(1, 3))page_index += 1num += comments['page']['size']if num >= comments['page']['count']:breakcomments: 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_commentsdef 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'] = repliesreturn upperelse:return Nonedef 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).textresponse = json.loads(response)sleep(randint(1, 3))replies.extend(response['data']['replies'])return repliesdef 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 xmlsdef 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))@staticmethoddef check_ad(upper):if upper['top']:if 'https://b23.tv/mall' in str(upper['top']['content']['jump_url']):return Trueelse: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 = 0comments: 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 = 0comments: 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 = commentswhile True:sleep(randint(1, 3))page_index += 1num += comments['page']['size']if num >= comments['page']['count']:breakcomments: 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).textresponse = 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 ([e-mail])
# @Time: 2024/7/12 15:19
import configparserfrom bilibili_api import Credential
from 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()