✨ 优化 Alist 模块，以 storage_id 获取存储路径参数

- 增加 storage_id_to_path 方法以处理存储 ID
- 更新构造函数以支持存储路径和根目录的缓存
- 修改 run 方法以使用新的存储路径逻辑
- 改进 get_info 方法以提供更清晰的输出信息
- 添加 get_storage_info 方法以获取存储信息
- 更新 refresh 方法以提供更详细的错误信息
- 增加 get_root_folder_full_path 方法以获取根文件夹完整路径

media_servers/alist.py

@@ -1,4 +1,6 @@
 import os
+import re
+import json
 import requests
 
 
@@ -7,27 +9,38 @@ class Alist:
     default_config = {
         "url": "",  # Alist服务器URL
         "token": "",  # Alist服务器Token
-        "quark_root_path": "/quark",  # 夸克根目录在Alist中的挂载路径
+        "storage_id": "",  # Alist 服务器夸克存储 ID
     }
     is_active = False
+    # 缓存参数
+    storage_mount_path = None
+    quark_root_dir = None
 
     def __init__(self, **kwargs):
         if kwargs:
-            for key, value in self.default_config.items():
+            for key, _ in self.default_config.items():
                 if key in kwargs:
                     setattr(self, key, kwargs[key])
                 else:
                     print(f"{self.__class__.__name__} 模块缺少必要参数: {key}")
             if self.url and self.token:
                 if self.get_info():
-                    self.is_active = True
+                    success, result = self.storage_id_to_path(self.storage_id)
+                    if success:
+                        self.storage_mount_path, self.quark_root_dir = result
+                        self.is_active = True
 
     def run(self, task):
-        if task.get("savepath"):
+        if task.get("savepath") and task.get("savepath").startswith(
-            full_path = os.path.normpath(
+            self.quark_root_dir
-                os.path.join(self.quark_root_path, task["savepath"].lstrip("/"))
+        ):
+            alist_path = os.path.normpath(
+                os.path.join(
+                    self.storage_mount_path,
+                    task["savepath"].replace(self.quark_root_dir, "", 1).lstrip("/"),
+                )
             ).replace("\\", "/")
-            self.refresh(full_path)
+            self.refresh(alist_path)
 
     def get_info(self):
         url = f"{self.url}/api/admin/setting/list"
@@ -39,15 +52,52 @@ class Alist:
             response = response.json()
             if response.get("code") == 200:
                 print(
-                    f"Alist: {response.get('data',[])[1].get('value','')} {response.get('data',[])[0].get('value','')}"
+                    f"Alist刷新: {response.get('data',[])[1].get('value','')} {response.get('data',[])[0].get('value','')}"
                 )
                 return True
             else:
-                print(f"Alist: 连接失败❌ {response.get('message')}")
+                print(f"Alist刷新: 连接失败❌ {response.get('message')}")
         except requests.exceptions.RequestException as e:
             print(f"获取Alist信息出错: {e}")
         return False
 
+    def storage_id_to_path(self, storage_id):
+        # 1. 检查是否符合 /aaa:/bbb 格式
+        match = re.match(r"^(\/[^:]*):(\/[^:]*)$", storage_id)
+        if match:
+            return True, (match.group(1), match.group(2))
+        # 2. 调用 Alist API 获取存储信息
+        storage_info = self.get_storage_info(storage_id)
+        if storage_info:
+            if storage_info["driver"] == "Quark":
+                addition = json.loads(storage_info["addition"])
+                # 存储挂载路径
+                storage_mount_path = storage_info["mount_path"]
+                # 夸克根文件夹
+                quark_root_dir = self.get_root_folder_full_path(
+                    addition["cookie"], addition["root_folder_id"]
+                )
+                if storage_mount_path and quark_root_dir:
+                    return True, (storage_mount_path, quark_root_dir)
+            else:
+                print(f"Alist刷新: 不支持[{storage_info['driver']}]驱动 ❌")
+
+    def get_storage_info(self, storage_id):
+        url = f"{self.url}/api/admin/storage/get"
+        headers = {"Authorization": self.token}
+        querystring = {"id": storage_id}
+        try:
+            response = requests.request("GET", url, headers=headers, params=querystring)
+            response.raise_for_status()
+            data = response.json()
+            if data.get("code") == 200:
+                return data.get("data", [])
+            else:
+                print(f"Alist刷新: 存储{storage_id}连接失败❌ {data.get('message')}")
+        except requests.exceptions.RequestException as e:
+            print(f"Alist刷新: 获取Alist存储出错 {e}")
+        return False
+
     def refresh(self, path, force_refresh=True):
         url = f"{self.url}/api/fs/list"
         headers = {"Authorization": self.token}
@@ -61,22 +111,55 @@ class Alist:
         try:
             response = requests.request("POST", url, headers=headers, json=payload)
             response.raise_for_status()
-            response = response.json()
+            data = response.json()
-            if response.get("code") == 200:
+            if data.get("code") == 200:
-                print(f"📁 刷新Alist目录：[{path}] 成功✅")
+                print(f"📁 Alist刷新：目录[{path}] 成功✅")
-                return response.get("data")
+                return data.get("data")
-            elif "object not found" in response.get("message", ""):
+            elif "object not found" in data.get("message", ""):
                 # 如果是根目录就不再往上查找
                 if path == "/" or path == self.quark_root_path:
-                    print(f"📁 刷新Alist目录：根目录不存在，请检查 Alist 配置")
+                    print(f"📁 Alist刷新：根目录不存在，请检查 Alist 配置")
                     return False
                 # 获取父目录
                 parent_path = os.path.dirname(path)
-                print(f"📁 刷新Alist目录：[{path}] 不存在，转父目录 [{parent_path}]")
+                print(f"📁 Alist刷新：[{path}] 不存在，转父目录 [{parent_path}]")
                 # 递归刷新父目录
                 return self.refresh(parent_path)
             else:
-                print(f"📁 刷新Alist目录：失败❌ {response.get('message')}")
+                print(f"📁 Alist刷新：失败❌ {data.get('message')}")
         except requests.exceptions.RequestException as e:
-            print(f"刷新Alist目录出错: {e}")
+            print(f"Alist刷新目录出错: {e}")
+        return False
+
+    def get_root_folder_full_path(self, cookie, pdir_fid):
+        if pdir_fid == "0":
+            return "/"
+        url = "https://drive-h.quark.cn/1/clouddrive/file/sort"
+        headers = {
+            "cookie": cookie,
+            "content-type": "application/json",
+        }
+        querystring = {
+            "pr": "ucpro",
+            "fr": "pc",
+            "uc_param_str": "",
+            "pdir_fid": pdir_fid,
+            "_page": 1,
+            "_size": "50",
+            "_fetch_total": "1",
+            "_fetch_sub_dirs": "0",
+            "_sort": "file_type:asc,updated_at:desc",
+            "_fetch_full_path": 1,
+        }
+        try:
+            response = requests.request(
+                "GET", url, headers=headers, params=querystring
+            ).json()
+            if response["code"] == 0:
+                file_names = [
+                    item["file_name"] for item in response["data"]["full_path"]
+                ]
+                return "/".join(file_names)
+        except requests.exceptions.RequestException as e:
+            print(f"Alist刷新: 获取Quark路径出错 {e}")
         return False