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

如何用Python高效爬取B站数据：bilibili-api-python实战指南

news 2026/4/16 19:47:32

如何用Python高效爬取B站数据：bilibili-api-python实战指南

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

bilibili-api-python是一个功能强大的Python异步库，专门用于调用B站的各种API接口，支持视频、用户、直播、动态、专栏、番剧等400+个API调用。对于需要批量处理B站数据的开发者来说，这个库能极大提升数据获取效率，同时提供完善的错误处理和异步并发支持。

痛点：传统爬虫在B站数据采集中的困境

大多数开发者在处理B站数据时都会遇到以下问题：

  1. API接口分散：B站官方API文档不完整，接口分散在不同域名下
  2. 反爬虫机制复杂：B站有完善的WAF防护和频率限制
  3. 数据结构不一致：不同模块返回的数据格式差异大
  4. 异步处理困难：传统同步请求无法充分利用网络资源

这些问题导致开发效率低下，代码维护困难。bilibili-api-python正是为解决这些问题而生。

核心功能架构解析

上图展示了B站投票功能的前端HTML结构，这反映了B站API返回数据的复杂性。bilibili-api-python通过统一的接口封装，将这种复杂的数据结构转化为易于使用的Python对象。

模块化设计

项目采用模块化设计，每个功能模块独立封装：

  • 视频模块(video.py)：处理视频信息、弹幕、字幕等
  • 用户模块(user.py)：用户信息、关注关系、动态等
  • 直播模块(live.py)：直播信息、弹幕WebSocket连接
  • 动态模块(dynamic.py)：动态发布、解析、互动
  • 认证模块(login_v2.py)：多种登录方式支持

实战场景：构建视频数据分析工具

场景一：批量获取视频统计信息

假设你需要分析某个UP主的所有视频数据，传统方法需要手动解析每个页面。使用bilibili-api-python可以轻松实现：

import asyncio from bilibili_api import video, user from bilibili_api.utils import Credential async def analyze_up_videos(uid: int, credential: Credential): """分析UP主所有视频数据""" up = user.User(uid, credential) # 获取用户所有视频 all_videos = [] page = 1 while True: videos_data = await up.get_videos(pn=page, ps=30) if not videos_data['list']['vlist']: break for v_info in videos_data['list']['vlist']: video_obj = video.Video(bvid=v_info['bvid'], credential=credential) video_detail = await video_obj.get_info() # 获取详细统计信息 stats = { 'bvid': v_info['bvid'], 'title': v_info['title'], 'views': video_detail['stat']['view'], 'likes': video_detail['stat']['like'], 'coins': video_detail['stat']['coin'], 'favorites': video_detail['stat']['favorite'], 'comments': video_detail['stat']['reply'], 'share': video_detail['stat']['share'], 'pubdate': video_detail['pubdate'], 'duration': video_detail['duration'], 'tags': await video_obj.get_tags() } all_videos.append(stats) page += 1 await asyncio.sleep(0.5) # 避免请求过快 return all_videos # 使用示例 async def main(): # 创建认证对象（需要先获取cookies） credential = Credential( sessdata="your_sessdata", bili_jct="your_bili_jct", buvid3="your_buvid3" ) # 分析UP主视频 videos_data = await analyze_up_videos(12345678, credential) # 数据统计 total_views = sum(v['views'] for v in videos_data) avg_like_rate = sum(v['likes'] for v in videos_data) / total_views * 100 print(f"总播放量: {total_views:,}") print(f"平均点赞率: {avg_like_rate:.2f}%") if __name__ == "__main__": asyncio.run(main())

场景二：实时监控直播弹幕

对于需要实时处理直播弹幕的应用，如舆情监控或互动分析：

import asyncio from bilibili_api import live from bilibili_api.utils import Credential class LiveDanmakuMonitor: """直播弹幕监控器""" def __init__(self, room_id: int, credential: Credential): self.room = live.LiveDanmaku(room_id, credential=credential) self.credential = credential async def start_monitoring(self): """开始监控弹幕""" @self.room.on('DANMU_MSG') async def on_danmaku(event): """处理弹幕消息""" info = event['data']['info'] user_info = info[2] danmaku_content = info[1] print(f"[弹幕] {user_info[1]}: {danmaku_content}") # 这里可以添加业务逻辑，如： # 1. 关键词过滤 # 2. 用户行为分析 # 3. 弹幕统计 # 4. 实时存储到数据库 @self.room.on('SEND_GIFT') async def on_gift(event): """处理礼物消息""" data = event['data']['data'] print(f"[礼物] {data['uname']} 赠送了 {data['giftName']} x{data['num']}") # 连接直播间 await self.room.connect() print(f"已连接到直播间 {self.room.get_room_id()}") # 保持连接 try: while True: await asyncio.sleep(1) except KeyboardInterrupt: await self.room.disconnect() print("监控已停止") # 使用示例 async def main(): credential = Credential( sessdata="your_sessdata", bili_jct="your_bili_jct" ) monitor = LiveDanmakuMonitor(123456, credential) await monitor.start_monitoring() if __name__ == "__main__": asyncio.run(main())

性能优化技巧

1. 异步并发请求优化

import asyncio from bilibili_api import video from typing import List async def batch_get_video_info(bvids: List[str], max_concurrent: int = 5): """批量获取视频信息，控制并发数""" semaphore = asyncio.Semaphore(max_concurrent) async def get_single_video(bvid: str): async with semaphore: v = video.Video(bvid=bvid) try: info = await v.get_info() return {'bvid': bvid, 'success': True, 'data': info} except Exception as e: return {'bvid': bvid, 'success': False, 'error': str(e)} finally: await asyncio.sleep(0.1) # 请求间隔 tasks = [get_single_video(bvid) for bvid in bvids] results = await asyncio.gather(*tasks, return_exceptions=True) return results

2. 请求客户端选择策略

bilibili-api-python支持多种HTTP客户端，根据场景选择：

客户端优点缺点适用场景
curl_cffi支持TLS指纹伪装依赖C库高防网站
aiohttp纯Python，异步性能好不支持WebSocket通用场景
httpxHTTP/2支持性能略低需要HTTP/2
from bilibili_api import select_client, request_settings # 选择最佳客户端 select_client("curl_cffi") # 推荐用于生产环境 # 设置代理（绕过IP限制） request_settings.set_proxy("http://your-proxy.com:8080") # 设置超时 request_settings.set_timeout(30.0)

3. 缓存策略实现

from functools import lru_cache import asyncio from bilibili_api import video class VideoCache: """视频信息缓存管理器""" def __init__(self, ttl: int = 300): # 5分钟缓存 self.ttl = ttl self.cache = {} async def get_video_info(self, bvid: str, force_refresh: bool = False): """获取视频信息（带缓存）""" current_time = asyncio.get_event_loop().time() if not force_refresh and bvid in self.cache: cached_data, timestamp = self.cache[bvid] if current_time - timestamp < self.ttl: return cached_data # 缓存未命中或过期，重新获取 v = video.Video(bvid=bvid) info = await v.get_info() # 更新缓存 self.cache[bvid] = (info, current_time) return info def clear_cache(self): """清空缓存""" self.cache.clear()

常见陷阱与调试技巧

陷阱1：频率限制触发

B站有严格的频率限制，过快请求会导致412错误：

# ❌ 错误做法：无间隔连续请求 async def bad_example(): for bvid in bvid_list: info = await video.Video(bvid=bvid).get_info() # 可能触发反爬 # ✅ 正确做法：添加适当延迟 async def good_example(): for bvid in bvid_list: info = await video.Video(bvid=bvid).get_info() await asyncio.sleep(1) # 1秒间隔

陷阱2：Cookie过期处理

from bilibili_api.exceptions import CookiesRefreshException from bilibili_api.utils import Credential async def safe_api_call(api_func, *args, **kwargs): """安全的API调用，处理Cookie过期""" try: return await api_func(*args, **kwargs) except CookiesRefreshException: print("Cookie已过期，需要重新登录") # 这里可以实现自动刷新逻辑 raise except ResponseCodeException as e: if e.code == -101: # 未登录 print("需要登录才能访问此API") raise else: raise

陷阱3：异步上下文管理

# ❌ 错误做法：忘记关闭连接 async def leak_example(): room = live.LiveDanmaku(123456) await room.connect() # 如果异常退出，连接不会关闭 # ✅ 正确做法：使用async with async def safe_example(): async with live.LiveDanmaku(123456) as room: @room.on('DANMU_MSG') async def on_danmaku(event): print(event) await asyncio.sleep(60) # 监听60秒 # 自动关闭连接

扩展应用场景

1. 内容监控系统

from bilibili_api import video, search import asyncio from datetime import datetime, timedelta class ContentMonitor: """内容监控系统""" def __init__(self, keywords: List[str], interval: int = 300): self.keywords = keywords self.interval = interval self.last_check = {} async def monitor_new_videos(self): """监控新发布的视频""" while True: for keyword in self.keywords: # 搜索最新视频 results = await search.search_by_type( keyword=keyword, search_type=search.SearchObjectType.VIDEO, order_type=search.OrderVideo.PUBDATE, time_range=1 # 最近1天 ) for item in results['result']: video_time = datetime.fromtimestamp(item['pubdate']) # 检查是否为新视频 if keyword not in self.last_check or video_time > self.last_check[keyword]: print(f"发现新视频: {item['title']}") # 发送通知或执行其他操作 self.last_check[keyword] = datetime.now() await asyncio.sleep(self.interval)

2. 数据分析流水线

import pandas as pd from bilibili_api import video, user from typing import Dict, List class BiliDataPipeline: """B站数据分析流水线""" def __init__(self): self.dataframes = {} async def collect_user_data(self, uid: int, credential) -> pd.DataFrame: """收集用户数据""" u = user.User(uid, credential) user_info = await u.get_user_info() up_stat = await u.get_up_stat() data = { 'uid': uid, 'name': user_info['name'], 'level': user_info['level'], 'fans': up_stat['follower'], 'following': up_stat['following'], 'video_count': up_stat['archive']['count'], 'total_views': up_stat['archive']['view'], 'collect_time': pd.Timestamp.now() } return pd.DataFrame([data]) async def collect_video_stats(self, bvid: str, credential) -> pd.DataFrame: """收集视频统计数据""" v = video.Video(bvid=bvid, credential=credential) info = await v.get_info() stat = info['stat'] data = { 'bvid': bvid, 'title': info['title'], 'views': stat['view'], 'likes': stat['like'], 'coins': stat['coin'], 'favorites': stat['favorite'], 'comments': stat['reply'], 'pubdate': pd.Timestamp(info['pubdate'], unit='s'), 'collect_time': pd.Timestamp.now() } return pd.DataFrame([data]) def analyze_trends(self, df: pd.DataFrame): """分析趋势""" # 计算增长率 df['view_growth'] = df['views'].pct_change() df['like_rate'] = df['likes'] / df['views'] df['coin_rate'] = df['coins'] / df['views'] return df

最佳实践总结

1. 认证信息管理

# 安全存储认证信息 import os from dotenv import load_dotenv from bilibili_api.utils import Credential load_dotenv() class CredentialManager: """认证信息管理器""" @staticmethod def get_credential(): """从环境变量获取认证信息""" return Credential( sessdata=os.getenv('BILI_SESSDATA'), bili_jct=os.getenv('BILI_JCT'), buvid3=os.getenv('BILI_BUVID3'), dedeuserid=os.getenv('BILI_DEDEUSERID') ) @staticmethod def validate_credential(credential: Credential) -> bool: """验证认证信息是否有效""" try: credential.check_valid() return True except: return False

2. 错误处理策略

import asyncio import logging from bilibili_api.exceptions import * logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class BiliAPIWrapper: """API包装器，统一错误处理""" def __init__(self, max_retries: int = 3): self.max_retries = max_retries async def call_with_retry(self, api_func, *args, **kwargs): """带重试的API调用""" for attempt in range(self.max_retries): try: return await api_func(*args, **kwargs) except NetworkException as e: logger.warning(f"网络错误，第{attempt+1}次重试: {e}") if attempt == self.max_retries - 1: raise await asyncio.sleep(2 ** attempt) # 指数退避 except ResponseCodeException as e: if e.code == -412: # 频率限制 logger.warning("触发频率限制，等待10秒") await asyncio.sleep(10) continue else: raise

3. 性能监控

import time from functools import wraps from typing import Callable def monitor_performance(func: Callable): """性能监控装饰器""" @wraps(func) async def wrapper(*args, **kwargs): start_time = time.time() try: result = await func(*args, **kwargs) elapsed = time.time() - start_time print(f"{func.__name__} 执行时间: {elapsed:.2f}秒") return result except Exception as e: elapsed = time.time() - start_time print(f"{func.__name__} 执行失败，耗时: {elapsed:.2f}秒，错误: {e}") raise return wrapper # 使用示例 @monitor_performance async def get_video_info_with_monitor(bvid: str): v = video.Video(bvid=bvid) return await v.get_info()

扩展阅读

相关模块深度探索

  1. 弹幕处理(bilibili_api/utils/danmaku.py)

    • 弹幕格式解析
    • ASS字幕生成
    • 弹幕时间轴处理

  2. 认证系统(bilibili_api/login_v2.py)

    • 二维码登录实现
    • 短信验证码登录
    • Cookie自动刷新机制

  3. 网络层(bilibili_api/utils/network.py)

    • 多客户端支持
    • 请求拦截器
    • WebSocket连接管理

项目架构建议

对于大型项目，建议采用以下架构：

project/ ├── src/ │ ├── core/ # 核心业务逻辑 │ │ ├── crawlers/ # 爬虫模块 │ │ ├── analyzers/ # 分析模块 │ │ └── exporters/ # 导出模块 │ ├── utils/ # 工具函数 │ │ ├── cache.py # 缓存管理 │ │ ├── logger.py # 日志配置 │ │ └── config.py # 配置管理 │ └── models/ # 数据模型 │ ├── video.py # 视频模型 │ ├── user.py # 用户模型 │ └── danmaku.py # 弹幕模型 ├── tests/ # 测试用例 │ ├── integration/ # 集成测试 │ └── unit/ # 单元测试 └── examples/ # 使用示例 └── config/ # 配置示例

通过合理使用bilibili-api-python，你可以构建出功能强大、性能优越的B站数据应用。无论是数据分析、内容监控还是自动化工具，这个库都能提供坚实的基础设施支持。

关键要点回顾：

  • 始终使用异步编程模式提升性能
  • 合理控制请求频率避免被封禁
  • 实现完善的错误处理和重试机制
  • 利用缓存减少重复请求
  • 根据场景选择合适的HTTP客户端

掌握这些技巧后，你将能够高效、稳定地处理B站的各种数据需求，构建出专业级的应用系统。

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

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

http://www.jsqmd.com/news/651676/

相关文章：

  • 从一次线上故障复盘：聊聊Nginx的upstream配置里，用IP和用服务名到底有啥区别？
  • Windows系统解锁HEIC缩略图：技术魔法让iPhone照片瞬间可见
  • Python自动化抢票脚本：3步搞定大麦网热门演出票务
  • GetQzonehistory终极指南：3步永久保存你的QQ空间青春记忆
  • Scanpy单细胞绘图功能实战全解：从核心函数到高级可视化
  • 厦门具身智能产业联盟：中国具身智能产业指数（EAII）——2026年度洞察报告
  • 如何在浏览器中轻松查看20多种3D模型格式？Online3DViewer完全指南
  • 【高届数EI稳定检索、IEEE出版，往届会后4个月EIScopus检索、多位IEEE Fellow主讲报告、优秀论文可获荐至合作期刊】第十二届传感云和边缘计算系统国际会议（SCECS 2026）
  • 5分钟掌握TMSpeech：Windows本地实时语音转文字的终极方案
  • CFCA精品可可设计师中级认证课程掌控：驾驭奶糖变量，构筑绝对可控的配方结构边界
  • 何超一行走访容积视觉 共探AI元宇宙与数字文旅融合发展
  • 从Shebang行到py.ini：彻底搞懂Windows上Python脚本的版本指定机制
  • 故障发现效率优异，告警响应速度有待优化
  • 小红书数据采集Python爬虫：3个核心问题与开源解决方案
  • AI Agent中的Memory机制：从理论到实践的全方位解析
  • 避坑指南：PADS9.5环境变量设置常见误区与正确破解姿势
  • 从脉冲密度到数字音频：深入解析PDM的编码奥秘与实现
  • 别再暴力求和了！用前缀和算法5分钟搞定LeetCode区间查询题（附Python/Java代码）
  • 构建基于Qwen Coder的上下文工程框架：标准化AI辅助开发的实践路径
  • 从源头到浏览器：net::ERR_INCOMPLETE_CHUNKED_ENCODING 200 (OK) 全链路排查指南
  • SVN：Checkout Depth
  • 【SPIE出版，往届已EI检索 | 复旦大学正式加入本次会议主办单位阵容 | 多所实验室高校加入会议支持单位 | 多位实力嘉宾加盟大会主讲】第二届先进半导体与通信国际学术会议（ICASC 2026）
  • 告别硬编码！用STM32F407+双向链表实现可无限扩展的菜单系统（附完整工程）
  • OneNote Md Exporter：轻松将OneNote笔记本转换为Markdown格式
  • 【语音识别】基于MFCC特征提取和机器学习分类技术语音信号情绪检测系统附Matlab代码
  • 鹏展-penggeon
  • 树--二叉树
  • 从jQuery到Vue3：我的项目架构升级踩坑记，聊聊MVC和MVVM的真实应用场景选择
  • 深度解析CaptfEncoder V3：跨平台网络安全工具套件的终极实战指南
  • AI-Shoujo HF Patch终极指南：5分钟解锁完整游戏体验