diff --git a/app/helper/downloader.py b/app/helper/downloader.py index 6ffcab13..56fe7751 100644 --- a/app/helper/downloader.py +++ b/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 @@ class DownloaderHelper(ServiceBaseHelper[DownloaderConf]): 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 diff --git a/app/helper/mediaserver.py b/app/helper/mediaserver.py index 1c17eaa0..f1178282 100644 --- a/app/helper/mediaserver.py +++ b/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 @@ class MediaServerHelper(ServiceBaseHelper[MediaServerConf]): 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 diff --git a/app/helper/notification.py b/app/helper/notification.py index e5fcd280..f0cd70f0 100644 --- a/app/helper/notification.py +++ b/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 @@ class NotificationHelper(ServiceBaseHelper[NotificationConf]): 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