feat(config): 优化配置变更事件处理机制

Author: wikrin

app/api/endpoints/system.py

@@ -248,13 +248,11 @@ async def set_env_setting(env: dict,
         )
 
     if success_updates:
-        for key in success_updates.keys():
-            # 发送配置变更事件
-            await eventmanager.async_send_event(etype=EventType.ConfigChanged, data=ConfigChangeEventData(
-                key=key,
-                value=getattr(settings, key, None),
-                change_type="update"
-            ))
+        # 发送配置变更事件
+        await eventmanager.async_send_event(etype=EventType.ConfigChanged, data=ConfigChangeEventData(
+            key=success_updates.keys(),
+            change_type="update"
+        ))
 
     return schemas.Response(
         success=True,

app/core/plugin.py

@@ -6,11 +6,11 @@
 import inspect
 import os
 import sys
+import threading
 import time
 import traceback
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from pathlib import Path
-import threading
 from typing import Any, Dict, List, Optional, Type, Union, Callable, Tuple
 
 from fastapi import HTTPException
@@ -20,24 +20,24 @@
 from app import schemas
 from app.core.cache import fresh, async_fresh
 from app.core.config import settings
-from app.core.event import eventmanager, Event
+from app.core.event import eventmanager
 from app.db.plugindata_oper import PluginDataOper
 from app.db.systemconfig_oper import SystemConfigOper
 from app.helper.plugin import PluginHelper
 from app.helper.sites import SitesHelper  # noqa
 from app.log import logger
 from app.schemas.types import EventType, SystemConfigKey
 from app.utils.crypto import RSAUtils
+from app.utils.mixins import ConfigReloadMixin
 from app.utils.object import ObjectUtils
 from app.utils.singleton import Singleton
 from app.utils.string import StringUtils
 from app.utils.system import SystemUtils
 
 
-class PluginManager(metaclass=Singleton):
-    """
-    插件管理器
-    """
+class PluginManager(ConfigReloadMixin, metaclass=Singleton):
+    """插件管理器"""
+    CONFIG_WATCH = {"DEV", "PLUGIN_AUTO_RELOAD"}
 
     def __init__(self):
         # 插件列表
@@ -250,20 +250,12 @@ def plugins(self) -> Dict[str, Any]:
         """
         return self._plugins
 
-    @eventmanager.register(EventType.ConfigChanged)
-    def handle_config_changed(self, event: Event):
-        """
-        处理配置变更事件
-        :param event: 事件对象
-        """
-        if not event:
-            return
-        event_data: schemas.ConfigChangeEventData = event.event_data
-        if event_data.key not in ['DEV', 'PLUGIN_AUTO_RELOAD']:
-            return
-        logger.info("配置变更，重新加载插件文件修改监测...")
+    def on_config_changed(self):
         self.reload_monitor()
 
+    def get_reload_name(self) -> str:
+        return "插件文件修改监测"
+
     def reload_monitor(self):
         """
         重新加载插件文件修改监测

app/helper/doh.py

@@ -14,10 +14,8 @@
 from typing import Dict, Optional
 
 from app.core.config import settings
-from app.core.event import Event, eventmanager
 from app.log import logger
-from app.schemas import ConfigChangeEventData
-from app.schemas.types import EventType
+from app.utils.mixins import ConfigReloadMixin
 from app.utils.singleton import Singleton
 
 # 定义一个全局线程池执行器
@@ -69,25 +67,23 @@ def _patched_getaddrinfo(host, *args, **kwargs):
         socket.getaddrinfo = _orig_getaddrinfo
 
 
-class DohHelper(metaclass=Singleton):
+class DohHelper(ConfigReloadMixin, metaclass=Singleton):
     """
     DoH帮助类，用于处理DNS over HTTPS解析。
     """
+    CONFIG_WATCH = {"DOH_ENABLE", "DOH_DOMAINS", "DOH_RESOLVERS"}
+
     def __init__(self):
         enable_doh(settings.DOH_ENABLE)
 
-    @eventmanager.register(EventType.ConfigChanged)
-    def handle_config_changed(self, event: Event):
-        if not event:
-            return
-        event_data: ConfigChangeEventData = event.event_data
-        if event_data.key not in ["DOH_ENABLE", "DOH_DOMAINS", "DOH_RESOLVERS"]:
-            return
+    def on_config_changed(self):
         with _doh_lock:
             # DOH配置有变动的情况下，清空缓存
             _doh_cache.clear()
         enable_doh(settings.DOH_ENABLE)
 
+    def get_reload_name(self):
+        return 'DoH'
 
 def _doh_query(resolver: str, host: str) -> Optional[str]:
     """

app/helper/redis.py

@@ -7,10 +7,8 @@
 from redis.asyncio import Redis
 
 from app.core.config import settings
-from app.core.event import eventmanager, Event
 from app.log import logger
-from app.schemas import ConfigChangeEventData
-from app.schemas.types import EventType
+from app.utils.mixins import ConfigReloadMixin
 from app.utils.singleton import Singleton
 
 # 类型缓存集合，针对非容器简单类型
@@ -74,16 +72,17 @@ def deserialize(value: bytes) -> Any:
         raise ValueError("Unknown serialization format")
 
 
-class RedisHelper(metaclass=Singleton):
+class RedisHelper(ConfigReloadMixin, metaclass=Singleton):
     """
     Redis连接和操作助手类，单例模式
-    
+
     特性：
     - 管理Redis连接池和客户端
     - 提供序列化和反序列化功能
     - 支持内存限制和淘��策略设置
     - 提供键名生成和区域管理功能
     """
+    CONFIG_WATCH = {"CACHE_BACKEND_TYPE", "CACHE_BACKEND_URL", "CACHE_REDIS_MAXMEMORY"}
 
     def __init__(self):
         """
@@ -114,25 +113,17 @@ def _connect(self):
             self.client = None
             raise RuntimeError("Redis connection failed") from e
 
-    @eventmanager.register(EventType.ConfigChanged)
-    def handle_config_changed(self, event: Event):
-        """
-        处理配置变更事件，更新Redis设置
-        :param event: 事件对象
-        """
-        if not event:
-            return
-        event_data: ConfigChangeEventData = event.event_data
-        if event_data.key not in ['CACHE_BACKEND_TYPE', 'CACHE_BACKEND_URL', 'CACHE_REDIS_MAXMEMORY']:
-            return
-        logger.info("配置变更，重连Redis...")
+    def on_config_changed(self):
         self.close()
         self._connect()
 
+    def get_reload_name(self):
+        return "Redis"
+
     def set_memory_limit(self, policy: Optional[str] = "allkeys-lru"):
         """
         动态设置Redis最大内存和内存淘汰策略
-        
+
         :param policy: 淘汰策略（如'allkeys-lru'）
         """
         try:
@@ -310,17 +301,18 @@ def close(self) -> None:
             logger.debug("Redis connection closed")
 
 
-class AsyncRedisHelper(metaclass=Singleton):
+class AsyncRedisHelper(ConfigReloadMixin, metaclass=Singleton):
     """
     异步Redis连接和操作助手类，单例模式
-    
+
     特性：
     - 管理异步Redis连接池和客户端
     - 提供序列化和反序列化功能
     - 支持内存限制和淘汰策略设置
     - 提供键名生成和区域管理功能
     - 所有操作都是异步的
     """
+    CONFIG_WATCH = {"CACHE_BACKEND_TYPE", "CACHE_BACKEND_URL", "CACHE_REDIS_MAXMEMORY"}
 
     def __init__(self):
         """
@@ -351,25 +343,17 @@ async def _connect(self):
             self.client = None
             raise RuntimeError("Redis async connection failed") from e
 
-    @eventmanager.register(EventType.ConfigChanged)
-    async def handle_config_changed(self, event: Event):
-        """
-        处理配置变更事件，更新Redis设置
-        :param event: 事件对象
-        """
-        if not event:
-            return
-        event_data: ConfigChangeEventData = event.event_data
-        if event_data.key not in ['CACHE_BACKEND_TYPE', 'CACHE_BACKEND_URL', 'CACHE_REDIS_MAXMEMORY']:
-            return
-        logger.info("配置变更，重连Redis (async)...")
+    async def on_config_changed(self):
         await self.close()
         await self._connect()
 
+    def get_reload_name(self):
+        return "Redis (async)"
+
     async def set_memory_limit(self, policy: Optional[str] = "allkeys-lru"):
         """
         动态设置Redis最大内存和内存淘汰策略
-        
+
         :param policy: 淘汰策略（如'allkeys-lru'）
         """
         try:

app/helper/system.py

@@ -8,35 +8,32 @@
 import docker
 
 from app.core.config import settings
-from app.core.event import eventmanager, Event
 from app.log import logger
-from app.schemas import ConfigChangeEventData
-from app.schemas.types import EventType
+from app.utils.mixins import ConfigReloadMixin
 from app.utils.system import SystemUtils
 
 
-class SystemHelper:
+class SystemHelper(ConfigReloadMixin):
     """
     系统工具类，提供系统相关的操作和判断
     """
+    CONFIG_WATCH = {
+        "DEBUG",
+        "LOG_LEVEL",
+        "LOG_MAX_FILE_SIZE",
+        "LOG_BACKUP_COUNT",
+        "LOG_FILE_FORMAT",
+        "LOG_CONSOLE_FORMAT",
+    }
 
     __system_flag_file = "/var/log/nginx/__moviepilot__"
 
-    @eventmanager.register(EventType.ConfigChanged)
-    def handle_config_changed(self, event: Event):
-        """
-        处理配置变更事件，更新日志设置
-        :param event: 事件对象
-        """
-        if not event:
-            return
-        event_data: ConfigChangeEventData = event.event_data
-        if event_data.key not in ['DEBUG', 'LOG_LEVEL', 'LOG_MAX_FILE_SIZE', 'LOG_BACKUP_COUNT',
-                                  'LOG_FILE_FORMAT', 'LOG_CONSOLE_FORMAT']:
-            return
-        logger.info("配置变更，更新日志设置...")
+    def on_config_changed(self):
         logger.update_loggers()
 
+    def get_reload_name(self):
+        return "日志设置"
+
     @staticmethod
     def can_restart() -> bool:
         """

app/modules/__init__.py

@@ -4,15 +4,22 @@
 from app.helper.service import ServiceConfigHelper
 from app.schemas import Notification, NotificationConf, MediaServerConf, DownloaderConf
 from app.schemas.types import ModuleType, DownloaderType, MediaServerType, MessageChannel, StorageSchema, \
-    OtherModulesType
+    OtherModulesType, SystemConfigKey
+from app.utils.mixins import ConfigReloadMixin
 
 
-class _ModuleBase(metaclass=ABCMeta):
+class _ModuleBase(ConfigReloadMixin, metaclass=ABCMeta):
     """
     模块基类，实现对应方法，在有需要时会被自动调用，返回None代表不启用该模块，将继续执行下一模块
     输入参数与输出参数一致的，或没有输出的，可以被多个模块重复实现
     """
 
+    def on_config_changed(self):
+        self.init_module()
+
+    def get_reload_name(self):
+        return self.get_name()
+
     @abstractmethod
     def init_module(self) -> None:
         """
@@ -177,6 +184,7 @@ class _MessageBase(ServiceBase[TService, NotificationConf]):
     """
     消息基类
     """
+    CONFIG_WATCH = {SystemConfigKey.Notifications.value}
 
     def __init__(self):
         """
@@ -224,6 +232,7 @@ class _DownloaderBase(ServiceBase[TService, DownloaderConf]):
     """
     下载器基类
     """
+    CONFIG_WATCH = {SystemConfigKey.Downloaders.value}
 
     def __init__(self):
         """
@@ -287,6 +296,7 @@ class _MediaServerBase(ServiceBase[TService, MediaServerConf]):
     """
     媒体服务器基类
     """
+    CONFIG_WATCH = {SystemConfigKey.MediaServers.value}
 
     def get_configs(self) -> Dict[str, MediaServerConf]:
         """

app/modules/emby/__init__.py

@@ -2,11 +2,11 @@
 
 from app import schemas
 from app.core.context import MediaInfo
-from app.core.event import eventmanager, Event
+from app.core.event import eventmanager
 from app.log import logger
 from app.modules import _MediaServerBase, _ModuleBase
 from app.modules.emby.emby import Emby
-from app.schemas.types import MediaType, ModuleType, ChainEventType, MediaServerType, SystemConfigKey, EventType
+from app.schemas.types import MediaType, ModuleType, ChainEventType, MediaServerType
 
 
 class EmbyModule(_ModuleBase, _MediaServerBase[Emby]):
@@ -18,20 +18,6 @@ def init_module(self) -> None:
         super().init_service(service_name=Emby.__name__.lower(),
                              service_type=lambda conf: Emby(**conf.config, sync_libraries=conf.sync_libraries))
 
-    @eventmanager.register(EventType.ConfigChanged)
-    def handle_config_changed(self, event: Event):
-        """
-        处理配置变更事件
-        :param event: 事件对象
-        """
-        if not event:
-            return
-        event_data: schemas.ConfigChangeEventData = event.event_data
-        if event_data.key not in [SystemConfigKey.MediaServers.value]:
-            return
-        logger.info("配置变更，重新初始化Emby模块...")
-        self.init_module()
-
     @staticmethod
     def get_name() -> str:
         return "Emby"