基本配置#

1. 创建对于文件夹

   1
   mkdir -p docker_data/bilisync/data && cd docker_data/bilisync/data
   

2. 下载BBDown、biliup-rs、ffmpeg二进制文件。

   1
   # 这里都以linux x86系统为例
   2
   wget https://github.com/nilaoda/BBDown/releases/download/1.6.3/BBDown_1.6.3_20240814_linux-x64.zip
   3
   unzip BBDown_1.6.3_20240814_linux-x64.zip && chmod +x BBDown && rm BBDown_1.6.3_20240814_linux-x64.zip
   4
   
   5
   wget https://github.com/biliup/biliup-rs/releases/download/v0.2.3/biliupR-v0.2.3-x86_64-linux.tar.xz
   6
   tar -xf biliupR-v0.2.3-x86_64-linux.tar.xz && mv biliupR-v0.2.3-x86_64-linux/biliup . && chmod +x biliup && rm biliupR-v0.2.3-x86_64-linux.tar.xz && rm -rf biliupR-v0.2.3-x86_64-linux
   7
   
   8
   wget https://github.com/BtbN/FFmpeg-Builds/releases/download/latest/ffmpeg-master-latest-linux64-gpl.tar.xz
   9
   tar -xf ffmpeg-master-latest-linux64-gpl.tar.xz && mv ffmpeg-master-latest-linux64-gpl/bin/ffmpeg . && chmod +x ffmpeg && rm ffmpeg-master-latest-linux64-gpl.tar.xz && rm -rf ffmpeg-master-latest-linux64-gpl
   

3. BBDown登录，设置配置文件

   1
   ./BBDown login
   2
   nano BBDown.config
   

   1
   #本文件是BBDown程序的配置文件
   2
   #以#开头的都会被程序忽略
   3
   #然后剩余非空白内容程序逐行读取，对于一个选项，其参数应当在下一行出现
   4
   
   5
   # 单个视频非分P的情况的文件下载后存储规则
   6
   --file-pattern
   7
   <videoTitle>/<videoTitle>
   8
   
   9
   # 单个视频多分P的情况的文件下载后存储规则
   10
   --multi-file-pattern
   11
   <videoTitle>/<pageTitle>
   12
   
   13
   # 下面设置下载多个分P时，每个分P的下载间隔为2秒
   14
   --delay-per-page
   15
   2
   16
   
   17
   # ffmpeg二进制文件路径
   18
   --ffmpeg-path
   19
   /app/data/ffmpeg
   20
   
   21
   # 如果你的机器下载很慢，可以尝试更改这个host，地址可以参考https://github.com/the1812/Bilibili-Evolved/issues/3234#issuecomment-1504764774
   22
   #--upos-host
   23
   #upos-sz-mirrorcosov.bilivideo.com
   24
   
   25
   # 跳过封面下载
   26
   --skip-cover
   27
   
   28
   # 跳过字幕下载
   29
   --skip-subtitle
   30
   
   31
   # 使用aria2下载引擎，不需要请注释
   32
   -aria2
   33
   
   34
   # 额外的aira2下载参数
   35
   --aria2c-args
   36
   -j48 # 48线程
   

4. biliup-rs登录

   1
   ./biliup login
   

   按照提示登录（这个账号是你希望把视频上传到哪个B站账号，和下载账号区分开来）

安全警告

运行登录命令后，会在 data 目录下生成包含您B站账户凭证的敏感文件（如 BBDown.data 和 cookies.json）。请妥善保管，切勿在任何地方（如GitHub、论坛）分享这些文件的内容！

代码配置#

1. 创建bilivideosync.py文件

   1
   nano bilivideosync.py
   

   1
   import json
   2
   import os
   3
   import subprocess
   4
   import sys
   5
   import re
   6
   import shutil
   7
   import time
   8
   import threading
   9
   from alist import AlistUploader
   10
   
   11
   # --- 配置区 ---
   12
   # 您的视频数据
   13
   # ❗只要保证每个元素有对应的bvid和title即可。获取方法非常多，说两个我用的
   14
   # 1. 去主页->投稿Tab，找到以`https://api.bilibili.com/x/space/wbi/arc/search?pn=2&ps=40&tid=0&special_type=`开头的请求，复制preview->data->list->vlist数据
   15
   # 2. 打开某一个合集，找到以`https://api.bilibili.com/x/polymer/web-space/seasons_archives_list?mid=`开头的请求，复制preview->data->archives数据
   16
   JSON_DATA = """
   17
   [
   18
     {
   19
       "aid": xxxxx,
   20
       "bvid": "xxxxx",
   21
       "ctime": xxxxx,
   22
       "duration": xxxx,
   23
       "enable_vt": false,
   24
       "interactive_video": false,
   25
       "pic": "http://xxxxx",
   26
       "playback_position": 0,
   27
       "pubdate": xxxxx,
   28
       "stat": {
   29
         "view": xxxxx,
   30
         "vt": 0
   31
       },
   32
       "state": 0,
   33
       "title": "xxxxxx",
   34
       "ugc_pay": 0,
   35
       "vt_display": "",
   36
       "is_lesson_video": 0
   37
     }
   38
   ]
   39
   """
   40
   # 日志文件名
   41
   SUCCESS_LOG_FILE = "success.log"
   42
   FAIL_LOG_FILE = "fail.log"
   43
   
   44
   # Alist配置
   45
   ALIST_BASE_URL = "http://172.17.0.1:65007"  # 修改为你的Alist服务器地址
   46
   ALIST_USERNAME = "xxxxxx"                   # ❗修改为你的Alist用户名
   47
   ALIST_PASSWORD = "xxxxxx"                   # ❗修改为你的Alist密码
   48
   
   49
   # --- 配置区结束 ---
   50
   
   51
   # 全局变量，用于存储一次性输入的用户信息
   52
   upload_details = {}
   53
   
   54
   # 初始化Alist上传器
   55
   alist_uploader = AlistUploader(ALIST_BASE_URL, ALIST_USERNAME, ALIST_PASSWORD)
   56
   
   57
   def check_executables():
   58
       """检查依赖的可执行文件是否存在且可执行"""
   59
       if not (os.path.exists('./BBDown') and os.access('./BBDown', os.X_OK)):
   60
           print("❌ 错误: './BBDown' 不存在或没有执行权限。")
   61
           sys.exit(1)
   62
       if not (os.path.exists('./biliup') and os.access('./biliup', os.X_OK)):
   63
           print("❌ 错误: './biliup' 不存在或没有执行权限。")
   64
           sys.exit(1)
   65
       print("✅ 依赖检查通过 (BBDown, biliup)。")
   66
   
   67
   def sanitize_filename(filename):
   68
       """移除Windows和Linux文件名中的非法字符以匹配BBDown的行为"""
   69
       return re.sub(r'[\\/:*?"<>|]', '_', filename)
   70
   
   71
   def find_video_files(directory):
   72
       """在指定目录中查找所有视频文件，并返回一个路径列表。"""
   73
       video_paths = []
   74
       if not os.path.isdir(directory):
   75
           return video_paths # 返回空列表
   76
   
   77
       # 排序以保证P序稳定
   78
       file_list = sorted(os.listdir(directory))
   79
   
   80
       for file in file_list:
   81
           if file.lower().endswith(('.mp4', '.flv', '.mkv', '.webm')):
   82
               video_paths.append(os.path.join(directory, file))
   83
       return video_paths
   84
   
   85
   def read_log_file(filepath):
   86
       """读取日志文件内容到集合中"""
   87
       if not os.path.exists(filepath):
   88
           return set()
   89
       with open(filepath, 'r', encoding='utf-8') as f:
   90
           return set(line.strip() for line in f if line.strip())
   91
   
   92
   def update_log_files(bvid, status):
   93
       """更新成功或失败日志"""
   94
       success_bvids = read_log_file(SUCCESS_LOG_FILE)
   95
       fail_bvids = read_log_file(FAIL_LOG_FILE)
   96
       if status == 'success':
   97
           success_bvids.add(bvid)
   98
           if bvid in fail_bvids:
   99
               fail_bvids.remove(bvid)
   100
       else: # status == 'fail'
   101
           fail_bvids.add(bvid)
   102
           if bvid in success_bvids:
   103
               success_bvids.remove(bvid)
   104
       with open(SUCCESS_LOG_FILE, 'w', encoding='utf-8') as f:
   105
           f.write('\n'.join(sorted(list(success_bvids))) + '\n')
   106
       with open(FAIL_LOG_FILE, 'w', encoding='utf-8') as f:
   107
           f.write('\n'.join(sorted(list(fail_bvids))) + '\n')
   108
   
   109
   def get_upload_details_from_user():
   110
       """从用户处获取上传所需信息，仅执行一次"""
   111
       if not upload_details:
   112
           print("\n--- 首次上传，请输入投稿信息 (后续将自动复用) ---")
   113
           upload_details['tid'] = input("请输入投稿分区TID (例如: 171代表单机游戏): ")
   114
           upload_details['desc'] = input("请输入视频简介: ")
   115
           upload_details['tags'] = input("请输入视频标签 (用英文逗号 ',' 分隔): ")
   116
   def download_video(bvid, title):
   117
       """
   118
       下载指定BVID的视频，并实时显示下载进度。
   119
       """
   120
       print(f"\n[📥] 正在下载: {title} (BVID: {bvid})")
   121
       print("-" * 40)
   122
       try:
   123
           command = ['./BBDown', bvid]
   124
           subprocess.run(command, check=True, text=True)
   125
           print("-" * 40)
   126
           print(f"✅ 下载成功: {title}")
   127
           return True
   128
       except subprocess.CalledProcessError:
   129
           print("-" * 40)
   130
           print(f"❌ 下载失败: {title} (BVID: {bvid})")
   131
           return False
   132
       except FileNotFoundError:
   133
           print("-" * 40)
   134
           print(f"❌ 命令执行失败: 找不到 './BBDown' 可执行文件。")
   135
           return False
   136
   
   137
   def upload_to_alist(title, video_paths):
   138
       """
   139
       上传视频文件到Alist指定路径
   140
   
   141
       参数:
   142
           title: 视频标题，用于生成Alist目录路径
   143
           video_paths: 视频文件路径列表
   144
   
   145
       返回:
   146
           成功上传所有文件返回True，否则返回False
   147
       """
   148
       if not video_paths:
   149
           print("⚠️ 没有找到需要上传到Alist的视频文件")
   150
           return False
   151
   
   152
       # 生成Alist目标路径
   153
       remote_dir_path = alist_uploader.generate_folder_path(title)
   154
       print(f"[🗂️] 创建Alist目录: {remote_dir_path}")
   155
   
   156
       # 创建目标文件夹
   157
       if not alist_uploader.create_new_folder(remote_dir_path):
   158
           print(f"❌ 创建Alist目录失败: {remote_dir_path}")
   159
           return False
   160
   
   161
       # 开始上传文件
   162
       print(f"[☁️] 开始上传 {len(video_paths)} 个文件到Alist...")
   163
       upload_complete_event = alist_uploader.upload_files(video_paths, remote_dir_path)
   164
   
   165
       # 等待所有上传完成
   166
       print("等待所有文件上传完成...")
   167
       upload_complete_event.wait()
   168
   
   169
       # 检查是否所有文件都上传成功
   170
       success = alist_uploader.all_uploads_successful()
   171
       if success:
   172
           print(f"✅ 所有文件都成功上传到Alist: {remote_dir_path}")
   173
       else:
   174
           print(f"❌ 部分文件上传Alist失败，请检查日志")
   175
   
   176
       return success
   177
   
   178
   def upload_video(bvid, title):
   179
       """上传指定视频并返回成功与否"""
   180
       sanitized_folder = sanitize_filename(title)
   181
       video_paths = find_video_files(sanitized_folder)
   182
       if not video_paths:
   183
           print(f"❌ 上传预处理失败: 在 '{sanitized_folder}' 目录中找不到视频文件。")
   184
           return False
   185
   
   186
       get_upload_details_from_user()
   187
   
   188
       print(f"\n[🚀] 正在上传: {title} (共 {len(video_paths)} 个分P)")
   189
       command = [
   190
           './biliup', 'upload',
   191
           '--copyright', '1',
   192
           '--line', 'txa',
   193
           '--limit', '20',
   194
           '--title', title,
   195
           '--tid', upload_details['tid'],
   196
           '--desc', upload_details['desc'],
   197
           '--tag', upload_details['tags'],
   198
       ]
   199
   
   200
       command.extend(video_paths)
   201
       try:
   202
           subprocess.run(command, check=True, text=True)
   203
           print(f"✅ 上传成功: {title}")
   204
           return True
   205
       except subprocess.CalledProcessError:
   206
           print(f"❌ 上传失败: {title}")
   207
           return False
   208
       except FileNotFoundError:
   209
           print(f"❌ 命令执行失败: 找不到 './biliup' 可执行文件。")
   210
           return False
   211
   
   212
   def process_video(bvid, title, download_first=True):
   213
       """完整的处理流程：下载 -> 上传到Alist -> 上传到B站 -> 清理 -> 记录"""
   214
       if download_first:
   215
           if not download_video(bvid, title):
   216
               update_log_files(bvid, 'fail')
   217
               return
   218
   
   219
       # 获取下载的视频文件路径
   220
       sanitized_folder = sanitize_filename(title)
   221
       video_paths = find_video_files(sanitized_folder)
   222
   
   223
       # 上传到Alist（可选步骤，失败不影响后续B站上传）
   224
       # ❗如果不需要上传到openlist，请把下面三行注释掉
   225
       print("\n[☁️] 开始上传到Alist...")
   226
       alist_success = upload_to_alist(title, video_paths)
   227
       print(f"[☁️] Alist上传{'成功' if alist_success else '失败'}")
   228
   
   229
       # 上传到B站
   230
       if upload_video(bvid, title):
   231
           update_log_files(bvid, 'success')
   232
           # 上传成功后，删除文件夹
   233
           try:
   234
               shutil.rmtree(sanitized_folder)
   235
               print(f"🗑️ 已清理文件夹: {sanitized_folder}")
   236
           except OSError as e:
   237
               print(f"⚠️ 清理文件夹失败: {sanitized_folder}, 错误: {e}")
   238
       else:
   239
           update_log_files(bvid, 'fail')
   240
   
   241
   def main():
   242
       """主执行函数"""
   243
       check_executables()
   244
   
   245
       try:
   246
           videos_data = json.loads(JSON_DATA)
   247
           video_map = {v['bvid']: v for v in videos_data}
   248
       except json.JSONDecodeError:
   249
           print("❌ 错误: JSON 数据格式不正确。")
   250
           sys.exit(1)
   251
   
   252
       successful_bvids = read_log_file(SUCCESS_LOG_FILE)
   253
       failed_bvids = read_log_file(FAIL_LOG_FILE)
   254
       processed_bvids = set()
   255
   
   256
       if failed_bvids:
   257
           print("\n--- 检查上次失败的任务 ---")
   258
           for bvid in list(failed_bvids):
   259
               if bvid not in video_map:
   260
                   print(f"ℹ️ 失败列表中的 {bvid} 在当前JSON数据中不存在，已跳过。")
   261
                   continue
   262
               video_info = video_map[bvid]
   263
               title = video_info['title']
   264
               sanitized_folder = sanitize_filename(title)
   265
   
   266
               print(f"\n[🔄] 正在重试失败任务: {title} (BVID: {bvid})")
   267
               if os.path.isdir(sanitized_folder):
   268
                   print("   └── 检测到已下载的文件夹，将直接尝试上传。")
   269
                   process_video(bvid, title, download_first=False)
   270
               else:
   271
                   print("   └── 未检测到文件夹，将重新下载并上传。")
   272
                   process_video(bvid, title, download_first=True)
   273
   
   274
               processed_bvids.add(bvid)
   275
   
   276
       print("\n--- 开始处理新的任务列表 ---")
   277
       for video in videos_data:
   278
           bvid = video.get("bvid")
   279
           title = video.get("title")
   280
           if not bvid or not title:
   281
               continue
   282
   
   283
           if bvid in processed_bvids:
   284
               print(f"ℹ️ {bvid} 已在失败重试阶段处理过，跳过。")
   285
               continue
   286
   
   287
           if bvid in successful_bvids:
   288
               print(f"ℹ️ {title} (BVID: {bvid}) 已在之前成功上传，跳过。")
   289
               continue
   290
   
   291
           process_video(bvid, title, download_first=True)
   292
   
   293
       print("\n\n🎉🎉🎉 所有任务处理完毕！🎉🎉🎉")
   294
       final_fails = read_log_file(FAIL_LOG_FILE)
   295
       if final_fails:
   296
           print(f"\n⚠️ 注意: 仍有 {len(final_fails)} 个任务失败，请检查 '{FAIL_LOG_FILE}' 文件。")
   297
   
   298
   if __name__ == '__main__':
   299
       main()
   

   注

   注意代码里的❗符号，需要修改或确认自己需要不需要

2. 创建alist.py文件

   1
   nano alist.py
   

   1
   import requests
   2
   import hashlib
   3
   import json
   4
   import urllib.parse
   5
   import os
   6
   import time
   7
   import threading
   8
   import queue
   9
   import re
   10
   from datetime import datetime
   11
   from tqdm import tqdm
   12
   import subprocess
   13
   
   14
   class AlistUploader:
   15
       def __init__(self, base_url, username, password):
   16
           self.base_url = base_url
   17
           self.username = username
   18
           self.password = password
   19
           self.token = None
   20
           self.upload_queue = queue.Queue()
   21
           self.upload_threads = []
   22
           self.max_threads = 6  # 最大并发上传线程数
   23
           self.upload_results = {}  # 存储上传结果
   24
           self.upload_complete_event = threading.Event()
   25
   
   26
       def get_token(self):
   27
           """获取 Alist 的 token"""
   28
           if self.token:
   29
               return self.token
   30
   
   31
           url = f"{self.base_url}/api/auth/login/hash"
   32
   
   33
           password_with_suffix = f"{self.password}-https://github.com/alist-org/alist"
   34
           hashed_password = hashlib.sha256(password_with_suffix.encode()).hexdigest()
   35
   
   36
           payload = {
   37
               "username": self.username,
   38
               "password": hashed_password
   39
           }
   40
   
   41
           headers = {"Content-Type": "application/json"}
   42
           response = requests.post(url, headers=headers, data=json.dumps(payload))
   43
   
   44
           if response.status_code == 200:
   45
               result = response.json()
   46
               if result["code"] == 200 and result["message"] == "success":
   47
                   self.token = result["data"]["token"]
   48
                   return self.token
   49
   
   50
           # print(f"获取 token 失败: {response.text}")
   51
           tqdm.write(f"获取 token 失败: {response.text}")
   52
           return None
   53
   
   54
       def create_new_folder(self, folder_path):
   55
           """创建文件夹"""
   56
           url = f"{self.base_url}/api/fs/mkdir"
   57
           token = self.get_token()
   58
           if not token:
   59
               return False
   60
   
   61
           headers = {
   62
               "Authorization": token,
   63
               "Content-Type": "application/json"
   64
           }
   65
           payload = {
   66
               "path": folder_path
   67
           }
   68
   
   69
           response = requests.post(url, headers=headers, json=payload)
   70
           if response.status_code == 200:
   71
               result = response.json()
   72
               if result["code"] == 200 and result["message"] == "success":
   73
                   # print(f"新建文件夹成功: {folder_path}")
   74
                   tqdm.write(f"新建文件夹成功: {folder_path}")
   75
                   return True
   76
               else:
   77
                   # print(f"新建文件夹失败: {result['message']}")
   78
                   tqdm.write(f"新建文件夹失败: {result['message']}")
   79
                   return False
   80
           else:
   81
               # print(f"新建文件夹请求失败，状态码: {response.status_code}, 响应内容: {response.text}")
   82
               tqdm.write(f"新建文件夹请求失败，状态码: {response.status_code}, 响应内容: {response.text}")
   83
               return False
   84
   
   85
       def extract_date_user_from_title(self, title):
   86
           """从标题中提取日期和用户名"""
   87
           # 提取用户名
   88
           user_match = re.search(r'【弹幕】(.*?)直播回放', title)
   89
           user = user_match.group(1) if user_match else "未知用户"
   90
   
   91
           # 提取日期
   92
           date_match = re.search(r'(\d{4})年(\d{2})月(\d{2})日', title)
   93
           if date_match:
   94
               year = date_match.group(1)
   95
               month = date_match.group(2)
   96
               day = date_match.group(3)
   97
           else:
   98
               # 尝试其他日期格式
   99
               date_match = re.search(r'(\d{4})[-.年](\d{1,2})[-.月](\d{1,2})', title)
   100
               if date_match:
   101
                   year = date_match.group(1)
   102
                   month = date_match.group(2).zfill(2)
   103
                   day = date_match.group(3).zfill(2)
   104
               else:
   105
                   # 使用当前日期
   106
                   now = datetime.now()
   107
                   year = str(now.year)
   108
                   month = str(now.month).zfill(2)
   109
                   day = str(now.day).zfill(2)
   110
   
   111
           return user, year, month, day
   112
   
   113
       def generate_folder_path(self, title):
   114
           """生成文件夹路径: /115/录播/{{user}}/{{yyyy}}年/{{MM}}-{{dd}}"""
   115
           user, year, month, day = self.extract_date_user_from_title(title)
   116
           return f"/115/录播/{user}/{year}年/{month}-{day}"
   117
   
   118
   
   119
       def upload_file(self, local_file_path, remote_dir_path, as_task=True, retry=0):
   120
           """
   121
           上传单个文件到 Alist，通过调用系统 curl 命令实现。
   122
           避免 Python 内存占用，并为并行上传做准备（并行性由调用方管理）。
   123
           """
   124
           token = self.get_token()
   125
           if not token:
   126
               tqdm.write("无法获取 AList 访问令牌。")
   127
               return False
   128
   
   129
           file_name = os.path.basename(local_file_path)
   130
           # 确保远程目录路径以 / 结尾，或者不以 / 结尾，然后拼接文件名为 /
   131
           # Alist 的 File-Path 头部需要一个完整的远程文件路径，例如 /path/to/dir/file.txt
   132
           remote_file_path = f"{remote_dir_path.rstrip('/')}/{file_name}"
   133
   
   134
           url = f"{self.base_url}/api/fs/form"
   135
   
   136
           # 构建 curl 命令
   137
           cmd_parts = [
   138
               "curl",
   139
               "-X", "PUT",  # HTTP 方法为 PUT
   140
               "--http1.1",
   141
               "--retry", "3",               # 失败自动重试3次
   142
               "--retry-delay", "5",          # 重试间隔5秒
   143
               "-H", f"Authorization: {token}",
   144
               # File-Path 头部的值需要 URL 编码
   145
               "-H", f"File-Path: {urllib.parse.quote(remote_file_path)}",
   146
           ]
   147
   
   148
           if as_task:
   149
               cmd_parts.extend(["-H", "As-Task: true"])
   150
   
   151
           cmd_parts.extend(["-s", "--show-error", "--fail"])
   152
           cmd_parts.extend(["-F", f"file=@{local_file_path}"])
   153
   
   154
           # 添加 URL
   155
           cmd_parts.append(url)
   156
   
   157
           # 可选：为了调试，你可以添加 -v (verbose) 或 --progress-bar
   158
           # cmd_parts.append("-v") # 输出更详细的请求/响应过程
   159
           # cmd_parts.append("--progress-bar") # 显示进度条
   160
   
   161
           try:
   162
               curl_response = subprocess.run(
   163
                   cmd_parts,
   164
                   capture_output=True, # 捕获 stdout 和 stderr
   165
                   text=True,
   166
                   check=False, # 不在非零退出码时抛出 CalledProcessError
   167
                   encoding='utf-8',
   168
               )
   169
   
   170
               # 优先解析 AList 的 JSON 响应
   171
               result = None
   172
               try:
   173
                   if curl_response.stdout: # 只有当 stdout 不为空时才尝试解析
   174
                       result = json.loads(curl_response.stdout)
   175
               except json.JSONDecodeError:
   176
                   # 如果 JSON 解析失败，则这是一个错误情况
   177
                   error_message = f"无法解析 AList 响应 JSON: {curl_response.stdout.strip() if curl_response.stdout else '无输出'}"
   178
                   if curl_response.stderr:
   179
                       error_message += f"\ncURL 错误输出: {curl_response.stderr.strip()}"
   180
   
   181
                   if retry < 3:
   182
                       tqdm.write(f"上传 {file_name} 失败: {error_message}，正在重试...")
   183
                       return self.upload_file(local_file_path, remote_dir_path, as_task, retry + 1)
   184
                   else:
   185
                       tqdm.write(f"上传 {file_name} 最终失败: {error_message}")
   186
                       return False
   187
   
   188
               # 根据 AList 的 JSON 响应判断成功或失败
   189
               if result and result.get("code") == 200 and result.get("message") == "success":
   190
                   # AList 明确表示成功，即使 curl 退出码非零 (如 92)，也视为成功
   191
                   if as_task and 'task' in result.get("data", {}):
   192
                       task_id = result["data"]["task"]["id"]
   193
                       # tqdm.write(f"文件 {file_name} 上传任务已创建，任务ID: {task_id}")
   194
                       return self.monitor_upload_task(task_id, local_file_path, remote_dir_path)
   195
   
   196
                   # tqdm.write(f"文件 {file_name} 上传成功。")
   197
                   return True
   198
               else:
   199
                   # AList 返回了错误代码或消息，或者没有返回有效的 JSON (已在上层捕获)
   200
                   alist_error_message = result.get("message", "未知错误或无AList响应") if result else "无AList响应或解析失败"
   201
   
   202
                   # 组合 AList 的错误信息和 curl 的退出码/错误输出，用于日志
   203
                   full_error_detail = f"AList错误: {alist_error_message}. "
   204
                   if curl_response.returncode != 0:
   205
                       full_error_detail += f"curl退出码: {curl_response.returncode}. "
   206
                   if curl_response.stderr:
   207
                       full_error_detail += f"cURL错误输出: {curl_response.stderr.strip()}."
   208
   
   209
                   if retry < 3:
   210
                       tqdm.write(f"上传 {file_name} 失败: {full_error_detail}，正在重试...")
   211
                       return self.upload_file(local_file_path, remote_dir_path, as_task, retry + 1)
   212
                   else:
   213
                       tqdm.write(f"上传 {file_name} 最终失败: {full_error_detail}")
   214
                       return False
   215
   
   216
           except FileNotFoundError:
   217
               # curl 命令本身没有找到
   218
               tqdm.write(f"错误: curl 命令未找到。请确保 curl 已安装并配置在系统 PATH 中。")
   219
               return False
   220
           except Exception as e:
   221
               # 捕获其他任何意外的Python异常
   222
               if retry < 3:
   223
                   tqdm.write(f"上传文件 {file_name} 时出错: {str(e)}，正在重试...")
   224
                   return self.upload_file(local_file_path, remote_dir_path, as_task, retry + 1)
   225
               else:
   226
                   tqdm.write(f"上传文件 {file_name} 时出错: {str(e)}")
   227
                   return False
   228
   
   229
       def get_task_info(self, task_id):
   230
           """获取指定任务的信息"""
   231
           token = self.get_token()
   232
           if not token:
   233
               return None
   234
   
   235
           url = f"{self.base_url}/api/task/upload/info"
   236
   
   237
           params = {"tid": task_id}
   238
           headers = {"Authorization": token}
   239
   
   240
           response = requests.post(url, headers=headers, params=params)
   241
   
   242
           if response.status_code == 200:
   243
               result = response.json()
   244
               if result["code"] == 200 and result["message"] == "success":
   245
                   return result["data"]
   246
               else:
   247
                   # print(f"获取任务信息失败: {result['message']}")
   248
                   tqdm.write(f"获取任务信息失败: {result['message']}")
   249
                   return None
   250
           else:
   251
               tqdm.write(f"获取任务信息失败，状态码: {response.status_code}, 响应内容: {response.text}")
   252
               return None
   253
   
   254
       def monitor_upload_task(self, task_id, local_file_path=None, remote_dir_path=None, poll_interval=2, timeout=7200, retry=0):
   255
           """
   256
           监控任务进度，直到任务完成，失败时支持重试
   257
   
   258
           参数:
   259
               task_id: 任务ID
   260
               local_file_path: 本地文件路径，用于重试
   261
               remote_dir_path: 远程目录路径，用于重试
   262
               poll_interval: 轮询间隔（秒）
   263
               timeout: 超时时间（秒）
   264
               retry: 剩余重试次数
   265
   
   266
           返回:
   267
               成功时返回任务信息，失败且重试次数用完时返回False
   268
           """
   269
           from tqdm import tqdm
   270
   
   271
           start_time = time.time()
   272
           pbar = None
   273
   
   274
           try:
   275
               while time.time() - start_time < timeout:
   276
                   task_info = self.get_task_info(task_id)
   277
   
   278
                   if not task_info:
   279
                       time.sleep(poll_interval)
   280
                       continue
   281
   
   282
                   # 获取进度
   283
                   progress = task_info.get('progress', 0)
   284
                   state = task_info.get('state', '')
   285
   
   286
                   # 初始化进度条
   287
                   if pbar is None:
   288
                       # pbar = tqdm(total=100, desc=f"上传任务 {task_id}", unit="%")
   289
                       pbar = tqdm(total=100, desc=f"{local_file_path.split('/')[-1]} 上传中", unit="%")
   290
   
   291
                   # 更新进度条
   292
                   pbar.n = int(progress * 100)
   293
                   pbar.refresh()
   294
   
   295
                   # 检查任务是否完成
   296
                   state = task_info.get('state', '')
   297
                   error = task_info.get('error', '')
   298
   
   299
                   if state == 2 and progress >= 1.0:
   300
                       # tqdm.write(f"任务完成！总共传输: {task_info.get('total_bytes', 0)} 字节")
   301
                       return True
   302
                   elif state == 2 and progress == 0:
   303
                       # print("任务已完成，秒传成功")
   304
                       # tqdm.write("任务已完成，秒传成功")
   305
                       return True
   306
                   elif error:
   307
                       retry += 1
   308
                       # 如果有错误且有重试次数，尝试重新上传
   309
                       if retry < 4 and local_file_path and remote_dir_path and os.path.exists(local_file_path):
   310
                           pbar.close()
   311
                           pbar = None
   312
                           # print(f"任务失败: {error}，剩余重试次数: {4 - retry}")
   313
                           tqdm.write(f"任务失败: {error}，剩余重试次数: {4 - retry}")
   314
   
   315
                           # 重新上传文件
   316
                           file_name = os.path.basename(local_file_path)
   317
                           remote_file_path = f"{remote_dir_path}/{file_name}"
   318
   
   319
                           url = f"{self.base_url}/api/fs/form"
   320
                           headers = {
   321
                               "Authorization": self.get_token(),
   322
                               "File-Path": urllib.parse.quote(remote_file_path),
   323
                               "As-Task": "true",
   324
                               "Content-Length": str(os.path.getsize(local_file_path))
   325
                           }
   326
   
   327
                           with open(local_file_path, 'rb') as file:
   328
                               # 核心修改：重试时也使用 data 参数进行流式上传
   329
                               response = requests.put(url, headers=headers, data=file)
   330
   
   331
                           if response.status_code == 200:
   332
                               result = response.json()
   333
                               if result["code"] == 200 and result["message"] == "success" and 'task' in result["data"]:
   334
                                   new_task_id = result["data"]["task"]["id"]
   335
                                   tqdm.write(f"重新创建上传任务，新任务ID: {new_task_id}")
   336
                                   # 递归调用自身监控新任务，减少重试次数
   337
                                   return self.monitor_upload_task(
   338
                                       new_task_id,
   339
                                       local_file_path,
   340
                                       remote_dir_path,
   341
                                       poll_interval,
   342
                                       timeout - (time.time() - start_time),  # 剩余超时时间
   343
                                       retry - 1
   344
                                   )
   345
   
   346
                       elif retry >= 4:
   347
                           tqdm.write(f"任务失败: {error}，所有重试次数已用完")
   348
                           return False
   349
                       else:
   350
                           tqdm.write(f"任务失败: {error}，无法重试")
   351
                           return False
   352
   
   353
                   time.sleep(poll_interval)
   354
   
   355
               tqdm.write(f"\n监控超时，已经等待 {timeout} 秒")
   356
               return False
   357
           finally:
   358
               # 确保进度条被关闭
   359
               if pbar is not None:
   360
                   pbar.close()
   361
   
   362
       def _worker(self):
   363
           """工作线程函数，处理上传队列中的任务"""
   364
           while True:
   365
               try:
   366
                   file_info = self.upload_queue.get(block=False)
   367
                   if file_info is None:  # 结束信号
   368
                       self.upload_queue.task_done()
   369
                       break
   370
   
   371
                   local_file_path, remote_dir_path = file_info
   372
                   file_name = os.path.basename(local_file_path)
   373
   
   374
                   # tqdm.write(f"开始上传文件: {file_name} 到 {remote_dir_path}")
   375
                   success = self.upload_file(local_file_path, remote_dir_path)
   376
   
   377
                   # 记录上传结果
   378
                   self.upload_results[local_file_path] = success
   379
   
   380
                   # tqdm.write(f"文件 {file_name} {'上传成功' if success else '上传失败'}")
   381
                   self.upload_queue.task_done()
   382
               except queue.Empty:
   383
                   # 队列为空，检查是否所有任务已完成
   384
                   if self.upload_queue.unfinished_tasks == 0:
   385
                       break
   386
                   time.sleep(0.5)
   387
               except Exception as e:
   388
                   tqdm.write(f"上传线程出错: {str(e)}")
   389
                   self.upload_queue.task_done()
   390
   
   391
       def upload_files(self, file_paths, remote_dir_path):
   392
           """
   393
           异步上传多个文件到 Alist 的指定目录
   394
   
   395
           参数:
   396
               file_paths: 本地文件路径列表
   397
               remote_dir_path: 远程目录路径
   398
   
   399
           返回:
   400
               一个事件对象，可以用于等待上传完成
   401
           """
   402
           # 清空之前的结果和事件状态
   403
           self.upload_results = {}
   404
           self.upload_complete_event.clear()
   405
   
   406
           # 将所有文件加入上传队列
   407
           for file_path in file_paths:
   408
               self.upload_queue.put((file_path, remote_dir_path))
   409
   
   410
           # 启动工作线程
   411
           self.upload_threads = []
   412
           for _ in range(min(self.max_threads, len(file_paths))):
   413
               thread = threading.Thread(target=self._worker)
   414
               thread.daemon = True
   415
               thread.start()
   416
               self.upload_threads.append(thread)
   417
   
   418
           # 启动监控线程，等待所有上传完成
   419
           monitor_thread = threading.Thread(target=self._monitor_upload_completion)
   420
           monitor_thread.daemon = True
   421
           monitor_thread.start()
   422
   
   423
           return self.upload_complete_event
   424
   
   425
       def _monitor_upload_completion(self):
   426
           """监控线程，等待所有上传任务完成"""
   427
           for thread in self.upload_threads:
   428
               thread.join()
   429
           self.upload_complete_event.set()
   430
   
   431
           # 汇总上传结果
   432
           total_files = len(self.upload_results)
   433
           success_count = sum(1 for success in self.upload_results.values() if success)
   434
           tqdm.write(f"\n上传完成: {success_count}/{total_files} 文件上传成功")
   435
   
   436
       def all_uploads_successful(self):
   437
           """检查所有文件是否都上传成功"""
   438
           if not self.upload_results:
   439
               return False
   440
           return all(self.upload_results.values())
   441
   
   442
   # 测试代码
   443
   if __name__ == "__main__":
   444
       uploader = AlistUploader(
   445
           base_url="http://xxxxx",
   446
           username="xxxxx",
   447
           password="xxxxx"
   448
       )
   449
   
   450
       # 测试创建文件夹
   451
       test_folder = "/115/录播/测试用户/2025年/05-31"
   452
       uploader.create_new_folder(test_folder)
   453
   
   454
       # 测试上传文件
   455
       test_files = ["/config/Desktop/bilibili.tar.gz"]
   456
       event = uploader.upload_files(test_files, test_folder)
   457
   
   458
       # 等待上传完成
   459
       event.wait()
   460
       tqdm.write(f"所有文件上传完成，全部成功: {uploader.all_uploads_successful()}")
   

配置Docker容器#

1. 创建Dockerfile和docker-compose.yaml

   1
   cd .. && nano Dockerfile && nano docker-compose.yaml`
   

   1
   FROM python:3.10-bookworm
   2
   
   3
   WORKDIR /app
   4
   
   5
   # 安装aria2和对应python包
   6
   RUN apt update && \
   7
       apt install -y \
   8
       aria2 && \
   9
       apt clean && \
   10
       rm -rf /var/lib/apt/lists/* && \
   11
       pip install --no-cache-dir tqdm requests
   12
   
   13
   
   14
   CMD ["python3", "bilivideosync.py"]
   

   1
   services:
   2
     bilivideosync:
   3
       build: .
   4
       volumes:
   5
         - ./data:/app
   6
       tty: true # 必须保留，否则第一个视频下载并同步到网盘结束后，无法输入所需的上传到B站的分区ID，视频简介和视频标签
   7
       stdin_open: true # 必须保留，否则第一个视频下载并同步到网盘结束后，无法输入所需的上传到B站的分区ID，视频简介和视频标签
   8
       network_mode: bridge
   

2. 启动容器docker compose up -d
3. 启动容器后，输入docker compose attach --sig-proxy=false bilivideosync进入容器内交互式终端，等待输入分区ID的提示出现，分别填写完分区ID、视频简介和视频Tags后，即可退出，后续会复用第一次填写的信息