mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-09-05 23:47:41 +08:00
feat(module): add type-checking methods
This commit is contained in:
@@ -1,5 +1,7 @@
|
|||||||
|
from typing import Optional
|
||||||
|
|
||||||
from app.helper.servicebase import ServiceBaseHelper
|
from app.helper.servicebase import ServiceBaseHelper
|
||||||
from app.schemas import DownloaderConf
|
from app.schemas import DownloaderConf, ServiceInfo
|
||||||
from app.schemas.types import SystemConfigKey
|
from app.schemas.types import SystemConfigKey
|
||||||
|
|
||||||
|
|
||||||
@@ -14,3 +16,27 @@ class DownloaderHelper(ServiceBaseHelper[DownloaderConf]):
|
|||||||
conf_type=DownloaderConf,
|
conf_type=DownloaderConf,
|
||||||
modules=["QbittorrentModule", "TransmissionModule"]
|
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
|
||||||
|
|||||||
@@ -1,5 +1,7 @@
|
|||||||
|
from typing import Optional
|
||||||
|
|
||||||
from app.helper.servicebase import ServiceBaseHelper
|
from app.helper.servicebase import ServiceBaseHelper
|
||||||
from app.schemas import MediaServerConf
|
from app.schemas import MediaServerConf, ServiceInfo
|
||||||
from app.schemas.types import SystemConfigKey
|
from app.schemas.types import SystemConfigKey
|
||||||
|
|
||||||
|
|
||||||
@@ -14,3 +16,39 @@ class MediaServerHelper(ServiceBaseHelper[MediaServerConf]):
|
|||||||
conf_type=MediaServerConf,
|
conf_type=MediaServerConf,
|
||||||
modules=["PlexModule", "EmbyModule", "JellyfinModule"]
|
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
|
||||||
|
|||||||
@@ -1,5 +1,7 @@
|
|||||||
|
from typing import Optional
|
||||||
|
|
||||||
from app.helper.servicebase import ServiceBaseHelper
|
from app.helper.servicebase import ServiceBaseHelper
|
||||||
from app.schemas import NotificationConf
|
from app.schemas import NotificationConf, ServiceInfo
|
||||||
from app.schemas.types import SystemConfigKey
|
from app.schemas.types import SystemConfigKey
|
||||||
|
|
||||||
|
|
||||||
@@ -12,6 +14,84 @@ class NotificationHelper(ServiceBaseHelper[NotificationConf]):
|
|||||||
super().__init__(
|
super().__init__(
|
||||||
config_key=SystemConfigKey.Notifications,
|
config_key=SystemConfigKey.Notifications,
|
||||||
conf_type=NotificationConf,
|
conf_type=NotificationConf,
|
||||||
modules=["WechatModule", "WebPushModule", "VoceChatModule", "TelegramModule", "SynologyChatModule",
|
modules=[
|
||||||
"SlackModule"]
|
"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
|
||||||
|
|||||||
Reference in New Issue
Block a user