个人学习需求,需要获取一些 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 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("&", "&").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 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()