B站评论API 2024实战Selenium Requests 双方案爬取与WBI签名破解指南1. 技术背景与挑战2024年B站对评论API进行了重大升级引入了WBI签名验证机制和动态分页策略。传统爬虫方法如直接请求API或解析网页源码已无法稳定获取数据。根据实测当前主要面临三大技术难点WBI签名验证所有请求需携带动态生成的w_rid和wts参数动态分页机制新版API采用pagination_str替代传统页码参数请求频率限制未登录状态下单个IP限制为5次/分钟重要提示爬取行为需遵守B站用户协议建议控制请求频率在3次/秒以下避免对服务器造成过大压力2. 环境准备与依赖安装2.1 基础环境配置# 创建Python虚拟环境 python -m venv bili_crawler source bili_crawler/bin/activate # Linux/Mac bili_crawler\Scripts\activate # Windows # 安装核心依赖 pip install selenium4.15.0 requests2.31.0 undetected-chromedriver3.5.0 pandas2.1.02.2 浏览器驱动配置浏览器驱动下载配置方法ChromeChromedriver添加PATH或指定路径EdgeEdgedriver同上FirefoxGeckodriver同上3. Selenium自动化方案实现3.1 完整登录流程from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC import undetected_chromedriver as uc def bilibili_login(username, password): options uc.ChromeOptions() options.add_argument(--disable-blink-featuresAutomationControlled) driver uc.Chrome(optionsoptions) driver.get(https://passport.bilibili.com/login) # 显式等待元素加载 wait WebDriverWait(driver, 15) user_elem wait.until(EC.presence_of_element_located((By.ID, login-username))) pwd_elem wait.until(EC.presence_of_element_located((By.ID, login-passwd))) # 模拟人类输入 for char in username: user_elem.send_keys(char) time.sleep(random.uniform(0.05, 0.2)) for char in password: pwd_elem.send_keys(char) time.sleep(random.uniform(0.05, 0.2)) # 点击登录按钮 login_btn wait.until(EC.element_to_be_clickable((By.CLASS_NAME, btn-login))) login_btn.click() # 验证登录成功 wait.until(EC.url_contains(bilibili.com)) return driver3.2 请求拦截与数据处理from selenium.webdriver.common.desired_capabilities import DesiredCapabilities def setup_request_interception(driver): # 开启网络请求记录 caps DesiredCapabilities.CHROME caps[goog:loggingPrefs] {performance: ALL} driver.execute_cdp_cmd(Network.enable, {}) # 添加请求拦截器 driver.execute_cdp_cmd(Network.setRequestInterception, { patterns: [{ urlPattern: *api.bilibili.com/x/v2/reply*, resourceType: XHR }] }) def extract_comments_from_logs(driver, video_url): driver.get(video_url) time.sleep(5) # 等待页面加载 # 获取网络日志 logs driver.get_log(performance) comments [] for entry in logs: try: message json.loads(entry[message])[message] if message[method] Network.responseReceived: url message[params][response][url] if reply in url: request_id message[params][requestId] response driver.execute_cdp_cmd(Network.getResponseBody, {requestId: request_id}) data json.loads(response[body]) comments.extend(process_comment_data(data)) except Exception as e: continue return comments4. Requests直接请求方案4.1 WBI签名破解实现import hashlib import time from urllib.parse import urlencode def get_wbi_keys(): 获取最新的img_key和sub_key nav_url https://api.bilibili.com/x/web-interface/nav response requests.get(nav_url) data response.json() img_key data[data][wbi_img][img_url].split(/)[-1].split(.)[0] sub_key data[data][wbi_img][sub_url].split(/)[-1].split(.)[0] return img_key, sub_key def generate_wbi_sign(params, img_key, sub_key): 生成WBI签名 mixin_key [ 46, 47, 18, 2, 53, 8, 23, 32, 15, 50, 10, 31, 58, 3, 45, 35, 27, 43, 5, 49, 33, 9, 42, 19, 29, 28, 14, 39, 12, 38, 41, 13, 37, 48, 7, 16, 24, 55, 40, 61, 26, 17, 0, 1, 60, 51, 30, 4, 22, 25, 54, 21, 56, 59, 6, 63, 57, 62, 11, 36, 20, 34, 44, 52 ] orig_key img_key sub_key mixed_key .join([orig_key[i] for i in mixin_key])[:32] params[wts] int(time.time()) sorted_params sorted(params.items()) query urlencode(sorted_params) sign hashlib.md5((query mixed_key).encode()).hexdigest() return f{query}w_rid{sign}4.2 分页爬取完整实现def get_bilibili_comments(oid, max_pages10): 获取B站视频评论新版API img_key, sub_key get_wbi_keys() base_url https://api.bilibili.com/x/v2/reply/wbi/main headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Referer: fhttps://www.bilibili.com/video/av{oid} } all_comments [] pagination_str None for _ in range(max_pages): params { oid: oid, type: 1, mode: 3, # 排序模式3热度 plat: 1, web_location: 1315875 } if pagination_str: params[pagination_str] pagination_str signed_url base_url ? generate_wbi_sign(params, img_key, sub_key) response requests.get(signed_url, headersheaders) if response.status_code 200: data response.json() comments data.get(data, {}).get(replies, []) all_comments.extend(process_comments(comments)) # 更新分页参数 pagination_str data.get(data, {}).get(cursor, {}).get(pagination_str) if not pagination_str: break time.sleep(1) # 请求间隔 else: print(f请求失败: {response.status_code}) break return all_comments5. 数据处理与存储优化5.1 评论数据结构化处理def process_comments(comments): processed [] for comment in comments: processed.append({ rpid: comment[rpid], oid: comment[oid], ctime: datetime.fromtimestamp(comment[ctime]).strftime(%Y-%m-%d %H:%M:%S), like: comment[like], content: comment[content][message], member: { mid: comment[member][mid], uname: comment[member][uname], level: comment[member][level_info][current_level], vip: comment[member][vip][vipStatus] }, reply_control: { location: comment[reply_control].get(location, ) } }) return processed5.2 数据存储方案对比存储方式优点缺点适用场景CSV简单易用无需额外服务查询效率低不支持复杂操作小规模数据快速导出SQLite轻量级单文件存储并发性能有限本地中规模数据存储MySQL支持复杂查询性能好需要独立服务大规模生产环境MongoDB灵活Schema适合非结构化数据内存占用较高评论内容分析场景SQLite存储示例import sqlite3 from contextlib import closing def init_db(db_pathcomments.db): with closing(sqlite3.connect(db_path)) as conn: conn.execute( CREATE TABLE IF NOT EXISTS comments ( rpid INTEGER PRIMARY KEY, oid INTEGER, ctime TEXT, like_count INTEGER, content TEXT, uname TEXT, user_level INTEGER, is_vip INTEGER, ip_location TEXT ) ) conn.commit() def save_to_sqlite(comments, db_pathcomments.db): with closing(sqlite3.connect(db_path)) as conn: cursor conn.cursor() for comment in comments: cursor.execute( INSERT OR REPLACE INTO comments VALUES (?,?,?,?,?,?,?,?,?) , ( comment[rpid], comment[oid], comment[ctime], comment[like], comment[content], comment[member][uname], comment[member][level], comment[member][vip], comment[reply_control][location] )) conn.commit()6. 双方案对比与性能优化6.1 技术指标对比指标Selenium方案Requests方案稳定性★★★★☆ (需处理动态加载)★★★☆☆ (依赖API稳定性)速度★★☆☆☆ (受浏览器性能限制)★★★★☆ (直接HTTP请求)反爬绕过能力★★★★★ (模拟真实浏览器)★★★☆☆ (需处理签名)资源消耗高 (启动浏览器实例)低 (纯网络请求)数据完整性★★★★☆ (可获取页面全部数据)★★★☆☆ (受API限制)6.2 性能优化技巧并发控制使用aiohttp实现异步请求import aiohttp import asyncio async def fetch_comments(session, url): async with session.get(url) as response: return await response.json() async def batch_fetch(urls): async with aiohttp.ClientSession() as session: tasks [fetch_comments(session, url) for url in urls] return await asyncio.gather(*tasks)请求缓存使用requests-cache减少重复请求from requests_cache import CachedSession session CachedSession(bili_cache, expire_after3600) # 缓存1小时 response session.get(api_url)智能重试使用tenacity实现指数退避重试from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def safe_request(url): response requests.get(url, timeout10) response.raise_for_status() return response7. 可视化分析实战7.1 基础数据统计import pandas as pd import matplotlib.pyplot as plt def analyze_comments(comments): df pd.DataFrame(comments) # 基础统计 print(f总评论数: {len(df)}) print(f平均点赞数: {df[like].mean():.1f}) print(fVIP用户占比: {(df[is_vip] 0).mean()*100:.1f}%) # 时间分布分析 df[ctime] pd.to_datetime(df[ctime]) df[hour] df[ctime].dt.hour hourly_counts df.groupby(hour).size() plt.figure(figsize(10, 5)) hourly_counts.plot(kindbar, color#FB7299) plt.title(评论时间分布) plt.xlabel(小时) plt.ylabel(评论数) plt.savefig(hourly_dist.png, dpi300, bbox_inchestight)7.2 高级情感分析from snownlp import SnowNLP def sentiment_analysis(comments): sentiments [] for comment in comments: s SnowNLP(comment[content]) sentiments.append(s.sentiments) plt.figure(figsize(8, 6)) plt.hist(sentiments, bins20, color#23ADE5, alpha0.7) plt.title(评论情感分布) plt.xlabel(情感极性) plt.ylabel(评论数) plt.savefig(sentiment_dist.png, dpi300) # 标记典型评论 most_positive max(comments, keylambda x: SnowNLP(x[content]).sentiments) most_negative min(comments, keylambda x: SnowNLP(x[content]).sentiments) return { avg_sentiment: sum(sentiments)/len(sentiments), most_positive: most_positive, most_negative: most_negative }8. 异常处理与监控8.1 常见异常处理from selenium.common.exceptions import TimeoutException, NoSuchElementException def safe_selenium_operation(driver): try: element WebDriverWait(driver, 10).until( EC.presence_of_element_located((By.ID, target-element)) ) return element.text except TimeoutException: print(元素加载超时) return None except NoSuchElementException: print(元素不存在) return None except Exception as e: print(f未知错误: {str(e)}) return None8.2 健康监控实现import logging from datetime import datetime def setup_monitoring(): logging.basicConfig( filenamecrawler.log, levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s ) def monitor_decorator(func): def wrapper(*args, **kwargs): start_time datetime.now() logging.info(f开始执行: {func.__name__}) try: result func(*args, **kwargs) duration (datetime.now() - start_time).total_seconds() logging.info(f执行成功: {func.__name__} [耗时: {duration:.2f}s]) return result except Exception as e: logging.error(f执行失败: {func.__name__} - {str(e)}) raise return wrapper return monitor_decorator9. 项目架构建议对于企业级应用推荐采用以下架构设计├── crawler/ # 爬虫核心模块 │ ├── browser/ # Selenium相关实现 │ ├── api/ # 直接请求API实现 │ ├── storage/ # 数据存储处理 │ └── utils/ # 通用工具类 ├── analysis/ # 数据分析模块 │ ├── basic_stats.py # 基础统计 │ ├── sentiment.py # 情感分析 │ └── visualization.py # 可视化 ├── config/ # 配置文件 │ ├── accounts.yaml # 账号配置 │ └── settings.py # 运行参数 ├── logs/ # 日志目录 ├── tests/ # 单元测试 └── main.py # 主入口配置管理示例config/settings.pyclass Config: # 请求配置 REQUEST_TIMEOUT 10 MAX_RETRIES 3 REQUEST_INTERVAL 1.5 # 浏览器配置 HEADLESS_MODE True USER_AGENT Mozilla/5.0 (Windows NT 10.0; Win64; x64) # 存储配置 STORAGE_TYPE sqlite # sqlite/mysql/mongodb DB_PATH data/comments.db # 代理配置 PROXY_ENABLED False PROXY_LIST []10. 法律合规与最佳实践遵守robots.txt协议B站目前允许对部分API进行合理爬取控制请求频率建议保持在1-2次/秒避免被封禁用户数据保护不应收集存储用户敏感信息如手机号、邮箱等数据使用限制爬取数据仅限个人学习研究使用版权注意事项评论内容可能具有版权需谨慎使用在实际项目中建议添加如下合规声明 本工具仅限技术学习交流使用不得用于任何商业用途 数据获取需遵守Bilibili用户协议和相关法律法规 使用者应对自己的行为负责与工具开发者无关