fix(api): 將主機資源告警收斂成脫敏事件卡

apps/api/src/services/telegram_gateway.py

@@ -133,6 +133,10 @@ def _resource_load_bar(load_text: str) -> str:
 
 def _compact_host_process_command(command: str) -> str:
     lowered = command.lower()
+    if "prisma" in lowered and "generate" in lowered:
+        return "Prisma generate"
+    if "prisma" in lowered and ("build/index.js" in lowered or "prisma/build" in lowered):
+        return "Prisma CLI child"
     if "jest-worker" in lowered or "processchild.js" in lowered:
         return "Next.js build worker"
     if "next build" in lowered:
@@ -145,7 +149,11 @@ def _compact_host_process_command(command: str) -> str:
         return "npm run build"
     if "node " in lowered:
         return "Node.js process"
-    return command.rsplit("/", 1)[-1][:80]
+    summary = re.sub(r"https?://\S+", "[redacted-url]", command)
+    summary = re.sub(r"\{.*", "", summary)
+    summary = re.sub(r"/(?:[\w@.+-]+/)+", "", summary)
+    summary = re.sub(r"\s+", " ", summary).strip()
+    return (summary.rsplit("/", 1)[-1] or "process")[:80]
 
 
 def _parse_host_process_lines(text: str) -> list[dict[str, str | float]]:
@@ -164,6 +172,7 @@ def _parse_host_process_lines(text: str) -> list[dict[str, str | float]]:
         processes.append(
             {
                 "pid": match.group("pid"),
+                "user": match.group("user"),
                 "cpu": cpu,
                 "command": _compact_host_process_command(command),
             }
@@ -191,6 +200,8 @@ def _host_resource_alert_impact(
         cpu = 0.0
 
     commands = " ".join(str(item.get("command", "")).lower() for item in processes)
+    if "prisma" in commands:
+        return "Prisma generate / package build 程序正在吃 CPU；先確認 CI/CD、runner job 與供應鏈來源，不可直接 kill"
     if "build" in commands:
         return "建置程序正在吃 CPU；先確認 CI/CD 或 runner 排程，不可直接重啟服務"
     if load >= 8 or cpu >= 90:
@@ -213,6 +224,11 @@ def _host_resource_automation_lane(
             "建立 CI load evidence packet，彙整 Gitea Actions run、runner queue、load/core 與 swap；不 kill process",
         )
     commands = " ".join(str(item.get("command", "")).lower() for item in processes)
+    if "prisma" in commands:
+        return (
+            "runner_prisma_generate_resource_pressure",
+            "建立 Prisma / package install 資源壓力候選，交由 AI 彙整 CI/CD run、套件來源、runner queue 與供應鏈 evidence",
+        )
     if "build" in commands:
         return (
             "runner_build_resource_pressure",
@@ -247,11 +263,16 @@ def _host_resource_alert_evidence_lines(
         lines: list[str] = []
         for item in processes[:3]:
             process_cpu = f"{float(item['cpu']):g}%"
+            user_suffix = (
+                " root"
+                if str(item.get("user", "")).lower() == "root"
+                else ""
+            )
             lines.append(
                 "├ "
                 f"<code>PID {html.escape(str(item['pid']))}</code> "
                 f"CPU <code>{html.escape(process_cpu)}</code>："
-                f"<code>{html.escape(str(item['command']))}</code>"
+                f"<code>{html.escape(str(item['command']))}{html.escape(user_suffix)}</code>"
             )
         lines[-1] = "└" + lines[-1][1:]
         return lines
@@ -286,12 +307,28 @@ def _host_resource_recommendation_lines(text: str) -> list[str]:
             "└ 若 CI 卡死，產出 owner packet 與 runner cleanup dry-run，再進維護窗口",
         ]
     return [
-        "├ 確認是否為 CI/CD / Actions / runner 正常建置窗口",
-        "├ 若持續超過門檻，先查 runner queue、build job、容器資源限制與服務 SLO",
+        "├ 確認是否為 CI/CD / Actions / runner 正常建置或 package install 窗口",
+        "├ 若持續超過門檻，先查 runner queue、build job、套件來源、容器資源限制與服務 SLO",
         "└ 同一 host/service 5 分鐘聚合一次，避免洗版",
     ]
 
 
+def _host_resource_alert_severity(load_text: str, cpu_text: str) -> str:
+    try:
+        load = float(load_text)
+    except (TypeError, ValueError):
+        load = 0.0
+    try:
+        cpu = float(cpu_text)
+    except (TypeError, ValueError):
+        cpu = 0.0
+    if load >= 8 or cpu >= 90:
+        return "P1"
+    if load >= 4 or cpu >= 70:
+        return "P2"
+    return "P3"
+
+
 def format_host_resource_alert_card(text: str) -> str:
     """把 host CPU/load raw dump 轉成值班者可讀的 Telegram HTML 卡。"""
     if not _is_host_resource_alert_text(text):
@@ -314,19 +351,23 @@ def format_host_resource_alert_card(text: str) -> str:
     impact = _host_resource_alert_impact(text, cpu, load, processes)
     automation_lane, automation_next_step = _host_resource_automation_lane(text, processes)
     load_bar = _resource_load_bar(load)
-    severity = "🔴" if load != "-" and load_bar.count("■") >= 7 else "⚠️"
+    severity = _host_resource_alert_severity(load, cpu)
+    root_process_count = sum(
+        1 for item in processes if str(item.get("user", "")).lower() == "root"
+    )
     evidence_lines = _host_resource_alert_evidence_lines(text, processes)
     recommendation_lines = _host_resource_recommendation_lines(text)
 
     return "\n".join(
         [
-            f"{severity} <b>主機資源告警｜{html.escape(target)}</b>",
+            f"<b>{severity} 主機資源壓力｜{html.escape(target)}</b>",
             "<code>ai_automation_alert_card_v1</code>",
             "",
-            "<b>影響判讀</b>",
+            "<b>一眼摘要</b>",
             f"├ CPU 使用率：<code>{html.escape(cpu)}%</code>",
             f"├ Load：<code>{html.escape(load)}</code> <code>{load_bar}</code>",
-            f"└ 狀態：<b>{html.escape(impact)}</b>",
+            f"├ 容器 root 進程：<code>{root_process_count}</code>",
+            f"└ 影響：<b>{html.escape(impact)}</b>",
             "",
             "<b>AI 自動化判讀</b>",
             f"├ Lane：<code>{html.escape(automation_lane)}</code>",
@@ -345,6 +386,13 @@ def format_host_resource_alert_card(text: str) -> str:
     )
 
 
+def normalize_alert_notification_payload(text: str, parse_mode: str) -> tuple[str, str]:
+    """最後出口統一脫敏監控告警；host 類 raw dump 一律改成 HTML 事件卡。"""
+    if _is_host_resource_alert_text(text):
+        return format_host_resource_alert_card(text), "HTML"
+    return text, parse_mode
+
+
 def _top_gateway_bucket(
     buckets: list[dict[str, object]],
     field: str,
@@ -8222,15 +8270,14 @@ class TelegramGateway:
         reply_markup: dict | None = None,
     ) -> dict:
         """發送告警型純文字通知到 SRE 戰情室群組。"""
-        safe_text = (
-            format_host_resource_alert_card(text)
-            if parse_mode.upper() == "HTML"
-            else text
+        safe_text, effective_parse_mode = normalize_alert_notification_payload(
+            text,
+            parse_mode,
         )
         payload: dict = {
             "chat_id": self.alert_chat_id,
             "text": safe_text[:4096],
-            "parse_mode": parse_mode,
+            "parse_mode": effective_parse_mode,
         }
         if reply_markup:
             payload["reply_markup"] = reply_markup
@@ -8263,10 +8310,14 @@ class TelegramGateway:
         Returns:
             dict: Telegram API 回應
         """
+        safe_text, effective_parse_mode = normalize_alert_notification_payload(
+            text,
+            parse_mode,
+        )
         payload: dict = {
             "chat_id": chat_id or self.alert_chat_id,
-            "text": text[:4096],
-            "parse_mode": parse_mode,
+            "text": safe_text[:4096],
+            "parse_mode": effective_parse_mode,
             "disable_web_page_preview": disable_web_page_preview,
         }
         return await self._send_request("sendMessage", payload)