520 lines
22 KiB
Python
520 lines
22 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""报警看板配置化（issue #52 / PRD 5.5「⑤ 配置化驾驶舱」）。
|
||
|
||
PRD 5.5 的告警面板（``alarm_panel``）在 #50/#51 里被实现为「固定订阅
|
||
``alarm_stream``、套用主题默认告警色」的硬编码组件。本模块把告警面板的
|
||
**展示语义外置为模板配置**（PRD line 152/171：阈值/规则外置 JSON，行业工程师
|
||
在配置台维护），让"换行业只换配置资产、前端代码零改动"这条验收口径也覆盖到
|
||
告警面板。
|
||
|
||
外置的配置点（均由行业模板在配置台维护）：
|
||
|
||
1. **严重度→颜色映射**（``severity_colors``）：P0/P1/P2 三级各自的前景色 /
|
||
背景色 / 图标，覆盖主题默认告警色；驾驶舱红色告警（PRD 5.3 ③ 场景A）
|
||
即由 ``P0`` 的 ``fg`` 决定，**换行业只改这张映射表**。
|
||
2. **告警规则 / 阈值源绑定**（``rules_source`` / ``thresholds_source``）：
|
||
告警面板订阅哪份规则资产（如 ``templates/ti-cl4/impurity-forecast/
|
||
config/alert_rules.template.yaml``）与哪份阈值包——把"看哪条规则"
|
||
也变成配置项，避免把规则 id 写死在前端。
|
||
3. **SOP 联动开关**（``show_sop``）：PRD 场景A「LLM 生成原因+处置建议 →
|
||
值班长确认」，是否在面板里展开处置 SOP（高利害人工确认，PRD line 333）。
|
||
4. **确认 / 静默行为**（``require_ack`` / ``ack_timeout_s`` / ``mute_lower``）：
|
||
关键告警是否强制人工确认、超时升级、是否静默低于某 severity 的提示。
|
||
5. **分组 / 排序 / 最大条数**（``group_by`` / ``sort_by`` / ``max_items``）：
|
||
大屏展示策略外置（与 #51 的 perf 策略互补：这里是"展示多少/怎么排"，
|
||
#51 是"怎么渲染得快"）。
|
||
|
||
设计要点
|
||
--------
|
||
- **零运行时依赖**：与 #50/#51、data-bus、rag-kb 一致，只用标准库；
|
||
配置资产是可 ``json.dumps`` 的纯 dict，便于配置台发布与审计。
|
||
- **声明式 + 强校验**：``validate_alarm_config`` 收集全部字段级错误
|
||
（沿用 #50 ``LayoutValidationResult`` 风格），``load_alarm_config`` 在校验
|
||
不通过时抛 ``AlarmConfigError`` 并携带错误清单，便于配置台「错误列表」展示。
|
||
- **与 #50/#51 解耦**：本模块不 import ``layout`` / ``renderer``（它们尚未合入
|
||
main），避免对未合并分支形成硬依赖；``render_alarm_panel_props`` 仅产出一份
|
||
``alarm_panel`` 组件的 props dict，由 #51 渲染层在 ``_build_props`` 里合并即可。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass, field
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 版本标识（被告警面板配置资产的 $schema 引用）
|
||
# ---------------------------------------------------------------------------
|
||
ALARM_CONFIG_SCHEMA_ID: str = "iAOP-cockpit-alarm-panel-v1"
|
||
ALARM_CONFIG_SCHEMA_VERSION: int = 1
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 合法性集合
|
||
# ---------------------------------------------------------------------------
|
||
# 严重度三级（与 templates/ti-cl4/impurity-forecast/alert_rules.py 的 AlertSeverity 对齐）。
|
||
VALID_SEVERITIES: Tuple[str, ...] = ("P0", "P1", "P2")
|
||
|
||
# severity 排序权重（越大越严重；用于 mute_lower / sort_by=severity 的排序基准）。
|
||
SEVERITY_RANK: Dict[str, int] = {"P0": 3, "P1": 2, "P2": 1}
|
||
|
||
# severity_colors 里每个 severity 必须声明的颜色键。
|
||
_REQUIRED_COLOR_KEYS: Tuple[str, ...] = ("fg", "bg")
|
||
# severity_colors 里允许额外声明的可选颜色键（图标 / 边框）。
|
||
_OPTIONAL_COLOR_KEYS: Tuple[str, ...] = ("icon", "border")
|
||
|
||
# rules_source / thresholds_source 允许的 ``kind`` 取值：
|
||
# - asset : 指向模板仓库内的一份配置资产路径（如 alert_rules.template.yaml）
|
||
# - inline : 内联在配置里（rules_source.data 直接给出规则声明列表）
|
||
VALID_SOURCE_KINDS: Tuple[str, ...] = ("asset", "inline")
|
||
|
||
# 告警面板允许的分组维度（PRD 5.5 能力：可按 severity / 工序 / 规则分组）。
|
||
VALID_GROUP_BY: Tuple[str, ...] = ("severity", "rule", "process")
|
||
|
||
# 告警面板允许的排序键（severity 按严重度；time 按触发时间倒序）。
|
||
VALID_SORT_BY: Tuple[str, ...] = ("severity", "time")
|
||
|
||
# 静默下限：低于该 severity 的告警不展示（默认 P2 = 提示级也展示，即不静默）。
|
||
DEFAULT_MUTE_LOWER: str = "P2"
|
||
|
||
# 单面板最大展示条数（PRD 5.5 大屏策略：超过则折叠 + 计数角标）。
|
||
DEFAULT_MAX_ITEMS: int = 50
|
||
MAX_ALLOWED_ITEMS: int = 500
|
||
|
||
# 确认超时下限（秒）：require_ack=True 时，超时未确认自动升级 severity。
|
||
DEFAULT_ACK_TIMEOUT_S: int = 300
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 异常 / 结果
|
||
# ---------------------------------------------------------------------------
|
||
class AlarmConfigError(ValueError):
|
||
"""告警面板配置校验失败。``load_alarm_config`` 在校验不通过时抛出。
|
||
|
||
``errors`` 收集全部字段级错误，便于配置台「错误列表（行号+原因）」展示，
|
||
沿用 #50 ``LayoutValidationError`` 的多错误聚合风格。
|
||
"""
|
||
|
||
def __init__(self, errors: List[str]):
|
||
super().__init__("; ".join(errors) if errors else "alarm panel config validation failed")
|
||
self.errors: List[str] = list(errors)
|
||
|
||
|
||
@dataclass
|
||
class AlarmConfigValidationResult:
|
||
"""``validate_alarm_config`` 的返回值，区分「是否合法」与「全部错误清单」。"""
|
||
|
||
ok: bool
|
||
errors: List[str] = field(default_factory=list)
|
||
# 校验通过后回填的规范化配置（补默认值后的 dict），便于直接发布/渲染。
|
||
normalized: Optional[Dict[str, Any]] = None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 内存模型（dataclass）
|
||
# ---------------------------------------------------------------------------
|
||
@dataclass
|
||
class SeverityColor:
|
||
"""单个严重度的展示配色（覆盖主题默认告警色）。
|
||
|
||
Attributes:
|
||
severity: P0 / P1 / P2。
|
||
fg: 前景色（告警文本 / 图标颜色；P0 的 fg 即驾驶舱「红色告警」）。
|
||
bg: 背景色（告警条底色）。
|
||
icon: 可选图标名（如 ``"alert-triangle"``）。
|
||
border: 可选左边框色（用于告警条强调）。
|
||
"""
|
||
|
||
severity: str
|
||
fg: str
|
||
bg: str
|
||
icon: Optional[str] = None
|
||
border: Optional[str] = None
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
d: Dict[str, Any] = {"severity": self.severity, "fg": self.fg, "bg": self.bg}
|
||
if self.icon is not None:
|
||
d["icon"] = self.icon
|
||
if self.border is not None:
|
||
d["border"] = self.border
|
||
return d
|
||
|
||
|
||
@dataclass
|
||
class AlarmRulesSource:
|
||
"""告警规则 / 阈值的来源绑定（PRD「阈值外置 JSON」）。
|
||
|
||
Attributes:
|
||
kind: ``asset``（模板仓库内资产路径）或 ``inline``（内联声明）。
|
||
ref: ``asset`` 时的资产路径（相对模板根，如
|
||
``ti-cl4/impurity-forecast/config/alert_rules.template.yaml``）。
|
||
data: ``inline`` 时的规则声明列表（每条是 {id, severity, ...} dict）。
|
||
"""
|
||
|
||
kind: str
|
||
ref: Optional[str] = None
|
||
data: Optional[List[Dict[str, Any]]] = None
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
d: Dict[str, Any] = {"kind": self.kind}
|
||
if self.ref is not None:
|
||
d["ref"] = self.ref
|
||
if self.data is not None:
|
||
d["data"] = list(self.data)
|
||
return d
|
||
|
||
|
||
@dataclass
|
||
class AlarmPanelConfig:
|
||
"""一份告警面板配置的内存模型（对应一份模板级配置资产）。
|
||
|
||
渲染层（Vue3 / 配置台）消费本对象即可驱动 ``AlarmPanel`` 组件的全部展示
|
||
行为：切换行业模板 = 加载另一份 ``AlarmPanelConfig``，**前端代码零改动**。
|
||
"""
|
||
|
||
severity_colors: List[SeverityColor]
|
||
rules_source: AlarmRulesSource
|
||
thresholds_source: Optional[AlarmRulesSource] = None
|
||
show_sop: bool = True
|
||
require_ack: bool = False
|
||
ack_timeout_s: int = DEFAULT_ACK_TIMEOUT_S
|
||
mute_lower: str = DEFAULT_MUTE_LOWER
|
||
group_by: str = "severity"
|
||
sort_by: str = "severity"
|
||
max_items: int = DEFAULT_MAX_ITEMS
|
||
schema: str = ALARM_CONFIG_SCHEMA_ID
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
"""序列化为可发布的配置资产 dict（结构对齐校验器输入）。"""
|
||
d: Dict[str, Any] = {
|
||
"$schema": self.schema,
|
||
"severityColors": [c.to_dict() for c in self.severity_colors],
|
||
"rulesSource": self.rules_source.to_dict(),
|
||
"showSop": self.show_sop,
|
||
"requireAck": self.require_ack,
|
||
"muteLower": self.mute_lower,
|
||
"groupBy": self.group_by,
|
||
"sortBy": self.sort_by,
|
||
"maxItems": self.max_items,
|
||
}
|
||
if self.thresholds_source is not None:
|
||
d["thresholdsSource"] = self.thresholds_source.to_dict()
|
||
if self.require_ack:
|
||
d["ackTimeoutS"] = self.ack_timeout_s
|
||
return d
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 校验器
|
||
# ---------------------------------------------------------------------------
|
||
def _is_str_nonempty(value: Any) -> bool:
|
||
return isinstance(value, str) and value.strip() != ""
|
||
|
||
|
||
def _validate_source(
|
||
source: Any, field_name: str, errors: List[str], required: bool
|
||
) -> Optional[Dict[str, Any]]:
|
||
"""校验一个 rules_source / thresholds_source dict。
|
||
|
||
返回规范化后的 dict（校验通过时），或 None（非法 / 缺失）。
|
||
"""
|
||
ctx = field_name
|
||
if source is None:
|
||
if required:
|
||
errors.append(f"{ctx}: 缺失（告警面板必须绑定 rulesSource）")
|
||
return None
|
||
if not isinstance(source, dict):
|
||
errors.append(f"{ctx}: 必须是对象（dict），实际为 {type(source).__name__}")
|
||
return None
|
||
|
||
kind = source.get("kind")
|
||
if kind not in VALID_SOURCE_KINDS:
|
||
errors.append(
|
||
f"{ctx}.kind: 非法 {kind!r}，合法值 {list(VALID_SOURCE_KINDS)}"
|
||
)
|
||
return None
|
||
|
||
normalized: Dict[str, Any] = {"kind": kind}
|
||
|
||
if kind == "asset":
|
||
ref = source.get("ref")
|
||
if not _is_str_nonempty(ref):
|
||
errors.append(f"{ctx}.ref: kind=asset 时必须给出非空资产路径")
|
||
else:
|
||
normalized["ref"] = ref
|
||
else: # inline
|
||
data = source.get("data")
|
||
if not isinstance(data, list) or not data:
|
||
errors.append(f"{ctx}.data: kind=inline 时必须给出非空规则声明列表")
|
||
else:
|
||
# 每条内联规则至少要有 id（便于驾驶舱引用 / 审计）
|
||
bad = [
|
||
str(i)
|
||
for i, item in enumerate(data)
|
||
if not isinstance(item, dict) or not _is_str_nonempty(item.get("id"))
|
||
]
|
||
if bad:
|
||
errors.append(
|
||
f"{ctx}.data: 内联规则项 {','.join(bad)} 缺失 id 或非对象"
|
||
)
|
||
else:
|
||
normalized["data"] = list(data)
|
||
|
||
return normalized
|
||
|
||
|
||
def validate_alarm_config(config: Any) -> AlarmConfigValidationResult:
|
||
"""对一份告警面板配置资产做结构 + 语义校验，返回校验结果。
|
||
|
||
非法资产不会提前返回：尽量收集全部字段级错误，便于配置台一次性展示
|
||
「错误列表（字段 + 原因）」，与 #50 ``validate_layout`` 行为一致。
|
||
"""
|
||
errors: List[str] = []
|
||
|
||
if not isinstance(config, dict):
|
||
return AlarmConfigValidationResult(
|
||
ok=False, errors=[f"配置必须是对象（dict），实际为 {type(config).__name__}"]
|
||
)
|
||
|
||
# $schema（选填，但若给出必须对齐版本标识）
|
||
schema = config.get("$schema")
|
||
if schema is not None and schema != ALARM_CONFIG_SCHEMA_ID:
|
||
errors.append(
|
||
f"$schema: 应为 {ALARM_CONFIG_SCHEMA_ID!r}，实际为 {schema!r}"
|
||
)
|
||
|
||
# severityColors：必填、至少覆盖 P0/P1/P2 三级、每级颜色键齐全
|
||
raw_colors = config.get("severityColors", config.get("severity_colors"))
|
||
color_by_sev: Dict[str, Dict[str, Any]] = {}
|
||
if not isinstance(raw_colors, list) or not raw_colors:
|
||
errors.append("severityColors: 缺失或非列表（至少需要 P0/P1/P2 三级配色）")
|
||
else:
|
||
for i, item in enumerate(raw_colors):
|
||
ctx = f"severityColors[{i}]"
|
||
if not isinstance(item, dict):
|
||
errors.append(f"{ctx}: 必须是对象（dict）")
|
||
continue
|
||
sev = item.get("severity")
|
||
if sev not in VALID_SEVERITIES:
|
||
errors.append(
|
||
f"{ctx}.severity: 非法 {sev!r}，合法值 {list(VALID_SEVERITIES)}"
|
||
)
|
||
continue
|
||
if sev in color_by_sev:
|
||
errors.append(f"{ctx}.severity: {sev!r} 重复声明")
|
||
continue
|
||
norm_color: Dict[str, Any] = {"severity": sev}
|
||
color_ok = True
|
||
for key in _REQUIRED_COLOR_KEYS:
|
||
v = item.get(key)
|
||
if not _is_str_nonempty(v):
|
||
errors.append(f"{ctx}.{key}: 缺失或非非空字符串")
|
||
color_ok = False
|
||
else:
|
||
norm_color[key] = v
|
||
for key in _OPTIONAL_COLOR_KEYS:
|
||
v = item.get(key)
|
||
if v is None:
|
||
continue
|
||
if not _is_str_nonempty(v):
|
||
errors.append(f"{ctx}.{key}: 给出则必须是非空字符串")
|
||
else:
|
||
norm_color[key] = v
|
||
if color_ok:
|
||
color_by_sev[sev] = norm_color
|
||
|
||
for sev in VALID_SEVERITIES:
|
||
if sev not in color_by_sev:
|
||
errors.append(f"severityColors: 缺少 {sev} 级配色（必须覆盖 P0/P1/P2）")
|
||
|
||
# rulesSource：必填
|
||
raw_rules = config.get("rulesSource", config.get("rules_source"))
|
||
norm_rules = _validate_source(raw_rules, "rulesSource", errors, required=True)
|
||
|
||
# thresholdsSource：选填（阈值可与规则同源，也可独立）
|
||
raw_thr = config.get("thresholdsSource", config.get("thresholds_source"))
|
||
norm_thr = _validate_source(raw_thr, "thresholdsSource", errors, required=False)
|
||
|
||
# showSop：布尔
|
||
show_sop = config.get("showSop", config.get("show_sop", True))
|
||
if not isinstance(show_sop, bool):
|
||
errors.append(f"showSop: 必须是布尔，实际为 {type(show_sop).__name__}")
|
||
|
||
# requireAck：布尔
|
||
require_ack = config.get("requireAck", config.get("require_ack", False))
|
||
if not isinstance(require_ack, bool):
|
||
errors.append(f"requireAck: 必须是布尔，实际为 {type(require_ack).__name__}")
|
||
|
||
# ackTimeoutS：require_ack 时才生效，给出则必须是正整数
|
||
ack_timeout = config.get("ackTimeoutS", config.get("ack_timeout_s", DEFAULT_ACK_TIMEOUT_S))
|
||
if isinstance(ack_timeout, bool) or not isinstance(ack_timeout, int) or ack_timeout <= 0:
|
||
errors.append(f"ackTimeoutS: 必须是正整数（秒），实际为 {ack_timeout!r}")
|
||
|
||
# muteLower：必须是合法 severity
|
||
mute_lower = config.get("muteLower", config.get("mute_lower", DEFAULT_MUTE_LOWER))
|
||
if mute_lower not in VALID_SEVERITIES:
|
||
errors.append(
|
||
f"muteLower: 非法 {mute_lower!r}，合法值 {list(VALID_SEVERITIES)}"
|
||
)
|
||
|
||
# groupBy / sortBy：枚举
|
||
group_by = config.get("groupBy", config.get("group_by", "severity"))
|
||
if group_by not in VALID_GROUP_BY:
|
||
errors.append(f"groupBy: 非法 {group_by!r}，合法值 {list(VALID_GROUP_BY)}")
|
||
sort_by = config.get("sortBy", config.get("sort_by", "severity"))
|
||
if sort_by not in VALID_SORT_BY:
|
||
errors.append(f"sortBy: 非法 {sort_by!r}，合法值 {list(VALID_SORT_BY)}")
|
||
|
||
# maxItems：1..MAX_ALLOWED_ITEMS
|
||
max_items = config.get("maxItems", config.get("max_items", DEFAULT_MAX_ITEMS))
|
||
if (
|
||
isinstance(max_items, bool)
|
||
or not isinstance(max_items, int)
|
||
or not (1 <= max_items <= MAX_ALLOWED_ITEMS)
|
||
):
|
||
errors.append(
|
||
f"maxItems: 必须是 1..{MAX_ALLOWED_ITEMS} 的整数，实际为 {max_items!r}"
|
||
)
|
||
|
||
if errors:
|
||
return AlarmConfigValidationResult(ok=False, errors=errors)
|
||
|
||
# 规范化输出（统一字段名 + 补默认值），便于直接发布 / 渲染。
|
||
normalized: Dict[str, Any] = {
|
||
"$schema": ALARM_CONFIG_SCHEMA_ID,
|
||
"severityColors": [color_by_sev[sev] for sev in VALID_SEVERITIES],
|
||
"rulesSource": norm_rules,
|
||
"showSop": show_sop,
|
||
"requireAck": require_ack,
|
||
"ackTimeoutS": ack_timeout if require_ack else DEFAULT_ACK_TIMEOUT_S,
|
||
"muteLower": mute_lower,
|
||
"groupBy": group_by,
|
||
"sortBy": sort_by,
|
||
"maxItems": max_items,
|
||
}
|
||
if norm_thr is not None:
|
||
normalized["thresholdsSource"] = norm_thr
|
||
|
||
return AlarmConfigValidationResult(ok=True, errors=[], normalized=normalized)
|
||
|
||
|
||
def load_alarm_config(config: Any) -> AlarmPanelConfig:
|
||
"""从已解析的 dict 构造 ``AlarmPanelConfig``（校验失败抛 ``AlarmConfigError``）。
|
||
|
||
与 #50 ``load_layout`` 对称：先 ``validate_alarm_config``，再按规范化结果
|
||
构造内存模型；校验不通过时把全部错误聚合到异常里。
|
||
"""
|
||
result = validate_alarm_config(config)
|
||
if not result.ok:
|
||
raise AlarmConfigError(result.errors)
|
||
assert result.normalized is not None # 校验通过一定回填 normalized
|
||
|
||
norm = result.normalized
|
||
|
||
severity_colors = [
|
||
SeverityColor(
|
||
severity=c["severity"],
|
||
fg=c["fg"],
|
||
bg=c["bg"],
|
||
icon=c.get("icon"),
|
||
border=c.get("border"),
|
||
)
|
||
for c in norm["severityColors"]
|
||
]
|
||
|
||
def _to_source(d: Dict[str, Any]) -> AlarmRulesSource:
|
||
return AlarmRulesSource(
|
||
kind=d["kind"],
|
||
ref=d.get("ref"),
|
||
data=d.get("data"),
|
||
)
|
||
|
||
thresholds_source = None
|
||
if norm.get("thresholdsSource") is not None:
|
||
thresholds_source = _to_source(norm["thresholdsSource"])
|
||
|
||
return AlarmPanelConfig(
|
||
severity_colors=severity_colors,
|
||
rules_source=_to_source(norm["rulesSource"]),
|
||
thresholds_source=thresholds_source,
|
||
show_sop=norm["showSop"],
|
||
require_ack=norm["requireAck"],
|
||
ack_timeout_s=norm["ackTimeoutS"],
|
||
mute_lower=norm["muteLower"],
|
||
group_by=norm["groupBy"],
|
||
sort_by=norm["sortBy"],
|
||
max_items=norm["maxItems"],
|
||
schema=ALARM_CONFIG_SCHEMA_ID,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 渲染辅助：把告警面板配置编译成 alarm_panel 组件的 props
|
||
# ---------------------------------------------------------------------------
|
||
def render_alarm_panel_props(config: AlarmPanelConfig) -> Dict[str, Any]:
|
||
"""把一份 ``AlarmPanelConfig`` 编译成 ``alarm_panel`` 组件的 props dict。
|
||
|
||
#51 渲染层在 ``_build_props`` 里对 ``alarm_panel`` 原本只写死
|
||
``subscribe="alarm_stream"``；合入本模块后改为合并本函数的输出即可，
|
||
前端 ``AlarmPanel`` 组件据此驱动「颜色 / 规则源 / SOP / 确认 / 分组排序」，
|
||
**切换行业模板只换配置资产，前端代码零改动**（PRD 5.5 验收口径）。
|
||
|
||
返回的 props 与 #51 的 props 风格一致：扁平、可直接 ``json.dumps``、
|
||
键名用前端友好的 camelCase。
|
||
"""
|
||
# severity → 展示描述（颜色 + 可选图标 / 边框），驱动告警条着色
|
||
severity_styles: Dict[str, Dict[str, Any]] = {}
|
||
for c in config.severity_colors:
|
||
style: Dict[str, Any] = {"fg": c.fg, "bg": c.bg}
|
||
if c.icon is not None:
|
||
style["icon"] = c.icon
|
||
if c.border is not None:
|
||
style["border"] = c.border
|
||
severity_styles[c.severity] = style
|
||
|
||
# 规则 / 阈值源：asset 透传 ref，inline 透传 data
|
||
def _source_props(src: AlarmRulesSource) -> Dict[str, Any]:
|
||
if src.kind == "asset":
|
||
return {"kind": "asset", "ref": src.ref or ""}
|
||
return {"kind": "inline", "data": list(src.data or [])}
|
||
|
||
props: Dict[str, Any] = {
|
||
# 保留 #51 原有的订阅语义（向后兼容）
|
||
"subscribe": "alarm_stream",
|
||
"severityStyles": severity_styles,
|
||
"rulesSource": _source_props(config.rules_source),
|
||
"showSop": config.show_sop,
|
||
"requireAck": config.require_ack,
|
||
"muteLower": config.mute_lower,
|
||
"groupBy": config.group_by,
|
||
"sortBy": config.sort_by,
|
||
"maxItems": config.max_items,
|
||
}
|
||
if config.thresholds_source is not None:
|
||
props["thresholdsSource"] = _source_props(config.thresholds_source)
|
||
if config.require_ack:
|
||
props["ackTimeoutS"] = config.ack_timeout_s
|
||
|
||
return props
|
||
|
||
|
||
def filter_alarms_by_mute(
|
||
alarms: List[Dict[str, Any]],
|
||
mute_lower: str = DEFAULT_MUTE_LOWER,
|
||
) -> List[Dict[str, Any]]:
|
||
"""按 ``mute_lower`` 过滤告警列表（演示配置如何驱动展示策略）。
|
||
|
||
给定一批告警 dict（每条含 ``severity``），返回严重度 ≥ ``mute_lower`` 的子集。
|
||
用于验证「静默下限」配置点真正生效——配置台改 ``muteLower`` 即可改变面板
|
||
展示内容，无需改前端代码。``sort_by=severity`` 时同时按严重度降序排序。
|
||
"""
|
||
if mute_lower not in SEVERITY_RANK:
|
||
raise AlarmConfigError([f"mute_lower 非法: {mute_lower!r}"])
|
||
floor = SEVERITY_RANK[mute_lower]
|
||
kept = [
|
||
a for a in alarms
|
||
if isinstance(a, dict) and SEVERITY_RANK.get(a.get("severity"), 0) >= floor
|
||
]
|
||
return sorted(
|
||
kept,
|
||
key=lambda a: SEVERITY_RANK.get(a.get("severity"), 0),
|
||
reverse=True,
|
||
)
|