之前在js动态解析引入高清B站视频中研究了B站的api,但是都是别人封装好的,使用方便但限制比较大。这次正好学了Python爬虫,又简单学习了多线程的使用,所以就去研究了bilibili的官方api,然后尝试封装了bilibili视频下载的类(造轮子)。

requirements

需要引入如下模块:(按需使用pip下载)

1
2
3
4
5
6
import re  # python自带的正则表达式库
import requests # http请求库,可能需要下载
import json # api返回的是json,需要转换为dict
from contextlib import closing # 创建上下文管理器
from multiprocessing.pool import Pool # 自带的多进程库
from tqdm import tqdm # 进度条库,需要下载

类的设计

  1. 类属性包括一些类中始终不变的变量,如请求头、cookie等。
  2. 一个类的实例对应一个视频或一个合集。
  3. 实例属性和视频相关,如视频的bvid、标题、直链等。
  4. 实例化对象时传入bilibili视频的url。
  5. 包含单个视频的下载和视频合集的下载两个实例方法。
  6. 因为两个实例方法都涉及到下载,所以可以把下载部分的公共代码设计成静态方法。

据此,类的框架为:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class BilibiliVideo(object):
# 类属性
header = {}
cookie = ''

# 构造函数
def __init__(self):
pass

# 下载单个视频
def download(self):
self.download_1p() # 调用下载的公共代码
pass

# 下载视频合集
def download_collection(self):
self.download_1p() # 调用下载的公共代码
pass

# 下载的公共代码
@staticmethod
def download_1p():
pass

api研究

  1. 第一个api根据cookie获取用户的信息:(前提是访问的header里面要有正确的cookie值,否则提示账号未登录。)

  2. 第二个api获取视频或合集的详细信息以及字幕、封面等静态资源的链接:

    可以使用json可视化工具更好地阅读api返回值

    提取出的关键信息有page、cid、pic、title(如果是合集就是合集的标题)、part(合集视频的分集标题)。

  3. 第三个api获取单个视频的动态地址:

    提取出的关键信息是url。

由于这些api获得的信息都和视频相关,所以将它们设置为实例属性,在构造函数里初始化。

进度条的实现

tqdm是一个进度条库,一个基本的使用例子如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
import time
from tqdm import tqdm
total = 100 # 总进度
step = 10 # 每次刷新的进度
# 总共要刷新的次数
flush_count = total//step

# 进度条对象
bar = tqdm(total=total, unit='MB', desc='下载中')
for _ in range(flush_count):
time.sleep(0.1) # 模拟下载的阻塞
bar.update(step)
bar.close()

一个文件的大小可以在http响应头中获取:int(response.headers['content-length'])(单位是字节)。下载的数据分块写入文件中,块的大小设置为1024B。那么下载时显示进度条的代码为:

1
2
3
4
5
6
7
8
9
10
chunk_size = 1024  # 单次请求最大值
content_size = int(response.headers['content-length']) # 内容体总大小,单位是字节

bar = tqdm(total=content_size/(1024**2), unit='MB', desc=f"下载文件 {info['title']}")
with open(f"./video/{str(info['pid']) + '. ' + info['title']}" + '.flv', mode='wb') as f:
# 写入分块文件
for chunk in response.iter_content(chunk_size=chunk_size):
f.write(chunk)
bar.update(chunk_size/(1024**2))
bar.close() # 关闭进度条

目标

  1. 继续学习Python异步,包括多线程下载多个文件和多线程下载一个文件后合并(如idm),优化下载部分的代码。
  2. 按照《Python高性能》这本书提供的思路来尝试优化Python代码。
  3. 确认无Bug后将这个脚本作为后端使用flask制作一个Bilibili下载器。
  4. 封装更多的功能,使用pyqt制作一个Bilibili客户端。(挖坟)

参考

代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
import re
import requests
import json
from contextlib import closing
from multiprocessing.pool import Pool
from tqdm import tqdm

url = "https://www.bilibili.com/video/BV1b5411c7Sa"

class BilibiliVideo(object):
"""
封装的bilibili视频类,一个对象对应一个视频或合集,实例化时需要传入url。
提供的接口:
1. download(): 下载单个视频
2. download_collection(): 下载视频合集,兼容download()
"""
header = {
'User-Agent': 'Mozilla/5.0 (iPhone; CPU iPhone OS 11_0 like Mac OS X) AppleWebKit/604.1.38 (KHTML,'
' like Gecko) Version/11.0 Mobile/15A372 Safari/604.1',
'Referer': 'https://www.bilibili.com' # bilibili新的反爬机制,需要设置Referer为bilibili主站
}
cookie = {'Cookie': '''粘贴你的COOKIE'''}
s = requests.session()

# 第一次实例化时需要传入COOKIE,不然为空
def __init__(self, url):
"""
:param url: bilibili视频的链接
video_id: 视频的bvid
vid: 视频或合集的详细信息
page: 多P视频选择的集数
cid: 视频的cid
video_name: 视频的标题
video_url: 视频的动态链接
"""
# 打印用户欢迎信息
my_info = json.loads(BilibiliVideo.s.get('http://api.bilibili.com/x/space/myinfo',
cookies=BilibiliVideo.cookie).text)
print("\n欢迎你!" + my_info['data']['name'])

# 初始化
self.video_id = re.findall("[\w.]*[\w:\-\+\%]", url)[3] # 从视频链接中获取bvid
self.vid = json.loads(
BilibiliVideo.s.get(f'https://api.bilibili.com/x/web-interface/view?bvid={self.video_id}',
headers=BilibiliVideo.header, cookies=BilibiliVideo.cookie).text
)
# 默认为单p视频,多P视频先初始化为第一个
self.page = 0
self.cid = self.vid['data']['pages'][self.page]['cid']

# 获取单个视频的信息
video_info = json.loads(
BilibiliVideo.s.get('https://api.bilibili.com/x/player/playurl?bvid=' +
self.video_id + '&cid=' + str(self.cid) + '&qn=80&otype=json',
headers=BilibiliVideo.header, cookies=BilibiliVideo.cookie).text
)
# 从视频信息中获取关键信息
self.video_name = self.vid['data']['title']
self.video_url = video_info['data']['durl'][0]['url']

# 下载单个视频
def download(self):
# 如果是多P视频,首先需要用户选择下载集数,然后更新视频标题和链接
if self.vid['data']['videos'] > 1:
print(f"这是一个多P视频,共{self.vid['data']['videos']}集,列表为:")
pid = 0
for page in self.vid['data']['pages']:
pid += 1
print(f"{pid}. {page['part']}")

page = int(input("\n请输入要下载第几集(从1开始):"))
self.page = page - 1
self.video_name = self.vid['data']['pages'][self.page]['part']

info = {
'pid': self.page + 1,
'cid': self.vid['data']['pages'][self.page]['cid'],
'title': self.vid['data']['pages'][self.page]['part'],
'bvid': self.video_id,
'url': self.video_url
}

self.download_1p(info)

# 多线程下载合集视频
def download_collection(self):
if self.vid['data']['videos'] == 1:
self.download()

infos = [] # 合集的数据字典
for pid in range(int(self.vid['data']['videos'])):
info = {
'pid': pid + 1,
'cid': self.vid['data']['pages'][pid]['cid'],
'title': self.vid['data']['pages'][pid]['part'],
'bvid': self.video_id
}
infos.append(info)

pool = Pool(3)
pool.map(self.download_1p, infos)
pool.close() # 关闭进程池,不再接受新的进程
pool.join() # 主进程阻塞等待子进程的退出

# 下载视频的公共代码,实现了一个简易进度条
# 存在的问题是下载合集视频时进度条会相互覆盖,还要想办法实现一个视频的下载对应一个进度条
@staticmethod
def download_1p(info: dict):
print("\n----------------------------------------------------------\n"
"开始下载视频{}".format(str(info['pid']) + '. ' + info['title']))

# 未传入url,说明这是一个多P视频,需要获取
try: info['url']
except KeyError:
detail_url = 'https://api.bilibili.com/x/player/playurl?bvid=' + \
info['bvid'] + '&cid=' + str(info['cid']) + '&qn=80&otype=json'
video_info = json.loads(
BilibiliVideo.s.get(detail_url, headers=BilibiliVideo.header, cookies=BilibiliVideo.cookie).text
)
info['url'] = video_info['data']['durl'][0]['url']

with closing(BilibiliVideo.s.get(info['url'], headers=BilibiliVideo.header, stream=True)) as response:
chunk_size = 1024 # 单次请求最大值
content_size = int(response.headers['content-length']) # 内容体总大小,单位是字节
data_count = 0 # 当前下载的大小,初始化为0

bar = tqdm(total=content_size/(1024**2), unit='MB', desc=f"下载文件 {info['title']}")
with open(f"./video/{str(info['pid']) + '. ' + info['title']}" + '.flv', mode='wb') as f:
# 写入分块文件
for chunk in response.iter_content(chunk_size=chunk_size):
f.write(chunk)
bar.update(chunk_size/(1024**2))
# 关闭进度条
bar.close()

if __name__ == "__main__":
# print(bvideo.__init__.__doc__)
bvideo = BilibiliVideo(url)
bvideo.download_collection()