374 lines
14 KiB
Python
374 lines
14 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",
|
||
}
|
||
|
||
capabilities: Dict[str, bool] = {
|
||
**BaseCloudDriveAdapter.capabilities,
|
||
"ensure_dir": True,
|
||
"save_files": True,
|
||
"rename": True,
|
||
"move_files": True,
|
||
"delete_files": True,
|
||
}
|
||
|
||
def __init__(self, config: PlatformConfig, transfer_config: TransferConfig):
|
||
# BaseCloudDriveAdapter.__init__ calls _setup_session(), so credential
|
||
# state must exist before super().__init__.
|
||
refresh_token = config.refresh_token or config.cookie or ""
|
||
self._credential = AliyunCredentialManager(refresh_token=refresh_token)
|
||
self._drive_id = ""
|
||
self._transfer: Optional[AliyunTransfer] = None
|
||
self._cleanup: Optional[AliyunCleanup] = None
|
||
super().__init__(config, transfer_config)
|
||
|
||
def _setup_session(self):
|
||
"""初始化 session 和凭证"""
|
||
refresh_token = getattr(self._credential, "refresh_token", "")
|
||
if 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,
|
||
)
|
||
|
||
# 确定目标目录：路径先解析/创建为 file_id。
|
||
to_parent = save_dir if save_dir and save_dir != "/" else "root"
|
||
if isinstance(to_parent, str) and to_parent.startswith("/"):
|
||
to_parent = self.ensure_dir(to_parent)
|
||
|
||
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)
|
||
|
||
# ─── 扩展功能 ──────────────────────────────────────────
|
||
|
||
|
||
# ─── Optional Drive API capability methods ─────────────────────
|
||
|
||
def get_fids(self, file_paths: List[str]) -> List[Dict]:
|
||
wanted = {p.rstrip("/") or "/" for p in file_paths}
|
||
found: List[Dict] = []
|
||
for path in wanted:
|
||
if path == "/":
|
||
found.append({"file_path": "/", "fid": "root"})
|
||
continue
|
||
parent = path.rsplit("/", 1)[0] or "/"
|
||
name = path.rsplit("/", 1)[-1]
|
||
for item in self.get_files(parent):
|
||
if item.is_dir and item.name == name:
|
||
found.append({"file_path": path, "fid": item.fid})
|
||
break
|
||
return found
|
||
|
||
def ensure_dir(self, dir_path: str) -> str:
|
||
normalized = "/" + (dir_path or "/").strip("/")
|
||
if normalized == "/":
|
||
return "root"
|
||
current = ""
|
||
last_fid = "root"
|
||
for part in [p for p in normalized.split("/") if p]:
|
||
current = f"{current}/{part}" if current else f"/{part}"
|
||
matches = self.get_fids([current])
|
||
if matches:
|
||
last_fid = matches[0].get("fid") or matches[0].get("file_id") or last_fid
|
||
continue
|
||
created = self.mkdir(current)
|
||
data = created.get("data", created) if isinstance(created, dict) else {}
|
||
last_fid = data.get("fid") or data.get("file_id") or data.get("id") or last_fid
|
||
return last_fid
|
||
|
||
def mkdir(self, dir_path: str) -> Dict:
|
||
parent = dir_path.rsplit("/", 1)[0] or "/"
|
||
name = dir_path.rstrip("/").rsplit("/", 1)[-1]
|
||
parent_fid = "root" if parent == "/" else self.ensure_dir(parent)
|
||
url = "https://api.aliyundrive.com/adrive/v2/file/createWithFolders"
|
||
body = {
|
||
"drive_id": self._drive_id or self._credential.get_drive_id(),
|
||
"parent_file_id": parent_fid,
|
||
"name": name,
|
||
"type": "folder",
|
||
"check_name_mode": "refuse",
|
||
}
|
||
resp = self._post(url, json_data=body, headers=self._credential.get_headers())
|
||
data = self._drive_api_json(resp, context="阿里云盘创建目录")
|
||
return {"code": 0, "status": 200, "data": {"fid": data.get("file_id", ""), **data}}
|
||
|
||
def rename(self, fid: str, file_name: str) -> Dict:
|
||
url = "https://api.aliyundrive.com/v3/file/update"
|
||
body = {"drive_id": self._drive_id or self._credential.get_drive_id(), "file_id": fid, "name": file_name, "check_name_mode": "refuse"}
|
||
data = self._drive_api_json(self._post(url, json_data=body, 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:
|
||
if isinstance(to_pdir_fid, str) and to_pdir_fid.startswith("/"):
|
||
to_pdir_fid = self.ensure_dir(to_pdir_fid)
|
||
drive_id = self._drive_id or self._credential.get_drive_id()
|
||
requests = [{"id": fid, "method": "POST", "url": "/file/move", "headers": {"Content-Type": "application/json"}, "body": {"drive_id": drive_id, "file_id": fid, "to_parent_file_id": to_pdir_fid}} for fid in fids]
|
||
data = self._drive_api_json(self._post("https://api.aliyundrive.com/adrive/v4/batch", json_data={"requests": requests, "resource": "file"}, headers=self._credential.get_headers()), context="阿里云盘移动文件")
|
||
return {"code": 0, "status": 200, "data": data}
|
||
|
||
def delete_files(self, fids: List[str]) -> Dict:
|
||
return {"code": 0, "status": 200} if self.delete(fids) else {"code": -1, "status": 500}
|
||
|
||
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
|