|
|
import requests
|
|
|
import re
|
|
|
import json
|
|
|
import pandas as pd
|
|
|
import jieba
|
|
|
from wordcloud import WordCloud
|
|
|
import matplotlib.pyplot as plt
|
|
|
from collections import Counter
|
|
|
import time
|
|
|
import random
|
|
|
from bs4 import BeautifulSoup
|
|
|
import numpy as np
|
|
|
from PIL import Image
|
|
|
|
|
|
# 设置中文显示
|
|
|
plt.rcParams["font.family"] = ["SimHei", "WenQuanYi Micro Hei", "Heiti TC"]
|
|
|
plt.rcParams['axes.unicode_minus'] = False # 解决负号显示问题
|
|
|
|
|
|
class BilibiliDanmakuAnalyzer:
|
|
|
def __init__(self):
|
|
|
# 初始化请求头，模拟浏览器访问
|
|
|
self.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",
|
|
|
"Accept": "*/*",
|
|
|
"Accept-Language": "zh-CN,zh;q=0.9",
|
|
|
"Connection": "keep-alive"
|
|
|
}
|
|
|
# 存储爬取的所有弹幕
|
|
|
self.all_danmakus = []
|
|
|
# 存储视频信息
|
|
|
self.video_info = []
|
|
|
|
|
|
def get_video_ids(self, keyword, page_count=36):
|
|
|
"""
|
|
|
根据关键词搜索视频，获取视频ID
|
|
|
每页约10个视频，36页共约360个视频
|
|
|
"""
|
|
|
video_ids = []
|
|
|
for page in range(1, page_count + 1):
|
|
|
try:
|
|
|
# B站搜索接口
|
|
|
url = f"https://api.bilibili.com/x/web-interface/search/type?keyword={keyword}&search_type=video&page={page}"
|
|
|
response = requests.get(url, headers=self.headers)
|
|
|
data = json.loads(response.text)
|
|
|
|
|
|
if data.get("code") == 0 and data.get("data"):
|
|
|
results = data["data"]["result"]
|
|
|
for item in results:
|
|
|
video_id = item["aid"]
|
|
|
video_ids.append(video_id)
|
|
|
# 保存视频信息
|
|
|
self.video_info.append({
|
|
|
"video_id": video_id,
|
|
|
"title": item["title"],
|
|
|
"up主": item["author"],
|
|
|
"播放量": item["play"],
|
|
|
"弹幕数": item["video_review"]
|
|
|
})
|
|
|
print(f"已获取第{page}页视频，累计{len(video_ids)}个视频ID")
|
|
|
|
|
|
# 随机休眠，避免被反爬
|
|
|
time.sleep(random.uniform(1, 3))
|
|
|
except Exception as e:
|
|
|
print(f"获取第{page}页视频ID失败: {str(e)}")
|
|
|
continue
|
|
|
|
|
|
return list(set(video_ids)) # 去重
|
|
|
|
|
|
def get_danmakus(self, video_id):
|
|
|
"""获取单个视频的弹幕"""
|
|
|
try:
|
|
|
# 获取cid（弹幕池ID）
|
|
|
url = f"https://api.bilibili.com/x/web-interface/view?aid={video_id}"
|
|
|
response = requests.get(url, headers=self.headers)
|
|
|
data = json.loads(response.text)
|
|
|
|
|
|
if data.get("code") == 0 and data.get("data"):
|
|
|
cid = data["data"]["cid"]
|
|
|
|
|
|
# 获取弹幕
|
|
|
danmaku_url = f"https://comment.bilibili.com/{cid}.xml"
|
|
|
response = requests.get(danmaku_url, headers=self.headers)
|
|
|
response.encoding = "utf-8"
|
|
|
|
|
|
# 解析XML格式的弹幕
|
|
|
soup = BeautifulSoup(response.text, "xml")
|
|
|
danmakus = soup.find_all("d")
|
|
|
|
|
|
# 提取弹幕文本
|
|
|
danmaku_texts = [danmaku.text.strip() for danmaku in danmakus]
|
|
|
print(f"视频ID {video_id} 获取到 {len(danmaku_texts)} 条弹幕")
|
|
|
|
|
|
return danmaku_texts
|
|
|
else:
|
|
|
print(f"获取视频 {video_id} 的cid失败")
|
|
|
return []
|
|
|
except Exception as e:
|
|
|
print(f"获取视频 {video_id} 弹幕失败: {str(e)}")
|
|
|
return []
|
|
|
|
|
|
def crawl_all_danmakus(self, keywords=["大语言模型", "大模型", "LLM"], max_videos=360):
|
|
|
"""爬取所有相关视频的弹幕"""
|
|
|
all_video_ids = []
|
|
|
|
|
|
# 从多个关键词获取视频ID
|
|
|
for keyword in keywords:
|
|
|
print(f"开始搜索关键词: {keyword}")
|
|
|
video_ids = self.get_video_ids(keyword)
|
|
|
all_video_ids.extend(video_ids)
|
|
|
time.sleep(2)
|
|
|
|
|
|
# 去重并限制最大数量
|
|
|
unique_video_ids = list(set(all_video_ids))[:max_videos]
|
|
|
print(f"共获取到 {len(unique_video_ids)} 个不重复的视频ID，开始爬取弹幕...")
|
|
|
|
|
|
# 爬取每个视频的弹幕
|
|
|
for i, video_id in enumerate(unique_video_ids):
|
|
|
danmakus = self.get_danmakus(video_id)
|
|
|
self.all_danmakus.extend(danmakus)
|
|
|
|
|
|
# 每爬取10个视频保存一次数据，防止意外丢失
|
|
|
if (i + 1) % 10 == 0:
|
|
|
self.save_danmakus_to_file()
|
|
|
print(f"已完成 {i + 1}/{len(unique_video_ids)} 个视频的弹幕爬取，累计弹幕数: {len(self.all_danmakus)}")
|
|
|
|
|
|
# 随机休眠，避免被反爬
|
|
|
time.sleep(random.uniform(1, 2))
|
|
|
|
|
|
# 最终保存一次
|
|
|
self.save_danmakus_to_file()
|
|
|
print(f"所有视频弹幕爬取完成，共获取 {len(self.all_danmakus)} 条弹幕")
|
|
|
|
|
|
# 保存视频信息
|
|
|
df = pd.DataFrame(self.video_info)
|
|
|
df.to_excel("视频信息.xlsx", index=False)
|
|
|
|
|
|
return self.all_danmakus
|
|
|
|
|
|
def save_danmakus_to_file(self, filename="弹幕数据.txt"):
|
|
|
"""保存弹幕数据到文件"""
|
|
|
with open(filename, "w", encoding="utf-8") as f:
|
|
|
for danmaku in self.all_danmakus:
|
|
|
f.write(danmaku + "\n")
|
|
|
|
|
|
def load_danmakus_from_file(self, filename="弹幕数据.txt"):
|
|
|
"""从文件加载弹幕数据"""
|
|
|
try:
|
|
|
with open(filename, "r", encoding="utf-8") as f:
|
|
|
self.all_danmakus = [line.strip() for line in f.readlines() if line.strip()]
|
|
|
print(f"从文件加载了 {len(self.all_danmakus)} 条弹幕数据")
|
|
|
return self.all_danmakus
|
|
|
except Exception as e:
|
|
|
print(f"加载弹幕数据失败: {str(e)}")
|
|
|
return []
|
|
|
|
|
|
def analyze_application_cases(self, top_n=8):
|
|
|
"""分析AI技术应用案例，统计排名前N的弹幕"""
|
|
|
# 常见的LLM应用领域关键词
|
|
|
application_keywords = [
|
|
|
"聊天机器人", "智能客服", "内容创作", "代码生成",
|
|
|
"教育辅导", "翻译", "数据分析", "医疗诊断",
|
|
|
"自动写作", "语音助手", "图像生成", "游戏开发",
|
|
|
"推荐系统", "法律咨询", "金融分析", "市场营销"
|
|
|
]
|
|
|
|
|
|
# 统计每个应用领域出现的次数
|
|
|
application_counts = {keyword: 0 for keyword in application_keywords}
|
|
|
|
|
|
for danmaku in self.all_danmakus:
|
|
|
for keyword in application_keywords:
|
|
|
if keyword in danmaku:
|
|
|
application_counts[keyword] += 1
|
|
|
|
|
|
# 按出现次数排序
|
|
|
sorted_applications = sorted(application_counts.items(), key=lambda x: x[1], reverse=True)
|
|
|
|
|
|
# 提取前N名
|
|
|
top_applications = sorted_applications[:top_n]
|
|
|
|
|
|
# 保存到Excel
|
|
|
df = pd.DataFrame(top_applications, columns=["应用案例", "出现次数"])
|
|
|
df.to_excel("LLM应用案例统计.xlsx", index=False)
|
|
|
|
|
|
return top_applications
|
|
|
|
|
|
def generate_wordcloud(self, mask=None, filename="弹幕词云图.png"):
|
|
|
"""生成词云图"""
|
|
|
# 合并所有弹幕文本
|
|
|
text = " ".join(self.all_danmakus)
|
|
|
|
|
|
# 使用jieba进行分词
|
|
|
words = jieba.cut(text)
|
|
|
words = [word for word in words if len(word) > 1] # 过滤单字
|
|
|
words_text = " ".join(words)
|
|
|
|
|
|
# 配置词云
|
|
|
wc = WordCloud(
|
|
|
font_path="simhei.ttf", # 确保有中文字体
|
|
|
background_color="white",
|
|
|
max_words=200,
|
|
|
mask=mask,
|
|
|
contour_width=1,
|
|
|
contour_color="steelblue"
|
|
|
)
|
|
|
|
|
|
# 生成词云
|
|
|
wc.generate(words_text)
|
|
|
|
|
|
# 显示词云
|
|
|
plt.figure(figsize=(12, 8))
|
|
|
plt.imshow(wc, interpolation="bilinear")
|
|
|
plt.axis("off")
|
|
|
plt.title("B站大语言模型相关视频弹幕词云")
|
|
|
plt.tight_layout(pad=0)
|
|
|
|
|
|
# 保存词云图
|
|
|
wc.to_file(filename)
|
|
|
print(f"词云图已保存为 {filename}")
|
|
|
|
|
|
plt.show()
|
|
|
|
|
|
def analyze_sentiment(self):
|
|
|
"""简单分析用户观点"""
|
|
|
# 成本相关关键词
|
|
|
cost_keywords = ["贵", "便宜", "成本", "收费", "免费", "价格"]
|
|
|
# 应用领域关键词
|
|
|
field_keywords = ["教育", "医疗", "工作", "学习", "娱乐", "创作", "办公"]
|
|
|
# 不利影响关键词
|
|
|
negative_keywords = ["失业", "取代", "错误", "偏见", "隐私", "风险", "依赖"]
|
|
|
# 积极影响关键词
|
|
|
positive_keywords = ["方便", "高效", "有用", "帮助", "创新", "进步", "强大"]
|
|
|
|
|
|
# 统计各类关键词出现次数
|
|
|
cost_count = sum(1 for danmaku in self.all_danmakus if any(kw in danmaku for kw in cost_keywords))
|
|
|
field_count = sum(1 for danmaku in self.all_danmakus if any(kw in danmaku for kw in field_keywords))
|
|
|
negative_count = sum(1 for danmaku in self.all_danmakus if any(kw in danmaku for kw in negative_keywords))
|
|
|
positive_count = sum(1 for danmaku in self.all_danmakus if any(kw in danmaku for kw in positive_keywords))
|
|
|
|
|
|
# 计算百分比
|
|
|
total = len(self.all_danmakus)
|
|
|
if total == 0:
|
|
|
return {}
|
|
|
|
|
|
result = {
|
|
|
"成本相关讨论占比": f"{cost_count/total*100:.2f}%",
|
|
|
"应用领域讨论占比": f"{field_count/total*100:.2f}%",
|
|
|
"不利影响讨论占比": f"{negative_count/total*100:.2f}%",
|
|
|
"积极影响讨论占比": f"{positive_count/total*100:.2f}%"
|
|
|
}
|
|
|
|
|
|
return result
|
|
|
|
|
|
def predict_trend(self):
|
|
|
"""预测大语言模型应用发展趋势"""
|
|
|
# 基于常见观点的简单趋势预测
|
|
|
trends = [
|
|
|
"1. 行业垂直化：大语言模型将更深入各个专业领域，如医疗、法律、教育等",
|
|
|
"2. 个性化增强：模型将更加了解用户需求，提供个性化服务",
|
|
|
"3. 多模态融合：文本、图像、语音等多模态能力将深度融合",
|
|
|
"4. 边缘部署增加：更多模型将在边缘设备上运行，提升响应速度和隐私性",
|
|
|
"5. 监管加强：随着应用广泛，相关法律法规将逐步完善",
|
|
|
"6. 低代码/无代码结合：降低AI应用门槛，使更多人能使用LLM能力"
|
|
|
]
|
|
|
return trends
|
|
|
|
|
|
def main():
|
|
|
# 创建分析器实例
|
|
|
analyzer = BilibiliDanmakuAnalyzer()
|
|
|
|
|
|
# 选择是爬取新数据还是加载已有数据
|
|
|
choice = input("请选择操作 (1: 爬取新数据, 2: 加载已有数据): ")
|
|
|
|
|
|
if choice == "1":
|
|
|
# 爬取弹幕数据
|
|
|
analyzer.crawl_all_danmakus()
|
|
|
else:
|
|
|
# 加载已有数据
|
|
|
analyzer.load_danmakus_from_file()
|
|
|
|
|
|
if not analyzer.all_danmakus:
|
|
|
print("没有可用的弹幕数据，程序退出")
|
|
|
return
|
|
|
|
|
|
# 分析应用案例并输出前8名
|
|
|
print("\n===== LLM应用案例排名前8 =====")
|
|
|
top_applications = analyzer.analyze_application_cases(8)
|
|
|
for i, (app, count) in enumerate(top_applications, 1):
|
|
|
print(f"{i}. {app}: {count}次")
|
|
|
|
|
|
# 生成词云图
|
|
|
print("\n===== 生成词云图 =====")
|
|
|
# 可以使用自定义形状作为词云掩码
|
|
|
try:
|
|
|
mask = np.array(Image.open("cloud_mask.png")) # 如果有掩码图片
|
|
|
analyzer.generate_wordcloud(mask)
|
|
|
except:
|
|
|
analyzer.generate_wordcloud()
|
|
|
|
|
|
# 分析用户观点
|
|
|
print("\n===== 用户观点分析 =====")
|
|
|
sentiment = analyzer.analyze_sentiment()
|
|
|
for key, value in sentiment.items():
|
|
|
print(f"{key}: {value}")
|
|
|
|
|
|
# 生成结论
|
|
|
print("\n===== 分析结论 =====")
|
|
|
print("1. 从弹幕讨论来看，B站用户最关注的大语言模型应用领域是：" +
|
|
|
", ".join([app for app, _ in top_applications[:3]]))
|
|
|
|
|
|
if float(sentiment["积极影响讨论占比"].rstrip('%')) > float(sentiment["不利影响讨论占比"].rstrip('%')):
|
|
|
print("2. 整体来看，用户对大语言模型的评价偏向积极，更多讨论其带来的便利和效率提升")
|
|
|
else:
|
|
|
print("2. 整体来看，用户对大语言模型存在较多担忧，主要集中在其可能带来的负面影响")
|
|
|
|
|
|
print("3. 应用领域的讨论最为广泛，说明用户普遍关注大语言模型的实际落地场景")
|
|
|
|
|
|
# 预测发展趋势
|
|
|
print("\n===== 大语言模型应用发展趋势预测 =====")
|
|
|
trends = analyzer.predict_trend()
|
|
|
for trend in trends:
|
|
|
print(trend)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
main() |
