bilibili 弹幕爬取及分析

一、项目概述

本项目旨在分析与 2024 年巴黎奥运会使用 AI 技术相关的哔哩哔哩视频弹幕。

二、依赖安装

pip install -r requirements.txt

三、项目结构

1. 业务逻辑

2. 代码的设计过程

本项目共设计四个类，分别是 bilibili 爬虫类（class BilibiliSpider）、数据分析类（class DataAnalysis）及两个单元测试类（class TestBilibiliSpider，class TestDataAnalysis）

① bilibili 爬虫类（class BilibiliSpider）

Bilibili 爬虫类 主要功能：

1. 通过关键词获取搜索结果：get_search_result
2. 通过 aid 获取 cid： get_cid
3. 通过 cid 获取视频弹幕： get_comments_by_cid
4. 通过 aid 获取视频弹幕： get_comments_by_aid

🅐 get_request 方法

def get_request(self, url: str, params=None, max_retries=5) -> requests.Response | None:
        """
        用于发送 Get 请求

        :param url: 请求路径
        :param params: 请求参数，默认为 None
        :param max_retries: 最大尝试次数，默认为 5 次
        :return: <class 'request.Response'>
        """
        for _ in range(max_retries):
            try:
                response = self.session.get(url, params=params)
                response.raise_for_status()
                response.encoding = "utf-8"
                return response
            except Exception as e:
                self.__logger("Error", f"Sending Request To {url}", str(e))
                time.sleep(1)
        return None

• 通用的 GET 请求方法，支持：

  ① 自动重试（默认为 5 次）；

  ② 自动设置响应编码为UTF-8；

  ③ 异常处理和日志记录

🅑 get_search_result 方法

def get_search_result(self, keyword: str, page: int, page_size: int) -> list | None:
        """
        通过关键词获取搜索结果

        :param keyword: 关键词
        :param page: 页码
        :param page_size: 分页大小
        :return: 搜索结果视频的 aid 列表
        """
        params = {
            "search_type": "video",
            "page": page,
            "page_size": page_size,
            "keyword": keyword,
        }
        url = "https://api.bilibili.com/x/web-interface/search/type"
        response = self.get_request(url, params=params)
        if response and response.status_code == 200 and response.json()["code"] == 0:
            response = response.json()
            self.__logger("Success", "GET Search Result",
                          "code: {}, message: {}, page: {}"
                          .format(response["code"], response["message"], response["data"]["page"]))
            result = [item['aid'] for item in response["data"]["result"]]
            return result
        else:
            self.__logger("Failed", "GET Search Result", str(response))
            return None

• 通过关键词获取搜索结果：

  ① 通过关键词、分页号和分页大小进行搜索；

  ② 返回搜索结果中视频的 aid（视频唯一标识符）列表 ；

  ③ 异常处理和日志记录

🅒 get_cid_by_aid 方法

def get_cid_by_aid(self, aid: int) -> int | None:
        """
        通过 aid 获取 cid

        :param aid: 视频的 aid
        :return 
        """
        url = f"https://api.bilibili.com/x/player/pagelist?aid={aid}"
        response = self.get_request(url)
        if response and response.status_code == 200 and response.json()["code"] == 0:
            cid = response.json()["data"][0]["cid"]
            self.__logger("Success", "GET CID", cid)
            return cid
        else:
            self.__logger("Failed", "GET CID", str(response))
            return None

• 获取视频的 cid：① cid 对于获取弹幕数据是必需的;

  ② 可以通过 aid 获取 cid；

  ③ 返回 cid；

  ④ 异常处理和日志记录

🅓 get_comments_by_cid 方法

def get_comments_by_cid(self, cid: int) -> list | None:
    """
    通过 cid 获取视频弹幕

    :param cid: 视频的 cid
    :return 视频的 aid
    """
    url = f"https://comment.bilibili.com/{cid}.xml"
    response = self.get_request(url)
    if response and response.status_code == 200:
        comments = re.findall("<d p=.+?>(.+?)</d>", response.text)
        self.__logger("Success", "GET Comments", f"cid: {cid}")
        return comments
    else:
        self.__logger("Failed", "GET Comments", f"cid: {cid}")
        return None

• 通过 cid 获取弹幕数据：

  ① bilibili 接口返回格式为 XML;

  ② 通过正则匹配解析XML，提取弹幕内容；

  ③ 返回弹幕数据列表；

  ④ 异常处理和日志记录

🅔 get_comments_by_aid 方法

def get_comments_by_aid(self, aid: int) -> list | None:
    """
    通过 aid 获取视频弹幕

    :param aid: 视频的 aid
    :return 无
    """
    cid = self.get_cid_by_aid(aid)
    if cid:
        return self.get_comments_by_cid(cid)
    return None

• 通过 aid 获取弹幕数据：

  ① 对 get_cid_by_aid 和 get_comments_by_cid 进行封装，方便后续的调用；

  ② 先通过aid获取cid，然后使用cid获取弹幕；

  ③ 返回弹幕数据列表；

  ④ 异常处理和日志记录

② 数据分析类（class DataAnalysis）

数据分析类 主要功能：

1. 统计弹幕中关键词出现的次数
2. 生成词云图

🅐 count_keywords 方法

@staticmethod
def count_keywords(data: list, keywords: list) -> list:
    """
    统计弹幕中关键词出现的次数

    :param data: 弹幕数据
    :param keywords: 关键词列表
    :return: 各关键词出现的次数列表
    """
    # 将所有关键词转换为小写，并创建一个模式
    keywords_lower = [keyword.lower() for keyword in keywords]
    pattern = re.compile('|'.join(re.escape(keyword) for keyword in keywords_lower))

    all_comments = ' '.join(temp.lower() for temp in data)  # 将所有评论拼接成一个大字符串
    matches = pattern.findall(all_comments)  # 使用正则表达式匹配所有关键词
    counter = Counter(matches)  # 使用 Counter 计数
    return [[keyword, counter[keyword.lower()]] for keyword in keywords]

• 统计弹幕中关键词出现的次数：

  ① 将所有关键词和弹幕转换为小写，以进行大小写不敏感的匹配；

  ②将所有关键词和弹幕数据拼接成一个字符串，以减少正则匹配的次数；

  ③ 使用Counter类计算每个关键词的出现次数；

  ④ 返回一个列表，其中包含每个关键词及其出现次数

🅑 make_wordcloud 方法

@staticmethod
def make_wordcloud(background, data: list, stopwords: list, output_path: str) -> None:
    """
    生成词云图

    :param background: 背景图片路径
    :param data: 弹幕数据
    :param stopwords: 停用词列表
    :param output_path: 词云图保存路径
    :return: 无
    """
    txt = ''.join(data)  # 合并所有评论为一个字符串
    seg_list = jieba.lcut(txt)  # 使用 jieba 分词
    txt = ' '.join(seg_list)  # 将分词结果合并成一个字符串
    mask = np.array(Image.open(background))  # 导入背景图片
    wordcloud = WordCloud(
        max_words=256,
        mask=mask,
        max_font_size=120,
        font_path='assets/微软雅黑.ttf',
        width=1024,
        height=1024,
        stopwords=set(stopwords)
    ).generate(txt)
    # 保存生成的词云图片
    wordcloud.to_file(output_path)

• 生成词云图:

  ① 将所有弹幕数据合并为一个大字符串，使用jieba库对合并后的文本进行分词，然后将分词结果重新组合为一个字符串，词与词之间用空格分隔；

  ② 使用 WordCloud 生成词云图；

  ③ 无返回值，但会在指定路径 output_file 生成一个词云图片文件

③ 两个单元测试类（class TestBilibiliSpider，class TestDataAnalysis）

• TestBilibiliSpider：完成对 BilibiliSpider 类的核心功能的测试

• TestDataAnalysis：完成对 DataAnalysis 类的核心功能的测试

3. 相关 Api

① 通过关键词获取搜索结果

https://api.bilibili.com/x/web-interface/search/type

Method: GET

Request Headers

Header	Description	Value
Content-Type	请求体的内容类型	application/json
Cookie	用户身份信息	string
User-Agent	标识发起请求的客户端软件	string

Request Parameters

Parameter	Type	Description	Required
search_type	string	搜索资源的类型	true
page	int	分页页码	true
page_size	int	分页大小（有默认值，最大为 50）	false
keyword	string	搜索关键字	true

Response: 仅保留了部分关键信息

{
    "code": 0,
    "message": "0",
	......
    "result": [
        {
            "type": "video",
            "id": 759935456,
			......
            "aid": 759935456,
            "bvid": "BV1D64y1q7ST",
			......
        },
		......
    }
}

② 通过 aid 获取 cid

https://api.bilibili.com/x/player/pagelist

Method: GET

Request Headers

Header	Description	Value
Content-Type	请求体的内容类型	application/json
Cookie	用户身份信息	string
User-Agent	标识发起请求的客户端软件	string

Request Parameters

Parameter	Type	Description	Required
aid	int	视频唯一标识符	true

Response: 仅保留了部分关键信息

{
  "code": 0,
  "message": "0",
  "ttl": 1,
  "data": [
      {
        "cid": 25640897684,
         ......
      }，
      ......
    }
  ]
}

③ 通过 cid 获取弹幕数据

https://comment.bilibili.com/{cid}.xml

Method: GET

Request Parameters: 直接将 cid 拼接到接口上即可

Response: 仅保留了部分关键信息

<i>
    ......
    <d p="78.43700,1,25,16777215,1723488980,0,90ab8a17,1647503990210351104,10">女王驾到</d>
    <d p="66.95800,1,25,16777215,1723470232,0,1f2ebe02,1647346721778676224,10">到陈梦这里欢呼小了好多</d>
    ......
</i>

四、结果展示

① 统计AI技术应用方面的每种弹幕数量，并输出数量排名前8的弹幕

import re
from collections import Counter

设计思路：

count_keywords 方法中：

• 将所有关键词转换为小写，以进行大小写不敏感的匹配

• 使用正则表达式创建一个模式，该模式可以匹配任何给定的关键词

• 将所有弹幕文本合并为一个大字符串，并转换为小写

• 使用正则表达式在合并后的文本中查找所有关键词

• 使用Counter类计算每个关键词的出现次数

• 返回一个列表，其中包含每个关键词及其出现次数

主函数中：

• 接收 count_keywords 返回的结果
• 使用 sorted(result, key=lambda x: x[1], reverse=True) 对接收到的结果进行排序
• 打印排名数量在前8的弹幕

② 将统计的数据利用编程工具或开发包自动写入Excel表中

在 ① 的基础上引入

import pandas as pd

使用以下代码将统计结果保存到 Excel 文件

# 将统计结果保存到 Excel 文件
df = pd.DataFrame(result, columns=['Keyword', 'Count']).sort_values(by='Count', ascending=False)
df.to_excel('output/counter.xlsx', index=False)

③ 制作词云图

import jieba
import numpy as np
from PIL import Image
from wordcloud import WordCloud

设计思路：

• 将所有弹幕文本合并为一个大字符串

• 使用jieba库对合并后的文本进行分词

• 将分词结果重新组合为一个字符串，词与词之间用空格分隔

• 使用 numpy 和 PIL.Image 加载蒙版图（使用小熊猫作为原始蒙版图）

• 创建 WordCloud 对象，设置各种参数（如最大词数、字体、大小等）

• 使用处理后的文本生成词云

• 将生成的词云图片保存到指定路径

• 原始蒙版图：

• 处理后蒙版图：

• 词云图：

④ 附加题展示

实现思路：

• 从API获取2024年巴黎奥运会的奖牌数据

• 将数据处理并嵌入到预设的HTML模板中

• 生成一个交互式的网页，使用ECharts库来可视化奖牌数据

• 自动在默认浏览器中打开生成的网页

创新点：

• 交互式可视化：使用ECharts创建动态饼图，允许用户切换查看不同类型的奖牌分布，允许排除任意国家
• 自动化流程：从数据获取到可视化展示的整个过程都是自动化的

解决的问题：

• 数据可视化：将复杂的数字数据转化为直观的图形，便于理解和分析
• 更加友好的用户界面