python

从 B 站弹幕到数据分析:Python 实战指南

2026-06-26 #python#爬虫#数据采集#数据分析#requests

在现代短视频与直播平台中,弹幕不仅是用户互动的工具,更是分析观众兴趣、情绪和行为模式的重要数据源。本文将以 B 站(Bilibili)为例,使用 Python 完整实现从弹幕抓取、数据处理到可视化分析的全流程。


1. 项目概述

本项目目标是抓取指定 B 站视频的所有分 P 弹幕,并进行文本分析与可视化,具体功能包括:

  1. 获取视频的分 P 信息(cid)。
  2. 抓取每个分 P 的弹幕 XML 数据。
  3. 将弹幕保存为 CSV 文件。
  4. 进行文本分析(词频统计、关键词统计)。
  5. 生成可视化图表(时间分布图、词云、弹幕长度分布图等)。

2. 弹幕抓取实现

2.1 配置与请求头

在抓取 B 站弹幕时,需要设置请求头以模拟浏览器访问:

1
2
3
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'
}

同时配置视频 BV 号和输出文件名:

1
2
BVID = "BV19R4y1K7yc"
OUTPUT_CSV = f"公孙田浩弹幕_{BVID}_danmu.csv"

2.2 获取所有分 P 的 cid

B 站视频可能存在多 P,每个分 P 对应一个 cid,通过官方 API 可以获取:

1
2
3
4
5
def fetch_cid_list(bvid):
cid_api = f"https://api.bilibili.com/x/player/pagelist?bvid={bvid}&jsonp=jsonp"
response = requests.get(cid_api, headers=headers, timeout=10)
response.raise_for_status()
return response.json()['data']

企业级提示:异常处理必不可少,网络请求失败或 JSON 解析异常都需要捕获,以保证程序健壮性。


2.3 获取弹幕 XML 数据

通过 cid 可以访问弹幕 XML 文件,并解析每条弹幕属性:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from lxml import etree

def fetch_danmu_for_cid(cid, page_index):
danmu_url = f"https://comment.bilibili.com/{cid}.xml"
xml_response = requests.get(danmu_url, headers=headers, timeout=10)
xml_response.encoding = 'utf-8'
root = etree.fromstring(xml_response.text.encode('utf-8'))
danmus = []
for d in root.xpath("//d"):
p_attr = d.attrib.get("p", "").split(",")
if len(p_attr) < 8: continue
danmus.append({
"page": page_index,
"time_in_video": float(p_attr[0]),
"mode": int(p_attr[1]),
"font_size": int(p_attr[2]),
"color": int(p_attr[3]),
"send_time": p_attr[4],
"danmu_pool": p_attr[5],
"user_hash": p_attr[6],
"danmu_id": p_attr[7],
"content": d.text
})
return danmus

注意:XML 解析错误和非法内容需捕获并记录,保证数据完整性。


2.4 主流程与数据保存

将抓取的弹幕统一保存到 CSV,方便后续分析:

1
2
3
4
5
6
7
8
9
10
import pandas as pd

pages = fetch_cid_list(BVID)
danmus = []
for page in pages:
danmus.extend(fetch_danmu_for_cid(page['cid'], page.get('page', 1)))

df = pd.DataFrame(danmus)
df.to_csv(OUTPUT_CSV, index=False, encoding="utf-8-sig")
print(f"弹幕已保存到 {OUTPUT_CSV}")

3. 弹幕数据分析

3.1 数据清洗与基本信息

使用 pandas 对 CSV 数据进行处理:

1
2
3
df['send_time'] = pd.to_datetime(df['send_time'], unit='s')
df['content'] = df['content'].astype(str)
print(f"总弹幕数量: {len(df)}")

3.2 高频词分析与词云

结合 jieba 分词库,过滤停用词后统计词频,并生成词云:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import jieba
from collections import Counter
from wordcloud import WordCloud
import matplotlib.pyplot as plt

stopwords = {'的','了','在','是','我','有','和','就','不','人','都','一','一个','上','也','很','到','说','要','去','你','会','着','没有','看','好','自己','这'}
all_text = ' '.join(df['content'].tolist())
words = [w for w in jieba.cut(all_text) if len(w) > 1 and w not in stopwords]
word_freq = Counter(words)

wordcloud = WordCloud(width=800, height=600, background_color='white', font_path='simhei.ttf').generate_from_frequencies(word_freq)
plt.imshow(wordcloud, interpolation='bilinear')
plt.axis('off')
plt.show()

企业级建议:词云可视化有助于快速捕捉弹幕热点主题和关键词。


3.3 弹幕时间分布

分析弹幕发送时间,绘制柱状图:

1
2
3
4
5
6
7
df['hour'] = df['send_time'].dt.hour
hourly_counts = df['hour'].value_counts().sort_index()
plt.bar(hourly_counts.index, hourly_counts.values, color='skyblue')
plt.xlabel('小时')
plt.ylabel('弹幕数量')
plt.title('弹幕时间分布(按小时)')
plt.show()

3.4 关键词统计与长度分布

针对核心关键词进行统计,并分析弹幕长度分布:

1
2
3
4
5
6
7
8
9
keywords = ['开盒','个人信息','电报','缓刑','犯罪','法律','处罚','网络','平台','主播']
keyword_counts = {kw: df['content'].str.contains(kw, na=False).sum() for kw in keywords}

plt.bar(keywords, list(keyword_counts.values()), color='lightcoral')
plt.show()

df['content_length'] = df['content'].apply(len)
plt.hist(df['content_length'], bins=30, color='lightgreen')
plt.show()

此方法可帮助分析用户关注的热点话题及内容互动特点。


4. 总结与企业级应用价值

通过以上流程,我们实现了从 B 站视频弹幕抓取到数据分析的闭环。该方法的应用场景包括:

  • 内容运营分析:了解用户最关注的内容点和互动时段。
  • 舆情监测:对关键词进行统计,快速捕捉负面信息或话题热点。
  • 推荐算法优化:分析弹幕情绪、长度、时间分布,为视频推荐提供数据支撑。

结语:弹幕中的社会脉搏

通过Python的强大生态,我们不仅可以抓取弹幕数据,更能从中解读出丰富的文化和社会信息。每一次弹幕的发送都是用户情感的即时表达,而将这些碎片化的表达汇聚起来,就能看到一幅生动的网络社会图景。

技术不是目的,而是理解世界的工具。希望本文为你打开了弹幕分析的大门,期待看到你更有创意的应用!

完整代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
import requests
from lxml import etree
import pandas as pd

# ---------- 配置 ----------
# BVID: B站视频的BV号,用于获取该视频的所有分P信息
BVID = "BV19R4y1K7yc"
# headers: 请求头配置,模拟浏览器访问
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'
}
# OUTPUT_CSV: 输出的CSV文件名,格式为“视频标题_弹幕.csv”
OUTPUT_CSV = f"公孙田浩弹幕_{BVID}_danmu.csv"

# ---------- 1. 获取所有 P 的 cid ----------
def fetch_cid_list(bvid):
"""
根据视频的BV号获取所有分P的cid列表。

参数:
bvid (str): 视频的BV号

返回:
list: 包含每个分P信息的字典列表,每个字典包含 'cid', 'page' 等字段

异常:
requests.exceptions.RequestException: 网络请求失败时抛出
requests.exceptions.JSONDecodeError: 响应内容无法解析为JSON时抛出
"""
cid_api = f"https://api.bilibili.com/x/player/pagelist?bvid={bvid}&jsonp=jsonp"
try:
response = requests.get(cid_api, headers=headers, timeout=10)
response.raise_for_status()
cid_data = response.json()
return cid_data['data']
except requests.exceptions.RequestException as e:
print("请求 cid 列表失败:", str(e))
raise
except requests.exceptions.JSONDecodeError:
print("无法解析JSON响应,响应内容为:", response.text)
print("状态码:", response.status_code)
raise

# ---------- 2. 获取弹幕数据 ----------
def fetch_danmu_for_cid(cid, page_index):
"""
根据cid获取指定分P的弹幕数据。

参数:
cid (int): 分P对应的cid
page_index (int): 当前分P的索引编号(从1开始)

返回:
list: 弹幕信息列表,每个元素是一个包含弹幕属性的字典
"""
danmu_url = f"https://comment.bilibili.com/{cid}.xml"
try:
xml_response = requests.get(danmu_url, headers=headers, timeout=10)
xml_response.raise_for_status()
xml_response.encoding = 'utf-8'
xml_text = xml_response.text.strip()

# 检查是否是合法的XML内容
if not (xml_text.startswith('<?xml') or xml_text.startswith('<')):
print(f"第 {page_index} P 弹幕获取异常,内容预览:")
print(xml_text[:500])
return []

root = etree.fromstring(xml_text.encode('utf-8'))
danmus = []
for d in root.xpath("//d"):
p_attr = d.attrib.get("p", "").split(",")
if len(p_attr) < 8:
continue
danmus.append({
"page": page_index,
"time_in_video": float(p_attr[0]), # 弹幕在视频中的出现时间(秒)
"mode": int(p_attr[1]), # 弹幕类型(1: 滚动, 4: 底部, 5: 顶部)
"font_size": int(p_attr[2]), # 字体大小
"color": int(p_attr[3]), # 颜色RGB值
"send_time": p_attr[4], # 发送时间戳
"danmu_pool": p_attr[5], # 弹幕池分类
"user_hash": p_attr[6], # 用户ID哈希值
"danmu_id": p_attr[7], # 弹幕唯一ID
"content": d.text # 弹幕文本内容
})
return danmus
except requests.exceptions.RequestException as e:
print(f"第 {page_index} P 请求失败:", str(e))
return []
except etree.XMLSyntaxError as e:
print(f"第 {page_index} P XML解析错误:", str(e))
print(xml_text[:1000])
return []

# ---------- 主流程 ----------
# 获取所有分P的cid信息
pages = fetch_cid_list(BVID)
danmus = []

# 遍历所有分P并抓取弹幕
for page in pages:
cid = page['cid']
page_index = page.get('page', 1)
print(f"正在抓取第 {page_index} P 弹幕,cid={cid} ...")
danmus.extend(fetch_danmu_for_cid(cid, page_index))

print(f"总共抓取弹幕数量: {len(danmus)}")

# ---------- 3. 保存到 CSV ----------
# 将弹幕数据转换为DataFrame并保存为CSV文件
df = pd.DataFrame(danmus)
df.to_csv(OUTPUT_CSV, index=False, encoding="utf-8-sig")
print(f"弹幕已保存到 {OUTPUT_CSV}")

# ---------- 4. 输出全部弹幕 ----------
# 打印所有弹幕内容,处理可能存在的编码问题
print(f"弹幕内容预览 (共 {len(danmus)} 条):")
for d in danmus:
safe_output = []
for item in d.values():
if isinstance(item, str):
try:
item.encode('gbk')
safe_output.append(item)
except UnicodeEncodeError:
safe_output.append(item.encode('gbk', errors='replace').decode('gbk'))
else:
safe_output.append(item)
print(tuple(safe_output))

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
import logging
from collections import Counter

import jieba
import matplotlib.pyplot as plt
import pandas as pd
from wordcloud import WordCloud

# 减少jieba日志输出
jieba.setLogLevel(logging.INFO)
# 解决中文显示问题
plt.rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei', 'DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False

# 读取CSV文件中的弹幕数据
file_path = "公孙田浩弹幕_BV19R4y1K7yc_danmu.csv"
df = pd.read_csv(file_path, encoding="utf-8-sig")

# 数据清洗和预处理
# 将 send_time 列转换为 datetime 类型,content 列确保为字符串类型
df['send_time'] = pd.to_datetime(df['send_time'], unit='s')
df['content'] = df['content'].astype(str)

# 显示所有弹幕数据的基本信息
print("弹幕数据基本信息:")
print(f"总弹幕数量: {len(df)}")
print(f"数据列数: {len(df.columns)}")
print(f"列名: {list(df.columns)}")

# 提取所有弹幕内容并拼接成一个字符串
contents = df['content'].tolist()
all_text = ' '.join(contents)

# 使用jieba进行中文分词
words = list(jieba.cut(all_text))

# 过滤掉单字符和无意义的词 (使用和之前一致的停用词)
stopwords = {'的', '了', '在', '是', '我', '有', '和', '就', '不', '人', '都', '一', '一个', '上', '也', '很', '到', '说', '要', '去', '你', '会', '着', '没有', '看', '好', '自己', '这'}
filtered_words = [word for word in words if len(word) > 1 and word not in stopwords]

# 统计词频
word_freq = Counter(filtered_words)

# 显示高频词 (恢复到Top 30)
print("\n高频词汇 Top 30:")
for word, freq in word_freq.most_common(30):
print(f"{word}: {freq}")

# 1. 时间分布图
# 绘制弹幕按小时的时间分布柱状图
plt.figure(figsize=(10, 6))
df['hour'] = df['send_time'].dt.hour
hourly_counts = df['hour'].value_counts().sort_index()
plt.bar(hourly_counts.index, hourly_counts.values, color='skyblue')
plt.xlabel('小时')
plt.ylabel('弹幕数量')
plt.title('弹幕时间分布(按小时)')
plt.xticks(range(0, 24))
plt.grid(axis='y', alpha=0.3)
plt.tight_layout()
plt.savefig('danmu_time_distribution.png', dpi=300, bbox_inches='tight')
plt.show()
print("\n时间分布图已保存为 danmu_time_distribution.png")

# 2. 词云图 (使用相同的词频数据)
# 根据词频生成词云图,并尝试使用指定字体
plt.figure(figsize=(10, 8))
try:
wordcloud = WordCloud(
width=800,
height=600,
background_color='white',
font_path='simhei.ttf', # Windows系统黑体
max_words=100,
relative_scaling=0.5,
random_state=42
).generate_from_frequencies(word_freq)
except OSError:
# 如果找不到指定字体,使用默认设置
wordcloud = WordCloud(
width=800,
height=600,
background_color='white',
max_words=100,
relative_scaling=0.5,
random_state=42
).generate_from_frequencies(word_freq)

plt.imshow(wordcloud, interpolation='bilinear')
plt.axis('off')
plt.title('弹幕词云图')
plt.tight_layout()
plt.savefig('danmu_wordcloud.png', dpi=300, bbox_inches='tight')
plt.show()
print("\n词云图已保存为 danmu_wordcloud.png")

# 3. 关键词统计图
# 对特定关键词在弹幕中出现的次数进行统计并绘制柱状图
plt.figure(figsize=(10, 6))
# 定义关键词
keywords = ['开盒', '个人信息', '电报', '缓刑', '犯罪', '法律', '处罚', '网络', '平台', '主播']
keyword_counts = {}

for keyword in keywords:
keyword_counts[keyword] = df['content'].str.contains(keyword, na=False).sum()

# 绘制关键词统计图
bars = plt.bar(keywords, list(keyword_counts.values()), color='lightcoral')
plt.xlabel('关键词')
plt.ylabel('出现次数')
plt.title('关键词统计')
plt.xticks(rotation=45)
plt.grid(axis='y', alpha=0.3)

# 在柱状图上添加数值标签
for bar in bars:
height = bar.get_height()
plt.text(bar.get_x() + bar.get_width()/2., height,
f'{int(height)}',
ha='center', va='bottom')

plt.tight_layout()
plt.savefig('danmu_keyword_statistics.png', dpi=300, bbox_inches='tight')
plt.show()
print("\n关键词统计图已保存为 danmu_keyword_statistics.png")

# 4. 弹幕长度分布
# 统计每条弹幕的字符长度并绘制直方图
plt.figure(figsize=(10, 6))
df['content_length'] = df['content'].apply(len)
plt.hist(df['content_length'], bins=30, color='lightgreen', edgecolor='black')
plt.xlabel('弹幕长度(字符数)')
plt.ylabel('弹幕数量')
plt.title('弹幕长度分布')
plt.grid(axis='y', alpha=0.3)
plt.tight_layout()
plt.savefig('danmu_length_distribution.png', dpi=300, bbox_inches='tight')
plt.show()
print("\n弹幕长度分布图已保存为 danmu_length_distribution.png")

# 输出统计信息
print(f"\n弹幕统计信息:")
print(f"平均每条弹幕长度: {df['content_length'].mean():.2f} 字符")
print(f"最长弹幕: {df['content_length'].max()} 字符")
print(f"最短弹幕: {df['content_length'].min()} 字符")

# 输出时间分布信息
print(f"\n时间分布信息:")
print(f"最早弹幕: {df['send_time'].min()}")
print(f"最晚弹幕: {df['send_time'].max()}")

数据来源:B站公开弹幕数据
工具栈:Python, requests, pandas, jieba, matplotlib, wordcloud

🙋‍♂️ 如果你觉得这篇文章有帮助:

  • 点赞 👍
  • 收藏 ⭐
  • 评论 💬
  • 关注我 👇 获取后续实战内容
评论
分享