Proj-04
Simple Bilibili Toolbox
简易 B 站工具箱
成品（并不）参见 BiliTools-Remake
still in beta stage...
我们终于决定要编写一个多功能的 B 站工具箱了（其实主要还是下载）。
在规划中，这个程序分为三层 —— 接口层、核心层、UI 层。
| Layer | Desc |
|---|---|
| 接口层 | 封装 B 站的 API 供其余部分调用 |
| 核心层 | 单个媒体的下载过程、来源解析等功能组件 |
| UI 层 | 与用户交互的部分，CLI 或 GUI |
Part.I 接口层
我们将要编写一个能供外界调用的模块。此时我们的用户是其他的程序员，所以我们需要写好类型标注、注释和 docstring 等，就像那些超好用的第三方库一样 ，但是我因为偷懒只写了一部分（逃
模块结构设计
最简单的结构，当然是用函数将 requests 的网络请求和 url 一同封装成函数，同时将接口中的参数选择性地转换为函数的参数。将这些函数分成不同部分放到不同的 py 文件中，最后写个 __init__.py 封顶，就大功告成了！就像这样：
biliapis
│ audio.py
│ bilicodes.py
│ error.py
│ login.py
│ manga.py
│ media.py
│ video.py
│ wbi.py
└─ __init__.py2
3
4
5
6
7
8
9
10
—— 但是这只是最简单的设想。
出现问题
模块分类
实际一做就会发现，有一些特殊的部分应该与别的部分区分开。比如上面的 bilicodes.py 中定义了 B 站 API 中的一些常见的状态码和枚举数据； error.py 中定义了 BiliError 异常类； wbi.py 会为别的部分提供签名服务而不是供用户调用。
那我们就将它们抽离出来，其余模块下沉一级：
biliapis
│ bilicodes.py
│ error.py
│ wbi.py
│ __init__.py
│
└─ apis
│ audio.py
│ login.py
│ manga.py
│ media.py
│ video.py
└─ __init__.py2
3
4
5
6
7
8
9
10
11
12
13
好！
代码复用
那么另一个问题是，要在每个分类的每个接口函数中都写一遍完整的请求流程吗？这也太不复用了吧……
这个也简单，我们将重复的部分抽取出来做成函数，每个接口函数中都使用这些会重复用到的代码包装成的函数，就好了！
在示例代码中我因为在写了先前的装饰器部分之后有点上头，所以这些「会重复用到的代码」全被我弄成了装饰器，所以实际的示例代码和我所述的可能会「稍微」有点不一样（逃）
登录会话
嗯还有一个问题，要想获取到 B 站的高画质和付费番剧等资源，你需要登录。登录信息保存在 Cookies 中，而 Cookies 保存在 requests.Session 中。稍微翻看一下 requests 的源代码发现，如果直接使用 requests 模块中的 get post head 等函数， requests 实际上会临时新建一个 Session 出来使用，Cookies 无法持久保留。
那怎么办呢？在每个接口函数的参数列表中额外增加一个 session 参数吗？感觉有点怪怪的 ，给我干哪来了这还是面向对象吗
那把 session 写成文件里的全局变量？感觉更怪了。但是除了这俩似乎没有更好的解决方法了…… 吗？
多登录会话共存
先来看再一个问题，假如使用我们的模块的开发人员想实现一个允许多个账号同时存在的工具箱（一个账号对于一个资源没有权限就换下一个账号去取那种），那用文件内全局变量的方法显然就不够灵活了。每个接口函数都传入 session 呢，又有点嫌繁琐。
唉唉这种时候就应该把提出问题的用户狠狠击倒在地
那就不能发挥下我们面向对象的特长吗，话说回来先前好像都一直没用到类 ww
解决方案（？）
我们可以将接口函数变成接口方法，session 作为类的私有方法存在。这样就只需要在接口类实例化的时候传入一遍，之后就可以对着接口实例随便调用。将每个分类的接口都做成接口类的实例方法。
class APIClass:
def __init__(self, session: requests.Session):
self._session = session
def api1(self, param1, param2) -> Any:
...2
3
4
5
6
大概像这样！
但是还是有一些部分会反复出现，比如这个 __init__ 特殊方法。那我们就通过类的继承来让它可以被复用。就像这样：
class APITemplate:
def __init__(self, session: requests.Session):
self._session = session
class APIClass1(APITemplate):
def api1(self, param1, param2) -> Any:
...
class APIClass2(APITemplate):
def api1(self, param1, param2) -> Any:
...2
3
4
5
6
7
8
9
10
11
在做了若干个这样的接口类之后，我们发现一个一个实例化它们还是有点麻烦，那我们就再做一个工厂函数，接收 session 产出接口实例，这些接口实例装在一个容器里共用一个 session。嗯做成一个容器类也不错。
这样既能一次性将接口类全实例化，也能单独实例化一个接口类。还是挺平衡的？
Well, actually ☝️🤓（）
你不觉得这样的代码跟
requests库的耦合程度有点太高了吗，尽管这个库是线程安全的并且也确实很好用，但是万一我们亲爱的用户们想弄个异步的应用程序呢（x就交给你来解决吧！（逃）
杂七杂八的东西
缓存
如果你觉得需要的话，可以弄。但是这么小的一个程序真的有必要吗……？
但是有个⑨就是弄了，还用了个 SQLite，虽然是 GPT 辅助的
得益于我们将接口的网络请求都收束到了一起，我们只需要修改组件们使用的函数就可以比较容易地实现缓存。
搓点代码
嗯实际上把思路弄清楚之后这部分的代码是很简单的，就略了 ww
Part.II 核心层
这部分我们编写下载媒体的功能函数，以及它所需的下载器。
现在我们的角色是接口层的用户了 w
下载器
先搓一个通用的下载器出来，需求：
- 如果服务器支持的话能断点续传
- 不主动阻塞当前线程
- 能够获取进度
- 能够暂停 / 继续
- 最好还能弄个多线程
唉唉沟槽的甲方
查阅资料得知，能否断点续传和单文件分块多线程，取决于服务器是否支持Range 请求头（可通过检查服务器的 Accept-Ranges 响应头是否为 bytes ）。在请求头中使用 Range 项，即可取到指定字节范围的数据，此时服务器会返回 206 状态码。
弄个简单的伪代码捋捋思路吧。
def download(url, 成品文件路径):
如果成品文件存在就直接终止
生成临时文件路径
预请求得到响应头
如果临时文件存在且不支持续传，直接删掉临时文件
如果临时文件存在就获取文件大小，设置到请求头中的Range里
发起请求:
如果请求是续传但服务器没有返回206状态码，直接报错 # 算是保险措施？
打开临时文件:
下载写入
将临时文件重命名为成品文件2
3
4
5
6
7
8
9
10
11
好。
现在再将整个过程投到 threading.Thread 中作为子线程运行，就能做到无阻塞了 —— 那要怎么获取进度信息和错误呢？
给函数传一个当作钩子的对象吗，像一个字典什么的？这样函数就能通过修改这个对象来向外界传递信息了。我我们顺着这个思路让它再直观一点，从 Thread 类派生出一个子类，让函数修改对象的私有属性，再添加一个方法用来查看这个属性。
再来。暂停和继续要如何实现呢？这时可以使用到 threading 中的一些用来通信的对象比如Event。 Event 对象在没有被设置的时候调用 wait() 方法会阻塞调用它的线程，设置后调用则不会阻塞。于是在下载函数的循环写入部分添加调用 event.wait() 的代码，这样当使用者将 event 取消设置时，下载过程就被暂停（阻塞）了。
一个替选方案是使用变量当作标识，修改它则表示暂停，下载循环便进入一个 while 套 time.sleep () 的循环直到这个标识变量被复原（这样做的话记得加锁）。
# === 子线程中 ===
class XXThread(threading.Thread):
...
def run(self):
...
for chunk in resp.iter_content(chunk_size=4096): # 下载循环
self._pause_event.wait()
if chunk:
fp.write(chunk)
...
# === 父线程中 ===
...
thread = XXThread(...)
thread.pause() # 写好的方法，会将`self._pause_event`设置
thread.resume() # 将event取消设置
...2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
同理，可以通过类似的方法实现取消下载，只是从阻塞线程变成了直接终止而已。
多线程下载器 (选)
这部分在示例程序中并没有用到
既然服务器都支持 Range 了，那我们可不可以多开几个线程分别负责文件的各个部分呢？
这里有两种任务分配方案：
- 不定量的线程负责固定大小的文件块，共创建
总大小//分块大小个线程，但限制同一时间正在运行的线程数 - 定量的线程负责固定比例的文件块，每个线程负责
总大小//线程数量大小的文件块
当然你还可以设计出更多更帅的，像 IDM 那样。但不管怎么分配任务，负责分配任务的函数都需要预请求得到总文件大小，然后将任务指派给子线程。
既然分配任务时已经预请求过了，再直接使用上文提到的那个函数就会导致重复的预请求。那我们就把上文的那个函数再拆分成预请求与下载部分。下载部分在需要时会被指派负责的字节范围，同时不再关心服务器是否支持 Range 操作（因为这部分工作现在由预请求部分来做，但可以做额外的检查）
具体实现可以借助标准库 concurrent ，也可以尝试自己动手搓搓线程管理
来源解析
在执行下载任务时，第一步是解析用户丢进来的内容，可能是链接也可能直接是内容 ID。不过在 B 站的链接中能够很容易地找到内容 ID。唯一值得特殊处理的是经由 APP 分享的短链接，需要经过一次重定向得到包含内容 ID 的链接。
B 站的内容 ID 们除了 BV 号以外都是纯数字，可以轻易地写出正则表达式。如果追求准确的话可以再搭配正则表达式的负向后顾。
下载流程
做出每种基础的下载流程。这些流程被写在一个继承于 threading.Thread 的类中，并添加了一些新方法来获取进度。
由于番剧、电影、电视剧、纪录片等（统称 PGC，专业生产内容，是 B 站接口中的说法）能够被解析成若干个普通视频，并且调用取流接口返回的数据结构也和普通视频的完全一致，所以我觉得没必要为它们专门设计一个下载流程。我个人的选择是将它们的实现放到 UI 部分，拆解为普通视频之后再交由普通视频的流程处理。
视频
按照 Proj-02，将流程复刻一遍。
除直接下载视频外，还可以提供仅下载视频音轨的选项。额外添加下载字幕的选项。
注意 FLAC 音轨无法被封装到 MP4 容器中，此时可以将容器格式换成 MKV，或者将音轨转换为 ALAC 再封装
音频
从接口取到流，然后下载，然后转码即可。
漫画
从接口得到图片路径和 token，拼接起来即为完整 url，下载即可。
Part.III UI 层
示例中只尝试弄了 CLI（逃
UI 层主要负责接收命令行参数，执行相应的操作。
同时将 PGC、合集等来源的输入转换为基本的媒体下载流程并执行。可以使用tqdm 库，一个很帅的进度条库，来跟踪进度。
持久化保存数据
UI 层同时还要负责保存登录会话和一些额外的东西到文件，以供下次启动时加载使用。
requests 自带的一个工具函数 requests.utils.dict_from_cookiejar() 可以将 Session 对象内部的 cookiejar 对象转换成 dict ，同时还有另一个位于同一位置的 cookiejar_from_dict 函数用来做相反的操作，这样就可以方便地将 cookies 保存到本地的 json 文件了…… 吗？
事实上，这样生成的 cookie dict ，只包含键值（ name 和 value ），缺失了很多元数据（比如 domain path expires 等等）。至少在我的实践中，B 站就是不承认像这样保存了又加载的 cookies 。
如果你成功找到了将 cookies 无损保存到 json 再加载并能为网站们所承认的方法，请告诉我……
如果没有别的方法，那么我们就要祭出下策了 —— pickle ！
警告：
pickle模块并不安全。你只应该对你信任的数据进行unpickle操作。构建恶意的pickle数据来在解封时执行任意代码是可能的。绝对不要对不信任来源的数据和可能被篡改过的数据进行解封。[1]
pickle 模块可以将 Python 对象序列化为 bytes 对象形式的二进制数据，并可以反序列化回 Python 对象。序列化出的二进制数据可以写入到文件。安全起见，我们可以添加数据校验（可以使用 Python 官方推荐的 hmac 库）
程序退出时使用 pickle 将 Session 对象序列化为二进制数据并生成校验数据，然后一并写入文件，下次启动时再读取即可。有很多机器学习的模型文件也是像这样生成的。
要注册函数以在程序退出时自动调用，可以借助 atexit 标准库实现。
引自官方文档
pickle模块部分 ↩︎