diff --git a/media_servers/alist.py b/media_servers/alist.py index 2a33c6f..f384dba 100644 --- a/media_servers/alist.py +++ b/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"): - full_path = os.path.normpath( - os.path.join(self.quark_root_path, task["savepath"].lstrip("/")) + if task.get("savepath") and task.get("savepath").startswith( + self.quark_root_dir + ): + 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() - if response.get("code") == 200: - print(f"📁 刷新Alist目录：[{path}] 成功✅") - return response.get("data") - elif "object not found" in response.get("message", ""): + data = response.json() + if data.get("code") == 200: + print(f"📁 Alist刷新：目录[{path}] 成功✅") + return data.get("data") + 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