从零开始用Python进行B站视频数据采集

从零开始用Python进行B站视频数据采集

【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs

你是否曾想深入分析B站热门视频的传播规律却受制于技术门槛？是否在寻找一种高效合规的方式获取视频元数据？本文将通过5个核心方法，带你掌握Python采集B站视频数据的完整流程，从API接口调用到数据存储，全方位解决实际采集需求。

方法1：构建B站API请求的基础框架

如何突破B站API的访问限制？B站提供了丰富的开放接口（Application Programming Interface），但直接调用往往面临权限和频率限制。通过Python构建请求框架是解决这一问题的基础。

核心原理

B站API采用RESTful设计风格，大部分接口需要验证信息。通过模拟浏览器请求头（Request Headers）和合理设置请求间隔，可以实现基础数据采集。核心是理解API的参数结构和响应格式。

关键代码

import requests import time from typing import Dict, Optional class BilibiliAPI: def __init__(self): # 初始化请求头，模拟Chrome浏览器 self.headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/98.0.4758.102 Safari/537.36", "Referer": "https://www.bilibili.com/" } self.base_url = "https://api.bilibili.com" self.rate_limit = 2 # 每秒最多请求数 def _request(self, endpoint: str, params: Dict = None) -> Optional[Dict]: """基础请求方法，处理限流和异常""" url = f"{self.base_url}{endpoint}" try: response = requests.get( url, params=params, headers=self.headers, timeout=10 ) response.raise_for_status() # 抛出HTTP错误 time.sleep(1 / self.rate_limit) # 限流控制 return response.json() except requests.exceptions.RequestException as e: print(f"请求错误: {str(e)}") return None def get_video_info(self, aid: int) -> Optional[Dict]: """获取视频基本信息""" return self._request( "/x/web-interface/view", params={"aid": aid} )

避坑指南

🔍操作步骤：

1. 访问B站robots.txt（https://www.bilibili.com/robots.txt）确认允许采集的路径
2. 使用浏览器开发者工具（F12）分析真实请求头
3. 初始设置较低的请求频率（建议rate_limit=2）

⚠️警告：

• 不要使用未经验证的第三方API密钥
• 避免短时间内对同一接口发起超过100次请求
• 商业用途需联系B站官方获取授权

💡优化技巧：

• 使用requests.Session()保持会话连接
• 实现请求重试机制（推荐tenacity库）
• 定期更新User-Agent列表避免特征识别

方法2：突破视频列表分页限制的高效方案

如何高效获取完整的视频列表数据？B站视频列表采用分页加载机制，传统循环分页常因参数错误或反爬限制导致采集中断。

核心原理

B站分页系统主要通过pn（页码）和ps（每页数量）参数控制，部分接口使用offset偏移量机制。理解不同接口的分页逻辑，结合增量采集策略，可以高效获取完整数据。

关键代码

def get_user_videos(self, mid: int, max_page: int = 5) -> list: """获取用户发布的视频列表""" videos = [] page = 1 while page <= max_page: print(f"采集第{page}页视频数据...") response = self._request( "/x/space/arc/search", params={ "mid": mid, # 用户ID "ps": 30, # 每页30条（最大限制） "pn": page, # 当前页码 "order": "pubdate" # 按发布时间排序 } ) if not response or response.get("code") != 0: break data = response.get("data", {}) items = data.get("list", {}).get("vlist", []) if not items: # 没有更多数据 break videos.extend(items) page += 1 return videos

避坑指南

🔍操作步骤：

1. 测试确定目标接口的最大ps值（通常为30或50）
2. 设置合理的max_page上限（建议单用户不超过20页）
3. 实现数据去重机制（基于aid字段）

⚠️警告：

• 单账号日采集上限300条，超过可能触发临时封禁
• 不要在短时间内切换多个用户ID进行采集
• 分页请求间隔应大于1秒

💡优化技巧：

• 记录最后采集时间，实现增量更新
• 使用ps=30（B站允许的最大每页数量）减少请求次数
• 对采集结果按created字段排序验证完整性

方法3：智能解析视频弹幕的完整方案

如何获取有价值的视频弹幕数据？弹幕作为B站特色互动内容，包含丰富的用户反馈信息，但直接采集面临格式解析和实时性挑战。

核心原理

B站弹幕数据存储在特殊格式的XML文件中，通过视频CID（弹幕池ID）获取。弹幕分为历史弹幕和实时弹幕，前者可直接获取，后者需要建立长连接。

关键代码

import xml.etree.ElementTree as ET from datetime import datetime def get_video_danmaku(self, cid: int) -> list: """获取视频历史弹幕""" url = f"https://comment.bilibili.com/{cid}.xml" try: response = requests.get(url, headers=self.headers, timeout=10) response.encoding = "utf-8" # 解析XML弹幕数据 root = ET.fromstring(response.text) danmakus = [] for d in root.findall("d"): # 解析弹幕属性 (格式: 时间,类型,字号,颜色,发送时间,池,用户ID,rowID) attrs = d.attrib["p"].split(",") danmakus.append({ "text": d.text, "time": float(attrs[0]), # 视频内时间(秒) "type": int(attrs[1]), # 弹幕类型(1:滚动,2:顶部,3:底部) "font_size": int(attrs[2]), "color": f"#{int(attrs[3]):06x}", # 颜色转十六进制 "timestamp": datetime.fromtimestamp(float(attrs[4])), "pool": int(attrs[5]) }) return danmakus except Exception as e: print(f"弹幕解析错误: {str(e)}") return []

避坑指南

🔍操作步骤：

1. 通过视频信息接口获取cid（弹幕池ID）
2. 解析XML时注意编码问题（指定utf-8）
3. 对弹幕类型进行分类统计（滚动/顶部/底部）

⚠️警告：

• 单个视频弹幕数量可能超过10万条，需注意内存占用
• 实时弹幕采集需要WebSocket连接，易触发反爬
• 弹幕数据包含用户行为信息，需匿名化处理后使用

💡优化技巧：

• 使用lxml库替代标准库提高XML解析速度
• 按时间切片采集长视频弹幕（如每30分钟一段）
• 过滤重复弹幕（通过内容和时间戳去重）

方法4：合规处理API限流的智能策略

如何在保证合规的前提下最大化采集效率？B站API有严格的访问频率限制，直接采集容易触发429错误（请求过于频繁）。

核心原理

通过动态调整请求间隔、实现智能退避算法和分布式请求策略，可以在遵守平台规则的前提下提高采集效率。关键是识别限流响应并作出自适应调整。

关键代码

from requests.adapters import HTTPAdapter from requests.packages.urllib3.util.retry import Retry import time import random def setup_session(self, retries=3, backoff_factor=0.5): """配置带重试和退避策略的会话""" session = requests.Session() # 配置重试策略 retry_strategy = Retry( total=retries, backoff_factor=backoff_factor, status_forcelist=[429, 500, 502, 503, 504] ) adapter = HTTPAdapter(max_retries=retry_strategy) session.mount("https://", adapter) session.headers.update(self.headers) return session def smart_request(self, session, url, params=None): """智能请求方法，处理限流和动态延迟""" try: response = session.get(url, params=params, timeout=10) # 处理限流响应 if response.status_code == 429: retry_after = int(response.headers.get("Retry-After", 60)) print(f"触发限流，将在{retry_after}秒后重试") time.sleep(retry_after + random.uniform(1, 3)) return self.smart_request(session, url, params) # 递归重试 response.raise_for_status() # 随机延迟，避免固定模式 time.sleep(random.uniform(0.5, 1.5) / self.rate_limit) return response.json() except Exception as e: print(f"智能请求错误: {str(e)}") return None

避坑指南

🔍操作步骤：

1. 实现基于状态码的重试机制（重点处理429状态）
2. 解析Retry-After响应头获取建议等待时间
3. 在请求间隔中加入随机扰动（±20%）

⚠️警告：

• 连续收到3次429响应后应暂停采集至少10分钟
• 不要使用代理IP池进行规避，可能导致账号封禁
• 晚间（23:00-7:00）可适当提高请求频率，白天需降低

💡优化技巧：

• 使用Redis记录请求频率，实现分布式限流
• 针对不同API接口设置差异化的请求频率
• 实现请求优先级队列，核心数据优先采集

方法5：视频评论情感分析的数据预处理

如何从海量评论中提取有效信息？评论数据包含丰富的用户反馈，但非结构化文本需要经过预处理才能用于分析。

核心原理

评论数据预处理包括清洗（去重、去噪）、分词、情感标注等步骤。通过Python的自然语言处理库，可以将原始评论转化为结构化数据，为后续分析奠定基础。

关键代码

import re import jieba import jieba.analyse from snownlp import SnowNLP from collections import defaultdict class CommentProcessor: def __init__(self): # 加载自定义停用词 self.stopwords = set() try: with open("stopwords.txt", "r", encoding="utf-8") as f: self.stopwords = set(f.read().splitlines()) except FileNotFoundError: print("未找到停用词文件，使用默认设置") def clean_comment(self, text: str) -> str: """清洗评论文本""" if not text: return "" # 移除HTML标签和特殊字符 text = re.sub(r"<[^>]*>", "", text) # 移除URL text = re.sub(r"https?://\S+", "", text) # 移除表情符号 text = re.sub(r"[\U00010000-\U0010ffff]", "", text) # 移除多余空格 text = re.sub(r"\s+", " ", text).strip() return text def analyze_sentiment(self, text: str) -> dict: """分析评论情感倾向""" if not text: return {"positive": 0.0, "negative": 0.0, "neutral": 1.0} s = SnowNLP(text) sentiment_score = s.sentiments # 0-1之间，越接近1越积极 if sentiment_score > 0.6: return {"positive": sentiment_score, "negative": 0, "neutral": 1-sentiment_score} elif sentiment_score < 0.4: return {"negative": 1-sentiment_score, "positive": 0, "neutral": sentiment_score} else: return {"neutral": 1.0, "positive": 0, "negative": 0} def extract_keywords(self, text: str, topK=5) -> list: """提取关键词""" if not text: return [] # 使用TF-IDF算法提取关键词 keywords = jieba.analyse.extract_tags( text, topK=topK, withWeight=False, allowPOS=("n", "v", "a") # 只提取名词、动词、形容词 ) return [k for k in keywords if k not in self.stopwords]

避坑指南

🔍操作步骤：

1. 准备中文停用词表（可从网络获取通用停用词表）
2. 安装必要的NLP库：pip install jieba snownlp
3. 对特殊符号和表情进行针对性过滤

⚠️警告：

• 情感分析结果仅作参考，需结合人工校验
• 处理超过10万条评论时需考虑性能优化
• 避免将情感分析结果用于商业决策的唯一依据

💡优化技巧：

• 使用jieba的自定义词典添加领域词汇
• 对长评论进行分段分析后综合结果
• 使用multiprocessing库并行处理评论数据

实战案例一：B站视频推荐系统原型

项目背景

构建一个基于内容特征的视频推荐系统，通过采集视频元数据和用户互动数据，实现相似视频推荐功能。

实现步骤

1. 数据采集模块

# 采集目标视频及其相关视频数据 def collect_recommendation_data(aid: int, depth: int = 2) -> dict: """采集推荐系统所需数据""" api = BilibiliAPI() processor = CommentProcessor() # 1. 获取目标视频信息 video_info = api.get_video_info(aid) if not video_info or video_info.get("code") != 0: print("无法获取视频信息") return {} data = { "target": video_info["data"], "related_videos": [], "comments_analysis": {} } # 2. 获取相关推荐视频 related = api.get_related_videos(aid) if related and related.get("code") == 0: # 只取前10个相关视频 for item in related["data"][:10]: # 递归采集相关视频的基本信息（深度为2） if depth > 0: sub_data = collect_recommendation_data(item["aid"], depth-1) data["related_videos"].append(sub_data) else: data["related_videos"].append(item) # 3. 分析目标视频评论 cid = video_info["data"]["cid"] comments = api.get_video_comments(cid)[:100] # 取前100条评论 # 情感分析和关键词提取 sentiment_stats = defaultdict(int) all_keywords = [] for comment in comments: cleaned = processor.clean_comment(comment["content"]) if not cleaned: continue # 情感分析 sentiment = processor.analyze_sentiment(cleaned) if sentiment["positive"] > 0.6: sentiment_stats["positive"] += 1 elif sentiment["negative"] > 0.6: sentiment_stats["negative"] += 1 else: sentiment_stats["neutral"] += 1 # 关键词提取 keywords = processor.extract_keywords(cleaned) all_keywords.extend(keywords) data["comments_analysis"] = { "sentiment": dict(sentiment_stats), "top_keywords": [k for k, _ in Counter(all_keywords).most_common(10)] } return data

2. 数据存储方案

import json import pandas as pd from sqlalchemy import create_engine def save_recommendation_data(data: dict, storage_type: str = "csv"): """保存推荐系统数据""" if storage_type == "json": # JSON格式，适合完整结构存储 with open("recommendation_data.json", "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) elif storage_type == "csv": # CSV格式，适合数据分析 # 处理视频基本信息 video_df = pd.DataFrame([data["target"]]) video_df.to_csv("video_info.csv", index=False, encoding="utf-8-sig") # 处理相关视频 related_df = pd.DataFrame(data["related_videos"]) related_df.to_csv("related_videos.csv", index=False, encoding="utf-8-sig") # 处理评论分析 comments_df = pd.DataFrame([data["comments_analysis"]["sentiment"]]) comments_df.to_csv("comments_analysis.csv", index=False, encoding="utf-8-sig") elif storage_type == "database": # 数据库存储，适合长期项目 engine = create_engine("sqlite:///bilibili_data.db") # 视频信息表 pd.DataFrame([data["target"]]).to_sql( "videos", engine, if_exists="append", index=False ) # 相关视频表 related_df = pd.DataFrame(data["related_videos"]) related_df["source_aid"] = data["target"]["aid"] related_df.to_sql("related_videos", engine, if_exists="append", index=False)

3. 推荐算法实现

from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics.pairwise import cosine_similarity def recommend_similar_videos(video_data: dict, top_n: int = 5) -> list: """基于内容的视频推荐""" # 提取视频特征：标题+标签+关键词 videos = [video_data["target"]] + video_data["related_videos"] # 创建文本特征 def create_feature_text(video): tags = " ".join(video.get("tags", [])) title = video.get("title", "") keywords = " ".join(video_data["comments_analysis"].get("top_keywords", [])) return f"{title} {tags} {keywords}" # 构建TF-IDF矩阵 texts = [create_feature_text(v) for v in videos] vectorizer = TfidfVectorizer() tfidf_matrix = vectorizer.fit_transform(texts) # 计算相似度（与目标视频的余弦相似度） target_index = 0 cosine_similarities = cosine_similarity( tfidf_matrix[target_index:target_index+1], tfidf_matrix ).flatten() # 获取相似度最高的前N个视频（排除自身） related_docs_indices = cosine_similarities.argsort()[:-top_n-1:-1] recommendations = [] for i in related_docs_indices: if i != target_index: # 排除目标视频本身 recommendations.append({ "aid": videos[i]["aid"], "title": videos[i]["title"], "similarity": float(cosine_similarities[i]) }) return recommendations[:top_n]

技术要点

• 采用混合存储策略：JSON用于完整结构、CSV用于数据分析、SQLite用于长期存储
• 特征工程：融合视频元数据（标题、标签）和评论关键词
• 相似度计算：使用余弦相似度衡量视频内容相似度

实战案例二：B站UP主内容运营助手

项目背景

开发一个帮助UP主分析内容表现和受众反馈的工具，通过采集视频数据和评论，提供数据驱动的内容优化建议。

实现步骤

1. 数据采集与存储

def collect_up_data(mid: int, days: int = 30) -> dict: """采集UP主近期数据""" api = BilibiliAPI() processor = CommentProcessor() # 1. 获取UP主基本信息 up_info = api.get_user_info(mid) if not up_info or up_info.get("code") != 0: print("无法获取UP主信息") return {} # 2. 获取近期视频（30天内） videos = api.get_user_videos(mid) recent_videos = [] cutoff_date = time.time() - days * 86400 # N天前的时间戳 for video in videos: if video["created"] > cutoff_date: # 获取视频详细数据 video_detail = api.get_video_info(video["aid"]) if video_detail and video_detail.get("code") == 0: video_data = video_detail["data"] # 获取评论情感分析 cid = video_data["cid"] comments = api.get_video_comments(cid)[:50] # 取前50条评论 sentiment_stats = defaultdict(int) for comment in comments: cleaned = processor.clean_comment(comment["content"]) if not cleaned: continue sentiment = processor.analyze_sentiment(cleaned) if sentiment["positive"] > 0.6: sentiment_stats["positive"] += 1 elif sentiment["negative"] > 0.6: sentiment_stats["negative"] += 1 else: sentiment_stats["neutral"] += 1 # 添加情感分析结果 video_data["sentiment_analysis"] = dict(sentiment_stats) recent_videos.append(video_data) return { "up_info": up_info["data"], "recent_videos": recent_videos, "collection_date": datetime.now().strftime("%Y-%m-%d %H:%M:%S") }

2. 数据分析与报告生成

import matplotlib.pyplot as plt import seaborn as sns def generate_content_report(data: dict, output_path: str = "content_report.html"): """生成内容分析报告""" if not data or not data["recent_videos"]: print("没有可分析的数据") return # 1. 基础数据统计 videos = data["recent_videos"] df = pd.DataFrame(videos) # 计算互动率 (弹幕数+评论数+点赞数)/播放量 df["interaction_rate"] = ( df["stat"].apply(lambda x: x["danmaku"]) + df["stat"].apply(lambda x: x["reply"]) + df["stat"].apply(lambda x: x["like"]) ) / df["stat"].apply(lambda x: x["view"]) # 2. 可视化分析 plt.figure(figsize=(12, 8)) # 播放量与互动率关系 plt.subplot(2, 2, 1) sns.scatterplot( x=df["stat"].apply(lambda x: x["view"]), y=df["interaction_rate"] ) plt.title("播放量与互动率关系") plt.xlabel("播放量") plt.ylabel("互动率") # 视频时长分布 plt.subplot(2, 2, 2) df["duration"].hist(bins=10) plt.title("视频时长分布") plt.xlabel("时长(秒)") plt.ylabel("视频数量") # 保存图表 plt.tight_layout() plt.savefig("content_analysis.png", dpi=300) # 3. 生成HTML报告 report = f""" <html> <head> <title>UP主内容分析报告</title> <meta charset="utf-8"> <style> body {{ font-family: Arial, sans-serif; margin: 20px; }} .header {{ background-color: #f5f5f5; padding: 10px; }} .metric {{ display: inline-block; margin: 10px; padding: 10px; border: 1px solid #ddd; }} .chart {{ margin: 20px 0; }} </style> </head> <body> <div class="header"> <h1>UP主内容分析报告</h1> <p>生成日期: {data['collection_date']}</p> <p>UP主: {data['up_info']['name']} (粉丝数: {data['up_info']['fans']})</p> </div> <div class="metrics"> <div class="metric"> <h3>平均播放量</h3> <p>{df['stat'].apply(lambda x: x['view']).mean():.0f}</p> </div> <div class="metric"> <h3>平均互动率</h3> <p>{df['interaction_rate'].mean():.2%}</p> </div> <div class="metric"> <h3>最高播放视频</h3> <p>{df.loc[df['stat'].apply(lambda x: x['view']).idxmax()]['title']}</p> </div> </div> <div class="chart"> <h3>内容分析图表</h3> <img src="content_analysis.png" alt="内容分析图表" width="800"> </div> </body> </html> """ with open(output_path, "w", encoding="utf-8") as f: f.write(report) print(f"报告已生成: {output_path}")

技术要点

• 数据指标设计：互动率 = (弹幕数+评论数+点赞数)/播放量
• 可视化方案：使用matplotlib和seaborn生成分析图表
• 报告输出：HTML格式便于查看和分享

伦理采集声明

本指南提供的技术方法仅用于学习研究目的。在进行任何数据采集活动时，请严格遵守以下原则：

1. 合规性原则：在采集前检查目标网站的robots.txt文件，遵守API使用条款，不采集非公开数据
2. 尊重版权：采集的数据不得用于商业用途，引用时需注明来源
3. 合理负载：控制请求频率，避免对目标服务器造成过度负担
4. 数据安全：采集过程中避免获取个人身份信息，对敏感数据进行匿名化处理
5. 法律边界：遵守《网络安全法》《个人信息保护法》等相关法律法规

技术是中性的，责任在于使用者。始终以尊重和负责任的态度进行数据采集活动，共同维护健康的网络生态环境。

通过本文介绍的5个核心方法和两个实战案例，你已经掌握了Python采集B站视频数据的完整流程。从API请求构建到数据存储分析，这些技能不仅适用于B站，也可迁移到其他平台的数据采集任务中。记住，优秀的采集者不仅要掌握技术，更要懂得尊重平台规则和数据伦理。

【免费下载链接】xhs基于小红书 Web 端进行的请求封装。https://reajason.github.io/xhs/项目地址: https://gitcode.com/gh_mirrors/xh/xhs