admin
83cbfaf03f
- 修复Redis认证 (配置密码) - 启动Python管理后台 (端口9531, 15个功能开关) - 统一版本号 0.2.7 - 更新docker-compose.yml (镜像版本/Redis URL/Admin服务)
254 lines
8.4 KiB
Python
254 lines
8.4 KiB
Python
"""
|
||
百度网盘适配器 — CloudSearch Transfer v1.0.0
|
||
参考 cloud-auto-save 的 BaiduNetDisk + netdisk 的 PanbaiduSave
|
||
|
||
完整的 5 步转存流程 + bdstoken 管理 + 路径删除 + 广告过滤
|
||
"""
|
||
|
||
import logging
|
||
from typing import List, Tuple
|
||
|
||
from ..base import BaseCloudDriveAdapter, FileInfo
|
||
from ...config import PlatformConfig, TransferConfig
|
||
from ...errors import TransferError, TransferErrorCode
|
||
|
||
from .credential import BaiduCredentialManager
|
||
from .transfer import BaiduTransfer
|
||
from .cleanup import BaiduCleanup
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class BaiduAdapter(BaseCloudDriveAdapter):
|
||
"""百度网盘适配器
|
||
|
||
完整的 Cookie + bdstoken 机制，支持:
|
||
- 验证分享链接 + 提取码
|
||
- 5 步转存到自己的网盘
|
||
- 创建新分享
|
||
- 按文件名删除文件
|
||
- 广告文件过滤
|
||
"""
|
||
|
||
PLATFORM_NAME = "百度网盘"
|
||
PLATFORM_KEY = "baidu"
|
||
URL_PATTERNS = [
|
||
r'pan\.baidu\.com/s/1([A-Za-z0-9_-]+)',
|
||
]
|
||
|
||
def __init__(self, config: PlatformConfig, transfer_config: TransferConfig):
|
||
super().__init__(config, transfer_config)
|
||
|
||
# 凭证管理器
|
||
self.credential = BaiduCredentialManager(
|
||
cookie=config.cookie,
|
||
session=self.session,
|
||
)
|
||
|
||
if not self.credential.validate():
|
||
raise TransferError(
|
||
TransferErrorCode.NOT_LOGIN,
|
||
message="百度网盘 Cookie 无效或太短 (需 >= 50 字符)",
|
||
platform=self.PLATFORM_KEY,
|
||
)
|
||
|
||
# 预热 bdstoken
|
||
try:
|
||
self.credential.get_bdstoken()
|
||
except TransferError as e:
|
||
logger.warning(f"预取 bdstoken 失败: {e}，将在首次使用时重试")
|
||
|
||
# 转存执行器 & 清理器
|
||
self._transfer = BaiduTransfer(self.session, self.credential)
|
||
self._cleanup = BaiduCleanup(
|
||
self.session, self.credential,
|
||
ad_keywords=config.banned_keywords or None,
|
||
)
|
||
|
||
# 暂存最近一次转存的文件信息（供 _filter_ads 使用）
|
||
self._last_transfer_files: List[dict] = []
|
||
|
||
# ─── session 初始化 ─────────────────────────────────────
|
||
|
||
def _setup_session(self):
|
||
"""设置 session 级别的 Cookie"""
|
||
if self.config.cookie:
|
||
self.session.headers["Cookie"] = self.config.cookie
|
||
self.session.headers["Referer"] = "https://pan.baidu.com/"
|
||
|
||
# ─── 核心抽象方法实现 ──────────────────────────────────
|
||
|
||
def _get_share_detail(self, pwd_id: str, passcode: str = "") -> dict:
|
||
"""获取百度分享详情（步骤 ①+②）
|
||
|
||
Args:
|
||
pwd_id: URL 中的 surl (s/1 后面的部分)
|
||
passcode: 提取码（可选）
|
||
|
||
Returns:
|
||
{"title": str, "fs_ids": [str], "filenames": [str], ...}
|
||
"""
|
||
bdstoken = self.credential.get_bdstoken()
|
||
|
||
# ① 验证提取码（如果有）
|
||
if passcode:
|
||
self._transfer._verify_password(pwd_id, passcode, bdstoken)
|
||
|
||
# ② 解析分享页
|
||
share_info = self._transfer._parse_share_page(pwd_id)
|
||
|
||
return {
|
||
"title": share_info.get("title", ""),
|
||
"shareid": share_info["shareid"],
|
||
"uk": share_info["uk"],
|
||
"fs_ids": share_info["fs_ids"],
|
||
"filenames": share_info["filenames"],
|
||
}
|
||
|
||
def _save_files(self, pwd_id: str, detail: dict,
|
||
save_dir: str) -> List[str]:
|
||
"""转存文件到自己的百度网盘（步骤 ③+④）
|
||
|
||
Args:
|
||
pwd_id: surl
|
||
detail: _get_share_detail 返回的 dict
|
||
save_dir: 目标目录
|
||
|
||
Returns:
|
||
转存后的新 fs_id 列表
|
||
"""
|
||
bdstoken = self.credential.get_bdstoken()
|
||
shareid = detail["shareid"]
|
||
uk = detail["uk"]
|
||
fs_ids = detail["fs_ids"]
|
||
filenames = detail.get("filenames", [])
|
||
|
||
# ③ 转存
|
||
self._transfer._transfer_files(shareid, uk, fs_ids, save_dir, bdstoken)
|
||
|
||
# ④ 列出目录匹配新 fs_id
|
||
new_fs_ids = self._transfer._list_and_match(save_dir, filenames, bdstoken)
|
||
|
||
# 暂存文件信息供 _filter_ads + _create_share 使用
|
||
self._last_transfer_files = [
|
||
{"fs_id": fid, "name": name}
|
||
for fid, name in zip(new_fs_ids, filenames)
|
||
if fid
|
||
]
|
||
|
||
return new_fs_ids
|
||
|
||
def _create_share(self, file_ids: List[str], title: str,
|
||
password: str = "") -> Tuple[str, str]:
|
||
"""创建百度分享（步骤 ⑤）
|
||
|
||
Args:
|
||
file_ids: 转存后的新 fs_id 列表
|
||
title: 原标题
|
||
password: 分享密码
|
||
|
||
Returns:
|
||
(new_share_url, share_password)
|
||
"""
|
||
# 如果 file_ids 中包含非数字，尝试从暂存信息中查找
|
||
numeric_ids = []
|
||
for fid in file_ids:
|
||
try:
|
||
int(fid)
|
||
numeric_ids.append(fid)
|
||
except ValueError:
|
||
logger.warning(f"忽略非数字 fs_id: {fid}")
|
||
|
||
return self._transfer.create_share(
|
||
fids=[int(x) for x in numeric_ids] if numeric_ids else [int(x) for x in file_ids],
|
||
password=password,
|
||
period=0, # 永久
|
||
)
|
||
|
||
# ─── 文件列表 & 删除 ────────────────────────────────────
|
||
|
||
def get_files(self, parent_fid: str = "0") -> List[FileInfo]:
|
||
"""列出百度网盘目录下的文件
|
||
|
||
GET /api/list?dir={parent_fid}
|
||
|
||
Args:
|
||
parent_fid: 目录路径 (默认 "0" = 根目录)
|
||
|
||
注意: parent_fid 对百度网盘而言是目录路径而非数字 ID。
|
||
根目录传 "/" 或 "0"。
|
||
"""
|
||
bdstoken = self.credential.get_bdstoken()
|
||
dir_path = parent_fid if parent_fid != "0" else "/"
|
||
|
||
url = "https://pan.baidu.com/api/list"
|
||
params = {"dir": dir_path, "bdstoken": bdstoken}
|
||
headers = self.credential.get_headers()
|
||
|
||
try:
|
||
resp = self._get(url, params=params, headers=headers)
|
||
data = resp.json()
|
||
except Exception as e:
|
||
logger.error(f"百度列出目录失败: {e}")
|
||
return []
|
||
|
||
errno = data.get("errno", -1)
|
||
if errno != 0:
|
||
logger.error(f"百度列出目录 errno={errno}: {data}")
|
||
return []
|
||
|
||
files = []
|
||
for item in data.get("list", []):
|
||
fid = str(item.get("fs_id", ""))
|
||
name = item.get("server_filename", "")
|
||
size = item.get("size", 0)
|
||
is_dir = item.get("isdir", 0) == 1
|
||
ext = ""
|
||
if not is_dir and "." in name:
|
||
ext = name.rsplit(".", 1)[-1]
|
||
|
||
files.append(FileInfo(
|
||
fid=fid,
|
||
name=name,
|
||
size=size,
|
||
is_dir=is_dir,
|
||
ext=ext,
|
||
))
|
||
|
||
return files
|
||
|
||
def delete(self, file_ids: List[str]) -> bool:
|
||
"""删除百度网盘文件（按路径）
|
||
|
||
file_ids 应为网盘中的完整路径，如 ["/dir/file.txt", "/dir/file2.zip"]
|
||
|
||
Args:
|
||
file_ids: 网盘路径列表
|
||
|
||
Returns:
|
||
True 删除成功（或文件不存在）
|
||
"""
|
||
return self._cleanup.delete_files(file_ids)
|
||
|
||
# ─── 广告过滤 ────────────────────────────────────────────
|
||
|
||
def _filter_ads(self, file_ids: List[str]) -> List[str]:
|
||
"""广告过滤 — 基于最近一次转存暂存的文件名"""
|
||
if not self._last_transfer_files:
|
||
return file_ids
|
||
|
||
names = []
|
||
for f in self._last_transfer_files:
|
||
if f["fs_id"] in file_ids:
|
||
names.append(f["name"])
|
||
else:
|
||
names.append("")
|
||
|
||
return self._cleanup.filter_ad_ids(file_ids, names)
|
||
|
||
# ─── 扩展方法 ────────────────────────────────────────────
|
||
|
||
def delete_paths(self, paths: List[str]) -> bool:
|
||
"""便捷删除方法（直接调用 cleanup）"""
|
||
return self._cleanup.delete_files(paths)
|