需求分析
本项目的目的是实现一个 Bilibili 视频下载器,能够根据给定的 Bilibili 视频 ID(bvid)列表,自动下载对应的视频和音频文件。
功能介绍
- 读取包含 Bilibili 视频 ID 的 CSV 文件。
- 通过 Bilibili API 获取视频和音频的 URL。
- 下载视频和音频文件,并将它们保存在本地文件夹中。
- 记录成功下载和下载失败的视频 ID。
代码讲解
导入必要的库
import re
import os
import requests
import json
import time
from lxml import etree
from tqdm import tqdm
import pandas as pd
import random
初始化变量和请求头
fold_path = '/Volumes/SSD/Data'
ERROR_CID_PATH = "/Volumes/SSD/Data/Video/error_bvids.csv"
SUCCESS_CID_PATH = "/Volumes/SSD/Data/Video/successful_bvids.csv"
user_agents = [
# ...
]
headers = {
# ...
}
bilibili_url = 'https://www.bilibili.com/video/'
videolist_url = 'https://api.bilibili.com/x/web-interface/web/channel/featured/list?'
这部分代码初始化了一些全局变量,如用于存储视频数据的文件夹路径、错误和成功的视频 ID 文件路径。同时,我们定义了请求头和 Bilibili 的 URL。
读取 CSV 文件
video_list = pd.read_csv('/Volumes/SSD/Data/getVideoinfo/getVideoinfo_byhot_1.csv')
bvid_list = video_list['bvid'].values.tolist()
cid_list = video_list['cid'].values.tolist()
这部分代码读取了包含 Bilibili 视频 ID 的 CSV 文件,并将其中的 bvid 和 cid 分别保存到两个列表中。
下载视频和音频
在以下代码中,我们遍历 bvid_list
中的每个 bvid,获取相应视频页面的 HTML 内容,并从中提取视频和音频的 URL。然后,我们将视频和音频文件分别保存到本地文件夹中。
for bvid in tqdm(bvid_list):
# ...
page_url = bilibili_url + bvid
# ...
video_content = json.loads(temp)
# ...
print(video_url)
print(audio_url)
# ...
with open(vfilename, 'wb') as f:
# ...
with open(afilename, 'wb') as f:
# ...
记录成功和失败的视频 ID
我们将成功下载的视频 ID 记录到 SUCCESS_CID_PATH
指定的文件中,将下载失败的视频 ID 记录到 ERROR_CID_PATH
指定的文件中。
with open(SUCCESS_CID_PATH, "a") as success_file:
# ...
with open(ERROR_CID_PATH, "a") as error_file:
# ...
使用方法
- 将包含 Bilibili 视频 ID 的 CSV 文件放到指定的文件夹中。
- 运行脚本,等待下载完成。
注意:本脚本仅用于个人学习和研究,请勿用于非法用途,尊重 Bilibili 平台的版权规定。
注意事项
- 确保你的 Python 环境中已安装
lxml
,requests
,pandas
和tqdm
库。 - 请根据需要更改脚本中的文件路径。
- 如果遇到下载速度慢或者失败的情况,可以尝试更换请求头中的 User-Agent 或 cookie 信息。
代码优化建议
- 使用函数和类对代码进行模块化,提高代码的可读性和可维护性。
- 增加异常处理,以避免程序因网络或其他原因而意外中断。
- 可以尝试使用多线程或多进程,以提高下载速度。
总结:本文档详细介绍了一个 Bilibili 视频下载器的需求分析、功能介绍和代码讲解。通过阅读本文档,你可以了解如何实现一个简单的 Bilibili 视频下载器,并根据需要进行扩展和优化。
完整代码(按需进行参考)
import re
import os
import requests
import json
import time
from lxml import etree
from tqdm import tqdm
import pandas as pd
import random
# 定义文件夹路径和错误信息文件路径
fold_path = '/Volumes/SSD/Data'
ERROR_CID_PATH = "/Volumes/SSD/Data/Video/error_bvids.csv"
SUCCESS_CID_PATH = "/Volumes/SSD/Data/Video/successful_bvids.csv"
# 设置请求头
user_agents = [
# ...(省略部分 User-Agent,可自行添加)
]
headers = {
'User-Agent': random.choice(user_agents),
'Referer': 'https://www.bilibili.com/',
# 添加其他请求头信息,例如 Cookie
# ...
}
# 定义 Bilibili 视频页面 URL 和生活区视频列表 API URL
bilibili_url = 'https://www.bilibili.com/video/'
videolist_url = 'https://api.bilibili.com/x/web-interface/web/channel/featured/list?'
# 读取 CSV 文件中的 bvid 和 cid
video_list = pd.read_csv('/Volumes/SSD/Data/getVideoinfo/getVideoinfo_byhot_1.csv')
bvid_list = video_list['bvid'].values.tolist()
cid_list = video_list['cid'].values.tolist()
# 遍历 bvid 列表,下载每个视频
for bvid in tqdm(bvid_list):
bvid_path = f"/Volumes/SSD/Data/Video/{bvid}"
# 如果已经下载过这个视频,跳过
if os.path.exists(bvid_path):
print(f"File for bvid {bvid} already exists. Skipping...")
with open(SUCCESS_CID_PATH, "a") as success_file:
success_file.write(f"{bvid}\n")
continue
# 请求视频页面
page_url = bilibili_url + bvid
response = requests.get(page_url, headers=headers)
page_content = response.text
page_tree = etree.HTML(page_content)
# 提取视频和音频的 URL
video_content = page_tree.xpath('/html/head/script[5]/text()')
pattern = r'\<script\>window\.__playinfo__=(.*?)\</script\>'
no_video_pattern = r'视频去哪了呢?'
# 如果无法获取视频,记录错误信息并跳过
if len(re.findall(no_video_pattern, page_content)) != 0:
with open(ERROR_CID_PATH, "a") as error_file:
error_file.write(f"{bvid}\n")
print(f"Cannot get video for bvid {bvid}. Skipping...")
continue
try:
temp = re.findall(pattern, page_content)[0]
except:
with open(ERROR_CID_PATH, "a") as error_file:
error_file.write(f"{bvid}\n")
print(f"Cannot get video for bvid {bvid}. Skipping...")
continue
video_content = json.loads(temp)
# 提取 Dash 格式的视频和音频 URL
if video_content['data'] is not None:
if 'dash' in video_content['data'].keys():
for item in video_content['data']['dash']['video']:
if 'baseUrl' in item.keys():
video_url = item['baseUrl']
continue
for item in video_content['data']['dash']['audio']:
if 'baseUrl' in item.keys():
audio_url = item['baseUrl']
continue
print(video_url)
print(audio_ url)
# 创建视频文件夹
videofold = fold_path + '/Video/' + str(bvid)
if not os.path.exists(videofold):
print(1)
os.mkdir(videofold)
# 处理视频标题,移除特殊字符
title = bvid
title = title.replace('/', '')
title = title.replace(' ', '')
title = title.replace('|', '')
title = title.replace(':', '')
title = title.replace(':', '')
# 定义视频和音频文件名
vfilename = videofold + '/' + title + '.mp4'
afilename = videofold + '/' + title + '.mp3'
# 更新请求头,添加 Origin 和 Referer
header = headers
header['Origin'] = 'https://www.bilibili.com'
header['Referer'] = page_url
# 下载视频
with open(vfilename, 'wb') as f:
with requests.get(url=video_url, headers=header, stream=True) as r:
for chunk in r.iter_content(chunk_size=512):
if chunk:
f.write(chunk)
# 下载音频
with open(afilename, 'wb') as f:
f.write(requests.get(url=audio_url, headers=header).content)
with open(SUCCESS_CID_PATH, "a") as success_file:
success_file.write(f"{bvid}\n")
continue
# 添加下载间隔,减轻对服务器的压力
# time.sleep(3)