import json import time import requests import re import shutil import subprocess from pathlib import Path from logger import get_system_logger, log_exception # ================= 配置区域 ================= SESSION_DIR = Path("./session") COOKIE_FILE = Path("./cookies.json") # 【这里填你 B 站网页上看到的合集 ID】 # 脚本会自动根据这两个 ID 去查找对应的 Section ID (小节ID) SEASON_ID_A = 7196643 # 合集 A (同名视频) SEASON_ID_B = 7196624 # 合集 B (Upload切片) # 自动寻找 biliup BILIUP_PATH = shutil.which("biliup") or "biliup" # =========================================== logger = get_system_logger("collection_manager") class BiliCollectionClient: def __init__(self): if not COOKIE_FILE.exists(): raise FileNotFoundError(f"Cookies 文件不存在: {COOKIE_FILE}") with open(COOKIE_FILE, "r", encoding="utf-8") as f: data = json.load(f) # 兼容处理 cookie 格式 if "cookie_info" in data: self.cookies = {c["name"]: c["value"] for c in data.get("cookie_info", {}).get("cookies", [])} else: self.cookies = data self.csrf = self.cookies.get("bili_jct") if not self.csrf: raise ValueError("Cookie 中缺少 bili_jct (CSRF Token)") self.session = requests.Session() self.session.cookies.update(self.cookies) # 使用你测试成功的 Headers self.session.headers.update({ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Referer": "https://member.bilibili.com/platform/upload-manager/distribution" }) def get_video_info(self, bvid): """通过 BVID 获取 AID, CID 和 Title""" url = "https://api.bilibili.com/x/web-interface/view" try: res = self.session.get(url, params={"bvid": bvid}, timeout=10).json() if res["code"] != 0: logger.error(f"查询视频信息失败 [{bvid}]: {res['message']}") return None data = res["data"] return { "aid": data["aid"], "cid": data["cid"], "title": data["title"] } except Exception as e: logger.error(f"获取视频信息异常: {e}") return None def resolve_section_id(self, target_season_id): """ 【关键逻辑】通过 Season ID (合集ID) 查找 Section ID (小节ID) """ url = "https://member.bilibili.com/x2/creative/web/seasons" params = {"pn": 1, "ps": 50} # 获取前50个合集 try: res = self.session.get(url, params=params, timeout=10).json() if res.get("code") != 0: logger.error(f"获取合集列表失败: {res.get('message')}") return None seasons = res.get("data", {}).get("seasons", []) for s in seasons: current_sid = s.get("season", {}).get("id") # 找到目标合集 if current_sid == target_season_id: title = s.get("season", {}).get("title", "未知标题") sections = s.get("sections", {}).get("sections", []) if sections: # 默认取第一个小节 first_section_id = sections[0]["id"] logger.info(f"✅ ID解析成功: 合集[{title}]({target_season_id}) -> 小节ID: {first_section_id}") return first_section_id else: logger.error(f"❌ 合集[{title}]({target_season_id}) 存在，但没有创建任何小节！") return None logger.error(f"❌ 未找到 Season ID 为 {target_season_id} 的合集，请检查 ID 是否正确。") return None except Exception as e: logger.error(f"解析 Section ID 异常: {e}") return None def add_video_to_section(self, section_id, video_info): """正式添加视频到合集""" url = "https://member.bilibili.com/x2/creative/web/season/section/episodes/add" # 参数必须包含 csrf params = {"csrf": self.csrf} payload = { "sectionId": section_id, "episodes": [{ "aid": video_info["aid"], "cid": video_info["cid"], "title": video_info["title"], "charging_pay": 0 }] } try: res = self.session.post(url, params=params, json=payload, timeout=15).json() if res["code"] == 0: logger.info(f"🎉 成功添加: {video_info['title']}") return True else: logger.error(f"添加失败: {res['message']} (Code: {res['code']})") return False except Exception as e: logger.error(f"添加请求异常: {e}") return False class CollectionWorker: def __init__(self, client, section_id_a, section_id_b): self.client = client self.section_id_a = section_id_a self.section_id_b = section_id_b self.ansi_escape = re.compile(r"\x1b\[[0-9;]*[A-Za-z]") def fetch_recent_videos(self): """获取最近投稿""" try: cmd = [str(BILIUP_PATH), "list", "--max-pages", "2"] res = subprocess.run(cmd, capture_output=True, text=True, encoding='utf-8') videos = [] for line in self.ansi_escape.sub("", res.stdout).splitlines(): parts = line.split() if len(parts) >= 2 and parts[0].startswith("BV"): raw_title = " ".join(parts[1:]) title = re.sub(r"(开放浏览|直播回放|审核中|-)$", "", raw_title).strip() videos.append({"bvid": parts[0], "title": title}) return videos except Exception: logger.warning("biliup list 执行失败，跳过同名视频匹配。") return [] def normalize(self, text): return re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9]', '', text).replace('点', '时').lower() def find_bvid_by_title(self, target_title, video_list): target_norm = self.normalize(target_title) for v in video_list: v_norm = self.normalize(v['title']) if target_norm in v_norm or v_norm in target_norm: return v['bvid'] return None def process_folder(self, folder: Path, video_list): flag_a = folder / "collection_a_done.flag" flag_b = folder / "collection_b_done.flag" # 任务 A: 同名视频 -> 合集 A if self.section_id_a and not flag_a.exists(): matched_bvid = self.find_bvid_by_title(folder.name, video_list) if matched_bvid: logger.info(f"任务A (同名): 匹配到 {matched_bvid}，尝试添加...") info = self.client.get_video_info(matched_bvid) if info and self.client.add_video_to_section(self.section_id_a, info): flag_a.touch() # 任务 B: Upload切片 -> 合集 B if self.section_id_b and not flag_b.exists(): bvid_file = folder / "bvid.txt" if bvid_file.exists(): bvid = bvid_file.read_text(encoding='utf-8').strip() logger.info(f"任务B (切片): 读取到 {bvid}，尝试添加...") info = self.client.get_video_info(bvid) if info and self.client.add_video_to_section(self.section_id_b, info): flag_b.touch() def main(): logger.info("启动合集管理模块 (基于成功测试版)...") try: client = BiliCollectionClient() except Exception as e: logger.error(f"客户端初始化失败: {e}") return # 1. 解析 ID (这是最关键的一步) logger.info("正在解析合集 ID...") real_section_a = None real_section_b = None if SEASON_ID_A > 0: real_section_a = client.resolve_section_id(SEASON_ID_A) if SEASON_ID_B > 0: real_section_b = client.resolve_section_id(SEASON_ID_B) if not real_section_a and not real_section_b: logger.error("没有解析到任何有效的 Section ID，脚本停止。") return # 2. 初始化 Worker worker = CollectionWorker(client, real_section_a, real_section_b) # 3. 扫描逻辑 logger.info("开始扫描目录...") recent_videos = worker.fetch_recent_videos() if SESSION_DIR.exists(): for folder in SESSION_DIR.iterdir(): if folder.is_dir(): worker.process_folder(folder, recent_videos) logger.info("扫描完成。") if __name__ == "__main__": main()