From 8ae45f9d6fe2db57b0a3f58d17e378073c3b12a5 Mon Sep 17 00:00:00 2001 From: Lee CQ Date: Sun, 18 Aug 2024 23:41:52 +0800 Subject: [PATCH] =?UTF-8?q?0.36.14:=20=20=20=20=201.=20Client=20API=20?= =?UTF-8?q?=E7=8E=B0=E5=9C=A8=E5=A2=9E=E5=8A=A0=E5=B1=9E=E6=80=A7=E8=8E=B7?= =?UTF-8?q?=E5=8F=96=E6=9C=8D=E5=8A=A1=E7=AB=AF=E7=89=88=E6=9C=AC:=20clien?= =?UTF-8?q?t.server=5Fversion=20=20=20=20=202.=20=E5=BC=82=E6=AD=A5?= =?UTF-8?q?=E5=AE=A2=E6=88=B7=E7=AB=AF=E6=B7=BB=E5=8A=A0=20client.login=5F?= =?UTF-8?q?username=20=E5=B1=9E=E6=80=A7=E3=80=82=20=20=20=20=203.=20?= =?UTF-8?q?=E7=A7=BB=E9=99=A4=E4=B8=BA=E4=BF=9D=E6=8C=81=E5=85=BC=E5=AE=B9?= =?UTF-8?q?=E7=9A=84=20alist=5Fsdk.path=5Flib=5Fold.py=20=E6=96=87?= =?UTF-8?q?=E4=BB=B6=20=20=20=20=204.=20=E6=B7=BB=E5=8A=A0=E6=B5=8B?= =?UTF-8?q?=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- alist_sdk/alistpath.py | 2 + alist_sdk/async_client.py | 18 ++- alist_sdk/client.py | 14 +- alist_sdk/models.py | 2 +- alist_sdk/path_lib.py | 11 +- alist_sdk/path_lib_old.py | 329 -------------------------------------- alist_sdk/version.py | 10 +- tests/test_client.py | 12 ++ tests/test_path_lib.py | 11 +- 9 files changed, 69 insertions(+), 340 deletions(-) delete mode 100644 alist_sdk/path_lib_old.py diff --git a/alist_sdk/alistpath.py b/alist_sdk/alistpath.py index 3d8fdfd..af27b88 100644 --- a/alist_sdk/alistpath.py +++ b/alist_sdk/alistpath.py @@ -14,6 +14,8 @@ __all__ = [*posixpath.__all__, "splitroot"] +_ = sep # 解决IDE报错：未使用的导入 + def splitroot(p): """Split a pathname into drive, root and tail. On Posix, drive is always diff --git a/alist_sdk/async_client.py b/alist_sdk/async_client.py index 5ddb458..cbfb05f 100644 --- a/alist_sdk/async_client.py +++ b/alist_sdk/async_client.py @@ -2,6 +2,7 @@ import logging import time import urllib.parse +from functools import cached_property from pathlib import Path, PurePosixPath from httpx import AsyncClient as HttpClient, Response @@ -55,9 +56,7 @@ async def verify_login_status(self) -> bool: """验证登陆状态,""" me = (await self.get("/api/me")).json() if me.get("code") != 200: - logger.error( - "异步客户端登陆失败[%d], %s", me.get("code"), me.get("message") - ) + logger.error("异步客户端登陆失败[%d], %s", me.get("code"), me.get("message")) return False username = me["data"].get("username") @@ -132,6 +131,10 @@ async def login(self, username, password, has_opt=False) -> bool: logger.warning("登陆失败[%d]：%s", res.status_code, res.text) return False + @cached_property + async def login_username(self): + return (await self.me()).data.username + # ================ FS 相关方法 ================= @@ -483,6 +486,15 @@ async def admin_setting_list(self, group: int = None): query = {"group": group} if group else {} return locals(), await self.get("/api/admin/setting/list", params=query) + @cached_property + async def service_version(self) -> tuple: + """返回服务器版本元组 (int, int, int)""" + settings: list[Setting] = (await self.admin_setting_list(group=1)).data + for s in settings: + if s.key == "version": + return tuple(map(int, s.value.strip("v").split(".", 2))) + raise ValueError("无法获取服务端版本") + class AsyncClient( _AsyncFS, diff --git a/alist_sdk/client.py b/alist_sdk/client.py index 09f7693..4f1804d 100644 --- a/alist_sdk/client.py +++ b/alist_sdk/client.py @@ -224,6 +224,9 @@ def get_item_info(self, path: str | PurePosixPath, password=None): "/api/fs/get", json={"path": path, "password": password} ) + def fs_get(self, path: str | PurePosixPath, password=None): + return self.get_item_info(path, password) + @verify() def search( self, @@ -500,6 +503,15 @@ def admin_setting_list(self, group: int = None): query = {"group": group} if group else {} return locals(), self.get("/api/admin/setting/list", params=query) + @cached_property + def service_version(self) -> tuple: + """返回服务器版本元组 (int, int, int)""" + settings: list[Setting] = self.admin_setting_list(group=1).data + for s in settings: + if s.key == "version": + return tuple(map(int, s.value.strip("v").split(".", 2))) + raise ValueError("无法获取服务端版本") + class Client( _SyncFs, @@ -530,7 +542,7 @@ def dict_files_items( logger.debug("缓存命中[times: %d]: %s", self._succeed_cache, path) return self._cached_path_list[path] - if len(self._cached_path_list) >= 10000: + if len(self._cached_path_list) >= 1000: self._cached_path_list.pop(0) # Python 3中的字典是按照插入顺序保存的 logger.debug("缓存未命中: %s", path) diff --git a/alist_sdk/models.py b/alist_sdk/models.py index f9b0666..9dec96a 100644 --- a/alist_sdk/models.py +++ b/alist_sdk/models.py @@ -201,7 +201,7 @@ class Setting(_BaseModel): """/api/admin/setting/list .data.[]""" key: str - value: Any + value: str help: str # 帮助信息 type: str # TODO 类型指导 options: str diff --git a/alist_sdk/path_lib.py b/alist_sdk/path_lib.py index 2fa7b83..5fe903c 100644 --- a/alist_sdk/path_lib.py +++ b/alist_sdk/path_lib.py @@ -13,7 +13,9 @@ from pydantic.json_schema import JsonSchemaValue from pydantic_core import core_schema -from alist_sdk import alistpath, Item, AlistError, RawItem +from alist_sdk import alistpath +from alist_sdk.models import Item, RawItem +from alist_sdk.err import AlistError from alist_sdk.py312_pathlib import PurePosixPath from alist_sdk.client import Client @@ -186,7 +188,12 @@ def raw_stat(self, retry=1, timeout=0.1) -> RawItem: def stat(self) -> Item | RawItem: def f_stat() -> Item | RawItem: - _r = self.client.dict_files_items(self.parent.as_posix()).get(self.name) + _r = ( + self.client.get_item_info(self.as_posix()).data + if self.as_posix() == "/" + else self.client.dict_files_items(self.parent.as_posix()).get(self.name) + ) + if not _r: raise FileNotFoundError(f"文件不存在: {self.as_posix()} ") self.set_stat(_r) diff --git a/alist_sdk/path_lib_old.py b/alist_sdk/path_lib_old.py deleted file mode 100644 index 4a125f1..0000000 --- a/alist_sdk/path_lib_old.py +++ /dev/null @@ -1,329 +0,0 @@ -#!/usr/bin/python3 -# -*- coding: utf-8 -*- -""" -@File Name : path_lib_old.py -@Author : LeeCQ -@Date-Time : 2024/2/25 12:58 - -path_lib的存档文件。 - -Path Lib 类实现 - -像使用Path一样的易于使用Alist中的文件 -""" -import fnmatch -import os -import posixpath -import re -import sys -from pathlib import PurePosixPath, PurePath -from functools import lru_cache, cached_property -from urllib.parse import quote_from_bytes as urlquote_from_bytes -from typing import Iterator - -from pydantic import BaseModel -from httpx import URL - -from alist_sdk import AlistError, RawItem -from alist_sdk.client import Client - - -class AlistServer(BaseModel): - server: str - token: str - kwargs: dict = {} - - -ALIST_SERVER_INFO: dict[str, AlistServer] = dict() - - -def login_server( - server: str, - token=None, - username=None, - password=None, - has_opt=False, - **kwargs, -): - """""" - if token is None: - client = Client( - server, - username=username, - password=password, - has_opt=has_opt, - **kwargs, - ) - token = client.headers.get("Authorization") - ALIST_SERVER_INFO[server] = AlistServer( - server=server, - token=token, - kwargs=kwargs, - ) - - -# noinspection PyMethodMayBeStatic -class _AlistFlavour: - sep = "/" - altsep = "" - has_drv = True - pathmod = posixpath - - is_supported = True - - def __init__(self): - self.join = self.sep.join - - def parse_parts(self, parts): - parsed = [] - sep = self.sep - altsep = self.altsep - drv = root = "" - - it = reversed(parts) - for part in it: - if not part: - continue - if altsep: - part = part.replace(altsep, sep) - drv, root, rel = self.splitroot(part) - if sep in rel: - for x in reversed(rel.split(sep)): - if x and x != ".": - parsed.append(sys.intern(x)) - else: - if rel and rel != ".": - parsed.append(sys.intern(rel)) - if drv or root: - if not drv: - # If no drive is present, try to find one in the previous - # parts. This makes the result of parsing e.g. - # ("C:", "/", "a") reasonably intuitive. - # noinspection PyAssignmentToLoopOrWithParameter - for part in it: - if not part: - continue - if altsep: - part = part.replace(altsep, sep) - drv = self.splitroot(part)[0] - if drv: - break - break - if drv or root: - parsed.append(drv + root) - parsed.reverse() - return drv, root, parsed - - def join_parsed_parts(self, drv, root, parts, drv2, root2, parts2): - """ - Join the two paths represented by the respective - (drive, root, parts) tuples. Return a new (drive, root, parts) tuple. - """ - if root2: - if not drv2 and drv: - return drv, root2, [drv + root2] + parts2[1:] - elif drv2: - if drv2 == drv or self.casefold(drv2) == self.casefold(drv): - # Same drive => second path is relative to the first - return drv, root, parts + parts2[1:] - else: - # Second path is non-anchored (common case) - return drv, root, parts + parts2 - return drv2, root2, parts2 - - def splitroot(self, part, sep=sep): - if part and part[0] == sep: - stripped_part = part.lstrip(sep) - # According to POSIX path resolution: - # http://pubs.opengroup.org/onlinepubs/009695399/basedefs/xbd_chap04.html#tag_04_11 - # "A pathname that begins with two successive slashes may be - # interpreted in an implementation-defined manner, although more - # than two leading slashes shall be treated as a single slash". - if len(part) - len(stripped_part) == 2: - return "", sep * 2, stripped_part - else: - return "", sep, stripped_part - else: - return "", "", part - - def casefold(self, s): - return s - - def casefold_parts(self, parts): - return parts - - def compile_pattern(self, pattern): - return re.compile(fnmatch.translate(pattern)).fullmatch - - def make_uri(self, path): - # We represent the path using the local filesystem encoding, - # for portability to other applications. - bpath = bytes(path) - return "file://" + urlquote_from_bytes(bpath) - - -# noinspection PyProtectedMember,PyUnresolvedReferences -class PureAlistPath(PurePosixPath): - _flavour = _AlistFlavour() - - @classmethod - def _parse_args(cls, args): - # This is useful when you don't want to create an instance, just - # canonicalize some constructor arguments. - parts = [] - for a in args: - if isinstance(a, PurePath): - parts += a._parts - else: - a = os.fspath(a) - if isinstance(a, str): - # Force-cast str subclasses to str (issue #21127) - parts.append(str(a)) - else: - raise TypeError( - "argument should be a str object or an os.PathLike " - "object returning str, not %r" % type(a) - ) - return cls._flavour.parse_parts(parts) - - @classmethod - def _from_parts(cls, args): - # We need to call _parse_args on the instance, to get the - # right flavour. - args = list(args) - self = object.__new__(cls) - if isinstance(args[0], str) and args[0].startswith("http"): - _u = URL(args[0]) - server = ( - f"{_u.scheme}://{_u.host}:{_u.port}".replace(":80", "") - .replace(":443", "") - .replace(":None", "") - ) - args[0] = _u.path - elif isinstance(args[0], cls): - server = args[0].server - else: - server = "" - - drv, root, parts = self._parse_args(args) - self._drv = drv or server - self._root = root - self._parts = parts - self.server = server - return self - - @classmethod - def _from_parsed_parts(cls, drv, root, parts): - self = object.__new__(cls) - self._drv = drv - self._root = root - self._parts = parts - return self - - def _make_child(self, args): - drv, root, parts = self._parse_args(args) - drv, root, parts = self._flavour.join_parsed_parts( - self._drv, self._root, self._parts, drv, root, parts - ) - return self._from_parsed_parts(drv, root, parts) - - def __new__(cls, *args): - return cls._from_parts(args) - - def as_posix(self) -> str: - return str(self).replace(self.drive, "") - - def as_uri(self): - if not self.is_absolute(): - raise ValueError("relative path can't be expressed as a file URI") - return str(self) - - def relative_to(self, *other): - raise NotImplementedError("AlistPath不支持relative_to") - - -class AlistPath(PureAlistPath): - """""" - - @cached_property - def _client(self) -> Client: - if self.drive == "": - raise AlistError("当前对象没有设置server") - - try: - _server = ALIST_SERVER_INFO[self.drive] - return Client( - _server.server, - token=_server.token, - **_server.kwargs, - ) - except KeyError: - raise AlistError(f"当前服务器[{self.drive}]尚未登陆") - - # def is_absolute(self) -> bool: - def as_download_uri(self): - if not self.is_absolute(): - raise ValueError("relative path can't be expressed as a file URI") - if self.is_dir(): - raise IsADirectoryError() - return self.drive + "/d" + self.as_posix() + "?sign=" + self.stat().sign - - @lru_cache() - def stat(self) -> RawItem: - _raw = self._client.get_item_info(self.as_posix()) - if _raw.code == 200: - data = _raw.data - return data - if _raw.code == 500 and ( - "object not found" in _raw.message or "storage not found" in _raw.message - ): - raise FileNotFoundError(_raw.message) - raise AlistError(_raw.message) - - def is_dir(self): - """""" - return self.stat().is_dir - - def is_file(self): - """""" - return not self.stat().is_dir - - def is_link(self): - raise NotImplementedError("AlistPath不支持连接.") - - def exists(self): - """""" - try: - return bool(self.stat()) - except FileNotFoundError: - return False - - def iterdir(self) -> Iterator["AlistPath"]: - """""" - if not self.is_dir(): - raise - - for item in ( - self._client.list_files(self.as_posix(), refresh=True).data.content or [] - ): - yield self.joinpath(item.name) - - def read_text(self): - """""" - return self._client.get(self.as_download_uri(), follow_redirects=True).text - - def read_bytes(self): - """""" - return self._client.get(self.as_download_uri(), follow_redirects=True).content - - def write_text(self, data: str, as_task=False): - """""" - return self.write_bytes(data.encode(), as_task=as_task) - - def write_bytes(self, data: bytes, as_task=False): - """""" - - _res = self._client.upload_file_put(data, self.as_posix(), as_task=as_task) - if _res.code == 200: - return self.stat() - return None diff --git a/alist_sdk/version.py b/alist_sdk/version.py index 7f4d4ab..677194d 100644 --- a/alist_sdk/version.py +++ b/alist_sdk/version.py @@ -42,7 +42,7 @@ 0.30.11: 1. BUGFIX: 从其他站点导入配置时，数据模型错误。 2. BUGFIX: http://localhost:5244 无法正常识别 - 3. UPDATE: 更新AplistPath.__repl__方法. + 3. UPDATE: 更新AlistPath.__repl__方法. 4. UPDATE: 更新AlistPath.添加新的方法 set_stat，可以自定义设置stat属性，加快速度。 5. UPDATE: 更新AlistPath.iterdir，在迭代时添加stat数据，加快速度。 6. UPDATE: 更新AlistPath.stat，不再使用/api/fs/get 接口 @@ -62,8 +62,14 @@ 2. 现在可以使用AlistPath(path, username="", password="", token="")的方式快速登录。 3. 登录失败现在抛出异常。 4. #3 Bugfix 为models中的全部可选字段添加默认值。 + +0.36.14: + 1. Client API 现在增加属性获取服务端版本: client.server_version + 2. 异步客户端添加 client.login_username 属性。 + 3. 移除为保持兼容的 alist_sdk.path_lib_old.py 文件 + 4. 添加测试 """ -__version__ = "0.36.13a4" +__version__ = "0.36.14a4" ALIST_VERSION = "v3.36.0" diff --git a/tests/test_client.py b/tests/test_client.py index df209a9..cfa6517 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -388,6 +388,12 @@ def test_admin_meta_list(self): res = self.run(self.client.admin_meta_list) assert res.code == 200 + def test_server_version(self): + assert self.client.service_version[0] == 3 + + def test_login_user(self): + assert self.client.login_username == "admin" + class TestAsyncClient(TestSyncClient): @property @@ -400,3 +406,9 @@ def run(self, func, *args, **kwargs): def test_async_client(self): assert isinstance(self.client, AsyncClient) + + def test_server_version(self): + assert asyncio.run(self.client.service_version)[0] == 3 + + def test_login_user(self): + assert asyncio.run(self.client.login_username) == "admin" diff --git a/tests/test_path_lib.py b/tests/test_path_lib.py index dfa1eaf..00c0c62 100644 --- a/tests/test_path_lib.py +++ b/tests/test_path_lib.py @@ -1,4 +1,6 @@ # 测试 path_lib.py +import time + import pytest from alist_sdk.path_lib import PureAlistPath, AlistPath, login_server @@ -54,6 +56,10 @@ def setup_class(self): def setup_method(self): self.client._cached_path_list = {} + def test_login(self): + _c = AlistPath("http://localhost:5245/", username="admin", password="123456") + assert _c.is_dir() + def test_read_text(self): DATA_DIR.joinpath("test.txt").write_text("123") path = AlistPath("http://localhost:5245/local/test.txt") @@ -74,9 +80,10 @@ def test_write_bytes(self): assert DATA_DIR.joinpath("test_write_bytes.txt").read_bytes() == b"123" def test_mkdir(self): - path = AlistPath("http://localhost:5245/local/test_mkdir") + dir_name = f"test_mkdir_{int(time.time())}" + path = AlistPath(f"http://localhost:5245/local/{dir_name}") path.mkdir() - assert DATA_DIR.joinpath("test_mkdir").is_dir() + assert DATA_DIR.joinpath(dir_name).is_dir() def test_touch(self): path = AlistPath("http://localhost:5245/local/test_touch.txt")