""" 通知服务的配置发现、实例管理与渠道选择。 """ 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"