♻️ 重构插件加载逻辑，优化配置更新

- 配置操作类函数整合到 class 中
- 容器运行时补充缺失配置

app/run.py

@@ -24,6 +24,7 @@ import os
 parent_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
 sys.path.insert(0, parent_dir)
 from quark_auto_save import Quark
+from quark_auto_save import Config
 
 
 def get_app_ver():
@@ -41,7 +42,9 @@ def get_app_ver():
 PYTHON_PATH = "python3" if os.path.exists("/usr/bin/python3") else "python"
 SCRIPT_PATH = os.environ.get("SCRIPT_PATH", "./quark_auto_save.py")
 CONFIG_PATH = os.environ.get("CONFIG_PATH", "./config/quark_config.json")
-DEBUG = os.environ.get("DEBUG", False)
+DEBUG = os.environ.get("DEBUG", True)
+
+task_plugins_config = {}
 
 app = Flask(__name__)
 app.config["APP_VERSION"] = get_app_ver()
@@ -145,6 +148,7 @@ def get_data():
         return redirect(url_for("login"))
     data = read_json()
     del data["webui"]
+    data["task_plugins_config"] = task_plugins_config
     return jsonify(data)
 
 
@@ -153,10 +157,10 @@ def get_data():
 def update():
     if not is_login():
         return "未登录"
-    data = read_json()
-    webui = data["webui"]
     data = request.json
-    data["webui"] = webui
+    data["webui"] = read_json()["webui"]
+    if "task_plugins_config" in data:
+        del data["task_plugins_config"]
     write_json(data)
     # 重新加载任务
     if reload_tasks():
@@ -279,6 +283,7 @@ def reload_tasks():
 
 
 def init():
+    global task_plugins_config
     logging.info(f">>> 初始化配置")
     # 检查配置文件是否存在
     if not os.path.exists(CONFIG_PATH):
@@ -287,6 +292,7 @@ def init():
         with open("quark_config.json", "rb") as src, open(CONFIG_PATH, "wb") as dest:
             dest.write(src.read())
     data = read_json()
+    Config.breaking_change_update(data)
     # 默认管理账号
     data["webui"] = {
         "username": os.environ.get("WEBUI_USERNAME")
@@ -297,6 +303,10 @@ def init():
     # 默认定时规则
     if not data.get("crontab"):
         data["crontab"] = "0 8,18,20 * * *"
+    # 初始化插件配置
+    _, plugins_config_default, task_plugins_config = Config.load_plugins()
+    plugins_config_default.update(data.get("plugins", {}))
+    data["plugins"] = plugins_config_default
     write_json(data)
 
 

quark_auto_save.py

@@ -39,18 +39,6 @@ MAGIC_REGEX = {
 }
 
 
-# 魔法正则匹配
-def magic_regex_func(pattern, replace, taskname=""):
-    keyword = pattern
-    if keyword in CONFIG_DATA["magic_regex"]:
-        pattern = CONFIG_DATA["magic_regex"][keyword]["pattern"]
-        if replace == "":
-            replace = CONFIG_DATA["magic_regex"][keyword]["replace"]
-    if taskname:
-        replace = replace.replace("$TASKNAME", taskname)
-    return pattern, replace
-
-
 # 发送通知消息
 def send_ql_notify(title, body):
     try:
@@ -75,28 +63,89 @@ def add_notify(text):
     return text
 
 
-# 下载配置
+class Config:
-def download_file(url, save_path):
+    # 下载配置
-    response = requests.get(url)
+    def download_file(url, save_path):
-    if response.status_code == 200:
+        response = requests.get(url)
-        with open(save_path, "wb") as file:
+        if response.status_code == 200:
-            file.write(response.content)
+            with open(save_path, "wb") as file:
-        return True
+                file.write(response.content)
-    else:
+            return True
-        return False
-
-
-# 读取CK
-def get_cookies(cookie_val):
-    if isinstance(cookie_val, list):
-        return cookie_val
-    elif cookie_val:
-        if "\n" in cookie_val:
-            return cookie_val.split("\n")
         else:
-            return [cookie_val]
+            return False
-    else:
+
-        return False
+    # 读取CK
+    def get_cookies(cookie_val):
+        if isinstance(cookie_val, list):
+            return cookie_val
+        elif cookie_val:
+            if "\n" in cookie_val:
+                return cookie_val.split("\n")
+            else:
+                return [cookie_val]
+        else:
+            return False
+
+    def load_plugins(plugins_config={}, plugins_dir="plugins"):
+        plugins_available = {}
+        task_plugins_config = {}
+        all_modules = [
+            f.replace(".py", "") for f in os.listdir(plugins_dir) if f.endswith(".py")
+        ]
+        # 调整模块优先级
+        priority_path = os.path.join(plugins_dir, "_priority.json")
+        try:
+            with open(priority_path, encoding="utf-8") as f:
+                priority_modules = json.load(f)
+            if priority_modules:
+                all_modules = [
+                    module for module in priority_modules if module in all_modules
+                ] + [module for module in all_modules if module not in priority_modules]
+        except (FileNotFoundError, json.JSONDecodeError):
+            priority_modules = []
+        for module_name in all_modules:
+            try:
+                module = importlib.import_module(f"{plugins_dir}.{module_name}")
+                ServerClass = getattr(module, module_name.capitalize())
+                # 检查配置中是否存在该模块的配置
+                if module_name in plugins_config:
+                    plugin = ServerClass(**plugins_config[module_name])
+                    plugins_available[module_name] = plugin
+                else:
+                    plugin = ServerClass()
+                    plugins_config[module_name] = plugin.default_config
+                # 检查插件是否支持单独任务配置
+                if hasattr(plugin, "default_task_config"):
+                    task_plugins_config[module_name] = plugin.default_task_config
+            except (ImportError, AttributeError) as e:
+                print(f"载入模块 {module_name} 失败: {e}")
+        print()
+        return plugins_available, plugins_config, task_plugins_config
+
+    def breaking_change_update(config_data):
+        if config_data.get("emby"):
+            print("🔼 Update config v0.3.6.1 to 0.3.7")
+            config_data.setdefault("media_servers", {})["emby"] = {
+                "url": config_data["emby"]["url"],
+                "token": config_data["emby"]["apikey"],
+            }
+            del config_data["emby"]
+            for task in config_data.get("tasklist", {}):
+                task["media_id"] = task.get("emby_id", "")
+                if task.get("emby_id"):
+                    del task["emby_id"]
+        if config_data.get("media_servers"):
+            print("🔼 Update config v0.3.8 to 0.3.9")
+            config_data["plugins"] = config_data.get("media_servers")
+            del config_data["media_servers"]
+            for task in config_data.get("tasklist", {}):
+                task["addition"] = {
+                    "emby": {
+                        "media_id": task.get("media_id", ""),
+                    }
+                }
+                if task.get("media_id"):
+                    del task["media_id"]
 
 
 class Quark:
@@ -418,6 +467,18 @@ class Quark:
     # ↑ 请求函数
     # ↓ 操作函数
 
+    # 魔法正则匹配
+    def magic_regex_func(self, pattern, replace, taskname=None):
+        magic_regex = CONFIG_DATA.get("magic_regex") or MAGIC_REGEX or {}
+        keyword = pattern
+        if keyword in magic_regex:
+            pattern = magic_regex[keyword]["pattern"]
+            if replace == "":
+                replace = magic_regex[keyword]["replace"]
+        if taskname:
+            replace = replace.replace("$TASKNAME", taskname)
+        return pattern, replace
+
     def get_id_from_url(self, url):
         url = url.replace("https://pan.quark.cn/s/", "")
         pattern = r"(\w+)(\?pwd=(\w+))?(#/list/share.*/(\w+))?"
@@ -578,7 +639,7 @@ class Quark:
             if share_file["dir"] and task.get("update_subdir", False):
                 pattern, replace = task["update_subdir"], ""
             else:
-                pattern, replace = magic_regex_func(
+                pattern, replace = self.magic_regex_func(
                     task["pattern"], task["replace"], task["taskname"]
                 )
             # 正则文件名匹配
@@ -671,7 +732,7 @@ class Quark:
         return tree
 
     def do_rename_task(self, task, subdir_path=""):
-        pattern, replace = magic_regex_func(
+        pattern, replace = self.magic_regex_func(
             task["pattern"], task["replace"], task["taskname"]
         )
         if not pattern or not replace:
@@ -707,44 +768,6 @@ class Quark:
         return is_rename_count > 0
 
 
-def load_plugins(plugins_config, plugins_dir="plugins"):
-    plugins_available = {}
-    task_plugins_config = {}
-    all_modules = [
-        f.replace(".py", "") for f in os.listdir(plugins_dir) if f.endswith(".py")
-    ]
-    # 调整模块优先级
-    priority_path = os.path.join(plugins_dir, "_priority.json")
-    try:
-        with open(priority_path, encoding="utf-8") as f:
-            priority_modules = json.load(f)
-        if priority_modules:
-            all_modules = [
-                module for module in priority_modules if module in all_modules
-            ] + [module for module in all_modules if module not in priority_modules]
-    except (FileNotFoundError, json.JSONDecodeError):
-        priority_modules = []
-    print(f"🧩 载入插件")
-    for module_name in all_modules:
-        try:
-            module = importlib.import_module(f"{plugins_dir}.{module_name}")
-            ServerClass = getattr(module, module_name.capitalize())
-            # 检查配置中是否存在该模块的配置
-            if module_name in plugins_config:
-                plugin = ServerClass(**plugins_config[module_name])
-                plugins_available[module_name] = plugin
-            else:
-                plugin = ServerClass()
-                plugins_config[module_name] = plugin.default_config
-            # 检查插件是否支持单独任务配置
-            if hasattr(plugin, "default_task_config"):
-                task_plugins_config[module_name] = plugin.default_task_config
-        except (ImportError, AttributeError) as e:
-            print(f"载入模块 {module_name} 失败: {e}")
-    print()
-    return plugins_available, task_plugins_config
-
-
 def verify_account(account):
     # 验证账号
     print(f"▶️ 验证第{account.index}个账号")
@@ -805,7 +828,8 @@ def do_sign(account):
 
 
 def do_save(account, tasklist=[]):
-    plugins, CONFIG_DATA["task_plugins_config"] = load_plugins(
+    print(f"🧩 载入插件")
+    plugins, CONFIG_DATA["plugins"], task_plugins_config = Config.load_plugins(
         CONFIG_DATA.get("plugins", {})
     )
     print(f"转存账号: {account.nickname}")
@@ -861,7 +885,7 @@ def do_save(account, tasklist=[]):
                 return result
 
             task["addition"] = merge_dicts(
-                task.get("addition", {}), CONFIG_DATA["task_plugins_config"]
+                task.get("addition", {}), task_plugins_config
             )
             # 调用插件
             if is_new_tree or is_rename:
@@ -874,33 +898,6 @@ def do_save(account, tasklist=[]):
     print()
 
 
-def breaking_change_update():
-    global CONFIG_DATA
-    if CONFIG_DATA.get("emby"):
-        print("🔼 Update config v0.3.6.1 to 0.3.7")
-        CONFIG_DATA.setdefault("media_servers", {})["emby"] = {
-            "url": CONFIG_DATA["emby"]["url"],
-            "token": CONFIG_DATA["emby"]["apikey"],
-        }
-        del CONFIG_DATA["emby"]
-        for task in CONFIG_DATA.get("tasklist", {}):
-            task["media_id"] = task.get("emby_id", "")
-            if task.get("emby_id"):
-                del task["emby_id"]
-    if CONFIG_DATA.get("media_servers"):
-        print("🔼 Update config v0.3.8 to 0.3.9")
-        CONFIG_DATA["plugins"] = CONFIG_DATA.get("media_servers")
-        del CONFIG_DATA["media_servers"]
-        for task in CONFIG_DATA.get("tasklist", {}):
-            task["addition"] = {
-                "emby": {
-                    "media_id": task.get("media_id", ""),
-                }
-            }
-            if task.get("media_id"):
-                del task["media_id"]
-
-
 def main():
     global CONFIG_DATA
     start_time = datetime.now()
@@ -921,20 +918,20 @@ def main():
         else:
             print(f"⚙️ 配置文件 {config_path} 不存在❌，正远程从下载配置模版")
             config_url = f"{GH_PROXY}https://raw.githubusercontent.com/Cp0204/quark_auto_save/main/quark_config.json"
-            if download_file(config_url, config_path):
+            if Config.download_file(config_url, config_path):
                 print("⚙️ 配置模版下载成功✅，请到程序目录中手动配置")
             return
     else:
         print(f"⚙️ 正从 {config_path} 文件中读取配置")
         with open(config_path, "r", encoding="utf-8") as file:
             CONFIG_DATA = json.load(file)
-            breaking_change_update()
+            Config.breaking_change_update(CONFIG_DATA)
         cookie_val = CONFIG_DATA.get("cookie")
         if not CONFIG_DATA.get("magic_regex"):
             CONFIG_DATA["magic_regex"] = MAGIC_REGEX
         cookie_form_file = True
     # 获取cookie
-    cookies = get_cookies(cookie_val)
+    cookies = Config.get_cookies(cookie_val)
     if not cookies:
         print("❌ cookie 未配置")
         return