简介:面向B站弹幕数据分析的综合项目资料包,整合爬虫采集、文本挖掘与可视化展示,适合Python学习者及计算机相关专业学生用于毕业设计、课程设计、项目演示等。项目基于Selenium实现弹幕爬取,完整覆盖词云分析、词频统计、情感分析与衍生指标构建,附有详细的代码注释、运行说明和可视化成果。资源共1202个文件,核心为1152个CSV词频数据表,涵盖全知识区、科学科普、社科人文、校园学习、财经等主题,另含16个Python源码脚本、22张PNG词云图、6个HTML可视化页面、2个Excel汇总表及文档说明,压缩包约26.08MB,层级分明便于查找。已有91人学习下载,项目代码经过实测运行成功,获导师指导并达到答辩高分水平,可直接作为毕设/课设作品提交,也可替换数据或修改参数以适配其他视频,是快速上手B站弹幕分析全流程的实用参考。
1. 从弹幕里挖出视频的“第二层内容”:这个项目到底能做什么
弹幕是 B 站视频里最容易被忽略的数据金矿。一个百万播放的视频,弹幕量动辄几十万条,这些短文本里藏着观众的情绪起伏、槽点密度、名场面坐标和二次传播的梗。如果你只把弹幕当成飘过去的白字,那等于扔掉了视频最有价值的行为数据。这个基于 bilibili 弹幕分析的项目,做的就是一套完整流水线:爬虫采数据、清洗入库、分词算词频、SnowNLP 做情感分析、再构造衍生指标,最后用 ECharts 拉出可视化大屏。它不是单个脚本,而是从零到一可复现的分析框架。
我最初做这个方向是为了给视频运营团队看“哪一段观众最激动、哪一句台词被刷屏、哪个角色口碑反转”,后来发现这套东西用在课程复盘、商品评测、甚至影视宣发都成立。适合的人群很明确:想入门 Python 爬虫和数据分析的在校生、需要做内容复盘的新媒体运营、以及想把文本分析接到业务里的初级工程师。项目亲测能跑通,但坑不少,尤其是风控和情感分析的准确度,这两块我会在后文直接给结论和替代方案。
2. 从视频页到弹幕文件:B站弹幕爬虫的正确打开方式
2.1 先搞清楚 B 站弹幕的请求链路:cid、oid 与分段接口
B 站的弹幕接口不是直接拿 bvid 就能请求的,中间至少有两层跳转。第一层是视频详情页,通过 bvid 换 cid(视频分 P 的 ID),第二层才是弹幕接口。老版本的comment.bilibili.com/{cid}.xml接口只能拿到 4000 条以内的历史弹幕,而且按时间倒序截取,这会导致你丢掉视频前中段的大量弹幕。新版接口api.bilibili.com/x/v1/dm/list.so?oid={cid}能拿到完整弹幕池,但返回的是 protobuf 编码,需要配套解析。
常见做法是先请求api.bilibili.com/x/web-interface/view?bvid={bvid}拿 cid 列表,再针对每个 cid 请求弹幕池。如果你的目标是全量弹幕,不要迷信单次请求,B 站服务端对单包弹幕量有上限,超长视频必须走分段参数。分段参数在不同接口里名字不同,有的叫segment_index,有的叫date。我做采集的时候习惯先拉一次完整接口看返回条数,如果接近上限就立刻切分段策略,否则你会拿到一份“缺中间”的脏数据。
import requests import re def get_cid(bvid: str) -> list: """通过 bvid 获取视频所有分P的 cid 列表""" api = "https://api.bilibili.com/x/web-interface/view" params = {"bvid": bvid} headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36", "Referer": "https://www.bilibili.com" } resp = requests.get(api, params=params, headers=headers, timeout=10) resp.raise_for_status() data = resp.json() if data["code"] != 0: raise RuntimeError(f"API 返回异常: {data['message']}") pages = data["data"]["pages"] return [{"cid": p["cid"], "page": p["page"], "part": p["part"]} for p in pages] print(get_cid("BV1xx411c7mD"))这里要把Referer带上,B 站接口对 referer 校验很严,不带直接 412。返回的pages列表里每一项对应一个分 P,如果你只分析单 P 视频,取第一项就行。我建议把page和part也存下来,后面做多 P 对比时不用回头再查一次接口。
拿到 cid 之后,弹幕池接口返回的是 protobuf,直接解析不现实。我一般加一层转换,先把弹幕池转成 XML 或 JSON 再入库。下面这段是转换逻辑的一部分:
def fetch_danmaku(cid: int) -> list: """拉取弹幕池并解析为结构化数据""" url = f"https://api.bilibili.com/x/v1/dm/list.so?oid={cid}" headers = { "User-Agent": "Mozilla/5.0", "Referer": f"https://www.bilibili.com/video/{bvid}" } resp = requests.get(url, headers=headers, timeout=10) resp.raise_for_status() # 注意:resp.content 是 protobuf,需要 dm.proto 定义才能解 # 临时做法:直接用第三方库 bilibili_dm 解析 from bilibili_dm import DanmakuParser parser = DanmakuParser(resp.content) return parser.parse()bilibili_dm不是官方库,如果你不想引入额外依赖,可以自己写 protobuf 解析,但成本太高。我的建议是:本地调试用第三方库,线上批量采集时把解析结果序列化成 JSON 落盘,避免重复请求。请求频率控制在每视频间隔 3-5 秒,不然 IP 会被临时封禁。
2.2 反爬绕不开的坎:wbi 签名、风控与线程池
B 站现在的 Web 端接口普遍要求 wbi 签名。所谓 wbi 签名,是把请求参数按 key 排序后拼接,再对img_key和sub_key做哈希混淆。第一次写爬虫时不知道这个机制,requests 直连接口,返回的永远是-352风控错误。后来研究了 B 站前端代码,才发现每个接口都要带w_rid参数,这个参数由固定算法生成。
import time import hashlib from functools import reduce def get_wbi_sign(params: dict, img_key: str, sub_key: str) -> dict: """生成 wbi 签名参数""" mixin_key = img_key + sub_key # 对 key 排序并过滤特殊字符 params = {k: v for k, v in params.items() if re.match(r"^[a-zA-Z0-9]+$", str(v))} params.update({"wts": int(time.time())}) sorted_params = dict(sorted(params.items(), key=lambda x: x[0])) query = reduce(lambda a, b: f"{a}&{b}", [f"{k}={v}" for k, v in sorted_params.items()]) w_rid = hashlib.md5((query + mixin_key).encode()).hexdigest() sorted_params["w_rid"] = w_rid return sorted_params生成img_key和sub_key要从api.bilibili.com/x/web-interface/nav拿,这个接口返回的wbi_img字段里有img_url和sub_url,从 URL 文件名里截取即可。这个签名算法会不定期更新,如果你发现某天开始批量 412,先别怀疑代码逻辑,去抓一下前端最新的nav响应看看 key 是否换了。
线程池方面,我踩过大坑。最初用 ThreadPoolExecutor 开 20 个线程并发拉弹幕,结果 5 分钟 IP 就被限制,所有请求返回 412。B 站对单 IP 的 QPS 限制比想象中严格。我把并发数压到 4-6,并且给每个线程设置独立的随机间隔,运行一晚上没再触发风控。另一个心得是:永远不要用同一套 User-Agent 跑全量采集,至少准备 20 个 UA 轮换,能明显降低被识别为爬虫的概率。
from concurrent.futures import ThreadPoolExecutor, as_completed import random def crawl_multi(cid_list: list, max_workers: int = 4): """多线程采集,每个线程独立延时""" results = {} def worker(cid): time.sleep(random.uniform(0.5, 1.5)) return cid, fetch_danmaku(cid) with ThreadPoolExecutor(max_workers=max_workers) as pool: futures = [pool.submit(worker, cid) for cid in cid_list] for future in as_completed(futures): cid, danmaku = future.result() results[cid] = danmaku return results这段代码的关键是max_workers=4和延时区间。如果你只想跑几十个视频,开 8 线程问题不大;但如果你要全站某个分区做批量分析,4 线程是最安全的起点。另外,fetch_danmaku里必须做异常捕获,网络抖动或风控返回不该让整个任务挂掉,我一般会在 worker 里加三次重试。
3. 弹幕入库与清洗:别让脏数据毁掉整张词云
3.1 清洗规则的四个边界坑:去重、模式覆盖、时区、用户哈希
弹幕数据拿回来不能直接用。第一个坑是重复弹幕,B 站弹幕池对同一用户在同一视频同一时间点的弹幕会做合并,但不同时间点的相同内容仍然存在,需要按用户哈希 + 时间点 + 内容做联合去重;第二个坑是弹幕模式,B 站弹幕分滚动、顶部、底部和逆向四种,不同模式的情感倾向分布差异极大,顶部弹幕通常是字幕或高能预警,算情感时会把“前方高能”这种中性词误判成负面;第三个坑是时间字段,弹幕接口返回的是毫秒级时间戳,直接入库后 MySQL 的DATETIME装不下,需要先转成秒级再用FROM_UNIXTIME;第四个坑是用户哈希,B 站为了保护隐私,弹幕里的用户信息是 MD5 后的哈希值,无法反查用户,但这不妨碍你用它做用户维度去重。
import hashlib from datetime import datetime def clean_danmaku(raw_list: list) -> list: """清洗弹幕:去重 + 模式过滤 + 时间格式化""" seen = set() cleaned = [] for item in raw_list: # item 结构: {"time": float, "mode": int, "uid_hash": str, "content": str} mode = item.get("mode", 1) if mode not in (1, 4): # 只保留滚动弹幕,顶部底部容易干扰情感分析 continue dedup_key = f"{item['uid_hash']}_{round(item['time'], 2)}_{item['content']}" if dedup_key in seen: continue seen.add(dedup_key) cleaned.append({ "vid": item["cid"], "play_time": round(item["time"], 2), "mode": mode, "uid_hash": item["uid_hash"], "content": item["content"].strip()[:50], "created_at": datetime.now() }) return cleaned第 8 行的mode not in (1, 4)是关键决策。滚动弹幕和逆向弹幕是观众主动发的,顶部底部弹幕大多是字幕组或告白者刷的,内容偏向不规范文本,直接删掉能减少后续分析的噪音。content[:50]是防止超长弹幕把存储撑爆,实际弹幕很少超 50 字,但防一手没坏处。
清洗后的数据建议直接进 MySQL,字段类型要提前设计好。play_time用FLOAT,uid_hash用CHAR(32),content用VARCHAR(100),cid用INT。索引方面,给(cid, play_time)建联合索引,后续做时间窗口分析时能省一半查询时间。
3.2 用 SQLAlchemy 把清洗结果落库:序列化与批量写入
数据量不大的时候直接pymysql逐条插入没问题,但弹幕量过万后逐条插入会产生巨大网络开销。我习惯用 SQLAlchemy 的ORMR映射加session.bulk_insert_mappings批量写入。这个项目的数据量级在几十万条以内,这个方案完全够用。
from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base Base = declarative_base() class Danmaku(Base): __tablename__ = "danmaku" id = Column(Integer, primary_key=True, autoincrement=True) cid = Column(Integer, index=True) play_time = Column(Float) mode = Column(Integer) uid_hash = Column(String(32)) content = Column(String(100)) created_at = Column(DateTime) def save_danmaku(cleaned_list: list): """批量写入清洗后的弹幕""" engine = create_engine("mysql+pymysql://user:pass@localhost/danmaku?charset=utf8mb4") Base.metadata.create_all(engine) Session = sessionmaker(bind=engine) session = Session() try: session.bulk_insert_mappings(Danmaku, cleaned_list) session.commit() except Exception: session.rollback() raise finally: session.close()bulk_insert_mappings不走 ORM 对象的实例化,性能比add_all高一个量级,缺点是没法拿到自增主键。如果你后续需要关联别的表,建议在清洗阶段就生成业务主键,比如用cid + play_time + uid_hash拼接字符串做逻辑主键。
入库后要立刻验证数据质量。我一般跑三条 SQL:查总条数、查play_time最大值最小值、查content长度分布。如果发现play_time的最大值超过视频时长,说明接口返回了广告弹幕或特殊位置弹幕,需要过滤。弹幕的play_time理论上不可能超出视频总时长,这条规则比任何清洗都硬。
4. 词频、词云与情感分析:让弹幕从“字”变成“态度”
4.1 分词与停用词:jieba 的词性过滤和自定义词典
弹幕文本和新闻、评论完全不同,它短、口语化、错别字率高、网络梗密集。“yyds”“绝绝子”“绷不住了”这类词,标准词典里根本没有,且带强烈情感色彩。直接用通用词典分词,会把“yyds”切成“yyd”“s”,把“绝绝子”切成“绝”“绝”“子”。词频分析的前提是分词准确,分词不准,后面的词云全是噪音。
解决方法分三步。第一步,准备自定义词典,把弹幕高频梗词录进去,每个词占一行;第二步,分完词后过滤单字和纯符号;第三步,加载停用词表,把“的、了、吗、啊”这类无意义词去掉。jieba 的词性标注在弹幕场景下其实不太靠谱,比如“救命”在弹幕里是表达强烈情绪,但在词性标注里是动词,所以词性过滤要慎用,我一般只过滤数词、量词和副词。
import jieba from collections import Counter def load_stopwords(path: str = "stopwords.txt") -> set: with open(path, "r", encoding="utf-8") as f: return {line.strip() for line in f if line.strip()} def word_frequency(danmaku_list: list, custom_dict: str = "danmaku_dict.txt") -> Counter: """分词并统计词频""" jieba.load_userdict(custom_dict) stopwords = load_stopwords() counter = Counter() for text in danmaku_list: words = jieba.lcut(text) for w in words: if len(w) < 2: # 过滤单字 continue if w in stopwords: # 过滤停用词 continue if not w.isalpha(): # 过滤非中文/英文 continue counter[w] += 1 return counter弹幕里“草”“好”“6”这类单字其实很有信息量,但单字入词频会让词云出现大量无意义的大字。我的折中方案是保留高频单字单独做一份统计,不在词云展示,但在情感分析里作为强特征。
4.2 情感分析的准确度问题:SnowNLP 打分与人工校准
弹幕情感分析是这个项目里准确度最“玄学”的部分。第一版我直接用 SnowNLP 对每条弹幕打分,发现负面弹幕被严重高估,原因是弹幕里的“不”字出现频率极高,比如“不是吧”“不会吧”表达的是惊讶而非否定。第二版改用大连理工大学的情感本体库,把弹幕切词后匹配情感词,准确率上去了一些,但“yyds”这种新词依然无法覆盖。
多模态情感分析听起来高级,但落到弹幕场景,主要是把文本情感和视频画面情绪做交叉验证。这个项目里不涉及视频帧提取,所以我只做了文本层面的优化:给 SnowNLP 的结果加一个置信度阈值,得分在 0.3 到 0.7 之间的弹幕标记为中性,不参与正负统计。
from snownlp import SnowNLP def sentiment_score(text: str) -> dict: """计算情感得分:snownlp + 阈值判定""" s = SnowNLP(text) score = s.sentiments # 0-1,越接近1越正向 if score > 0.7: label = "positive" elif score < 0.3: label = "negative" else: label = "neutral" return {"text": text, "score": round(score, 3), "label": label}阈值 0.7/0.3 是我在自己的数据集上调出来的。不同视频类型的最佳阈值不同,比如鬼畜视频的“负面”弹幕大量是调侃,阈值要调到 0.8 才不误杀。我的建议是:先跑一遍全量数据,抽取 500 条人工标注,再用标注结果回调阈值。这个过程痛苦但值得做,否则你的情感趋势图只能看个大概。
情感分析结果入库时,注意和原始弹幕表分开存,用cid + play_time做关联键。这样后续做时间线可视化时,可以直接 JOIN 出每个时间点的情感得分均值。
5. 衍生指标构造与可视化大屏:把分析结果变成能看的业务结论
5.1 三个值得做的衍生指标:弹幕密度、高能峰值、情感波动率
原始弹幕数据只有文本和时间,直接画折线图看不出名堂。我一般构造五个衍生指标,其中三个最值得做。第一个是弹幕密度,按每 10 秒窗口统计弹幕条数,这个指标直接反映视频的节奏感——B 站观众习惯在剧情高潮、名场面、反转处密集发弹幕,弹幕密度曲线和视频完播率强相关。第二个是高能峰值,把弹幕密度超过全视频均值三倍的时间段标记为“高能片段”,运营可以直接跳到这些时间点做二次剪辑。第三个是情感波动率,用滑动窗口计算情感得分的标准差,波动率高的段落说明观众观点撕裂,这类内容在评论区通常也吵得厉害。
-- 每10秒窗口的弹幕密度与情感均值 SELECT FLOOR(play_time / 10) * 10 AS time_window, COUNT(*) AS cnt, AVG(s.score) AS avg_sentiment FROM danmaku d LEFT JOIN danmaku_sentiment s ON d.cid = s.cid AND d.play_time = s.play_time WHERE d.cid = 123456 GROUP BY time_window ORDER BY time_window;这个 SQL 的FLOOR(play_time / 10) * 10是把秒级时间归一到 10 秒窗口的常见做法。要注意LEFT JOIN的关联条件必须带上cid,否则不同视频的弹幕会串。如果弹幕量超过十万,这条查询会有点慢,建议在play_time和cid上建联合索引。
弹幕密度算完后,高能峰值直接用 SQL 的窗口函数找:按密度降序排,取超过均值三倍的窗口。这里的“三倍”是经验值,你可以按视频类型调。短视频高能阈值高一些,长视频低一些,因为长视频的弹幕分布更均匀。
5.2 可视化:ECharts 时间线热力图与词云的参数调整
可视化部分我用 ECharts,不自己造轮子。弹幕密度曲线用折线图,情感趋势用面积图,高能片段用热力图叠加在时间线上。这套组合在业务侧比饼图和柱状图好用得多,运营能直接看到“哪一分钟观众炸了”。下面是核心配置:
option = { title: { text: '弹幕密度与情感趋势' }, tooltip: { trigger: 'axis' }, legend: { data: ['弹幕密度', '情感得分'] }, xAxis: { type: 'time', name: '播放时间' }, yAxis: [ { type: 'value', name: '弹幕密度' }, { type: 'value', name: '情感得分', max: 1 } ], series: [ { name: '弹幕密度', type: 'line', data: densityData, areaStyle: { opacity: 0.3 } }, { name: '情感得分', type: 'line', yAxisIndex: 1, data: sentimentData, lineStyle: { color: '#ff6b81' } } ] };双 Y 轴配置是必须的,因为弹幕密度的量级是几百,情感得分是 0 到 1,放同一根轴会互相压制。densityData和sentimentData的格式是[timestamp, value]对,我通常从后端接口直接生成这个结构。词云组件用 ECharts 的wordCloud系列,但要注意词频数据量过大时渲染会卡,我会在传给前端前把词频截断到 Top 200。
可视化大屏不是把图表堆一起就完事。我一般分四个区域:左上角放视频基本信息和高能片段列表,右上角放情感趋势面积图,中部放弹幕密度热力图,底部放词云。顶部的“高能片段”按钮可以直接跳转到视频对应时间点,这个功能运营反馈最多,因为省去了来回拖进度条的时间。
5.3 本地跑通的完整姿势:从码到图的全链路验证
拿到源码包后,我建议按这个顺序验证:先跑爬虫拿 3 个视频的弹幕,验证数据样貌;再跑清洗入库,验证 MySQL 连接;然后跑词频和情感分析,输出 CSV;最后启动可视化服务。任何一步失败了,都能通过日志快速定位。我摊上过一次可笑的翻车——爬虫跑完,清洗入库也正常,但词云全是一个个孤立汉字,排查到最后发现是jieba.load_userdict加载的是空文件,自定义词典没有生效。
6. 从能跑到跑好:进阶用法与避坑经验
6.1 避坑清单:三个反复出现的实际问题
现象一:前端词云出现大量“=”“/”等符号词。原因:清洗阶段没有过滤纯符号弹幕。解决:在分词时加入if not re.match(r"^[\u4e00-\u9fa5a-zA-Z]+$", w)让非中英文直接淘汰。现象二:批量采集时请求偶尔返回 412,重试后恢复。原因:单位时间请求量超过接口限制。解决:把线程数降下来,并且在重试前增加指数退避而不是固定等待。现象三:情感分析结果中弹幕字数越多,得分越接近 0.5。原因:SnowNLP 内置模型对长文本趋向中性,弹幕场景下让它算短句精度尚可,算长句没意义。解决:超过 20 字的弹幕直接截断到前 20 字再算,或者在预处理阶段就把超长弹幕标记为中性,不参与打分。
6.2 进阶:把衍生指标接到定时任务与告警
跑通一次只是开始,你大概率会想每天跑增量弹幕。我建议写一个定时脚本,每天凌晨拉取昨日新增弹幕,增量更新词频和情感表。数据量小时直接全量重算,数据量大时用MAX(play_time)做增量游标。到了这一步,你才真正把“弹幕分析”从一次性项目变成了长期数据资产。
我的个人习惯是,最后做一次全量导出,把词频表、情感表、衍生指标表分别导成 CSV 存档。因为每次代码升级或词典更新都会改变分析结果,历史存档能让你对比新旧版本的差异,不至于“旧数据找不回”。这个习惯救过我一次:有次我把停用词表改坏了,导致词频统计崩盘,回滚后全靠前一天导出的 CSV 兜底。
希望这些踩坑经验能让你少走几步冤枉路,如果在高能片段定位或情感阈值调优上有更好的思路,也值得多试几个版本再定。这套框架的核心价值不是一次性的报表,而是把弹幕这道“观众情绪暗流”变成可量化、可追踪、可对比的指标,让你的视频选题和内容调整有据可依。
本文还有配套的精品资源,点击获取