""" CloudSearch Transfer — 迅雷网盘适配器 v1.0.0 PLATFORM_KEY = 'xunlei' 迅雷网盘使用 refresh_token + captcha_token 双重认证。 """ from __future__ import annotations import logging from typing import List, Optional, Tuple, Dict from ..base import ( BaseCloudDriveAdapter, FileInfo, TransferResult, VerifyResult, ) from ...config import PlatformConfig, TransferConfig from ...errors import TransferError, TransferErrorCode from .credential import XunleiCredentialManager from .transfer import XunleiTransfer from .cleanup import XunleiCleanup logger = logging.getLogger(__name__) class XunleiAdapter(BaseCloudDriveAdapter): """迅雷网盘适配器""" PLATFORM_NAME = "迅雷网盘" PLATFORM_KEY = "xunlei" URL_PATTERNS = [r"pan\.xunlei\.com/s/([A-Za-z0-9]+)"] capabilities: Dict[str, bool] = { **BaseCloudDriveAdapter.capabilities, "ensure_dir": True, "save_files": True, "poll_task": True, "rename": True, "move_files": True, # batchDelete is permanent on Xunlei, so do not advertise it as generic delete_files. "delete_files": False, } def __init__(self, config: PlatformConfig, transfer_config: TransferConfig): # BaseCloudDriveAdapter.__init__ calls _setup_session(), so credential # state must exist before super().__init__. self._credential = XunleiCredentialManager(config) self._transfer_engine: Optional[XunleiTransfer] = None self._cleanup = XunleiCleanup(self._credential) super().__init__(config, transfer_config) def _setup_session(self): """初始化 session 认证头""" headers = self._credential.get_auth_headers() if headers: self.session.headers.update(headers) def _ensure_auth(self): """确保认证头是最新的""" headers = self._credential.get_auth_headers() self.session.headers.update(headers) @property def _transfer(self) -> XunleiTransfer: """懒加载转存引擎""" if self._transfer_engine is None: self._transfer_engine = XunleiTransfer( credential=self._credential, timeout=self.transfer_config.request_timeout, poll_interval=self.transfer_config.task_poll_interval, poll_max_attempts=self.transfer_config.task_poll_max_attempts, ) self._transfer_engine.session = self.session return self._transfer_engine # ─── 抽象方法实现 ────────────────────────────── def _get_share_detail(self, pwd_id: str, passcode: str = "") -> dict: self._ensure_auth() return self._transfer._get_share_info(pwd_id, passcode) def _save_files(self, pwd_id: str, detail: dict, save_dir: str) -> List[str]: self._ensure_auth() target = self.ensure_dir(save_dir) if save_dir and save_dir.startswith("/") else (save_dir or "") files = detail.get("files", []) file_ids = [f.get("file_id") or f.get("fid") or f.get("id") for f in files if f.get("file_id") or f.get("fid") or f.get("id")] if not file_ids: raise RuntimeError("无法从分享中提取文件ID") task_id = self._transfer._restore_files(pwd_id, detail.get("pass_code_token", ""), file_ids, parent_id=target) mapping = self._transfer._poll_restore_task(task_id) return [mapping.get(fid, "") for fid in file_ids if mapping.get(fid, "")] def _create_share(self, file_ids: List[str], title: str, password: str = "") -> Tuple[str, str]: self._ensure_auth() return self._transfer._create_share(file_ids, password=password) def _extract_file_list(self, detail: dict) -> List[FileInfo]: files = detail.get("files", []) return [ FileInfo(fid=f.get("id", ""), name=f.get("name", ""), size=f.get("size", 0), is_dir=f.get("is_dir", False)) for f in files ] def _filter_ads(self, file_ids: List[str]) -> List[str]: banned = self._get_banned_keywords() return self._cleanup.filter_ad_ids( file_ids, getattr(self._transfer, "_last_file_names", []), banned, ) def get_files(self, parent_fid: str = "0") -> List[FileInfo]: self._ensure_auth() url = "https://api-pan.xunlei.com/drive/v1/files" params = {"parent_id": "" if parent_fid in ("0", "/") else parent_fid} data = self._drive_api_json(self._get(url, params=params, headers=self._credential.get_headers()), context="迅雷网盘列目录") items = data.get("files", data.get("list", [])) return [FileInfo(fid=item.get("id") or item.get("file_id", ""), name=item.get("name", ""), size=item.get("size", 0), is_dir=item.get("kind") == "drive#folder" or item.get("is_dir", False)) for item in items] def ensure_dir(self, dir_path: str) -> str: normalized = "/" + (dir_path or "/").strip("/") if normalized == "/": return "" parent_id = "" current = "" for part in [p for p in normalized.split("/") if p]: current = f"{current}/{part}" if current else f"/{part}" existing = next((item for item in self.get_files(parent_id or "0") if item.is_dir and item.name == part), None) if existing: parent_id = existing.fid continue result = self.mkdir(current if not parent_id else part, parent_id=parent_id) parent_id = result.get("data", {}).get("fid", parent_id) return parent_id def get_fids(self, file_paths: List[str]) -> List[Dict]: """Resolve existing file/directory paths without creating anything.""" results: List[Dict] = [] for path in file_paths: normalized = "/" + (path or "").strip("/") if normalized == "/": results.append({"file_path": path, "fid": ""}) continue parent_id = "" found: Optional[FileInfo] = None missing = False parts = [p for p in normalized.split("/") if p] for index, part in enumerate(parts): found = next((item for item in self.get_files(parent_id or "0") if item.name == part), None) if not found: missing = True break if index < len(parts) - 1 and not found.is_dir: missing = True break parent_id = found.fid if not missing and found: results.append({"file_path": path, "fid": found.fid}) return results def mkdir(self, dir_path: str, parent_id: str = "") -> Dict: name = dir_path.rstrip("/").rsplit("/", 1)[-1] body = {"kind": "drive#folder", "name": name, "parent_id": parent_id or ""} data = self._drive_api_json(self._post("https://api-pan.xunlei.com/drive/v1/files", json_data=body, headers=self._credential.get_headers_with_captcha(action="mkdir")), context="迅雷网盘创建目录") return {"code": 0, "status": 200, "data": {"fid": data.get("id") or data.get("file_id", ""), **data}} def rename(self, fid: str, file_name: str) -> Dict: data = self._drive_api_json(self._post(f"https://api-pan.xunlei.com/drive/v1/files/{fid}", json_data={"name": file_name}, headers=self._credential.get_headers()), context="迅雷网盘重命名") return {"code": 0, "status": 200, "data": data} def move_files(self, fids: List[str], to_pdir_fid: str) -> Dict: target = self.ensure_dir(to_pdir_fid) if to_pdir_fid.startswith("/") else to_pdir_fid data = self._drive_api_json(self._post("https://api-pan.xunlei.com/drive/v1/files:batchMove", json_data={"ids": fids, "parent_id": target}, headers=self._credential.get_headers()), context="迅雷网盘移动文件") return {"code": 0, "status": 200, "data": data} def query_task(self, task_id: str) -> Dict: return self.poll_task(task_id) def poll_task(self, task_id: str) -> Dict: return {"code": 0, "status": 200, "data": self._transfer._poll_restore_task(task_id)} def delete_files(self, fids: List[str]) -> Dict: return {"code": 0, "status": 200} if self.delete(fids) else {"code": -1, "status": 500} def delete(self, file_ids: List[str]) -> bool: self._ensure_auth() return self._cleanup.delete_files(file_ids) def _get_banned_keywords(self) -> List[str]: return self.config.banned_keywords or self.transfer_config.default_banned_keywords def close(self): self.session.close() def __repr__(self): return f""