feat(storages): 新增 AList 存储类型

app/modules/filemanager/storages/alist.py

@@ -47,7 +47,7 @@ class Alist(StorageBase, metaclass=WeakSingleton):
         """
         初始化
         """
-        self.__generate_token.cache_clear()  # noqa
+        self.__login_token.cache_clear()  # noqa
 
     def _delay_get_item(
         self, path: Path, /, refresh: bool = False
@@ -117,22 +117,32 @@ class Alist(StorageBase, metaclass=WeakSingleton):
         """
         return self.__generate_token()
 
-    @cached(maxsize=1, ttl=60 * 60 * 24 * 2 - 60 * 5, skip_empty=True)
     def __generate_token(self) -> str:
         """
         如果设置永久令牌则返回永久令牌，否则使用账号密码生成一个临时 token
-        缓存2天，提前5分钟更新
         """
         conf = self.get_conf()
         token = conf.get("token")
         if token:
             return str(token)
+        return self.__login_token(
+            self.__get_base_url, conf.get("username"), conf.get("password")
+        )
+
+    @cached(maxsize=8, ttl=60 * 60 * 24 * 2 - 60 * 5, skip_empty=True)
+    def __login_token(
+        self, base_url: str, username: Optional[str], password: Optional[str]
+    ) -> str:
+        """
+        使用账号密码生成一个临时 token
+        缓存2天，提前5分钟更新
+        """
         resp = RequestUtils(headers={"Content-Type": "application/json"}).post_res(
-            self.__get_api_url("/api/auth/login"),
+            UrlUtils.adapt_request_url(base_url, "/api/auth/login"),
             data=json.dumps(
                 {
-                    "username": conf.get("username"),
-                    "password": conf.get("password"),
+                    "username": username,
+                    "password": password,
                 }
             ),
         )

app/modules/filemanager/storages/alistgo.py

@@ -0,0 +1,12 @@
+from app.modules.filemanager.storages.alist import Alist
+from app.schemas.types import StorageSchema
+
+
+class AlistGo(Alist):
+    """
+    AList相关操作
+
+    API 文档：https://docs.alistgo.com/
+    """
+
+    schema = StorageSchema.AlistGo

app/schemas/types.py

@@ -426,6 +426,7 @@ class StorageSchema(Enum):
     U115 = "u115"
     Rclone = "rclone"
     Alist = "alist"
+    AlistGo = "alistgo"
     SMB = "smb"
 
 

tests/test_alistgo_storage.py

@@ -0,0 +1,63 @@
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from app.modules.filemanager.storages import alist as alist_module
+from app.modules.filemanager.storages.alist import Alist
+from app.modules.filemanager.storages.alistgo import AlistGo
+from app.schemas.types import StorageSchema
+
+
+class _FakeResponse:
+    def __init__(self, payload: dict, status_code: int = 200):
+        self._payload = payload
+        self.status_code = status_code
+
+    def json(self):
+        return self._payload
+
+
+@pytest.fixture
+def clear_token_cache():
+    Alist().init_storage()
+    yield
+    Alist().init_storage()
+
+
+def test_alistgo_schema_registered():
+    assert AlistGo.schema == StorageSchema.AlistGo
+    assert StorageSchema.AlistGo.value == "alistgo"
+
+
+def test_alistgo_singleton_isolated_from_alist():
+    alist = Alist()
+    alistgo = AlistGo()
+    assert alistgo is not alist
+    assert isinstance(alistgo, Alist)
+
+
+def test_alistgo_token_isolated_from_alist(clear_token_cache):
+    def _conf(storage):
+        return {
+            "url": f"http://{storage.schema.value}.test",
+            "username": "user",
+            "password": "pass",
+        }
+
+    responses = [
+        _FakeResponse({"code": 200, "message": "success", "data": {"token": "token-alist"}}),
+        _FakeResponse({"code": 200, "message": "success", "data": {"token": "token-alistgo"}}),
+    ]
+    request_utils = MagicMock()
+    request_utils.post_res.side_effect = responses
+
+    alist = Alist()
+    alistgo = AlistGo()
+    with patch.object(Alist, "get_conf", _conf):
+        with patch.object(alist_module, "RequestUtils", return_value=request_utils):
+            assert alist._Alist__generate_token() == "token-alist"  # noqa
+            assert alistgo._Alist__generate_token() == "token-alistgo"  # noqa
+            assert alist._Alist__generate_token() == "token-alist"  # noqa
+            assert alistgo._Alist__generate_token() == "token-alistgo"  # noqa
+
+    assert request_utils.post_res.call_count == 2