refactor: unify service configuration boundary

app/application/mediaserver.py

@@ -304,3 +304,12 @@ class MediaServerHelper(ServiceBaseHelper[MediaServerConf]):
         """判断给定服务或服务名称是否属于指定媒体服务器类型。"""
         service = service or self.get_service(name=name)
         return bool(service and service.type == service_type)
+
+
+def get_mediaserver_configs(
+    include_disabled: bool = False,
+) -> list[MediaServerConf]:
+    """返回媒体服务器配置列表，并按调用方需要决定是否包含禁用项。"""
+    return list(
+        MediaServerHelper().get_configs(include_disabled=include_disabled).values()
+    )

app/application/notification.py

@@ -1,9 +1,9 @@
 from typing import Optional
 
-from app.application.service import ServiceBaseHelper
-from app.schemas.system import NotificationConf
+from app.application.service import ServiceBaseHelper, get_service_configs
+from app.schemas.system import NotificationConf, NotificationSwitchConf
 from app.schemas.system import ServiceInfo
-from app.schemas.types import ModuleType, SystemConfigKey
+from app.schemas.types import MessageType, ModuleType, SystemConfigKey
 
 
 class NotificationHelper(ServiceBaseHelper[NotificationConf]):
@@ -33,3 +33,23 @@ class NotificationHelper(ServiceBaseHelper[NotificationConf]):
         """
         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) -> Optional[str]:
+    """返回指定通知场景的目标范围。"""
+    for switch in get_service_configs(
+        SystemConfigKey.NotificationSwitchs,
+        NotificationSwitchConf,
+    ):
+        if switch.type == mtype.value:
+            return switch.action
+    return None

app/application/service.py

@@ -41,6 +41,14 @@ def configure_service_directory(
     _module_loader = modules
 
 
+def get_service_configs(
+    config_key: SystemConfigKey,
+    conf_type: Type[TConf],
+) -> list[TConf]:
+    """通过组合根登记的读取器返回已校验服务配置。"""
+    return _config_loader(config_key, conf_type)
+
+
 class ServiceBaseHelper(Generic[TConf]):
     """通过应用端口查询服务配置和对应运行实例。"""
 
@@ -57,7 +65,7 @@ class ServiceBaseHelper(Generic[TConf]):
 
     def get_configs(self, include_disabled: bool = False) -> Dict[str, TConf]:
         """返回按名称索引的有效服务配置。"""
-        configs = _config_loader(self.config_key, self.conf_type)
+        configs = get_service_configs(self.config_key, self.conf_type)
         return {
             config.name: config
             for config in configs