Merge pull request #2796 from InfinityPacer/feature/module

app/helper/downloader.py

@@ -1,5 +1,7 @@
+from typing import Optional
+
 from app.helper.servicebase import ServiceBaseHelper
-from app.schemas import DownloaderConf
+from app.schemas import DownloaderConf, ServiceInfo
 from app.schemas.types import SystemConfigKey
 
 
@@ -14,3 +16,27 @@ def __init__(self):
             conf_type=DownloaderConf,
             modules=["QbittorrentModule", "TransmissionModule"]
         )
+
+    def is_qbittorrent(self, service: Optional[ServiceInfo] = None, name: Optional[str] = None) -> bool:
+        """
+        判断指定的下载器是否为 qbittorrent 类型，需要传入 `service` 或 `name` 中的任一参数
+
+        :param service: 要判断的服务信息
+        :param name: 服务的名称
+        :return: 如果服务类型为 qbittorrent，返回 True；否则返回 False。
+        """
+        if not service:
+            service = self.get_service(name=name)
+        return service.type == "qbittorrent" if service else False
+
+    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

app/helper/mediaserver.py

@@ -1,5 +1,7 @@
+from typing import Optional
+
 from app.helper.servicebase import ServiceBaseHelper
-from app.schemas import MediaServerConf
+from app.schemas import MediaServerConf, ServiceInfo
 from app.schemas.types import SystemConfigKey
 
 
@@ -14,3 +16,39 @@ def __init__(self):
             conf_type=MediaServerConf,
             modules=["PlexModule", "EmbyModule", "JellyfinModule"]
         )
+
+    def is_plex(self, service: Optional[ServiceInfo] = None, name: Optional[str] = None) -> bool:
+        """
+        判断指定的媒体服务器是否为 Plex 类型，需要传入 `service` 或 `name` 中的任一参数
+
+        :param service: 要判断的服务信息
+        :param name: 服务的名称
+        :return: 如果服务类型为 plex，返回 True；否则返回 False。
+        """
+        if not service:
+            service = self.get_service(name=name)
+        return service.type == "plex" if service else False
+
+    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

app/helper/notification.py

@@ -1,5 +1,7 @@
+from typing import Optional
+
 from app.helper.servicebase import ServiceBaseHelper
-from app.schemas import NotificationConf
+from app.schemas import NotificationConf, ServiceInfo
 from app.schemas.types import SystemConfigKey
 
 
@@ -12,6 +14,84 @@ def __init__(self):
         super().__init__(
             config_key=SystemConfigKey.Notifications,
             conf_type=NotificationConf,
-            modules=["WechatModule", "WebPushModule", "VoceChatModule", "TelegramModule", "SynologyChatModule",
-                     "SlackModule"]
+            modules=[
+                "WechatModule",
+                "WebPushModule",
+                "VoceChatModule",
+                "TelegramModule",
+                "SynologyChatModule",
+                "SlackModule"
+            ]
         )
+
+    def is_wechat(self, service: Optional[ServiceInfo] = None, name: Optional[str] = None) -> bool:
+        """
+        判断指定的消息通知服务是否为 Wechat 类型，需要传入 `service` 或 `name` 中的任一参数
+
+        :param service: 要判断的服务信息
+        :param name: 服务的名称
+        :return: 如果服务类型为 wechat，返回 True；否则返回 False。
+        """
+        if not service:
+            service = self.get_service(name=name)
+        return service.type == "wechat" if service else False
+
+    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