add ModuleType Schema

2026-09-06 16:07:01 +08:00 · 2024-10-12 11:50:00 +08:00
parent fcfeeb09d3
commit 603ab97665
27 changed files with 248 additions and 148 deletions
@@ -2,7 +2,7 @@ from typing import Optional

 from app.helper.servicebase import ServiceBaseHelper
 from app.schemas import DownloaderConf, ServiceInfo
-from app.schemas.types import SystemConfigKey
+from app.schemas.types import SystemConfigKey, ModuleType


 class DownloaderHelper(ServiceBaseHelper[DownloaderConf]):
@@ -14,29 +14,24 @@ class DownloaderHelper(ServiceBaseHelper[DownloaderConf]):
        super().__init__(
            config_key=SystemConfigKey.Downloaders,
            conf_type=DownloaderConf,
-            modules=["QbittorrentModule", "TransmissionModule"]
+            module_type=ModuleType.Downloader
        )

-    def is_qbittorrent(self, service: Optional[ServiceInfo] = None, name: Optional[str] = None) -> bool:
+    def is_downloader(
+            self,
+            service_type: Optional[str] = None,
+            service: Optional[ServiceInfo] = None,
+            name: Optional[str] = None,
+    ) -> bool:
        """
-        判断指定的下载器是否为 qbittorrent 类型，需要传入 `service` 或 `name` 中的任一参数
-
+        通用的下载器类型判断方法
+        :param service_type: 下载器的类型名称（如 'qbittorrent', 'transmission'）
        :param service: 要判断的服务信息
        :param name: 服务的名称
-        :return: 如果服务类型为 qbittorrent，返回 True；否则返回 False。
+        :return: 如果服务类型或实例为指定类型，返回 True；否则返回 False
        """
-        if not service:
-            service = self.get_service(name=name)
-        return service.type == "qbittorrent" if service else False
+        # 如果未提供 service 则通过 name 获取服务
+        service = service or self.get_service(name=name)

-    def is_transmission(self, service: Optional[ServiceInfo] = None, name: Optional[str] = None) -> bool:
-        """
-        判断指定的下载器是否为 transmission 类型，需要传入 `service` 或 `name` 中的任一参数
-
-        :param service: 要判断的服务信息
-        :param name: 服务的名称
-        :return: 如果服务类型为 transmission，返回 True；否则返回 False。
-        """
-        if not service:
-            service = self.get_service(name=name)
-        return service.type == "transmission" if service else False
+        # 判断服务类型是否为指定类型
+        return bool(service and service.type == service_type)
@@ -2,7 +2,7 @@ from typing import Optional

 from app.helper.servicebase import ServiceBaseHelper
 from app.schemas import MediaServerConf, ServiceInfo
-from app.schemas.types import SystemConfigKey
+from app.schemas.types import SystemConfigKey, ModuleType


 class MediaServerHelper(ServiceBaseHelper[MediaServerConf]):
@@ -14,41 +14,24 @@ class MediaServerHelper(ServiceBaseHelper[MediaServerConf]):
        super().__init__(
            config_key=SystemConfigKey.MediaServers,
            conf_type=MediaServerConf,
-            modules=["PlexModule", "EmbyModule", "JellyfinModule"]
+            module_type=ModuleType.MediaServer
        )

-    def is_plex(self, service: Optional[ServiceInfo] = None, name: Optional[str] = None) -> bool:
+    def is_media_server(
+            self,
+            service_type: Optional[str] = None,
+            service: Optional[ServiceInfo] = None,
+            name: Optional[str] = None,
+    ) -> bool:
        """
-        判断指定的媒体服务器是否为 Plex 类型，需要传入 `service` 或 `name` 中的任一参数
-
+        通用的媒体服务器类型判断方法
+        :param service_type: 媒体服务器的类型名称（如 'plex', 'emby', 'jellyfin'）
        :param service: 要判断的服务信息
        :param name: 服务的名称
-        :return: 如果服务类型为 plex，返回 True；否则返回 False。
+        :return: 如果服务类型或实例为指定类型，返回 True；否则返回 False
        """
-        if not service:
-            service = self.get_service(name=name)
-        return service.type == "plex" if service else False
+        # 如果未提供 service 则通过 name 获取服务
+        service = service or self.get_service(name=name)

-    def is_emby(self, service: Optional[ServiceInfo] = None, name: Optional[str] = None) -> bool:
-        """
-        判断指定的媒体服务器是否为 Emby 类型，需要传入 `service` 或 `name` 中的任一参数
-
-        :param service: 要判断的服务信息
-        :param name: 服务的名称
-        :return: 如果服务类型为 emby，返回 True；否则返回 False。
-        """
-        if not service:
-            service = self.get_service(name=name)
-        return service.type == "emby" if service else False
-
-    def is_jellyfin(self, service: Optional[ServiceInfo] = None, name: Optional[str] = None) -> bool:
-        """
-        判断指定的媒体服务器是否为 Jellyfin 类型，需要传入 `service` 或 `name` 中的任一参数
-
-        :param service: 要判断的服务信息
-        :param name: 服务的名称
-        :return: 如果服务类型为 jellyfin，返回 True；否则返回 False。
-        """
-        if not service:
-            service = self.get_service(name=name)
-        return service.type == "jellyfin" if service else False
+        # 判断服务类型是否为指定类型
+        return bool(service and service.type == service_type)
@@ -2,7 +2,7 @@ from typing import Optional

 from app.helper.servicebase import ServiceBaseHelper
 from app.schemas import NotificationConf, ServiceInfo
-from app.schemas.types import SystemConfigKey
+from app.schemas.types import SystemConfigKey, ModuleType


 class NotificationHelper(ServiceBaseHelper[NotificationConf]):
@@ -14,84 +14,25 @@ class NotificationHelper(ServiceBaseHelper[NotificationConf]):
        super().__init__(
            config_key=SystemConfigKey.Notifications,
            conf_type=NotificationConf,
-            modules=[
-                "WechatModule",
-                "WebPushModule",
-                "VoceChatModule",
-                "TelegramModule",
-                "SynologyChatModule",
-                "SlackModule"
-            ]
+            module_type=ModuleType.Notification
        )

-    def is_wechat(self, service: Optional[ServiceInfo] = None, name: Optional[str] = None) -> bool:
+    def is_notification(
+            self,
+            service_type: Optional[str] = None,
+            service: Optional[ServiceInfo] = None,
+            name: Optional[str] = None,
+    ) -> bool:
        """
-        判断指定的消息通知服务是否为 Wechat 类型，需要传入 `service` 或 `name` 中的任一参数
+        通用的消息通知服务类型判断方法

+        :param service_type: 消息通知服务的类型名称（如 'wechat', 'voicechat', 'telegram', 等）
        :param service: 要判断的服务信息
        :param name: 服务的名称
-        :return: 如果服务类型为 wechat，返回 True；否则返回 False。
+        :return: 如果服务类型或实例为指定类型，返回 True；否则返回 False
        """
-        if not service:
-            service = self.get_service(name=name)
-        return service.type == "wechat" if service else False
+        # 如果未提供 service 则通过 name 获取服务
+        service = service or self.get_service(name=name)

-    def is_webpush(self, service: Optional[ServiceInfo] = None, name: Optional[str] = None) -> bool:
-        """
-        判断指定的消息通知服务是否为 WebPush 类型，需要传入 `service` 或 `name` 中的任一参数
-
-        :param service: 要判断的服务信息
-        :param name: 服务的名称
-        :return: 如果服务类型为 webpush，返回 True；否则返回 False。
-        """
-        if not service:
-            service = self.get_service(name=name)
-        return service.type == "webpush" if service else False
-
-    def is_voicechat(self, service: Optional[ServiceInfo] = None, name: Optional[str] = None) -> bool:
-        """
-        判断指定的消息通知服务是否为 VoiceChat 类型，需要传入 `service` 或 `name` 中的任一参数
-
-        :param service: 要判断的服务信息
-        :param name: 服务的名称
-        :return: 如果服务类型为 voicechat，返回 True；否则返回 False。
-        """
-        if not service:
-            service = self.get_service(name=name)
-        return service.type == "voicechat" if service else False
-
-    def is_telegram(self, service: Optional[ServiceInfo] = None, name: Optional[str] = None) -> bool:
-        """
-        判断指定的消息通知服务是否为 Telegram 类型，需要传入 `service` 或 `name` 中的任一参数
-
-        :param service: 要判断的服务信息
-        :param name: 服务的名称
-        :return: 如果服务类型为 telegram，返回 True；否则返回 False。
-        """
-        if not service:
-            service = self.get_service(name=name)
-        return service.type == "telegram" if service else False
-
-    def is_synologychat(self, service: Optional[ServiceInfo] = None, name: Optional[str] = None) -> bool:
-        """
-        判断指定的消息通知服务是否为 SynologyChat 类型，需要传入 `service` 或 `name` 中的任一参数
-
-        :param service: 要判断的服务信息
-        :param name: 服务的名称
-        :return: 如果服务类型为 synologychat，返回 True；否则返回 False。
-        """
-        if not service:
-            service = self.get_service(name=name)
-        return service.type == "synologychat" if service else False
-
-    def is_slack(self, service: Optional[ServiceInfo] = None, name: Optional[str] = None) -> bool:
-        """
-        判断指定的消息通知服务是否为 Slack 类型，需要传入 `service` 或 `name` 中的任一参数
-
-        :param service: 要判断的服务信息
-        :param name: 服务的名称
-        :return: 如果服务类型为 slack，返回 True；否则返回 False。
-        """
-        if not service:
-            service = self.get_service(name=name)
-        return service.type == "slack" if service else False
+        # 判断服务类型是否为指定类型
+        return bool(service and service.type == service_type)
@@ -3,7 +3,7 @@ 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
+from app.schemas.types import SystemConfigKey, ModuleType

 TConf = TypeVar("TConf")

@@ -13,11 +13,11 @@ class ServiceBaseHelper(Generic[TConf]):
    通用服务帮助类，抽象获取配置和服务实例的通用逻辑
    """

-    def __init__(self, config_key: SystemConfigKey, conf_type: Type[TConf], modules: List[str]):
+    def __init__(self, config_key: SystemConfigKey, conf_type: Type[TConf], module_type: ModuleType):
        self.modulemanager = ModuleManager()
        self.config_key = config_key
        self.conf_type = conf_type
-        self.modules = modules
+        self.module_type = module_type

    def get_configs(self, include_disabled: bool = False) -> Dict[str, TConf]:
        """
@@ -47,8 +47,8 @@ class ServiceBaseHelper(Generic[TConf]):
        迭代所有模块的实例及其对应的配置，返回 ServiceInfo 实例
        """
        configs = self.get_configs()
-        for module_name in self.modules:
-            module = self.modulemanager.get_running_module(module_name)
+        modules = self.modulemanager.get_running_type_modules(self.module_type)
+        for module in modules:
            if not module:
                continue
            module_instances = module.get_instances()
@@ -81,9 +81,10 @@ class ServiceBaseHelper(Generic[TConf]):
        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) and
-               (name_filters_set is None or service_info.name in name_filters_set)
+            if service_info.config and (
+                    type_filter is None or service_info.type == type_filter
+            ) and (
+                    name_filters_set is None or service_info.name in name_filters_set)
        }

    def get_service(self, name: str, type_filter: Optional[str] = None) -> Optional[ServiceInfo]: