diff --git a/apps/api/src/services/drift_narrator_service.py b/apps/api/src/services/drift_narrator_service.py index 1806f398..36e92104 100644 --- a/apps/api/src/services/drift_narrator_service.py +++ b/apps/api/src/services/drift_narrator_service.py @@ -436,28 +436,73 @@ class DriftNarratorService: if val is None: return "(未設)" s = str(val) - # 嘗試判斷是不是 JSON 字串 + if s in ("{}", "[]"): + return "空" if s.startswith("[") and s.endswith("]"): return f"[清單 {s.count(',')+1 if s != '[]' else 0} 項]" if s.startswith("{") and s.endswith("}"): - # 粗估欄位數 return f"{{物件 {s.count(':')} 欄位}}" if len(s) > 40: return s[:37] + "..." return s + def _is_trivial_drift(self, git_val, actual_val) -> bool: + """ + 判斷是否為 K8s controller 自動補齊的噪音 + (例: None ↔ {} / None ↔ [] / {} ↔ [] 等視為無實質變更) + """ + def _is_empty(v): + if v is None: + return True + s = str(v).strip() + return s in ("", "{}", "[]", "null", "None", "false", "False", "0") + return _is_empty(git_val) and _is_empty(actual_val) + + def _summarize_item(self, item) -> str: + """ + 生成一筆 drift 的人話摘要 (fallback 用) + - 空 vs 空 → 標註為 controller 自動補齊 + - None → 新增 → 顯示新值摘要 + - 有值 → 有值 → 顯示前後變化 + """ + git_val = item.git_value + actual_val = item.actual_value + + if self._is_trivial_drift(git_val, actual_val): + return "K8s 預設值補齊 (無實質變更)" + + from_val = self._smart_shorten(git_val) + to_val = self._smart_shorten(actual_val) + + # None → 有值: 新增 + if git_val is None and actual_val is not None: + return f"新增 {to_val}" + # 有值 → None: 刪除 + if git_val is not None and actual_val is None: + return f"已刪除 (原: {from_val})" + # 一般變化 + return f"{from_val} → {to_val}" + def _fallback_items(self, report: "DriftReport") -> list[dict]: - """LLM 失敗時的 Python 智能摘要(取代舊 str()[:30])""" + """ + LLM 失敗時的 Python 智能摘要 (取代舊 str()[:30]) + - 過濾白名單 + - 優先 HIGH + - trivial drift 標註為「預設值補齊」 + """ + # 按 level 排序 (HIGH 優先) 並過濾白名單 + filtered = [ + it for it in report.items + if not it.is_allowlisted and it.field_path not in _HPA_ALLOWLIST_PATHS + ] + filtered.sort(key=lambda x: 0 if x.drift_level.value == "high" else 1) + items = [] - for item in report.items[:5]: - if item.is_allowlisted or item.field_path in _HPA_ALLOWLIST_PATHS: - continue - from_val = self._smart_shorten(item.git_value) - to_val = self._smart_shorten(item.actual_value) + for item in filtered[:5]: items.append({ "level": item.drift_level.value, "field": item.field_path[:60], - "summary": f"{from_val} → {to_val}", + "summary": self._summarize_item(item), }) return items @@ -518,6 +563,20 @@ class DriftNarratorService: except Exception as e: logger.warning("drift_narrator_telegram_error", error=str(e)) + def _count_nontrivial_drift(self, report: "DriftReport") -> int: + """ + 計算非白名單、非 trivial (K8s 自動補齊) 的 drift 數 + 用於 Telegram 底部「還有 N 項」顯示實際可操作數量 + """ + n = 0 + for item in report.items: + if item.is_allowlisted or item.field_path in _HPA_ALLOWLIST_PATHS: + continue + if self._is_trivial_drift(item.git_value, item.actual_value): + continue + n += 1 + return n + def _render_telegram_body( self, report: "DriftReport", @@ -538,15 +597,21 @@ class DriftNarratorService: ... 還有 27 項 """ lines = [f"🤖 AI 研判\n{narrative}\n"] - lines.append(f"📊 漂移明細 (HIGH: {report.high_count} | MEDIUM: {report.medium_count})") - for it in items: - emoji = "🔴" if it.get("level") == "high" else "🟡" - lines.append(f"{emoji} {it['field']}: {it['summary']}") - total = report.high_count + report.medium_count + # 用非 trivial + 非白名單 的實際可操作數顯示 + actionable = self._count_nontrivial_drift(report) + lines.append(f"📊 漂移明細 (HIGH: {report.high_count} | MEDIUM: {report.medium_count} | 可操作: {actionable})") + + if not items: + lines.append(" (全部為白名單或 K8s 預設值補齊,無實質變更)") + else: + for it in items: + emoji = "🔴" if it.get("level") == "high" else "🟡" + lines.append(f"{emoji} {it['field']}: {it['summary']}") + shown = len(items) - if total > shown: - lines.append(f"... 還有 {total - shown} 項 (按 🔍 查看 Diff)") + if actionable > shown: + lines.append(f"... 還有 {actionable - shown} 項 (按 🔍 查看 Diff)") return "\n".join(lines)