MoviePilot/app/application/notification.py

"""
通知服务的配置发现、实例管理与渠道选择。
"""

from typing import Optional

from app.application.service import ServiceBaseHelper, get_service_configs
from app.schemas.system import NotificationConf, NotificationSwitchConf, ServiceInfo
from app.schemas.types import MessageType, ModuleType, SystemConfigKey


class NotificationHelper(ServiceBaseHelper[NotificationConf]):
    """
    提供按持久化配置发现通知服务的能力。
    """

    def __init__(self) -> None:
        """
        绑定通知配置和通知模块类型。
        """
        super().__init__(
            config_key=SystemConfigKey.Notifications,
            conf_type=NotificationConf,
            module_type=ModuleType.Notification,
        )

    def is_notification(
        self,
        service_type: Optional[str] = None,
        service: Optional[ServiceInfo] = None,
        name: Optional[str] = None,
    ) -> bool:
        """
        判断通知服务是否属于指定类型。

        :param service_type: 消息通知服务的类型名称
        :param service: 要判断的服务信息
        :param name: 未传入服务信息时用于查询的服务名称
        :return: 服务存在且类型匹配时返回 True
        """
        service = service or self.get_service(name=name)
        return bool(service and service.type == service_type)


def get_notification_configs(
    include_disabled: bool = False,
) -> list[NotificationConf]:
    """
    返回通知配置列表，并按调用方需要决定是否包含禁用项。
    """
    return list(
        NotificationHelper().get_configs(include_disabled=include_disabled).values()
    )


def get_notification_switch(mtype: MessageType) -> str:
    """
    返回指定通知场景的目标范围，缺少配置时默认仅管理员。
    """
    for switch in get_service_configs(
        SystemConfigKey.NotificationSwitchs,
        NotificationSwitchConf,
    ):
        if switch.type == mtype.value:
            return switch.action or "admin"
    return "admin"