Python爬虫抓取Bilibili弹幕并生成词云

小白学大数据

发布于 2025-05-10 00:15:05

18700

代码可运行

文章被收录于专栏：python进阶学习python进阶学习

运行总次数：0

代码可运行

1. 引言

Bilibili（B站）是国内知名的视频分享平台，拥有海量的弹幕数据。弹幕是B站的核心特色之一，用户通过弹幕进行实时互动，这些数据对于分析视频热度、用户情感倾向等具有重要价值。

本文将介绍如何利用Python爬虫技术抓取Bilibili视频的弹幕数据，并使用WordCloud库生成词云，直观展示弹幕中的高频词汇。

2. 技术栈

Python：主编程语言
Requests：HTTP请求库，用于获取网页数据
BeautifulSoup / lxml：HTML/XML解析库
re（正则表达式）：提取弹幕数据
WordCloud / jieba：生成词云并进行中文分词
Matplotlib / PIL：可视化展示

3. 分析B站弹幕数据来源

B站的弹幕数据通常存储在XML文件中，每个视频对应一个弹幕文件（**cid**决定）。我们需要：

获取视频的**cid**（弹幕ID）
请求弹幕API（如 **https://comment.bilibili.com/{cid}.xml**）
解析XML数据，提取弹幕文本

3.1 获取视频的`cid`

B站的视频页面（如 **https://www.bilibili.com/video/BV1xxxxxx**）中，**cid**通常可以通过以下方式获取：

解析网页源码，查找 **window.__playinfo__** 或 **cid** 相关字段
调用B站API（如 **https://api.bilibili.com/x/web-interface/view?bvid=BV1xxxxxx**）

本文采用 API方式 获取 **cid**，更加稳定。

4. 代码实现

4.1 安装依赖

4.2 获取视频`cid`

import requests

def get_cid(bvid):
    """通过B站API获取视频的cid"""
    url = f"https://api.bilibili.com/x/web-interface/view?bvid={bvid}"
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
    }
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
        data = response.json()
        return data["data"]["cid"]
    else:
        raise Exception("Failed to fetch cid")

# 示例：获取视频 BV1GJ411x7h7 的 cid
bvid = "BV1GJ411x7h7"  # 替换为目标视频的BV号
cid = get_cid(bvid)
print(f"视频的cid: {cid}")

4.3 抓取弹幕数据

B站的弹幕文件通常存储在 **https://comment.bilibili.com/{cid}.xml**，我们需要解析XML并提取弹幕文本。

from bs4 import BeautifulSoup

def fetch_danmaku(cid):
    """获取弹幕XML并解析"""
    url = f"https://comment.bilibili.com/{cid}.xml"
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"
    }
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
        soup = BeautifulSoup(response.content, "lxml")
        danmaku_list = [d.text for d in soup.find_all("d")]
        return danmaku_list
    else:
        raise Exception("Failed to fetch danmaku")

# 获取弹幕
danmaku_list = fetch_danmaku(cid)
print(f"共获取 {len(danmaku_list)} 条弹幕")

4.4 数据清洗（可选）

弹幕可能包含无意义的符号、表情等，可以使用正则表达式过滤：

import re

def clean_text(text):
    """清洗弹幕文本"""
    # 去除特殊符号、空格、换行等
    text = re.sub(r'[^\w\s]', '', text)  # 去除非字母数字汉字
    text = re.sub(r'\s+', ' ', text)     # 合并多个空格
    return text.strip()

cleaned_danmaku = [clean_text(d) for d in danmaku_list]

4.5 生成词云

使用 **jieba** 进行中文分词，并用 **WordCloud** 生成词云：

from wordcloud import WordCloud
import jieba
import matplotlib.pyplot as plt
from PIL import Image
import numpy as np

def generate_wordcloud(text_list, output_path="wordcloud.png"):
    """生成词云"""
    # 合并所有弹幕
    text = " ".join(text_list)
    
    # 使用 jieba 分词
    words = " ".join(jieba.cut(text))
    
    # 设置词云参数
    wc = WordCloud(
        font_path="msyh.ttc",  # 支持中文的字体（Windows可用）
        width=800,
        height=600,
        background_color="white",
        max_words=200,
        collocations=False,  # 避免重复词
    )
    
    # 生成词云
    wc.generate(words)
    
    # 保存词云图片
    wc.to_file(output_path)
    print(f"词云已生成: {output_path}")
    
    # 显示词云
    plt.imshow(wc, interpolation="bilinear")
    plt.axis("off")
    plt.show()

# 生成词云
generate_wordcloud(cleaned_danmaku)

4.6 完整代码整合

import requests
from bs4 import BeautifulSoup
import re
from wordcloud import WordCloud
import jieba
import matplotlib.pyplot as plt

# 代理配置
proxyHost = "www.16yun.cn"
proxyPort = "5445"
proxyUser = "16QMSOML"
proxyPass = "280651"

# 代理格式整理
proxyMeta = f"http://{proxyUser}:{proxyPass}@{proxyHost}:{proxyPort}"
proxies = {
    "http": proxyMeta,
    "https": proxyMeta,
}

def get_cid(bvid):
    """获取视频cid"""
    url = f"https://api.bilibili.com/x/web-interface/view?bvid={bvid}"
    headers = {"User-Agent": "Mozilla/5.0"}
    try:
        # 添加 proxies 参数
        response = requests.get(url, headers=headers, proxies=proxies, timeout=10)
        if response.status_code == 200:
            return response.json()["data"]["cid"]
        else:
            raise Exception(f"API请求失败，状态码：{response.status_code}")
    except Exception as e:
        raise Exception(f"获取cid时出错：{str(e)}")

def fetch_danmaku(cid):
    """获取弹幕XML并解析"""
    url = f"https://comment.bilibili.com/{cid}.xml"
    headers = {"User-Agent": "Mozilla/5.0"}
    try:
        # 添加 proxies 参数
        response = requests.get(url, headers=headers, proxies=proxies, timeout=10)
        if response.status_code == 200:
            soup = BeautifulSoup(response.content, "lxml")
            return [d.text for d in soup.find_all("d")]
        else:
            raise Exception(f"弹幕请求失败，状态码：{response.status_code}")
    except Exception as e:
        raise Exception(f"获取弹幕时出错：{str(e)}")

def clean_text(text):
    """清洗弹幕文本"""
    text = re.sub(r'[^\w\s]', '', text)
    text = re.sub(r'\s+', ' ', text)
    return text.strip()

def generate_wordcloud(text_list, output_path="wordcloud.png"):
    """生成词云"""
    text = " ".join(text_list)
    words = " ".join(jieba.cut(text))
    
    wc = WordCloud(
        font_path="msyh.ttc",
        width=800,
        height=600,
        background_color="white",
        max_words=200,
        collocations=False,
    )
    wc.generate(words)
    wc.to_file(output_path)
    
    plt.imshow(wc, interpolation="bilinear")
    plt.axis("off")
    plt.show()

if __name__ == "__main__":
    try:
        bvid = "BV1GJ411x7h7"  # 替换为目标视频BV号
        cid = get_cid(bvid)
        print(f"成功获取视频CID: {cid}")
        danmaku_list = fetch_danmaku(cid)
        print(f"共获取 {len(danmaku_list)} 条弹幕")
        cleaned_danmaku = [clean_text(d) for d in danmaku_list]
        generate_wordcloud(cleaned_danmaku)
    except Exception as e:
        print(f"程序运行出错: {str(e)}")