feat(module): add ServiceBaseHelper for service and instance

app/helper/downloader.py

@@ -1,5 +1,16 @@
-class DownloaderHelper:
+from app.helper.servicebase import ServiceBaseHelper
+from app.schemas import DownloaderConf
+from app.schemas.types import SystemConfigKey
+
+
+class DownloaderHelper(ServiceBaseHelper[DownloaderConf]):
     """
     下载器帮助类
     """
-    pass
+
+    def __init__(self):
+        super().__init__(
+            config_key=SystemConfigKey.Downloaders,
+            conf_type=DownloaderConf,
+            modules=["QbittorrentModule", "TransmissionModule"]
+        )

app/helper/mediaserver.py

@@ -1,5 +1,16 @@
-class MediaServerHelper:
+from app.helper.servicebase import ServiceBaseHelper
+from app.schemas import MediaServerConf
+from app.schemas.types import SystemConfigKey
+
+
+class MediaServerHelper(ServiceBaseHelper[MediaServerConf]):
     """
     媒体服务器帮助类
     """
-    pass
+
+    def __init__(self):
+        super().__init__(
+            config_key=SystemConfigKey.MediaServers,
+            conf_type=MediaServerConf,
+            modules=["PlexModule", "EmbyModule", "JellyfinModule"]
+        )

app/helper/notification.py

@@ -1,5 +1,17 @@
-class NotificationHelper:
+from app.helper.servicebase import ServiceBaseHelper
+from app.schemas import NotificationConf
+from app.schemas.types import SystemConfigKey
+
+
+class NotificationHelper(ServiceBaseHelper[NotificationConf]):
     """
-    消息通知渠道帮助类
+    消息通知帮助类
     """
-    pass
+
+    def __init__(self):
+        super().__init__(
+            config_key=SystemConfigKey.Notifications,
+            conf_type=NotificationConf,
+            modules=["WechatModule", "WebPushModule", "VoceChatModule", "TelegramModule", "SynologyChatModule",
+                     "SlackModule"]
+        )

app/helper/servicebase.py

@@ -0,0 +1,97 @@
+from typing import Dict, List, Optional, Type, TypeVar, Generic, Iterator
+
+from app.core.module import ModuleManager
+from app.helper.serviceconfig import ServiceConfigHelper
+from app.schemas import ServiceInfo
+from app.schemas.types import SystemConfigKey
+
+TConf = TypeVar("TConf")
+
+
+class ServiceBaseHelper(Generic[TConf]):
+    """
+    通用服务帮助类，抽象获取配置和服务实例的通用逻辑
+    """
+
+    def __init__(self, config_key: SystemConfigKey, conf_type: Type[TConf], modules: List[str]):
+        self.modulemanager = ModuleManager()
+        self.config_key = config_key
+        self.conf_type = conf_type
+        self.modules = modules
+
+    def get_configs(self, include_disabled: bool = False) -> Dict[str, TConf]:
+        """
+        获取配置列表
+
+        :param include_disabled: 是否包含禁用的配置，默认 False（仅返回启用的配置）
+        :return: 配置字典
+        """
+        configs: List[TConf] = ServiceConfigHelper.get_configs(self.config_key, self.conf_type)
+        return {
+            config.name: config
+            for config in configs
+            if (config.name and config.type and config.enabled) or include_disabled
+        } if configs else {}
+
+    def get_config(self, name: str) -> Optional[TConf]:
+        """
+        获取指定名称配置
+        """
+        if not name:
+            return None
+        configs = self.get_configs()
+        return configs.get(name)
+
+    def iterate_module_instances(self) -> Iterator[ServiceInfo]:
+        """
+        迭代所有模块的实例及其对应的配置，返回 ServiceInfo 实例
+        """
+        configs = self.get_configs()
+        for module_name in self.modules:
+            module = self.modulemanager.get_running_module(module_name)
+            if not module:
+                continue
+            module_instances = module.get_instances()
+            if not isinstance(module_instances, dict):
+                continue
+            for name, instance in module_instances.items():
+                if not instance:
+                    continue
+                config = configs.get(name)
+                service_info = ServiceInfo(
+                    name=name,
+                    instance=instance,
+                    module=module,
+                    type=config.type if config else None,
+                    config=config
+                )
+                yield service_info
+
+    def get_services(self, type_filter: Optional[str] = None) -> Dict[str, ServiceInfo]:
+        """
+        获取服务信息列表，并根据类型过滤
+
+        :param type_filter: 需要过滤的服务类型
+        :return: 过滤后的服务信息字典
+        """
+        return {
+            service_info.name: service_info
+            for service_info in self.iterate_module_instances()
+            if service_info.config and (type_filter is None or service_info.type == type_filter)
+        }
+
+    def get_service(self, name: str, type_filter: Optional[str] = None) -> Optional[ServiceInfo]:
+        """
+        获取指定名称的服务信息，并根据类型过滤
+
+        :param name: 服务名称
+        :param type_filter: 需要过滤的服务类型
+        :return: 对应的服务信息，若不存在或类型不匹配则返回 None
+        """
+        if not name:
+            return None
+        for service_info in self.iterate_module_instances():
+            if service_info.name == name:
+                if service_info.config and (type_filter is None or service_info.type == type_filter):
+                    return service_info
+        return None

app/helper/serviceconfig.py

@@ -56,7 +56,7 @@ class ServiceConfigHelper:
     @staticmethod
     def get_notification_switch(mtype: NotificationType) -> Optional[str]:
         """
-        获取消息通知场景开关
+        获取指定类型的消息通知场景的开关
         """
         switchs = ServiceConfigHelper.get_notification_switches()
         for switch in switchs:

app/modules/__init__.py

@@ -90,6 +90,14 @@ class ServiceBase(Generic[TService, TConf], metaclass=ABCMeta):
                         # 如果传入的是工厂函数，直接调用工厂函数
                         self._instances[conf.name] = service_type(conf)
 
+    def get_instances(self) -> Dict[str, TService]:
+        """
+        获取服务实例列表
+
+        :return: 返回服务实例列表
+        """
+        return self._instances
+
     def get_instance(self, name: str) -> Optional[TService]:
         """
         获取服务实例

app/schemas/system.py

@@ -1,8 +1,26 @@
-from typing import Optional
+from dataclasses import dataclass
+from typing import Optional, Any
 
 from pydantic import BaseModel
 
 
+@dataclass
+class ServiceInfo:
+    """
+    封装服务相关信息的数据类
+    """
+    # 名称
+    name: Optional[str] = None
+    # 实例
+    instance: Optional[Any] = None
+    # 模块
+    module: Optional[Any] = None
+    # 类型
+    type: Optional[str] = None
+    # 配置
+    config: Optional[Any] = None
+
+
 class MediaServerConf(BaseModel):
     """
     媒体服务器配置