基于cat-catch传递信息的bilibili视频下载服务端（可拓展）

好久没写blog了，想起来把最近做的这个东西发上来，个人用起来还是很满意的

使用场景

下B站视频，要能最高画质，不然我大会员白开了，顺带直接存nas

前置准备

我们需要一个工具来解析视频的下载地址，而且要很方便不需要我单独去找一个软件来下，所以我就想到用浏览器拓展来干这个事情。猫抓(cat-catch)刚好是这么一个现成的流媒体抓取拓展，可以很方便的抓到B站的视频下载链接，同时还支持多端，在手机的edge同样可以使用。注意到猫抓拓展还集成有数据发送的功能，即可以将抓取到的数据通过json格式发送到远程服务器，我想到可以在我的nas上部署一个这个服务器，就可以随时下载需要下载的视频了。

架构分析

可以知道的是，B站视频流分了两个文件，一个只传输视频，另一个只传输音频，所以下载的视频需要处理。为了服务的持久性，我不再将这个server放到我的本地电脑上而是移交到nas进行处理。在此之前我们需要分析一下传递的json的结构：

1
{
2
  "action": "",         // 无用信息，忽略
3
  "data": {
4
    "url": "",          // 这里是解析到的下载url，关键
5
    "referer": "",      // 通常是"https://www.bilibili.com"
6
    "origin": "",       // 同上
7
    "initiator": "",    // 同上
8
    "webUrl": "",       // 抓取源网页的url，含有Bv号的信息，关键
9
    "title": "",        // 抓取源网页的标题
10
    "cookie": "",       // 一般是空的，忽略
11
    "tabId": $Id,       // 标签页id
12
    "year": ,
13
    "month": ,
14
    "date": "",
15
    "day": "",
16
    "fullDate": "",
17
    "time": "",
18
    "hours": "",
19
    "minutes": ,
20
    "seconds": ,
21
    "now": ,
22
    "timestamp": "",
23
    "fullFileName": "", // 完整文件名称
24
    "fileName": "",     // 文件名
25
    "ext": "",          // 文件后缀名
26
  },
27
  "tabId": ""           // 和上面那个一样
28
}

为了保证不产生错误的音视频流匹配，我们需要充分利用浏览器拓展使用json为我们提供的信息，我使用的是标签页id，只有通过相同标签页发送的两个json才允许进行合并另外，使用FIFO的数据结构对存入的json缓存不过出于技术性考量，没有做更进一步的匹配设计，只要在客户端做正确的输入，基本上不会出现问题另外，我让ai使用多线程进行操作，因为猫抓同时勾选两个进行发送，发送的是两次json，且几乎没有延时，需要分两个线程来处理不然会来不及接收丢包

代码组成

主函数多线程

1
def main():
2
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
3
        s.bind((HOST, PORT))
4
        s.listen()
5
        print(f"Listening on port {PORT}...\n")
6

7
        while True:
8
            conn, addr = s.accept()
9
            t = threading.Thread(target=handle_client, args=(conn, addr))
10
            t.daemon = True
11
            t.start()

接收处理json

1
def handle_client(conn, addr):
2
    print(f"Connected by {addr}")
3
    buffer = ""
4

5
    with conn:
6
        while True:
7
            data = conn.recv(4096)
8
            if not data:
9
                break
10

11
            buffer += data.decode('utf-8', errors='ignore')
12
            json_packet = extract_json(buffer)
13
            if json_packet:
14
                print(f"收到 JSON 来自 {addr}: {json_packet}")
15

16
                process_incoming_json(json_packet)
17

18
                buffer = ""  # 清空，避免重复解析

分析json数据，进行配对

1
def process_incoming_json(json_packet):
2
    """处理已解析的 JSON 包，同时执行 tabId 配对逻辑"""
3
    tabId = json_packet.get("tabId")
4
    data = json_packet.get("data")
5
    if tabId is None or data is None:
6
        print("收到 JSON 但缺少 tabId 或 data")
7
        return
8

9
    # 只接受来自 https://www.bilibili.com 的 origin
10
    origin_val = data.get("origin") or data.get("originUrl") or data.get("Referer") or ''
11
    if origin_val != "https://www.bilibili.com":
12
        print(f"忽略非 B站来源（origin={origin_val}） tabId={tabId}")
13
        return
14

15
    new_item = Item(
16
        tabId=tabId,
17
        data=data,
18
        timestamp=time.time()
19
    )
20

21
    # 检查是否已有同 tabId 数据
22
    if tabId in index:
23
        old_item = index.pop(tabId)
24
        # 从 pending 中移除旧 item
25
        for item in pending:
26
            if item.tabId == tabId:
27
                pending.remove(item)
28
                break
29
        # 成对数据 → 触发事件
30
        handle_pair([old_item, new_item])
31
    else:
32
        # 插入新数据
33
        pending.append(new_item)
34
        index[tabId] = new_item
35
        print(f"已存入（等待配对） tabId={tabId}")

配对数据后进行处理

1
def handle_pair(pair_list):
2
    # pair_list contains two Item objects (来自 process_incoming_json)
3
    left = pair_list[0]
4
    right = pair_list[1]
5

6
    url1 = left.data.get("url")
7
    url2 = right.data.get("url")
8
    filename1 = left.data.get("fullFileName")
9
    filename2 = right.data.get("fullFileName")
10
    source = left.data.get("origin", "")
11
    title = left.data.get("title", "")
12
    subtitle = left.data.get("webUrl", "")
13

14
    # 固定 Referer 为 bilibili，并强制使用默认 User-Agent；不发送 Origin
15
    headers_list = [
16
        "Referer: https://www.bilibili.com",
17
        f"User-Agent: {DEFAULT_USER_AGENT}"
18
    ]
19

20
    # 直接在后台线程下载两个文件并合并
21
    t = threading.Thread(target=download_and_merge_thread, args=(url1, url2, filename1, filename2, title, subtitle, headers_list))
22
    t.daemon = True
23
    t.start()

下载视频到缓存文件夹并触发ffmpeg合并

1
def download_and_merge_thread(url1, url2, filename1, filename2, title, weburl, headers_list):
2
    """后台线程：依次下载两个文件并合并，合并后删除源文件。"""
3
    path1 = download_direct(url1, filename1, headers_list)
4
    path2 = download_direct(url2, filename2, headers_list)
5
    if not path1 or not path2:
6
        print("至少有一个下载失败，跳过合并")
7
        return
8

9
    output_base = build_output_name(title, weburl)
10
    # 确保输出目录存在并写入到 OUTPUT_DIR
11
    try:
12
        os.makedirs(OUTPUT_DIR, exist_ok=True)
13
    except Exception as e:
14
        print(f"创建输出目录失败 {OUTPUT_DIR}: {e}")
15
        output_dir = os.getcwd()
16
    else:
17
        output_dir = OUTPUT_DIR
18

19
    output_file_name = f"{output_base}.mkv"
20
    output_file = os.path.join(output_dir, output_file_name)
21
    ok = merge_with_ffmpeg(path1, path2, output=output_file)
22
    if ok:
23
        for p in (path1, path2):
24
            try:
25
                if os.path.exists(p):
26
                    os.remove(p)
27
                    print(f"已删除源文件: {p}")
28
            except Exception as e:
29
                print(f"删除文件失败 {p}: {e}")

1
def merge_with_ffmpeg(file_a, file_b, output=None):
2
    """使用 ffmpeg 将两个流合并（不转码，直接拷贝）。返回 True/False。
3
    输出文件默认使用第一个文件名加后缀 `_merged.mkv`。
4
    """
5
    if output is None:
6
        base = os.path.splitext(file_a)[0]
7
        output = f"{base}_merged.mkv"
8

9
    cmd = [
10
        "ffmpeg",
11
        "-y",
12
        "-i", file_a,
13
        "-i", file_b,
14
        "-c", "copy",
15
        output
16
    ]
17
    try:
18
        print(f"运行 ffmpeg 合并: {' '.join(cmd)}")
19
        proc = subprocess.run(cmd, capture_output=True, text=True)
20
        if proc.returncode == 0:
21
            print(f"合并完成: {output}")
22
            return True
23
        else:
24
            print(f"ffmpeg 合并失败 (code={proc.returncode}): {proc.stderr}")
25
            return False
26
    except FileNotFoundError:
27
        print("ffmpeg 未找到，请先安装 ffmpeg 并确保其在 PATH 中。")
28
        return False