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 pprint

from bilibili_api import video


async 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:22
from pprint import pprint

from bilibili_api import video, sync


def 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 configparser


def 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 json

from time import sleep
from random import randint
from tqdm import trange, tqdm

from 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 ass

import requests


HEADERS = {
    '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:19
import configparser

from bilibili_api import Credential
from video import Video

import json


def 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


def 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()