Python+Selenium实战:自动化登录与高清视频爬取技术解析
1. 项目概述与核心价值
最近在做一个数据分析项目,需要一批特定领域的视频数据,B站自然成了我的首选目标。直接手动下载几十上百个视频显然不现实,而市面上那些现成的下载工具要么有水印,要么功能受限,要么就是收费。作为一个老程序员,我的第一反应就是自己动手写一个。这次我选择的是Python+Selenium的组合,目标很明确:不仅要能稳定爬取B站的高清视频,还要解决最烦人的登录验证问题,实现Cookie的自动化管理。这个方案听起来简单,但实操起来,从绕过B站的反爬机制到处理动态加载的视频流,再到让登录状态持久化,每一步都有不少门道。如果你也受困于手动收集视频数据的低效,或者想学习如何用自动化工具处理复杂的Web交互,那这篇从实战中总结出来的经验,应该能帮你省下不少摸索的时间。
2. 技术选型与整体设计思路
2.1 为什么是Selenium,而不是Requests?
很多人在做爬虫时,第一个想到的可能是requests库,因为它轻量、快速。但对于B站这样的现代网站,直接使用requests会遇到几个难以逾越的障碍。首先,B站的大量页面内容(包括视频信息、评论列表)是通过JavaScript动态渲染的,简单的HTTP GET请求拿不到完整的HTML。其次,视频播放页有复杂的反爬措施,比如对XHR请求的参数进行加密校验。最后,也是最关键的,登录状态维持和验证码处理,用纯requests模拟会异常繁琐。
Selenium的优势就在这里。它本质上是一个浏览器自动化工具,可以驱动一个真实的浏览器(如Chrome)来访问网页,并执行点击、输入等操作。这意味着,网站“看到”的访问者是一个真实的用户在使用浏览器,从而完美绕过了JS渲染和大多数基于请求特征的反爬。我们可以用Selenium来完成登录操作(包括处理滑动验证码),获取登录后的Cookie,然后再用这个Cookie去发起后续的请求,甚至结合requests来下载文件,实现“自动化登录 + 高效抓取”的组合拳。
2.2 核心工具链搭建
工欲善其事,必先利其器。这个项目的基础环境搭建有几个关键点:
- Python环境:建议使用Python 3.8及以上版本,兼容性和库支持都更好。使用虚拟环境(如
venv或conda)来管理依赖是个好习惯,能避免包冲突。 - Selenium库:通过pip安装即可:
pip install selenium。 - 浏览器驱动:这是Selenium能控制浏览器的桥梁。我选用Chrome,所以需要下载对应版本的ChromeDriver。这里有个大坑:ChromeDriver的版本必须与你电脑上安装的Chrome浏览器主版本号完全一致。你可以通过Chrome的“设置->关于Chrome”查看版本,然后去淘宝镜像源等网站下载匹配的驱动。将下载的
chromedriver.exe放在项目目录下,或者将其路径添加到系统的环境变量中。 - 辅助库:
requests: 用于在获取到视频真实地址后,进行高效的文件下载。json: 用于解析B站接口返回的数据。time/datetime: 用于添加延时,模拟真人操作,避免请求过快被封。os: 用于创建本地目录来保存下载的视频。
注意:在正式编写爬虫前,请务必阅读并遵守B站的
robots.txt协议以及相关服务条款。本技术分享仅用于个人学习与研究,请勿用于大规模、商业化的数据抓取,以免对目标网站造成负担,并引发法律风险。
3. 核心环节一:Cookie的自动化获取与持久化
登录是爬取个人相关数据或高清视频的第一步。手动登录再复制Cookie太麻烦,我们的目标是全自动。
3.1 使用Selenium模拟登录流程
B站的登录页面有多种方式,为了通用性和稳定性,我们选择扫描二维码登录。虽然看起来复杂,但Selenium可以让我们自动化完成“打开登录页->等待扫码->登录成功”的全过程。
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 time import json def auto_login(): # 初始化浏览器,并添加一些选项以优化体验和避免检测 options = webdriver.ChromeOptions() # 避免浏览器弹出“正受到自动测试软件控制”的提示(某些版本可能需要) options.add_experimental_option("excludeSwitches", ["enable-automation"]) options.add_experimental_option('useAutomationExtension', False) # 可选项:无头模式,不显示浏览器界面,适合在服务器后台运行 # options.add_argument('--headless') driver = webdriver.Chrome(options=options) driver.get('https://passport.bilibili.com/login') try: # 等待页面加载并找到二维码图片元素 wait = WebDriverWait(driver, 30) qr_img = wait.until( EC.presence_of_element_located((By.CLASS_NAME, 'qrcode-img')) ) print("请使用B站APP扫描屏幕上的二维码进行登录...") # 循环检测登录是否成功:通过判断页面是否跳转,或者某个登录后才会出现的元素是否存在 # 这里以检查页面URL是否包含‘/h5/’或跳转到首页为例,更可靠的是检查用户头像等元素 while True: time.sleep(3) # 每3秒检查一次 current_url = driver.current_url # 如果URL不再是登录页,或者页面中出现了“我的”等登录后元素,则认为登录成功 if 'h5/' in current_url or driver.find_elements(By.CLASS_NAME, 'header-avatar'): print("登录成功!") break except Exception as e: print(f"登录过程出现异常: {e}") driver.quit() return None # 登录成功后,获取所有的Cookies cookies = driver.get_cookies() # 将cookies转换为requests库可用的字典格式 cookie_dict = {} for cookie in cookies: cookie_dict[cookie['name']] = cookie['value'] # 将cookies保存到本地文件,供后续使用 with open('bilibili_cookies.json', 'w', encoding='utf-8') as f: json.dump(cookies, f, ensure_ascii=False, indent=2) print("Cookies已保存至本地文件。") driver.quit() return cookie_dict if __name__ == '__main__': my_cookies = auto_login()实操心得:
- 等待策略:使用
WebDriverWait配合expected_conditions是Selenium最佳实践之一,比死板的time.sleep更智能、更高效。它会在指定时间内持续检查条件是否满足,一旦满足就立即执行后续代码。 - 登录成功判定:判定登录成功不能只看URL,因为B站的登录跳转逻辑可能变化。更稳健的方法是等待一个登录后必定存在的页面元素出现,比如用户昵称、头像等。上述代码中的判断条件可能需要根据B站页面实际结构进行调整。
- Cookie保存:获取到的Cookie列表是一个包含字典的列表,每个字典有
name,value,domain,path等字段。我们将其保存为JSON格式,方便下次直接读取,实现“一次登录,长期使用”。
3.2 Cookie的加载与复用
我们不可能每次运行脚本都扫码登录一次。将Cookie保存到文件后,下次就可以直接加载使用。
import json import requests def load_cookies_from_file(filename='bilibili_cookies.json'): """从文件加载cookies并构建requests可用的session""" try: with open(filename, 'r', encoding='utf-8') as f: cookies_list = json.load(f) # 创建一个requests Session对象,用于保持会话 session = requests.Session() for cookie in cookies_list: # 为session的cookies属性设置每一个cookie session.cookies.set(cookie['name'], cookie['value']) print("Cookies加载成功。") return session except FileNotFoundError: print("Cookie文件未找到,请先运行登录程序。") return None # 使用示例 session = load_cookies_from_file() if session: # 使用这个带cookie的session去访问需要登录的页面,如个人中心 response = session.get('https://api.bilibili.com/x/web-interface/nav') user_info = response.json() if user_info['code'] == 0: print(f"当前登录用户:{user_info['data']['uname']}") else: print("Cookie可能已失效,需要重新登录。")注意事项:
- Cookie有效期:B站的登录Cookie是有有效期的,可能几天或几周。当脚本发现用旧Cookie访问API返回错误码(如
-101表示未登录)时,就需要重新触发auto_login函数来更新Cookie文件。 - Session对象:使用
requests.Session()可以自动管理Cookie,在同一个Session内发起的请求都会携带之前获得的Cookie,非常方便。
4. 核心环节二:高清视频地址的解析与爬取策略
登录问题解决后,就进入了核心环节:如何找到并下载高清视频。B站的视频播放采用了分段传输(M4S)和动态密钥,直接找.mp4或.flv链接的时代早已过去。
4.1 解析视频播放页,获取aid、bvid和cid
B站的每个视频都有几个关键ID:
bvid:现在视频链接中的主要标识符,如BV1xx411c7mh。aid:传统的数字ID,与bvid有一一对应关系。cid:分P视频中每一P的唯一标识符,是获取视频流地址的关键。
我们的第一步是从视频播放页的源代码中提取这些信息。这些信息通常藏在一个名为__INITIAL_STATE__的JavaScript变量中。
import re import json def get_video_ids_by_selenium(url, driver): """使用Selenium打开页面,并解析出视频ID""" driver.get(url) time.sleep(2) # 等待页面JS加载完成 # 获取页面源码 page_source = driver.page_source # 使用正则表达式查找 __INITIAL_STATE__ 变量 pattern = r'window\.__INITIAL_STATE__\s*=\s*({.*?});' match = re.search(pattern, page_source, re.DOTALL) if match: init_state_str = match.group(1) try: init_state = json.loads(init_state_str) # 解析数据结构,获取bvid, aid, cid bvid = init_state.get('bvid') aid = init_state.get('aid') # 对于多P视频,需要获取当前播放的cid,通常位于videoData或videoInfo中 # 这里是一个常见路径,实际结构可能需要微调 cid = init_state.get('videoData', {}).get('cid') if not cid: cid = init_state.get('videoInfo', {}).get('cid') print(f"解析成功: bvid={bvid}, aid={aid}, cid={cid}") return {'bvid': bvid, 'aid': aid, 'cid': cid} except json.JSONDecodeError as e: print(f"解析JSON失败: {e}") else: print("未在页面中找到 __INITIAL_STATE__ 信息。") return None4.2 调用B站API获取视频流信息
拿到aid和cid后,我们就可以模拟播放器,调用B站内部获取视频流信息的API。这个API返回的数据包含了不同清晰度(如1080P、720P)对应的音视频分段地址、密钥等信息。
def get_video_play_info(aid, cid, session): """通过API获取视频播放信息(含视频流地址)""" api_url = 'https://api.bilibili.com/x/player/playurl' params = { 'avid': aid, 'cid': cid, 'qn': 80, # 清晰度标识,80代表1080P,64代表720P,32代表480P 'fnval': 80, # 视频流格式标识,80代表获取dash格式(音视频分离) 'fnver': 0, 'fourk': 1, # 是否允许4K } headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', 'Referer': 'https://www.bilibili.com', # 必须携带Referer,否则可能403 } response = session.get(api_url, params=params, headers=headers) if response.status_code == 200: play_info = response.json() if play_info['code'] == 0: return play_info['data'] else: print(f"API返回错误: {play_info['message']}") else: print(f"请求API失败,状态码: {response.status_code}") return None关键参数解析:
qn(Quality Number): 决定视频清晰度。常用值:120(4K),116(1080P60),112(1080P+),80(1080P),64(720P),32(480P)。数值越大通常清晰度越高,但并非所有视频都有所有档位。fnval: 决定返回的视频流格式。1是老的FLV格式,16是DASH格式(MPEG-4),80或4048是DASH格式并包含杜比视界等信息。我们选择80获取通用的DASH格式,它通常能提供最好的清晰度和音质,并且音视频是分开的。Referer: 这个请求头非常重要,B站服务器会校验它,必须设置为B站的主站域名,否则会返回403 Forbidden错误。
4.3 解析DASH流并下载音视频
从API获取的play_info中,dash字段包含了视频流(video)和音频流(audio)的数组。每个流都有baseUrl(或backupUrl)和size等信息。
def download_dash_video(play_info, title, session): """下载DASH格式的音视频并合并(简易示例)""" if not play_info or 'dash' not in play_info: print("未获取到DASH流信息。") return dash = play_info['dash'] # 通常选择第一个视频流和第一个音频流(可根据codecid、bandwidth等选择最高质量) video_url = dash['video'][0]['baseUrl'] audio_url = dash['audio'][0]['baseUrl'] video_size = dash['video'][0]['size'] audio_size = dash['audio'][0]['size'] print(f"视频流大小: {video_size / (1024**2):.2f} MB") print(f"音频流大小: {audio_size / (1024**2):.2f} MB") headers = { 'User-Agent': 'Mozilla/5.0 ...', 'Referer': 'https://www.bilibili.com', # 对于视频流,Range请求头很重要,支持断点续传 'Range': 'bytes=0-' } # 下载视频部分 print(f"正在下载视频流...") video_resp = session.get(video_url, headers=headers, stream=True) video_filename = f"{title}_video.m4s" with open(video_filename, 'wb') as f: for chunk in video_resp.iter_content(chunk_size=8192): if chunk: f.write(chunk) print(f"视频流下载完成: {video_filename}") # 下载音频部分 print(f"正在下载音频流...") audio_resp = session.get(audio_url, headers=headers, stream=True) audio_filename = f"{title}_audio.m4s" with open(audio_filename, 'wb') as f: for chunk in audio_resp.iter_content(chunk_size=8192): if chunk: f.write(chunk) print(f"音频流下载完成: {audio_filename}") # 合并音视频(此处为示意,实际合并推荐使用ffmpeg) print(f"请使用FFmpeg合并音视频文件,命令示例:") print(f'ffmpeg -i "{video_filename}" -i "{audio_filename}" -c:v copy -c:a copy "{title}.mp4"')重要提示:
- 流式下载:使用
stream=True和iter_content来下载大文件,可以避免内存占用过高。 - 合并工具:下载下来的
.m4s文件是音视频分离的容器格式,需要使用专业工具合并。FFmpeg是命令行下最强大、最通用的选择。上述代码最后打印了合并命令。你需要先在系统上安装FFmpeg,并确保其在环境变量中。 - 清晰度选择:代码中固定了
qn=80。在实际应用中,你可以先获取所有可用的清晰度列表(API返回的accept_quality和accept_description),然后让用户选择或根据规则自动选择最高可用清晰度。
5. 实战流程整合与优化
将上述模块组合起来,就形成了一个完整的爬虫工作流。我们还需要考虑一些工程化优化。
5.1 完整的单视频爬取函数
def download_bilibili_video(video_url, save_dir='downloads', quality=80): """ 主函数:给定B站视频链接,下载指定清晰度的视频。 """ # 0. 准备工作 if not os.path.exists(save_dir): os.makedirs(save_dir) # 1. 初始化Selenium WebDriver (用于解析页面,如果需要) options = webdriver.ChromeOptions() options.add_argument('--headless') # 无头模式,不显示浏览器窗口 driver = webdriver.Chrome(options=options) # 2. 获取视频ID print(f"正在解析页面: {video_url}") ids = get_video_ids_by_selenium(video_url, driver) if not ids: driver.quit() return driver.quit() # 获取到ID后就可以关闭浏览器了 # 3. 加载Cookie会话 session = load_cookies_from_file() if not session: print("无法加载Cookie,尝试重新登录...") # 这里可以添加自动调用登录函数的逻辑 return # 4. 获取视频播放信息 print(f"正在获取视频流信息 (清晰度: {quality})...") play_info = get_video_play_info(ids['aid'], ids['cid'], session, quality) if not play_info: print("获取播放信息失败。") return # 5. 提取视频标题,并清理非法文件名字符 title = play_info.get('title', ids['bvid']) # 移除文件名中不允许的字符 import re title = re.sub(r'[<>:"/\\|?*]', '_', title) # 6. 下载DASH流 print(f"开始下载: {title}") download_dash_video(play_info, os.path.join(save_dir, title), session) print(f"下载流程结束,请检查 {save_dir} 目录下的文件并使用FFmpeg合并。")5.2 处理多P视频与合集
B站很多视频是分P的,或者属于一个合集。我们的爬虫需要能够处理这种情况。
def get_video_cid_list(aid, session): """通过视频aid获取其所有分P的cid列表""" api_url = 'https://api.bilibili.com/x/player/pagelist' params = {'aid': aid} response = session.get(api_url, params=params) if response.status_code == 200: data = response.json() if data['code'] == 0: # 返回一个列表,每个元素包含'cid'和'page'(分P序号)等信息 return data['data'] return [] # 在主函数中,可以先获取cid列表,然后循环下载每一个cid对应的视频 cid_list = get_video_cid_list(target_aid, session) for index, part in enumerate(cid_list): print(f"开始下载第 {part['page']} P: {part['part']}") play_info = get_video_play_info(target_aid, part['cid'], session) # ... 后续下载逻辑,文件名可以加上 part['page'] 以示区分5.3 添加延时与异常处理
为了避免请求过快被B站的风控系统识别为爬虫,必须在请求之间添加随机延时。同时,健壮的程序必须有完善的异常处理。
import random import time def safe_request_with_delay(session, url, **kwargs): """包装请求函数,添加随机延时和重试机制""" delay = random.uniform(1, 3) # 随机延时1-3秒 time.sleep(delay) max_retries = 3 for attempt in range(max_retries): try: response = session.get(url, timeout=10, **kwargs) # 设置超时 response.raise_for_status() # 如果状态码不是200,抛出HTTPError return response except requests.exceptions.RequestException as e: print(f"请求失败 (尝试 {attempt+1}/{max_retries}): {e}") if attempt < max_retries - 1: retry_delay = random.uniform(5, 10) print(f"等待 {retry_delay:.1f} 秒后重试...") time.sleep(retry_delay) else: print("已达到最大重试次数,放弃请求。") raise return None6. 常见问题排查与进阶技巧
在实际操作中,你肯定会遇到各种各样的问题。这里记录了几个我踩过的坑和解决方案。
6.1 常见错误码与应对
| 错误现象 | 可能原因 | 排查与解决思路 |
|---|---|---|
API返回-101 | 未登录或Cookie失效。 | 检查Cookie文件是否存在、是否过期。重新运行登录函数获取新Cookie。 |
API返回-400 | 请求参数错误。 | 检查aid、cid、qn、fnval等参数是否正确,尤其是cid是否对应了正确的分P。 |
API返回-404 | 视频不存在或无权限。 | 检查视频链接是否正确,视频是否已被UP主删除或设置为私密。 |
下载链接返回403 | 请求头不完整,特别是缺少Referer或User-Agent。 | 确保在下载请求中携带了正确的Referer(B站域名)和模拟真实浏览器的User-Agent。 |
| 下载速度极慢或中断 | 网络问题,或目标服务器限流。 | 使用stream=True分块下载。添加重试机制。考虑在请求头中添加Accept-Encoding(但注意requests会自动解压)。 |
| Selenium找不到元素 | 页面加载未完成或元素定位方式改变。 | 增加等待时间,使用更稳健的WebDriverWait。检查元素的CSS选择器或XPath是否因网站改版而失效。 |
6.2 关于清晰度qn的深入理解
不是所有视频都支持所有清晰度。更稳妥的做法是先查询可用清晰度。在get_video_play_info的API返回数据中,有一个accept_quality列表,里面包含了该视频支持的所有qn值。你应该编写逻辑,优先选择你期望的清晰度(如80),如果不在支持列表中,则降级选择列表中可用的最高清晰度。
6.3 使用FFmpeg合并的细节
打印出FFmpeg命令让用户自己操作只是权宜之计。更好的做法是在Python脚本中直接调用FFmpeg进行合并。
import subprocess def merge_audio_video(video_path, audio_path, output_path): """使用FFmpeg合并音视频""" command = [ 'ffmpeg', '-i', video_path, '-i', audio_path, '-c:v', 'copy', # 视频流直接复制,不重新编码(速度快,无损) '-c:a', 'copy', # 音频流直接复制 '-y', # 覆盖已存在的输出文件 output_path ] try: print(f"正在合并文件: {output_path}") result = subprocess.run(command, check=True, capture_output=True, text=True) print("合并成功!") # 可选:合并成功后删除临时的.m4s文件 # os.remove(video_path) # os.remove(audio_path) except subprocess.CalledProcessError as e: print(f"FFmpeg合并失败: {e}") print(f"标准错误输出: {e.stderr}") except FileNotFoundError: print("未找到FFmpeg,请确保已安装并将其添加到系统环境变量PATH中。")6.4 应对反爬升级:User-Agent池与IP代理
如果进行大规模爬取,即使有Cookie和延时,单一IP和User-Agent也可能被限制。这时需要考虑:
- User-Agent池:准备一个列表,存放多种浏览器和设备的User-Agent字符串,每次请求随机选取一个。
- IP代理:使用代理IP池来轮换请求的出口IP。这涉及到付费代理服务或自建代理,复杂度较高,仅在必要时考虑。对于个人学习和少量数据抓取,通常不需要走到这一步。
整个项目从环境搭建到最终下载合并,涉及了Web自动化、网络请求、数据解析、文件处理等多个环节。最关键的体会是,理解网站的数据加载逻辑和接口参数含义,远比盲目写代码更重要。多利用浏览器的开发者工具(F12)中的“网络(Network)”面板,观察真实播放视频时浏览器发起了哪些请求,这些请求携带了哪些参数和请求头,这是逆向解析任何网站数据接口的通用法门。最后再次强调,技术用于学习,请合理控制爬取频率,尊重网站规则和创作者版权。