media_servers/README.md

@@ -6,7 +6,7 @@
 
 * 模块位于 `media_servers` 目录下.
 * 每个模块是一个 `.py` 文件 (例如 `emby.py`, `plex.py`)，文件名小写。
-* 每个模块文件包含一个与文件名对应的首字母大写命名类（例如 `emby.py` 中的 `Emby` 类）。
+* 每个模块文件包含一个与文件名对应的大驼峰命名法类（例如 `emby.py` 中的 `Emby` 类）。
 
 ## 模块要求
 
@@ -27,7 +27,7 @@
 
 * **`run(self, task)`**：整个模块入口函数，处理模块逻辑。
   * `task` 是一个字典，包含任务信息。如果需要修改任务参数，返回修改后的 `task` 字典；
-  * 无修改则不返回或返回 `None`。
+  * 无修改则不返回或返回 `False`。
 
 ## 模块示例
 
@@ -84,10 +84,4 @@ docker run -d \
 }
 ```
 
-当模块代码正确赋值 `default_config` 时，首次运行会自动补充缺失的键。
-
-## 🤝 贡献者
-
-| 模块    | 说明                 | 贡献者                                  |
-| ------- | -------------------- | --------------------------------------- |
-| plex.py | 自动刷新 Plex 媒体库 | [zhazhayu](https://github.com/zhazhayu) |
+当模块代码正确赋值 `default_config` 时，首次运行会自动补充缺失的键。

media_servers/alist.py

@@ -1,14 +1,11 @@
 import os
+import re
 import requests
 
 
 class Alist:
 
-    default_config = {
-        "url": "",  # Alist服务器URL
-        "token": "",  # Alist服务器Token
-        "quark_root_path": "/quark",  # 夸克根目录在Alist中的挂载路径
-    }
+    default_config = {"url": "", "token": "", "path_prefix": "/quark"}
     is_active = False
 
     def __init__(self, **kwargs):
@@ -24,10 +21,8 @@ class Alist:
 
     def run(self, task):
         if task.get("savepath"):
-            full_path = os.path.normpath(
-                os.path.join(self.quark_root_path, task["savepath"].lstrip("/"))
-            ).replace("\\", "/")
-            self.refresh(full_path)
+            path = self._normalize_path(task["savepath"])
+            self.refresh(path)
 
     def get_info(self):
         url = f"{self.url}/api/admin/setting/list"
@@ -59,20 +54,22 @@ class Alist:
             "per_page": 0,
         }
         try:
-            response = requests.request("POST", url, headers=headers, json=payload)
+            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}] 成功✅")
+                print(f"📁 刷新Alist目录：{path} 成功✅")
                 return response.get("data")
             elif "object not found" in response.get("message", ""):
                 # 如果是根目录就不再往上查找
-                if path == "/" or path == self.quark_root_path:
+                if path == "/" or path == self.path_prefix:
                     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:
@@ -80,3 +77,9 @@ class Alist:
         except requests.exceptions.RequestException as e:
             print(f"刷新Alist目录出错: {e}")
         return False
+
+    def _normalize_path(self, path):
+        """标准化路径格式"""
+        if not path.startswith(self.path_prefix):
+            path = f"/{self.path_prefix}/{path}"
+        return re.sub(r"/{2,}", "/", path)

media_servers/plex.py

@@ -1,96 +0,0 @@
-import os
-import requests
-
-
-class Plex:
-
-    default_config = {
-        "url": "",  # Plex服务器URL
-        "token": "",  # Plex Token，可F12在请求中抓取
-        "quark_root_path": "",  # 夸克根目录在Plex中的路径；假设夸克目录/media/tv在plex中对应的路径为/quark/media/tv，则为/quark
-    }
-    is_active = False
-    _libraries = None  # 缓存库信息
-
-    def __init__(self, **kwargs):
-        if kwargs:
-            for key, value 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 and self.quark_root_path:
-                if self.get_info():
-                    self.is_active = True
-
-    def run(self, task):
-        if task.get("savepath"):
-            # 检查是否已缓存库信息
-            if self._libraries is None:
-                self._libraries = self._get_libraries()
-            # 拼接完整路径
-            full_path = os.path.normpath(
-                os.path.join(self.quark_root_path, task["savepath"].lstrip("/"))
-            ).replace("\\", "/")
-            self.refresh(full_path)
-
-    def get_info(self):
-        """获取Plex服务器信息"""
-        headers = {"Accept": "application/json", "X-Plex-Token": self.token}
-        try:
-            response = requests.get(f"{self.url}/", headers=headers)
-            if response.status_code == 200:
-                info = response.json()["MediaContainer"]
-                print(
-                    f"Plex媒体库: {info.get('friendlyName','')} v{info.get('version','')}"
-                )
-                return True
-            else:
-                print(f"Plex媒体库: 连接失败❌ 状态码：{response.status_code}")
-        except Exception as e:
-            print(f"获取Plex媒体库信息出错: {e}")
-        return False
-
-    def refresh(self, folder_path):
-        """刷新指定文件夹"""
-        if not folder_path:
-            return False
-        headers = {"Accept": "application/json", "X-Plex-Token": self.token}
-        try:
-            for library in self._libraries:
-                for location in library.get("Location", []):
-                    if (
-                        os.path.commonpath([folder_path, location["path"]])
-                        == location["path"]
-                    ):
-                        refresh_url = f"{self.url}/library/sections/{library['key']}/refresh?path={folder_path}"
-                        refresh_response = requests.get(refresh_url, headers=headers)
-                        if refresh_response.status_code == 200:
-                            print(
-                                f"🎞️ 刷新Plex媒体库：{library['title']} [{folder_path}] 成功✅"
-                            )
-                            return True
-                        else:
-                            print(
-                                f"🎞️ 刷新Plex媒体库：刷新请求失败❌ 状态码：{refresh_response.status_code}"
-                            )
-            print(f"🎞️ 刷新Plex媒体库：{folder_path} 未找到匹配的媒体库❌")
-        except Exception as e:
-            print(f"刷新Plex媒体库出错: {e}")
-        return False
-
-    def _get_libraries(self):
-        """获取Plex媒体库信息"""
-        url = f"{self.url}/library/sections"
-        headers = {"Accept": "application/json", "X-Plex-Token": self.token}
-        try:
-            response = requests.get(url, headers=headers)
-            if response.status_code == 200:
-                libraries = response.json()["MediaContainer"].get("Directory", [])
-                return libraries
-            else:
-                print(f"🎞️ 获取Plex媒体库信息失败❌ 状态码：{response.status_code}")
-                return None
-        except Exception as e:
-            print(f"获取Plex媒体库信息出错: {e}")
-            return None