feat(netx-bridge): use English UME alarm text for WhatsApp push

Replace Chinese NetX labels with UME Alarm Raised/Updated/Cleared and
English field names in proactive alert messages.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-06-22 20:37:51 +08:00
parent eaeb9fd33d
commit 538945fb66
2 changed files with 32 additions and 8 deletions

View file

@ -34,7 +34,7 @@ def _default_tenant_id() -> str:
def _format_alarm_text(payload: dict[str, Any]) -> str:
action = str(payload.get("action") or "").strip().lower()
label = str(payload.get("rule_label") or payload.get("native_probable_cause") or "关键告警").strip()
label = str(payload.get("rule_label") or payload.get("native_probable_cause") or "Key alarm").strip()
ne = payload.get("ne") if isinstance(payload.get("ne"), dict) else {}
host = str(ne.get("host_name") or payload.get("host_name") or "").strip()
ip = str(ne.get("ip_address") or "").strip()
@ -42,14 +42,18 @@ def _format_alarm_text(payload: dict[str, Any]) -> str:
device = host or ne_name or str(payload.get("ne_id") or "").strip()
if ip:
device = f"{device} ({ip})" if device else ip
action_label = {"inserted": "新增", "updated": "更新", "deleted": "清除"}.get(action, action or "告警")
action_label = {
"inserted": "Alarm Raised",
"updated": "Alarm Updated",
"deleted": "Alarm Cleared",
}.get(action, action or "Alarm")
lines = [
f"[NetX {action_label}] {label}",
f"设备: {device or '-'}",
f"对象: {str(payload.get('object_name') or '-').strip()}",
f"级别: {str(payload.get('perceived_severity') or '-').strip()}",
f"原因: {str(payload.get('native_probable_cause') or '-').strip()}",
f"时间: {str(payload.get('time_created') or '-').strip()}",
f"[UME {action_label}] {label}",
f"Device: {device or '-'}",
f"Object: {str(payload.get('object_name') or '-').strip()}",
f"Severity: {str(payload.get('perceived_severity') or '-').strip()}",
f"Cause: {str(payload.get('native_probable_cause') or '-').strip()}",
f"Time: {str(payload.get('time_created') or '-').strip()}",
f"notificationId: {str(payload.get('notification_id') or '-').strip()}",
]
return "\n".join(lines)

View file

@ -8,6 +8,7 @@ from unittest.mock import patch
from fastapi.testclient import TestClient
from interfaces.http.fastapi_app import create_app
from interfaces.ws.netx_bridge import _format_alarm_text
from svc.persistence.assistant_store import reset_assistant_store_singleton
@ -22,6 +23,25 @@ class NetxBridgeTests(unittest.TestCase):
def tearDown(self) -> None:
reset_assistant_store_singleton()
def test_format_alarm_text_english(self) -> None:
text = _format_alarm_text(
{
"action": "inserted",
"rule_label": "Fan",
"object_name": "ME{abc},FAN={/module=0}",
"perceived_severity": "major",
"native_probable_cause": "Fan The fan speed level is abnormally high",
"time_created": "2026-06-22T19:29:41.086+07:00",
"notification_id": "1680996323029",
"ne": {"host_name": "LPG-BKM-AN1-ZM3SP", "ip_address": "114.0.24.178"},
}
)
self.assertIn("[UME Alarm Raised] Fan", text)
self.assertIn("Device: LPG-BKM-AN1-ZM3SP (114.0.24.178)", text)
self.assertIn("Severity: major", text)
self.assertNotIn("设备", text)
self.assertNotIn("NetX", text)
def test_netx_bridge_auth_and_alarm_ack(self) -> None:
with self.client.websocket_connect("/ws/netx-bridge") as ws:
ws.send_json({"type": "auth", "token": "bad"})