155 lines
5.9 KiB
Python
Executable File
155 lines
5.9 KiB
Python
Executable File
import json
|
||
import time
|
||
import requests
|
||
import subprocess
|
||
import re
|
||
import shutil
|
||
import random
|
||
from pathlib import Path
|
||
|
||
# ================= 配置区域 =================
|
||
COOKIE_FILE = Path("./cookies.json")
|
||
TARGET_SEASON_ID = 7196643
|
||
# 必须包含的关键词
|
||
MUST_KEYWORDS = []
|
||
# 必须排除的关键词
|
||
EXCLUDE_KEYWORD = "纯享"
|
||
|
||
BILIUP_PATH = shutil.which("biliup") or "biliup"
|
||
# ===========================================
|
||
|
||
class BiliCollectionBatchTool:
|
||
def __init__(self):
|
||
self.load_cookies()
|
||
self.session = requests.Session()
|
||
self.session.headers.update({
|
||
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
|
||
"Referer": "https://member.bilibili.com/platform/upload-manager/distribution"
|
||
})
|
||
|
||
def load_cookies(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)
|
||
self.cookies = {c["name"]: c["value"] for c in data.get("cookie_info", {}).get("cookies", [])} if "cookie_info" in data else data
|
||
self.csrf = self.cookies.get("bili_jct")
|
||
|
||
def get_existing_bvids(self, season_id):
|
||
"""拉取合集内所有已存在的 BVID，确保去重 100% 准确"""
|
||
print(f"📡 正在拉取合集 {season_id} 的现有视频数据...")
|
||
self.session.cookies.update(self.cookies)
|
||
try:
|
||
# 1. 获取 Section ID
|
||
list_url = "https://member.bilibili.com/x2/creative/web/seasons"
|
||
res_list = self.session.get(list_url, params={"pn": 1, "ps": 50}).json()
|
||
section_id = None
|
||
for s in res_list.get("data", {}).get("seasons", []):
|
||
if s.get("season", {}).get("id") == season_id:
|
||
sections = s.get("sections", {}).get("sections", [])
|
||
if sections: section_id = sections[0]["id"]
|
||
break
|
||
|
||
if not section_id: return None, set()
|
||
|
||
# 2. 获取该小节详细列表
|
||
detail_url = "https://member.bilibili.com/x2/creative/web/season/section"
|
||
res_detail = self.session.get(detail_url, params={"id": section_id}).json()
|
||
|
||
existing = set()
|
||
if res_detail.get("code") == 0:
|
||
for ep in res_detail.get("data", {}).get("episodes", []):
|
||
existing.add(ep.get("bvid"))
|
||
|
||
print(f"📊 查重参考：合集内已有 {len(existing)} 个视频。")
|
||
return section_id, existing
|
||
except Exception as e:
|
||
print(f"❌ 查重逻辑失败: {e}")
|
||
return None, set()
|
||
|
||
def fetch_filtered_videos(self, existing_set):
|
||
"""
|
||
核心逻辑修改：
|
||
1. 包含 王海颖, 唱歌, 录播
|
||
2. 不包含 纯享
|
||
3. 不在合集 existing_set 中
|
||
"""
|
||
print(f"🔍 扫描符合条件且不含“{EXCLUDE_KEYWORD}”的视频...")
|
||
try:
|
||
res = subprocess.run([BILIUP_PATH, "list", "--max-pages", "20"], capture_output=True, text=True, encoding='utf-8')
|
||
output = re.sub(r"\x1b\[[0-9;]*[A-Za-z]", "", res.stdout)
|
||
|
||
to_add_bvids = []
|
||
for line in output.splitlines():
|
||
if line.startswith("BV"):
|
||
parts = line.split()
|
||
bvid = parts[0]
|
||
title = " ".join(parts[1:])
|
||
|
||
# 判断逻辑
|
||
is_match = all(kw in title for kw in MUST_KEYWORDS)
|
||
is_excluded = EXCLUDE_KEYWORD in title
|
||
|
||
if is_match and not is_excluded:
|
||
if bvid in existing_set:
|
||
continue
|
||
to_add_bvids.append(bvid)
|
||
return to_add_bvids
|
||
except Exception as e:
|
||
print(f"❌ biliup 调用失败: {e}")
|
||
return []
|
||
|
||
def get_metadata(self, bv_list):
|
||
episodes = []
|
||
for bvid in bv_list:
|
||
url = "https://api.bilibili.com/x/web-interface/view"
|
||
try:
|
||
res = self.session.get(url, params={"bvid": bvid}).json()
|
||
if res["code"] == 0:
|
||
d = res["data"]
|
||
episodes.append({
|
||
"aid": d["aid"], "cid": d["cid"],
|
||
"title": d["title"], "charging_pay": 0
|
||
})
|
||
time.sleep(0.3)
|
||
except: pass
|
||
return episodes
|
||
|
||
def run(self):
|
||
# 1. 深度查重
|
||
section_id, existing_set = self.get_existing_bvids(TARGET_SEASON_ID)
|
||
if not section_id:
|
||
print("❌ 无法解析合集，任务终止。")
|
||
return
|
||
|
||
# 2. 条件过滤 + 查重剔除
|
||
target_bvids = self.fetch_filtered_videos(existing_set)
|
||
|
||
if not target_bvids:
|
||
print("✨ 扫描完毕：没有符合条件的新视频。")
|
||
return
|
||
|
||
print(f"💡 过滤后，确认有 {len(target_bvids)} 个视频待加入合集。")
|
||
|
||
# 3. 解析元数据
|
||
final_list = self.get_metadata(target_bvids)
|
||
|
||
# 4. 一次性全量提交
|
||
if final_list:
|
||
print(f"🚀 正在发送合并添加请求...")
|
||
add_url = "https://member.bilibili.com/x2/creative/web/season/section/episodes/add"
|
||
res = self.session.post(add_url, params={"csrf": self.csrf}, json={
|
||
"sectionId": section_id,
|
||
"episodes": final_list
|
||
}).json()
|
||
|
||
if res["code"] == 0:
|
||
print(f"🎉 成功！已补齐 {len(final_list)} 个不含“纯享”的录播视频。")
|
||
else:
|
||
print(f"❌ 批量失败: {res['message']} (Code: {res['code']})")
|
||
else:
|
||
print("❌ 未能获取有效的视频详情。")
|
||
|
||
if __name__ == "__main__":
|
||
tool = BiliCollectionBatchTool()
|
||
tool.run() |