fix(drift-narrator): fallback 強化 — 標註 K8s 預設值補齊 + 可操作數獨立計算
All checks were successful
CD Pipeline / build-and-deploy (push) Successful in 10m37s

2026-04-18 下午(台北時區)—— ogt + Claude Opus 4.7 (1M)

統帥實彈測試回報: 卡片顯示「securityContext: (未設) → {物件 0 欄位}」毫無意義。

根因: _fallback_items 對「K8s controller 自動補齊空物件」的噪音
     誤當成真實變更輸出。且「還有 29 項」數字包含白名單 + trivial。

修復 3 項:

1. _is_trivial_drift() 新判定函數
   None/空字串/{}/[]/false/0 等互相視為「無實質變更」
   捕捉 K8s controller 自動補齊場景

2. _summarize_item() 替代原本 smart_shorten
   - trivial → "K8s 預設值補齊 (無實質變更)"
   - None → value → "新增 xxx"
   - value → None → "已刪除 (原: xxx)"
   - 其他 → "from → to"

3. _fallback_items() 改進
   - 按 level 排序 (HIGH 優先)
   - 白名單 + HPA allowlist 先過濾

4. _count_nontrivial_drift() + Telegram 呈現
   - 新增「可操作」計數 (去掉白名單 + trivial)
   - 「還有 N 項」用可操作數,不會誤導
   - items 為空時顯示「全為白名單或預設值補齊」

預期效果:
  之前: "... 還有 29 項" (其實只 1 個是真實 drift)
  現在: "... 還有 0 項" 或 "(全部為白名單或預設值補齊)"

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
OG T
2026-04-18 16:29:49 +08:00
parent 1606093dd2
commit f3960f36d2

View File

@@ -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)