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 (cuifulai@mail.hfut.edu.cn)
# @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 (cuifulai@mail.hfut.edu.cn)
# @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("&", "&").replace("<", "<").replace(">", ">")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 (cuifulai@mail.hfut.edu.cn)
# @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()