feat(adapters): add 123Pan account adapter

This commit is contained in:
shiyu committed 2026-10-03 17:56:40 +08:00
1 parent f856f1a320
commit d7f9a0b304
8 files changed
+587 -2

No files matched your search

+1 -1
View File
@@ -40,7 +40,7 @@ Utilizes an extensible adapter pattern to easily integrate various storage types
| Category | Adapters |
|---|---|
| **Standard Protocols** | Local, S3-compatible, WebDAV, SFTP, FTP |
| **Cloud Drives** | Google Drive, OneDrive, Dropbox, Quark, UC |
| **Cloud Drives** | Google Drive, OneDrive, Dropbox, Quark, UC, 123Pan |
| **Special** | Telegram, AList, Foxel-to-Foxel |
### 🔍 AI-Powered Semantic Search
+1 -1
View File
@@ -40,7 +40,7 @@
| 分类 | 适配器 |
|---|---|
| **标准协议** | 本地存储、S3 兼容存储、WebDAV、SFTP、FTP |
| **网盘服务** | Google Drive、OneDrive、Dropbox、夸克网盘、UC 网盘 |
| **网盘服务** | Google Drive、OneDrive、Dropbox、夸克网盘、UC 网盘、123 网盘 |
| **特殊类型** | Telegram、AList、Foxel 互联 |
### 🔍 AI 语义搜索
+208
View File
@@ -0,0 +1,208 @@
import asyncio
import mimetypes
import posixpath
import tempfile
from datetime import datetime, timezone
from urllib.parse import urlsplit
import httpx
from fastapi import HTTPException
from fastapi.responses import StreamingResponse
from starlette.background import BackgroundTask
def clean_path(path):
parts = str(path or "").replace("\\", "/").split("/")
if any(part in ("..", ".") for part in parts):
raise ValueError("Relative path components are not allowed")
return "/".join(part for part in parts if part)
def join_path(root, rel):
return "/" + "/".join(filter(None, (clean_path(root), clean_path(rel))))
def timestamp(value):
if isinstance(value, (int, float)):
return int(value / 1000 if value > 10**11 else value)
if not value:
return 0
try:
return int(value)
except (ValueError, TypeError):
try:
date = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
return int(date.replace(tzinfo=date.tzinfo or timezone.utc).timestamp())
except ValueError:
return 0
def entry(name, is_dir, size=0, mtime=0, **extra):
return {"name": name, "is_dir": bool(is_dir), "type": "dir" if is_dir else "file",
"size": 0 if is_dir else int(size or 0), "mtime": timestamp(mtime), **extra}
def paginate(items, page_num, page_size, sort_by, sort_order):
field = sort_by if sort_by in ("name", "size", "mtime") else "name"
items = sorted(items, key=lambda item: str(item["name"]).casefold() if field == "name" else item[field],
reverse=sort_order.lower() == "desc")
items.sort(key=lambda item: not item["is_dir"])
start = (max(1, page_num) - 1) * max(1, page_size)
return items[start:start + max(1, page_size)], len(items)
class RemoteFileAdapter:
def _client(self, **kwargs):
return httpx.AsyncClient(timeout=getattr(self, "timeout", 60), **kwargs)
async def _download(self, root, rel):
raise NotImplementedError
async def read_file(self, root, rel):
url, headers = await self._download(root, rel)
async with self._client(follow_redirects=True) as client:
response = await client.get(url, headers=headers)
if response.status_code == 404:
raise FileNotFoundError(rel)
response.raise_for_status()
return response.content
async def stream_file(self, root, rel, range_header):
url, headers = await self._download(root, rel)
if urlsplit(url).scheme not in ("https", "http"):
raise HTTPException(502, "Invalid download URL")
headers = {**headers, "Accept-Encoding": "identity"}
if range_header:
headers["Range"] = range_header
client = self._client(follow_redirects=True)
response = None
async def close():
if response is not None:
await response.aclose()
await client.aclose()
try:
response = await client.send(client.build_request("GET", url, headers=headers), stream=True)
if response.status_code == 416:
raise HTTPException(416, "Range not satisfiable", headers={
"Content-Range": response.headers.get("content-range", "bytes */*")})
if response.status_code == 404:
raise FileNotFoundError(rel)
response.raise_for_status()
except BaseException:
await close()
raise
async def body():
try:
async for chunk in response.aiter_raw():
yield chunk
finally:
await close()
forwarded = {key: response.headers[key] for key in
("content-length", "content-range", "accept-ranges", "etag", "last-modified")
if key in response.headers}
return StreamingResponse(body(), status_code=response.status_code, headers=forwarded,
media_type=response.headers.get("content-type") or
mimetypes.guess_type(rel)[0] or "application/octet-stream",
background=BackgroundTask(close))
async def write_file(self, root, rel, data):
async def chunks():
yield data
return await self.write_file_stream(root, rel, chunks())
async def write_file_stream(self, root, rel, data_iter):
with tempfile.TemporaryFile() as file:
size = 0
async for chunk in data_iter:
if chunk:
await asyncio.to_thread(file.write, chunk)
size += len(chunk)
file.seek(0)
await self.write_upload_file(root, rel, file, posixpath.basename(rel), size)
return size
async def exists(self, root, rel):
try:
await self.stat_file(root, rel)
return True
except FileNotFoundError:
return False
async def stat_path(self, root, rel):
try:
return {**await self.stat_file(root, rel), "exists": True, "path": rel}
except FileNotFoundError:
return {"exists": False, "is_dir": None, "path": rel}
async def rename(self, root, src_rel, dst_rel):
return await self.move(root, src_rel, dst_rel)
async def copy(self, root, src_rel, dst_rel, overwrite=False):
src_rel, dst_rel = clean_path(src_rel), clean_path(dst_rel)
if not src_rel or not dst_rel or dst_rel == src_rel or dst_rel.startswith(src_rel + "/"):
raise ValueError("Invalid copy destination")
item = await self.stat_file(root, src_rel)
if await self.exists(root, dst_rel):
if not overwrite:
raise FileExistsError(dst_rel)
await self.delete(root, dst_rel)
if item["is_dir"]:
await self.mkdir(root, dst_rel)
page = 1
while True:
children, total = await self.list_dir(root, src_rel, page, 100)
for child in children:
await self.copy(root, src_rel + "/" + child["name"], dst_rel + "/" + child["name"])
if page * 100 >= total:
break
page += 1
else:
response = await self.stream_file(root, src_rel, None)
try:
await self.write_file_stream(root, dst_rel, response.body_iterator)
finally:
await response.body_iterator.aclose()
if response.background:
await response.background()
class IDFileAdapter(RemoteFileAdapter):
def get_effective_root(self, sub_path):
return "/".join(filter(None, (str(self.root_id), clean_path(sub_path))))
async def _resolve(self, root, rel):
base, _, sub_path = (root or str(self.root_id)).partition("/")
current = entry("", True, id=base)
for name in clean_path(join_path(sub_path, rel)).split("/"):
if not name:
continue
if not current["is_dir"]:
raise NotADirectoryError(name)
children = await self._children(current["id"])
current = next((child for child in children if child["name"] == name), None)
if current is None:
raise FileNotFoundError(rel)
return current
async def _parent(self, root, rel):
path = clean_path(rel)
if not path:
raise ValueError("Cannot modify mount root")
parent, name = posixpath.split(path)
folder = await self._resolve(root, parent)
if not folder["is_dir"]:
raise NotADirectoryError(parent)
return folder, name
async def stat_file(self, root, rel):
return await self._resolve(root, rel)
async def list_dir(self, root, rel, page_num=1, page_size=50, sort_by="name", sort_order="asc"):
folder = await self._resolve(root, rel)
if not folder["is_dir"]:
raise NotADirectoryError(rel)
return paginate(await self._children(folder["id"]), page_num, page_size, sort_by, sort_order)
+214
View File
@@ -0,0 +1,214 @@
import asyncio
import base64
import hashlib
import secrets
import time
import zlib
from datetime import datetime, timezone, timedelta
from urllib.parse import parse_qs, urlsplit
import aioboto3
from botocore.config import Config
from fastapi import HTTPException
from ._remote import IDFileAdapter, clean_path, entry
class Pan123Adapter(IDFileAdapter):
api_base = "https://yun.123pan.com/b/api"
part_size = 16 * 1024 * 1024
def __init__(self, record):
self.record = record
cfg = record.config or {}
self.username = cfg.get("username")
self.password = cfg.get("password")
self.root_id = str(cfg.get("root_id", "0"))
self.timeout = float(cfg.get("timeout", 60))
if not self.username or not self.password:
raise ValueError("123Pan requires username and password")
self._token = None
self._login_lock = asyncio.Lock()
def _headers(self):
return {"Origin": "https://yun.123pan.com", "Referer": "https://yun.123pan.com/",
"User-Agent": "Mozilla/5.0", "platform": "web", "app-version": "3"}
async def _login(self):
async with self._login_lock:
if self._token:
return
payload = {"mail": self.username, "password": self.password, "type": 2} if "@" in self.username else {
"passport": self.username, "password": self.password, "remember": True}
async with self._client() as client:
response = await client.post("https://login.123pan.com/api/user/sign_in",
headers=self._headers(), json=payload)
response.raise_for_status()
data = response.json()
if data.get("code") != 200 or not (data.get("data") or {}).get("token"):
raise HTTPException(502, "123Pan login failed; check credentials or complete account verification")
self._token = data["data"]["token"]
@staticmethod
def _signature(path):
# The web API signs the request path and a timestamp, not the account credentials.
now = int(time.time())
digits = datetime.fromtimestamp(now, timezone(timedelta(hours=8))).strftime("%Y%m%d%H%M")
encoded = digits.translate(str.maketrans("0123456789", "adefghlmyi"))
key = str(zlib.crc32(encoded.encode()))
nonce = str(secrets.randbelow(10**7))
checksum = zlib.crc32(f"{now}|{nonce}|{path}|web|3|{key}".encode())
return {key: f"{now}-{nonce}-{checksum}"}
async def _api(self, method, path, *, params=None, json=None):
for attempt in range(2):
await self._login()
url = self.api_base + path
async with self._client() as client:
response = await client.request(method, url, json=json,
params={**(params or {}), **self._signature(urlsplit(url).path)},
headers={**self._headers(), "Authorization": "Bearer " + self._token})
if response.status_code == 401 and attempt == 0:
self._token = None
continue
response.raise_for_status()
payload = response.json()
code = payload.get("code")
if code == 401 and attempt == 0:
self._token = None
continue
if code != 0:
raise HTTPException(502, f"123Pan API failed (code={code})")
return payload.get("data") or {}
raise HTTPException(502, "123Pan authentication expired")
async def _children(self, parent_id):
items, page = [], 1
while True:
result = await self._api("GET", "/file/list/new", params={
"parentFileId": parent_id, "driveId": 0, "limit": 100, "Page": page,
"next": 0, "trashed": "false", "orderBy": "file_id", "orderDirection": "desc",
"SearchData": "", "OnlyLookAbnormalFile": 0, "event": "homeListFile",
"operateType": 4, "inDirectSpace": "false"})
batch = result.get("InfoList") or []
items.extend(entry(f["FileName"], f["Type"] == 1, f.get("Size"), f.get("UpdateAt"),
id=str(f["FileId"]), raw=f) for f in batch)
if not batch or str(result.get("Next")) == "-1" or len(items) >= int(result.get("Total", len(items))):
return items
page += 1
async def _download(self, root, rel):
item = await self._resolve(root, rel)
if item["is_dir"]:
raise IsADirectoryError(rel)
f = item["raw"]
result = await self._api("POST", "/file/download_info", json={
"driveId": 0, "fileId": f["FileId"], "fileName": f["FileName"], "size": f["Size"],
"etag": f.get("Etag", ""), "s3keyFlag": f.get("S3KeyFlag", ""), "type": 0})
url = result.get("DownloadUrl", "")
encoded = parse_qs(urlsplit(url).query).get("params")
if encoded:
url = base64.b64decode(encoded[0]).decode()
if urlsplit(url).scheme not in ("https", "http"):
raise HTTPException(502, "123Pan returned no download URL")
async with self._client(follow_redirects=False) as client:
response = await client.send(client.build_request("GET", url, headers={"Referer": "https://yun.123pan.com/"}), stream=True)
try:
if response.is_redirect:
url = str(response.url.join(response.headers["location"]))
elif "application/json" in response.headers.get("content-type", ""):
await response.aread()
url = (response.json().get("data") or {}).get("redirect_url", "")
else:
response.raise_for_status()
finally:
await response.aclose()
if not url:
raise HTTPException(502, "123Pan returned no download URL")
return url, {"Referer": "https://yun.123pan.com/"}
async def mkdir(self, root, rel):
parent, name = await self._parent(root, rel)
await self._api("POST", "/file/upload_request", json={"driveId": 0, "etag": "", "fileName": name,
"parentFileId": int(parent["id"]), "size": 0, "type": 1})
async def delete(self, root, rel):
await self._parent(root, rel)
item = await self._resolve(root, rel)
await self._api("POST", "/file/trash", json={"driveId": 0, "operation": True,
"fileTrashInfoList": [item["raw"]]})
async def move(self, root, src_rel, dst_rel):
src_rel, dst_rel = clean_path(src_rel), clean_path(dst_rel)
if not src_rel or not dst_rel or dst_rel.startswith(src_rel + "/"):
raise ValueError("Invalid move destination")
if src_rel == dst_rel:
return
src_parent, _ = await self._parent(root, src_rel)
item = await self._resolve(root, src_rel)
parent, name = await self._parent(root, dst_rel)
if await self.exists(root, dst_rel):
raise FileExistsError(dst_rel)
if parent["id"] != src_parent["id"]:
await self._api("POST", "/file/mod_pid", json={"fileIdList": [{"FileId": int(item["id"])}],
"parentFileId": int(parent["id"])})
if name != item["name"]:
await self._api("POST", "/file/rename", json={"driveId": 0, "fileId": int(item["id"]), "fileName": name})
async def write_upload_file(self, root, rel, file_obj, filename=None, file_size=None, content_type=None):
parent, name = await self._parent(root, rel)
def measure():
file_obj.seek(0)
digest, size = hashlib.md5(), 0
while chunk := file_obj.read(1024 * 1024):
digest.update(chunk)
size += len(chunk)
file_obj.seek(0)
return digest.hexdigest(), size
md5, size = await asyncio.to_thread(measure)
upload = await self._api("POST", "/file/upload_request", json={
"driveId": 0, "duplicate": 2, "etag": md5, "fileName": name,
"parentFileId": int(parent["id"]), "size": size, "type": 0})
if upload.get("Reuse"):
return {"size": size}
if not upload.get("Key"):
raise HTTPException(502, "123Pan returned incomplete upload information")
if upload.get("AccessKeyId"):
async with aioboto3.Session().client("s3", endpoint_url=upload["EndPoint"], region_name="123pan",
aws_access_key_id=upload["AccessKeyId"], aws_secret_access_key=upload["SecretAccessKey"],
aws_session_token=upload["SessionToken"], config=Config(s3={"addressing_style": "path"})) as client:
await client.upload_fileobj(file_obj, upload["Bucket"], upload["Key"])
await self._api("POST", "/file/upload_complete", json={"fileId": upload["FileId"]})
else:
count = max(1, (size + self.part_size - 1) // self.part_size)
async with self._client() as client:
for number in range(1, count + 1):
signed = await self._api("POST", "/file/s3_repare_upload_parts_batch" if count > 1 else "/file/s3_upload_object/auth",
json={"StorageNode": upload["StorageNode"], "bucket": upload["Bucket"], "key": upload["Key"],
"uploadId": upload["UploadId"], "partNumberStart": number, "partNumberEnd": number + 1})
url = (signed.get("presignedUrls") or {}).get(str(number))
if not url:
raise HTTPException(502, "123Pan returned no upload URL")
response = await client.put(url, content=await asyncio.to_thread(file_obj.read, self.part_size))
response.raise_for_status()
await self._api("POST", "/file/upload_complete/v2", json={
"StorageNode": upload["StorageNode"], "bucket": upload["Bucket"], "key": upload["Key"],
"uploadId": upload["UploadId"], "fileId": upload["FileId"], "fileSize": size, "isMultipart": count > 1})
return {"size": size}
async def get_usage(self, root):
data = await self._api("GET", "/user/info")
used = int(data["SpaceUsed"])
total = int(data.get("SpacePermanent", 0)) + int(data.get("SpaceTemp", 0))
return {"used_bytes": used, "total_bytes": total, "free_bytes": total - used,
"scope": "account", "source": "pan123"}
ADAPTER_TYPE = "pan123"
ADAPTER_FACTORY = Pan123Adapter
CONFIG_SCHEMA = [
{"key": "username", "label": "Username", "type": "string", "required": True},
{"key": "password", "label": "Password", "type": "password", "required": True},
{"key": "root_id", "label": "Root ID", "type": "string", "default": "0"},
{"key": "timeout", "label": "Timeout (seconds)", "type": "number", "default": 60},
]
+101
View File
@@ -0,0 +1,101 @@
import json
import io
from types import SimpleNamespace
import unittest
import httpx
from domain.adapters.providers.pan123 import Pan123Adapter
class Pan123AdapterTests(unittest.IsolatedAsyncioTestCase):
def make_adapter(self, handler):
adapter = Pan123Adapter(SimpleNamespace(config={"username": "user", "password": "password"}))
adapter._client = lambda **kwargs: httpx.AsyncClient(transport=httpx.MockTransport(handler), **kwargs)
return adapter
async def test_login_retry_and_subdirectory_resolution(self):
logins, lists = [], []
def handler(request):
if request.url.host == "login.123pan.com":
logins.append(json.loads(request.content))
return httpx.Response(200, json={"code": 200, "data": {"token": f"token{len(logins)}"}})
self.assertTrue(any("-" in value for value in request.url.params.values()))
self.assertEqual(request.headers["platform"], "web")
parent = request.url.params["parentFileId"]
lists.append(parent)
if len(lists) == 1:
return httpx.Response(200, json={"code": 401})
item = {"FileName": "folder" if parent == "0" else "file.txt",
"FileId": 1 if parent == "0" else 2, "Type": 1 if parent == "0" else 0, "Size": 5}
return httpx.Response(200, json={"code": 0, "data": {"InfoList": [item], "Total": 1}})
adapter = self.make_adapter(handler)
items, total = await adapter.list_dir(adapter.get_effective_root("folder"), "")
self.assertEqual(total, 1)
self.assertEqual(items[0]["name"], "file.txt")
self.assertEqual(len(logins), 2)
self.assertEqual(logins[0]["passport"], "user")
self.assertEqual(lists, ["0", "0", "1"])
async def test_multipart_upload_order_and_finalize(self):
calls, uploaded = [], []
def handler(request):
if request.url.host == "upload.test":
uploaded.append(request.content)
return httpx.Response(200)
body = json.loads(request.content) if request.content else {}
calls.append((request.url.path, body))
if request.url.path.endswith("upload_request"):
self.assertEqual(body["etag"], "e80b5017098950fc58aad83c8c14978e")
result = {"Key": "key", "StorageNode": "node", "Bucket": "bucket", "UploadId": "upload", "FileId": 1}
elif request.url.path.endswith("s3_repare_upload_parts_batch"):
n = body["partNumberStart"]
self.assertEqual(body["partNumberEnd"], n + 1)
result = {"presignedUrls": {str(n): f"https://upload.test/{n}"}}
else:
result = {}
return httpx.Response(200, json={"code": 0, "data": result})
adapter = self.make_adapter(handler)
adapter._token = "token"
adapter.part_size = 2
result = await adapter.write_upload_file("0", "file", io.BytesIO(b"abcdef"))
self.assertEqual(result, {"size": 6})
self.assertEqual(uploaded, [b"ab", b"cd", b"ef"])
self.assertEqual(calls[-1][0], "/b/api/file/upload_complete/v2")
self.assertTrue(calls[-1][1]["isMultipart"])
async def test_failed_part_does_not_finalize(self):
finalized = []
def handler(request):
if request.url.host == "upload.test":
return httpx.Response(500)
if request.url.path.endswith("upload_request"):
data = {"Key": "key", "StorageNode": "node", "Bucket": "bucket", "UploadId": "upload", "FileId": 1}
elif request.url.path.endswith("auth"):
data = {"presignedUrls": {"1": "https://upload.test/1"}}
else:
finalized.append(request.url.path)
data = {}
return httpx.Response(200, json={"code": 0, "data": data})
adapter = self.make_adapter(handler)
adapter._token = "token"
with self.assertRaises(httpx.HTTPStatusError):
await adapter.write_upload_file("0", "file", io.BytesIO(b"a"))
self.assertFalse(finalized)
async def test_root_escape_is_rejected(self):
adapter = self.make_adapter(lambda request: self.fail("Unexpected request"))
with self.assertRaises(ValueError):
await adapter.stat_file("0", "../other")
async def test_invalid_move_is_rejected_before_network_calls(self):
adapter = self.make_adapter(lambda request: self.fail("Unexpected request"))
with self.assertRaises(ValueError):
await adapter.move("0", "dir", "dir/subdir")
await adapter.move("0", "dir", "dir")
+60
View File
@@ -0,0 +1,60 @@
import unittest
import httpx
from fastapi import HTTPException
from domain.adapters.providers._remote import RemoteFileAdapter, join_path, paginate
class DownloadAdapter(RemoteFileAdapter):
async def _download(self, root, rel):
return "https://download.test/file", {"Cookie": "session=test"}
class RemoteAdapterTests(unittest.IsolatedAsyncioTestCase):
async def test_range_response_and_connection_cleanup(self):
clients = []
class Body(httpx.AsyncByteStream):
closed = False
async def __aiter__(self):
yield b"ab"
async def aclose(self):
self.closed = True
body = Body()
def handler(request):
self.assertEqual(request.headers["Range"], "bytes=1-2")
return httpx.Response(206, headers={"Content-Range": "bytes 1-2/4", "Content-Length": "2"}, stream=body)
adapter = DownloadAdapter()
def client(**kwargs):
instance = httpx.AsyncClient(transport=httpx.MockTransport(handler), **kwargs)
clients.append(instance)
return instance
adapter._client = client
response = await adapter.stream_file("/", "file", "bytes=1-2")
self.assertFalse(clients[0].is_closed)
self.assertEqual(response.status_code, 206)
self.assertEqual(response.headers["content-range"], "bytes 1-2/4")
self.assertEqual(b"".join([part async for part in response.body_iterator]), b"ab")
self.assertTrue(clients[0].is_closed)
self.assertTrue(body.closed)
async def test_unsatisfiable_range(self):
adapter = DownloadAdapter()
adapter._client = lambda **kwargs: httpx.AsyncClient(transport=httpx.MockTransport(
lambda request: httpx.Response(416, headers={"Content-Range": "bytes */4"})), **kwargs)
with self.assertRaises(HTTPException) as caught:
await adapter.stream_file("/", "file", "bytes=8-")
self.assertEqual(caught.exception.status_code, 416)
def test_paths_and_directories_first_in_descending_order(self):
self.assertEqual(join_path("/root", "sub/file"), "/root/sub/file")
with self.assertRaises(ValueError):
join_path("/root", "../file")
items, total = paginate([{"name": "file", "is_dir": False}, {"name": "dir", "is_dir": True}], 1, 10, "name", "desc")
self.assertEqual(total, 2)
self.assertTrue(items[0]["is_dir"])
+1
View File
@@ -591,6 +591,7 @@
"adapter.type.telegram": "Telegram",
"adapter.type.quark": "Quark Drive",
"adapter.type.uc": "UC Drive",
"adapter.type.pan123": "123Pan (Account)",
"adapter.type.alist": "AList",
"adapter.type.openlist": "OpenList",
"Automation Tasks": "Automation Tasks",
+1
View File
@@ -582,6 +582,7 @@
"adapter.type.dropbox": "Dropbox",
"adapter.type.quark": "夸克网盘",
"adapter.type.uc": "UC 网盘",
"adapter.type.pan123": "123 网盘(账号)",
"adapter.type.alist": "AList",
"adapter.type.openlist": "OpenList",
"Automation Tasks": "自动化任务",