当前位置：首页 > news >正文

开源项目 bilibili-api 评论系统深度探索与实战指南

news 2026/8/24 9:07:52

开源项目 bilibili-api 评论系统深度探索与实战指南

【免费下载链接】bilibili-api哔哩哔哩常用API调用。支持视频、番剧、用户、频道、音频等功能。原仓库地址：https://github.com/MoyuScript/bilibili-api项目地址: https://gitcode.com/gh_mirrors/bi/bilibili-api

在当今的社交媒体和视频平台中，评论系统是用户互动和数据获取的核心功能。对于开发者而言，如何高效、稳定地获取和处理评论数据成为了技术挑战。本文将深入探索开源项目 bilibili-api 的评论系统实现，从架构设计到高级应用，为开发者提供完整的实战指南。

架构设计：模块化与异步化的完美结合

bilibili-api 的评论系统采用了高度模块化的设计，将不同类型的评论资源抽象为统一的接口。通过CommentResourceType枚举类，系统支持视频、专栏、动态、音频、歌单、课程、小黑屋、漫画、活动等九种资源类型的评论操作。

class CommentResourceType(Enum): VIDEO = 1 ARTICLE = 12 DYNAMIC_DRAW = 11 DYNAMIC = 17 AUDIO = 14 AUDIO_LIST = 19 CHEESE = 33 BLACK_ROOM = 6 MANGA = 22 ACTIVITY = 4

这种设计使得开发者可以通过统一的接口处理不同类型的评论，大大降低了学习成本和使用复杂度。每个资源类型都对应着B站内部不同的API端点，但对外提供了完全一致的调用方式。

核心模块拆解：双接口策略的智慧

传统分页接口：get_comments

传统接口采用经典的页码分页模式，适合对分页逻辑有精确控制需求的场景：

async def get_comments( oid: int, type_: CommentResourceType, page_index: int = 1, order: OrderType = OrderType.TIME, credential: Union[Credential, None] = None, ) -> dict:

该接口直接对应B站的API["comment"]["get"]端点，通过page_index参数控制分页，逻辑直观易懂。然而，随着B站反爬机制的升级，这种接口在某些情况下可能面临403错误的风险。

新一代懒加载接口：get_comments_lazy

针对传统接口的局限性，项目引入了更先进的get_comments_lazy接口，采用游标式分页设计：

async def get_comments_lazy( oid: int, type_: CommentResourceType, offset: str = "", order: OrderType = OrderType.TIME, credential: Union[Credential, None] = None, ) -> dict:

这个接口的核心创新在于offset参数的使用。每次请求返回的结果中都包含下一次请求所需的偏移量，形成一个单向链表结构。这种设计不仅更符合现代API的设计理念，还能更好地应对B站的反爬机制。

图示：B站富文本投票模块的前端实现，展示了评论系统中互动功能的复杂性

实战应用：构建智能评论监控系统

实时评论流处理

利用get_comments_lazy接口的游标特性，我们可以构建高效的实时评论监控系统：

import asyncio from bilibili_api import comment, Credential from datetime import datetime class CommentMonitor: def __init__(self, video_id: int, credential: Credential): self.video_id = video_id self.credential = credential self.last_offset = "" self.comment_cache = {} async def monitor_comments(self, interval: int = 30): """实时监控视频评论""" while True: try: result = await comment.get_comments_lazy( oid=self.video_id, type_=comment.CommentResourceType.VIDEO, offset=self.last_offset, credential=self.credential ) new_comments = self._process_new_comments(result) self._analyze_comment_trends(new_comments) self.last_offset = result["cursor"]["pagination_reply"]["next_offset"] await asyncio.sleep(interval) except Exception as e: print(f"监控异常: {e}") await asyncio.sleep(interval * 2) def _process_new_comments(self, result: dict) -> list: """处理新评论并去重""" new_comments = [] for reply in result.get("replies", []): rpid = reply["rpid"] if rpid not in self.comment_cache: self.comment_cache[rpid] = { "content": reply["content"]["message"], "user": reply["member"]["uname"], "timestamp": datetime.now() } new_comments.append(reply) return new_comments

评论情感分析集成

结合第三方NLP库，我们可以构建评论情感分析系统：

from textblob import TextBlob from bilibili_api import comment class CommentSentimentAnalyzer: def __init__(self): self.sentiment_cache = {} async def analyze_video_comments(self, video_id: int, max_pages: int = 10): """分析视频评论情感倾向""" all_comments = [] offset = "" page_count = 0 while page_count < max_pages: result = await comment.get_comments_lazy( oid=video_id, type_=comment.CommentResourceType.VIDEO, offset=offset ) comments = result.get("replies", []) sentiment_scores = [] for cmt in comments: text = cmt["content"]["message"] sentiment = self._analyze_sentiment(text) sentiment_scores.append(sentiment) all_comments.append({ "text": text, "sentiment": sentiment, "user": cmt["member"]["uname"] }) if not result["cursor"]["pagination_reply"]["next_offset"]: break offset = result["cursor"]["pagination_reply"]["next_offset"] page_count += 1 return self._generate_sentiment_report(all_comments)

性能优化与调优技巧

1. 连接池管理优化

对于大规模评论抓取任务，合理的连接池配置至关重要：

import aiohttp from bilibili_api import network class OptimizedCommentFetcher: def __init__(self, max_connections: int = 10): connector = aiohttp.TCPConnector( limit=max_connections, ttl_dns_cache=300, force_close=False ) self.session = aiohttp.ClientSession(connector=connector) async def fetch_comments_batch(self, video_ids: list): """批量获取多个视频的评论""" tasks = [] for vid in video_ids: task = self._fetch_single_video_comments(vid) tasks.append(task) results = await asyncio.gather(*tasks, return_exceptions=True) return results

2. 智能重试机制

针对B站API的不稳定性，实现智能重试策略：

import random from tenacity import retry, stop_after_attempt, wait_exponential class ResilientCommentClient: @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10) ) async def get_comments_with_retry(self, oid: int, type_, offset=""): """带指数退避的重试机制""" try: return await comment.get_comments_lazy( oid=oid, type_=type_, offset=offset ) except Exception as e: if "403" in str(e): # 添加随机延迟避免触发反爬 await asyncio.sleep(random.uniform(2, 5)) raise

问题诊断：从底层原理角度分析

1. 403错误的根本原因分析

403错误通常源于B站的WBI签名验证机制。bilibili-api在底层自动处理了WBI签名，但开发者仍需注意：

认证信息时效性：Cookies的有效期有限，需要定期刷新
请求频率限制：B站对未认证请求有严格的频率限制
User-Agent策略：合理的User-Agent轮换可以减少被识别为爬虫的风险

2. 评论数据不完整的排查

当获取的评论数据不完整时，可以从以下角度排查：

async def debug_comment_fetching(oid: int): """调试评论获取问题""" # 检查认证状态 credential = Credential(sessdata="your_sessdata") if not credential.has_sessdata(): print("警告：认证信息可能已过期") # 测试基础API连通性 try: result = await comment.get_comments_lazy( oid=oid, type_=comment.CommentResourceType.VIDEO, credential=credential ) # 分析返回数据结构 print(f"总评论数: {result.get('cursor', {}).get('all_count', 0)}") print(f"实际获取数: {len(result.get('replies', []))}") print(f"下一页偏移量: {result.get('cursor', {}).get('pagination_reply', {}).get('next_offset', '无')}") except Exception as e: print(f"API调用失败: {e}") # 检查网络代理设置 print("建议检查网络代理或尝试使用不同的网络环境")

生态整合：与其他数据处理工具的无缝对接

1. 与Pandas的数据分析集成

import pandas as pd from bilibili_api import comment class CommentDataFrameBuilder: async def build_comment_dataframe(self, video_id: int, pages: int = 5): """将评论数据转换为Pandas DataFrame""" comments_data = [] offset = "" for _ in range(pages): result = await comment.get_comments_lazy( oid=video_id, type_=comment.CommentResourceType.VIDEO, offset=offset ) for reply in result.get("replies", []): comments_data.append({ "rpid": reply["rpid"], "user": reply["member"]["uname"], "user_id": reply["member"]["mid"], "content": reply["content"]["message"], "like_count": reply["like"], "ctime": pd.to_datetime(reply["ctime"], unit='s'), "reply_count": reply.get("rcount", 0) }) offset = result["cursor"]["pagination_reply"]["next_offset"] if not offset: break return pd.DataFrame(comments_data)

2. 与Elasticsearch的搜索集成

from elasticsearch import AsyncElasticsearch from bilibili_api import comment class CommentSearchEngine: def __init__(self, es_client: AsyncElasticsearch): self.es = es_client async def index_video_comments(self, video_id: int, index_name: str = "bilibili_comments"): """将视频评论索引到Elasticsearch""" offset = "" batch_size = 100 while True: result = await comment.get_comments_lazy( oid=video_id, type_=comment.CommentResourceType.VIDEO, offset=offset ) comments = result.get("replies", []) if not comments: break # 批量索引评论 actions = [] for cmt in comments: doc = { "video_id": video_id, "rpid": cmt["rpid"], "user": cmt["member"]["uname"], "content": cmt["content"]["message"], "like_count": cmt["like"], "timestamp": cmt["ctime"] } actions.append({ "_index": index_name, "_source": doc }) await self.es.bulk(actions) offset = result["cursor"]["pagination_reply"]["next_offset"]