""" 阿里云盘适配器 v1.0.0 AliyunAdapter — 继承 BaseCloudDriveAdapter，实现阿里云盘全部转存能力。 组件： - AliyunCredentialManager: refresh_token 刷新 + 缓存 - AliyunTransfer: 4 步批量转存 - AliyunCleanup: 回收站清理 URL 匹配: aliyundrive.com/s/ """ import re import logging from typing import List, Dict, Tuple, Optional from ..base import BaseCloudDriveAdapter, FileInfo, match_url from ..config import PlatformConfig, TransferConfig from ..errors import TransferError, TransferErrorCode from .credential import AliyunCredentialManager from .transfer import AliyunTransfer from .cleanup import AliyunCleanup logger = logging.getLogger(__name__) class AliyunAdapter(BaseCloudDriveAdapter): """阿里云盘适配器""" PLATFORM_NAME = "阿里云盘" PLATFORM_KEY = "aliyun" URL_PATTERNS = [ r'aliyundrive\.com/s/([a-zA-Z0-9]+)', r'alipan\.com/s/([a-zA-Z0-9]+)', ] DEFAULT_HEADERS = { "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/135.0.0.0 Safari/537.36" ), "Accept": "application/json, text/plain, */*", "Content-Type": "application/json", "Referer": "https://aliyundrive.com", } def __init__(self, config: PlatformConfig, transfer_config: TransferConfig): super().__init__(config, transfer_config) # 创建凭证管理器（AliyunCredentialManager） refresh_token = config.refresh_token or config.cookie or "" self._credential = AliyunCredentialManager(refresh_token=refresh_token) # 初始化 drive_id self._drive_id = "" # 创建子模块 self._transfer: Optional[AliyunTransfer] = None self._cleanup: Optional[AliyunCleanup] = None def _setup_session(self): """初始化 session 和凭证""" if self._credential.refresh_token: # 验证 refresh_token 并获取 drive_id if self._credential.validate(): self._drive_id = self._credential.get_drive_id() logger.info( f"[AliyunAdapter] 凭证验证成功, drive_id={self._drive_id[:8]}..." ) else: logger.warning("[AliyunAdapter] 凭证验证失败，转存功能可能不可用") else: logger.warning("[AliyunAdapter] 未配置 refresh_token") # ─── 核心抽象方法实现 ────────────────────────────────── def _get_share_detail(self, pwd_id: str, passcode: str = "") -> dict: """ 获取分享详情。 步骤①②: 先获取匿名分享信息，再获取 share_token。 Returns: { "title": "分享标题", "share_id": "...", "share_token": "...", "files": [{"file_id": "...", "name": "...", "size": 0, "type": "file"}, ...], } """ try: transfer = self._get_transfer() # ① 获取分享信息（匿名） share_info = transfer._get_share_info(pwd_id) if not share_info: raise TransferError( TransferErrorCode.SHARE_NOT_EXIST, platform=self.PLATFORM_KEY, ) # ② 获取分享令牌（Auth） share_token = transfer._get_share_token(pwd_id, passcode) if not share_token: raise TransferError( TransferErrorCode.PASSCODE_WRONG if passcode else TransferErrorCode.SHARE_NOT_EXIST, platform=self.PLATFORM_KEY, message="获取分享令牌失败（可能需要提取码）", ) return { "title": share_info.get("share_name", share_info.get("share_title", "")), "share_id": pwd_id, "share_token": share_token, "files": share_info.get("file_infos", []), "creator_name": share_info.get("creator_name", ""), } except TransferError: raise except Exception as e: logger.exception(f"[AliyunAdapter] 获取分享详情失败: {e}") raise TransferError( TransferErrorCode.NETWORK_ERROR, message=str(e), platform=self.PLATFORM_KEY, ) def _save_files(self, pwd_id: str, detail: dict, save_dir: str) -> List[str]: """ 步骤③: 批量复制文件到自己的网盘。 Args: pwd_id: 分享 ID detail: _get_share_detail 的返回值 save_dir: 目标目录（根目录用 "root"） Returns: 新文件 ID 列表 """ share_token = detail.get("share_token", "") files = detail.get("files", []) if not share_token: raise TransferError( TransferErrorCode.SHARE_NOT_EXIST, message="缺少 share_token", platform=self.PLATFORM_KEY, ) if not files: raise TransferError( TransferErrorCode.RESOURCE_EMPTY, platform=self.PLATFORM_KEY, ) file_ids = [f.get("file_id", "") for f in files if f.get("file_id")] if not file_ids: raise TransferError( TransferErrorCode.RESOURCE_EMPTY, message="无法提取文件 ID", platform=self.PLATFORM_KEY, ) # 确定目标目录 to_parent = save_dir if save_dir and save_dir != "/" else "root" transfer = self._get_transfer() new_ids = transfer._batch_copy(pwd_id, share_token, file_ids, to_parent) if not new_ids: raise TransferError( TransferErrorCode.NETWORK_ERROR, message="批量转存失败，所有文件复制均失败", platform=self.PLATFORM_KEY, ) return new_ids def _create_share( self, file_ids: List[str], title: str, password: str = "" ) -> Tuple[str, str]: """ 步骤④: 创建新分享链接。 Returns: (share_url, share_password) """ if not file_ids: raise TransferError( TransferErrorCode.RESOURCE_EMPTY, platform=self.PLATFORM_KEY, ) transfer = self._get_transfer() result = transfer._create_share(file_ids, password) share_url = result.get("share_url", "") share_pwd = result.get("share_pwd", password) if not share_url: raise TransferError( TransferErrorCode.SHARE_LINK_FAIL, message="创建分享链接失败", platform=self.PLATFORM_KEY, ) return share_url, share_pwd def get_files(self, parent_fid: str = "0") -> List[FileInfo]: """ 列出网盘目录下的文件。 NOTE: 当前实现为占位。如需完整功能，请调用阿里云盘 /adrive/v3/file/list API。 """ logger.warning("[AliyunAdapter] get_files() 未完整实现，返回空列表") return [] def delete(self, file_ids: List[str]) -> bool: """ 删除文件（移入回收站）。 Args: file_ids: 要删除的文件 ID 列表 Returns: 是否全部删除成功 """ if not file_ids: return True cleanup = self._get_cleanup() result = cleanup.delete_files(file_ids) return result.get("success", False) # ─── 扩展功能 ────────────────────────────────────────── def cleanup_files(self, file_ids: List[str]) -> Dict: """ 清理文件（移入回收站），返回详细结果。 Returns: AliyunCleanup.delete_files() 的返回字典 """ cleanup = self._get_cleanup() return cleanup.delete_files(file_ids) def force_refresh_token(self) -> bool: """强制刷新 access_token""" return self._credential.refresh() def get_credential_status(self) -> Dict: """获取当前凭证状态""" return self._credential.to_dict() # ─── 文件列表提取 ────────────────────────────────────── def _extract_file_list(self, detail: dict) -> List[FileInfo]: """从分享详情中提取 FileInfo 列表""" files = detail.get("files", []) result = [] for f in files: result.append(FileInfo( fid=f.get("file_id", ""), name=f.get("name", ""), size=int(f.get("size", 0)), is_dir=f.get("type", "") == "folder", ext=f.get("file_extension", ""), )) return result # ─── 内部辅助方法 ────────────────────────────────────── def _get_transfer(self) -> AliyunTransfer: """懒加载获取 AliyunTransfer 实例""" if self._transfer is None: drive_id = self._drive_id or self._credential.get_drive_id() self._transfer = AliyunTransfer( credential=self._credential, drive_id=drive_id, to_parent_file_id=self.config.save_dir or "root", request_timeout=self.transfer_config.request_timeout, ) return self._transfer def _get_cleanup(self) -> AliyunCleanup: """懒加载获取 AliyunCleanup 实例""" if self._cleanup is None: drive_id = self._drive_id or self._credential.get_drive_id() self._cleanup = AliyunCleanup( credential=self._credential, drive_id=drive_id, request_timeout=self.transfer_config.request_timeout, ) return self._cleanup