在当今短视频盛行的时代,有时我们会遇到一些想要保存下来的精彩视频内容。本文将带你深入解析一个B站视频批量下载工具,并教你如何扩展其功能,使其更加强大实用。
这个Python脚本可以自动批量下载B站热门小视频,通过调用B站官方API获取视频列表,然后逐个下载保存到本地。下面我们来逐部分解析代码并探讨如何扩展功能。
import requests # 网络请求模块
import time # 时间模块
import random # 随机模块
import os # 操作系统模块
import re # 正则表达式
这些是脚本运行的基础模块。我们可以考虑增加一些模块来扩展功能:
import json
from urllib.parse import quote
import logging
from tqdm import tqdm # 进度条显示
# 哔哩哔哩小视频json地址
json_url = 'http://api.vc.bilibili.com/board/v1/ranking/top?page_size=10&next_offset={page}1&tag=%E4%BB%8A%E6%97%A5%E7%83%AD%E9%97%A8&platform=pc'
class Crawl():
def __init__(self):
# 创建头部信息
self.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:66.0) Gecko/20100101 Firefox/66.0',
'Referer': 'https://www.bilibili.com/'
}
这里我们添加了Referer头部,这是很多网站包括B站的反爬措施之一。我们还可以进一步扩展:
def __init__(self, save_path='video'):
self.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:66.0) Gecko/20100101 Firefox/66.0',
'Referer': 'https://www.bilibili.com/'
}
self.save_path = save_path
# 设置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s: %(message)s')
self.logger = logging.getLogger(__name__)
def get_json(self, json_url):
try:
response = requests.get(json_url, headers=self.headers, timeout=10)
# 判断请求是否成功
if response.status_code == 200:
return response.json() # 返回json信息
else:
self.logger.error(f'获取json信息失败,状态码:{response.status_code}')
return None
except Exception as e:
self.logger.error(f'获取json信息时发生错误:{str(e)}')
return None
我们增加了异常处理和超时设置,使程序更加健壮。同时使用日志记录替代简单的print语句,便于调试和监控。
def download_video(self, video_url, title):
if not title:
title = f'untitled_{int(time.time())}'
# 进一步清洗文件名,防止过长
title = self.clean_filename(title)
# 确保保存目录存在
if not os.path.exists(self.save_path):
os.makedirs(self.save_path)
file_path = os.path.join(self.save_path, f'{title}.mp4')
# 如果文件已存在,则跳过下载
if os.path.exists(file_path):
self.logger.info(f'文件已存在,跳过下载:{title}')
return True
try:
# 添加进度条显示
response = requests.get(video_url, headers=self.headers, stream=True, timeout=30)
if response.status_code == 200:
total_size = int(response.headers.get('content-length', 0))
with open(file_path, 'wb') as f, tqdm(
desc=title[:20], # 进度条描述,取前20个字符
total=total_size,
unit='B',
unit_scale=True,
unit_divisor=1024,
) as bar:
for data in response.iter_content(chunk_size=1024):
size = f.write(data)
bar.update(size)
self.logger.info(f'下载完成:{title}')
return True
else:
self.logger.error(f'视频下载失败,状态码:{response.status_code}')
return False
except Exception as e:
self.logger.error(f'下载视频时发生错误:{str(e)}')
# 删除可能已损坏的文件
if os.path.exists(file_path):
os.remove(file_path)
return False
def clean_filename(self, filename):
# 只保留标题中英文、数字与汉字,其它符号会影响写入文件
comp = re.compile('[^A-Z^a-z^0-9^u4e00-u9fa5]')
filename = comp.sub('', filename)
# 限制文件名长度
if len(filename) > 100:
filename = filename[:100]
# 如果清洗后文件名为空,使用时间戳
if not filename:
filename = f'video_{int(time.time())}'
return filename
这里我们做了大量改进:
添加了进度条显示,让用户清楚知道下载进度增强了文件名清洗功能,防止文件名过长或无效添加了文件存在检查,避免重复下载增加了更完善的异常处理
def get_video_info(self, json_data):
"""从JSON数据中提取视频信息"""
if not json_data or 'data' not in json_data or 'items' not in json_data['data']:
return []
video_infos = []
infos = json_data['data']['items']
for info in infos:
try:
title = info['item']['description']
video_url = info['item']['video_playurl']
# 验证必要字段是否存在
if title and video_url:
video_infos.append({
'title': title,
'url': video_url
})
except KeyError as e:
self.logger.warning(f'解析视频信息时缺少字段:{str(e)}')
continue
return video_infos
def run(self, start_page=0, page_count=10):
"""运行爬虫"""
self.logger.info('开始下载B站热门视频')
total_downloaded = 0
for page in range(start_page, start_page + page_count):
self.logger.info(f'正在获取第{page+1}页数据...')
json_data = self.get_json(json_url.format(page=page))
if not json_data:
self.logger.warning(f'第{page+1}页数据获取失败,跳过')
continue
video_infos = self.get_video_info(json_data)
if not video_infos:
self.logger.warning(f'第{page+1}页没有找到视频信息')
continue
self.logger.info(f'第{page+1}页找到{len(video_infos)}个视频')
for video_info in video_infos:
title = video_info['title']
video_url = video_info['url']
self.logger.info(f'开始下载:{title}')
if self.download_video(video_url, title):
total_downloaded += 1
# 随机延时,避免请求过于频繁
delay = random.randint(3, 6)
self.logger.info(f'等待{delay}秒后继续...')
time.sleep(delay)
self.logger.info(f'下载完成,共成功下载{total_downloaded}个视频')
主程序部分我们将其重构为更清晰的
run方法,并添加了视频信息提取的专门方法,使代码更加模块化。
if __name__ == '__main__':
# 可以在这里添加更多配置选项
save_path = 'bilibili_videos' # 保存路径
start_page = 0 # 起始页码
page_count = 5 # 要下载的页数
crawler = Crawl(save_path=save_path)
crawler.run(start_page=start_page, page_count=page_count)
通过这个完整的B站视频下载工具,我们不仅学习了如何与Web API交互、处理网络请求、文件操作等Python编程技能,还探讨了如何通过模块化设计和异常处理提高代码的健壮性。你可以基于这个基础版本,根据实际需求添加更多实用功能。
希望这个教程对你有所帮助,如果你有任何问题或改进建议,欢迎交流讨论!