admin
83cbfaf03f
- 修复Redis认证 (配置密码) - 启动Python管理后台 (端口9531, 15个功能开关) - 统一版本号 0.2.7 - 更新docker-compose.yml (镜像版本/Redis URL/Admin服务)
298 lines
10 KiB
Python
298 lines
10 KiB
Python
"""
|
||
阿里云盘适配器 v1.0.0
|
||
AliyunAdapter — 继承 BaseCloudDriveAdapter，实现阿里云盘全部转存能力。
|
||
|
||
组件：
|
||
- AliyunCredentialManager: refresh_token 刷新 + 缓存
|
||
- AliyunTransfer: 4 步批量转存
|
||
- AliyunCleanup: 回收站清理
|
||
|
||
URL 匹配: aliyundrive.com/s/<share_id>
|
||
"""
|
||
|
||
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
|