0.32.12:

    1. 添加Client.dict_files_items方法，以获取目录下的文件列表。
    2. 解决AListPath使用URL编码问题，现在可以正常使用中文路径。
    3. 修复AlistPath.stat() & AlistPath.raw_stat() 的缓存问题。

alist_sdk/alistpath.py

@@ -8,6 +8,7 @@
 """
 import os
 import posixpath
+from urllib.parse import unquote
 
 from posixpath import *
 
@@ -37,7 +38,9 @@ def splitroot(p):
         # Absolute path, e.g.: 'http://server/path/to/file'
         if p.count("/") < 3:
             p += "/"
-        return "/".join(p.split("/", 3)[:3]), "/", p.split("/", 3)[-1]
+        p = unquote(p)
+        sps = p.split("/", 3)
+        return "/".join(sps[:3]), "/", (sps[-1])
 
     elif p[:1] != _sep:
         # Relative path, e.g.: 'foo'

alist_sdk/async_client.py

@@ -55,7 +55,9 @@ 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")
@@ -328,11 +330,7 @@ async def copy(
             json={
                 "src_dir": str(src_dir),
                 "dst_dir": str(dst_dir),
-                "names": [
-                    files,
-                ]
-                if isinstance(files, str)
-                else files,
+                "names": [files] if isinstance(files, str) else files,
             },
         )
 
@@ -494,4 +492,33 @@ class AsyncClient(
     _AsyncAdminStorage,
     _AsyncAdminUser,
 ):
-    pass
+    _cached_path_list = dict()
+    _succeed_cache = 0
+
+    async def dict_files_items(
+        self,
+        path: str | PurePosixPath,
+        password="",
+        refresh=False,
+    ) -> dict[str, Item]:
+        """列出文件目录"""
+        path = str(path)
+        if refresh:
+            self._cached_path_list.pop(path, None)
+
+        if path in self._cached_path_list:
+            self._succeed_cache += 1
+            logger.debug("缓存命中[times: %d]: %s", self._succeed_cache, path)
+            return self._cached_path_list[path]
+
+        if len(self._cached_path_list) >= 10000:
+            self._cached_path_list.pop(0)  # Python 3中的字典是按照插入顺序保存的
+
+        logger.debug("缓存未命中: %s", path)
+        _res = await self.list_files(path, password, refresh=True)
+        if _res.code == 200:
+            _ = {d.name: d for d in _res.data.content or []}
+            if _:  # 有数据才缓存
+                self._cached_path_list[path] = _
+            return _
+        return {}

alist_sdk/client.py

@@ -509,22 +509,34 @@ class Client(
     _SyncAdminTask,
 ):
     _cached_path_list = dict()
+    _succeed_cache = 0
 
-    def dict_files_item(self, path: str | PurePosixPath, password="", refresh=False):
+    def dict_files_items(
+        self,
+        path: str | PurePosixPath,
+        password="",
+        refresh=False,
+        cache_empty=False,
+    ) -> dict[str, Item]:
         """列出文件目录"""
-
-        def _dict_files(_path: str | PurePosixPath, _password="") -> dict[str, Item]:
-            _res = self.list_files(_path, _password, refresh=True)
-            if _res.code == 200:
-                _ = {d.name: d for d in _res.data.content or []}
-                self._cached_path_list[_path] = _
-                return _
-            return {}
-
         path = str(path)
         if refresh:
             self._cached_path_list.pop(path, None)
-        if len(self._cached_path_list) >= 100:
+
+        if path in self._cached_path_list:
+            if not self._cached_path_list[path] or cache_empty:
+                self._succeed_cache += 1
+                logger.debug("缓存命中[times: %d]: %s", self._succeed_cache, path)
+                return self._cached_path_list[path]
+
+        if len(self._cached_path_list) >= 10000:
             self._cached_path_list.pop(0)  # Python 3中的字典是按照插入顺序保存的
 
-        return self._cached_path_list.get(path, _dict_files(path, password))
+        logger.debug("缓存未命中: %s", path)
+        _res = self.list_files(path, password, refresh=True)
+        if _res.code == 200:
+            _ = {d.name: d for d in _res.data.content or []}
+            if _ or cache_empty:  # 有数据才缓存
+                self._cached_path_list[path] = _
+            return _
+        return {}

alist_sdk/path_lib.py

@@ -4,7 +4,7 @@
 """
 
 import time
-from functools import lru_cache, cached_property
+from functools import cached_property
 from pathlib import Path
 from typing import Iterator, Annotated, Any
 
@@ -67,7 +67,6 @@ def is_absolute(self):
         """True if the path is absolute (has both a root and, if applicable,
         a drive)."""
         if self._flavour is alistpath:
-            # ntpath.isabs() is defective - see GH-44626.
             return bool(self.drive and self.root)
 
         else:
@@ -147,37 +146,39 @@ def get_download_uri(self) -> str:
     def as_download_uri(self):
         return self.get_download_uri()
 
-    @lru_cache()
-    def _raw_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 raw_stat(self, force=False, retry=1, timeout=0.1) -> RawItem:
-        if force:
-            self._raw_stat.cache_clear()
+    def raw_stat(self, retry=1, timeout=0.1) -> RawItem:
+
         try:
-            return self._raw_stat()
+            _raw = self.client.get_item_info(self.as_posix())
+            if _raw.code == 200:
+                data = _raw.data
+                self.set_stat(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)
         except FileNotFoundError as _e:
             if retry > 0:
                 time.sleep(timeout)
-                return self.raw_stat(force, retry - 1)
+                return self.raw_stat(retry - 1)
             raise _e
 
-    def stat(self) -> Item:
-        def _stat() -> Item:
-            _r = self.client.dict_files_item(self.parent.as_posix()).get(self.name)
+    def stat(self) -> Item | RawItem:
+        def f_stat() -> Item | RawItem:
+            _r = 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)
             return _r
 
-        return getattr(self, "_stat", _stat())
+        _stat = getattr(self, "_stat", None)
+        if isinstance(_stat, Item | RawItem):
+            return _stat
+
+        return f_stat()
 
     def set_stat(self, value: RawItem | Item):
         # noinspection PyAttributeOutsideInit
@@ -187,7 +188,7 @@ def re_stat(self, retry=2, timeout=1) -> Item:
         if hasattr(self, "_stat"):
             delattr(self, "_stat")
         self.client.list_files(self.parent.as_posix(), per_page=1, refresh=True)
-        return self.raw_stat(force=True, retry=retry, timeout=timeout)
+        return self.raw_stat(retry=retry, timeout=timeout)
 
     def is_dir(self):
         """"""
@@ -212,9 +213,9 @@ 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 []
-        ):
+        for item in self.client.dict_files_items(
+            self.as_posix(), refresh=True
+        ).values():
             _ = self.joinpath(item.name)
             _.set_stat(item)
             yield _
@@ -244,7 +245,7 @@ def write_bytes(self, data: bytes | Path, 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 self.re_stat()
         raise AlistError()
 
     def mkdir(self, parents=False, exist_ok=False):

alist_sdk/version.py

@@ -48,8 +48,13 @@
     6. UPDATE: 更新AlistPath.stat，不再使用/api/fs/get 接口
     7. UPDATE: 限制alist_sdk.Client 和 alist_sdk.AsyncClient在多线程和协程中的并发量，默认30.
 
+
+0.32.12:
+    1. 添加Client.dict_files_items方法，以获取目录下的文件列表。
+    2. 解决AListPath使用URL编码问题，现在可以正常使用中文路径。
+    3. 修复AlistPath.stat() & AlistPath.raw_stat() 的缓存问题。
 """
 
-__version__ = "0.30.11"
+__version__ = "0.32.12a3"
 
 ALIST_VERSION = "v3.32.0"

tests/init_alist.sh

@@ -15,5 +15,5 @@ fi
 
 rm -rf alist/data/ alist/test_dir/ alist/test_dir_dst/
 ./alist admin set 123456
-sed -i 's/: 5244/: 5245/' data/config.json
+sed -i '' 's/: 5244/: 5245/' data/config.json
 ./alist restart

tests/test_client.py

@@ -95,8 +95,10 @@ def setup_method(self) -> None:
     def run(self, func, *args, **kwargs):
         print("run", func.__name__, end=": ")
         res = func(*args, **kwargs)
-        res: models.Resp
-        print("RESP: ", res.model_dump_json())
+        if isinstance(res, models.Resp):
+            print("RESP: ", res.model_dump_json())
+        else:
+            print(res)
         return res
 
     def test_login(self):
@@ -159,6 +161,16 @@ def test_list_dir(self):
         assert isinstance(res.data, models.ListItem)
         assert isinstance(res.data.content[0], models.Item)
 
+    def test_dict_list_dir(self):
+        DATA_DIR.joinpath("test_list_dir_dir").mkdir()
+        DATA_DIR.joinpath("test_list_dir_file").write_text("test")
+        res = self.run(
+            self.client.dict_files_items,
+            "/local",
+        )
+        assert isinstance(res, dict)
+        assert len(res) == 2
+
     def test_list_dir_null(self):
         DATA_DIR.joinpath("test_list_dir_null").mkdir()
         res = self.run(

tests/test_path_lib.py

@@ -51,7 +51,11 @@ def setup_class(self):
         )
         DATA_DIR.joinpath("test.txt").write_text("123")
 
+    def setup_method(self):
+        self.client._cached_path_list = {}
+
     def test_read_text(self):
+        DATA_DIR.joinpath("test.txt").write_text("123")
         path = AlistPath("http://localhost:5245/local/test.txt")
         assert path.read_text() == "123"
 
@@ -69,7 +73,6 @@ def test_write_bytes(self):
         path.write_bytes(b"123")
         assert DATA_DIR.joinpath("test_write_bytes.txt").read_bytes() == b"123"
 
-
     def test_mkdir(self):
         path = AlistPath("http://localhost:5245/local/test_mkdir")
         path.mkdir()