1606093dd2
Some checks failed
CD Pipeline / build-and-deploy (push) Has been cancelled
2026-04-18 下午（台北時區）—— ogt + Claude Opus 4.7 (1M)
Live-fire test (report_id=80a34b58) 暴露兩個 bug:
## Bug 1: LLM JSON 被 NEMOTRON wrapper 吞掉
根因: openclaw.call() 經 NEMOTRON 路由時強制回 {description,...} 結構,
我的 prompt 要 {narrative, items} 無法穿透。
(同 1ff3405 早前碰過的 JSON 裸奔問題根源)
修復: 三路 fallback 解析
- Path 1: 直接我們的 {narrative, items}（Ollama 或 LLM 守規矩）
- Path 2: NEMOTRON wrapper,description 巢狀 JSON 含我們結構
- Path 3: description 是純敘述 → 當 narrative + Python fallback_items
## Bug 2: tags 參數 asyncpg DataError
根因: 傳 '{drift,type4d,llm_summary}' 字面量字串,asyncpg 要求 Python list
'(a sized iterable container expected (got type str))'
修復: tags 改傳 ['drift','type4d','llm_summary'] Python list,移除 CAST AS text[]
asyncpg 自動推斷 text[]
Live-fire 結果驗證:
- narrative ✅ 生成(fallback path)
- items ⚠️ 只 1 筆(NEMOTRON 未吐我們結構)
- DB write ❌ tags 型別錯
- Telegram ✅ 送出(雖 fallback 內容但視覺 OK)
本 commit 後預期:
- LLM 回應走 Path 2/3 → narrative + Python fallback items(5 筆 smart summary)
- DB write 成功 → automation_operation_log + ai_collaboration_trace 皆有記錄
- 若 LLM 未來學會走 Path 1(給我們 {narrative, items}),自動升級
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
566 lines
22 KiB
Python
566 lines
22 KiB
Python
"""
|
||
Drift Narrator Service - Phase 30
|
||
===================================
|
||
職責：將 DriftReport 轉為繁體中文人話，推送 Telegram
|
||
|
||
設計邊界：
|
||
- 只負責「敘述」，不做分析、不生成修復指令
|
||
- 觸發條件：high_count > 0 or medium_count > 2
|
||
- 模型：qwen2.5:7b-instruct (Ollama 111, 90s timeout)
|
||
- Redis 快取：drift_narrative:{report_id} TTL 1h，避免重複推送
|
||
- HPA replicas 自動調整在白名單，不觸發摘要
|
||
|
||
版本: v1.0
|
||
建立: 2026-04-10 (台北時區)
|
||
建立者: Claude Code (Phase 30 ADR-067)
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import TYPE_CHECKING
|
||
|
||
import structlog
|
||
|
||
from src.core.redis_client import get_redis
|
||
from src.services.model_registry import get_model
|
||
from src.services.openclaw import get_openclaw
|
||
|
||
if TYPE_CHECKING:
|
||
from src.models.drift import DriftInterpretation, DriftReport
|
||
|
||
logger = structlog.get_logger(__name__)
|
||
|
||
# ============================================================
|
||
# 設定
|
||
# ============================================================
|
||
OLLAMA_URL = "http://192.168.0.111:11434"
|
||
# D1 集中化 2026-04-11: 從 models.json providers.ollama.models.drift_summary 讀取
|
||
NARRATOR_MODEL = get_model("ollama", "drift_summary")
|
||
NARRATOR_TIMEOUT = 90.0 # seconds
|
||
CACHE_TTL = 3600 # 1 小時
|
||
CACHE_PREFIX = "drift_narrative:"
|
||
|
||
# HPA 自動調整白名單 field_path，不納入敘述
|
||
_HPA_ALLOWLIST_PATHS = {
|
||
"spec.replicas",
|
||
}
|
||
|
||
# 觸發條件
|
||
TRIGGER_HIGH_MIN = 1
|
||
TRIGGER_MEDIUM_MIN = 3
|
||
|
||
|
||
# ============================================================
|
||
# Prompt
|
||
# ============================================================
|
||
# 2026-04-18 ogt + Claude Opus 4.7: B 方案 — LLM 驅動智能摘要（取代 Python str()[:30] 截斷）
|
||
# 架構鐵律: 捨棄 Python 寫死字串解析，結構化 diff 直接餵 LLM,由 LLM 產出繁中 Top 5 摘要
|
||
_NARRATIVE_PROMPT = """你是 AWOOOI SRE 維運助理。以下是 K8s Config Drift 報告的原始結構化資料。
|
||
|
||
## 漂移項目原始資料（JSON）
|
||
{drift_items_json}
|
||
|
||
## 意圖分析
|
||
{intent_summary}
|
||
|
||
## 輸出規格（必須是合法 JSON，不得有任何前後文字）
|
||
{{
|
||
"narrative": "4-5 行繁體中文敘述,說明漂移了哪些資源/嚴重程度/可能原因/建議動作",
|
||
"items": [
|
||
{{
|
||
"level": "high 或 medium",
|
||
"field": "簡化後的欄位路徑 (40 字內)",
|
||
"summary": "30 字內繁體中文口語摘要,說明從什麼變成什麼"
|
||
}}
|
||
]
|
||
}}
|
||
|
||
## 規則
|
||
- 繁體中文
|
||
- items 最多挑 5 筆最重要的（HIGH 優先）
|
||
- summary 要讓非技術人員看懂「改了什麼」,例如:
|
||
- "新增 repair-ssh-key secret 掛載"（而非 repr 一長串）
|
||
- "(未設) → awoooi-executor"
|
||
- "新增 pod anti-affinity 規則"
|
||
- 禁止 markdown、反引號、emoji
|
||
- 只輸出純 JSON,不要包在 code block 裡
|
||
"""
|
||
|
||
|
||
class DriftNarratorService:
|
||
"""
|
||
Drift 報告人話摘要服務
|
||
|
||
職責邊界：
|
||
✅ 呼叫 qwen2.5:7b-instruct 生成繁中摘要
|
||
✅ Redis 快取（避免重複推送）
|
||
✅ 推送 Telegram
|
||
❌ 不做漂移分析
|
||
❌ 不生成修復指令
|
||
"""
|
||
|
||
async def narrate_and_notify(
|
||
self,
|
||
report: "DriftReport",
|
||
interpretation: "DriftInterpretation | None" = None,
|
||
) -> None:
|
||
"""
|
||
生成人話摘要並推送 Telegram
|
||
|
||
只在 high_count > 0 or medium_count >= TRIGGER_MEDIUM_MIN 時執行
|
||
"""
|
||
if not self._should_narrate(report):
|
||
logger.debug(
|
||
"drift_narrator_skip",
|
||
report_id=report.report_id,
|
||
high=report.high_count,
|
||
medium=report.medium_count,
|
||
)
|
||
return
|
||
|
||
# Redis 快取檢查（同 report_id 不重複推送）
|
||
cache_key = f"{CACHE_PREFIX}{report.report_id}"
|
||
redis = await get_redis()
|
||
if await redis.exists(cache_key):
|
||
logger.debug("drift_narrator_cache_hit", report_id=report.report_id)
|
||
return
|
||
|
||
# 2026-04-18 B 方案: LLM 同時產 narrative + 結構化 items（取代 str()[:30]）
|
||
narrative, items = await self._generate_narrative_and_items(report, interpretation)
|
||
await self._send_telegram(report, narrative, items)
|
||
|
||
# 寫入 DB narrative_text (Phase 30 ADR-067)
|
||
try:
|
||
from src.repositories.drift_repository import get_drift_repository
|
||
await get_drift_repository().update_narrative(report.report_id, narrative)
|
||
except Exception as e:
|
||
logger.warning("drift_narrator_db_write_failed", error=str(e))
|
||
|
||
# 寫入快取
|
||
await redis.set(cache_key, narrative[:500], ex=CACHE_TTL)
|
||
|
||
logger.info(
|
||
"drift_narrator_sent",
|
||
report_id=report.report_id,
|
||
high=report.high_count,
|
||
medium=report.medium_count,
|
||
)
|
||
|
||
def _should_narrate(self, report: "DriftReport") -> bool:
|
||
"""觸發條件：high >= 1 or medium >= 3"""
|
||
# 過濾 HPA 白名單後重算
|
||
non_hpa_items = [
|
||
item for item in report.items
|
||
if item.field_path not in _HPA_ALLOWLIST_PATHS
|
||
and not item.is_allowlisted
|
||
]
|
||
high = sum(1 for i in non_hpa_items if i.drift_level.value == "high")
|
||
medium = sum(1 for i in non_hpa_items if i.drift_level.value == "medium")
|
||
return high >= TRIGGER_HIGH_MIN or medium >= TRIGGER_MEDIUM_MIN
|
||
|
||
async def _generate_narrative_and_items(
|
||
self,
|
||
report: "DriftReport",
|
||
interpretation: "DriftInterpretation | None",
|
||
) -> tuple[str, list[dict]]:
|
||
"""
|
||
2026-04-18 ogt + Claude Opus 4.7: B 方案 — LLM 產生 narrative + 結構化 items
|
||
|
||
回傳 (narrative, items):
|
||
narrative: 繁中 4-5 行敘述
|
||
items: [{level, field, summary}, ...] 最多 5 筆
|
||
|
||
LLM 失敗則 fallback 到 Python 智能截斷（不是 str()[:30] 暴力砍）
|
||
|
||
2026-04-18 ADR-090-C: 每次呼叫同步寫入 automation_operation_log +
|
||
ai_collaboration_trace(不論成功或 fallback),完整 L4 稽核。
|
||
"""
|
||
import json as _json
|
||
import time
|
||
|
||
drift_items_json = self._format_drift_for_llm(report)
|
||
intent_summary = self._format_intent_summary(interpretation)
|
||
|
||
prompt = _NARRATIVE_PROMPT.format(
|
||
drift_items_json=drift_items_json,
|
||
intent_summary=intent_summary,
|
||
)
|
||
|
||
started_ms = time.time()
|
||
narrative: str = ""
|
||
items: list[dict] = []
|
||
raw_response: str | None = None
|
||
provider: str = "unknown"
|
||
status: str = "failed"
|
||
llm_accepted: bool = False
|
||
|
||
try:
|
||
openclaw = get_openclaw()
|
||
text, _provider, success = await openclaw.call(prompt)
|
||
provider = _provider or "unknown"
|
||
raw_response = text if text else None
|
||
|
||
if success and text and text.strip():
|
||
_raw = text.strip()
|
||
if _raw.startswith("```"):
|
||
_raw = _raw.strip("`").lstrip("json").strip()
|
||
|
||
# 解析策略: 3 路 fallback
|
||
# Path 1: 直接我們的 {narrative, items} 結構 (純 Ollama 或 LLM 守規矩)
|
||
# Path 2: NEMOTRON wrapper {description,...} 且 description 內含我們的結構
|
||
# Path 3: NEMOTRON wrapper,description 是純敘述 → 用它當 narrative + Python fallback items
|
||
_parsed_narrative = None
|
||
_parsed_items = None
|
||
try:
|
||
_parsed = _json.loads(_raw)
|
||
if isinstance(_parsed, dict):
|
||
# Path 1
|
||
if "narrative" in _parsed and isinstance(_parsed.get("items"), list):
|
||
_parsed_narrative = str(_parsed["narrative"]).strip()
|
||
_parsed_items = _parsed["items"]
|
||
else:
|
||
# Path 2 / Path 3: NEMOTRON wrapper
|
||
_desc = (
|
||
_parsed.get("description")
|
||
or _parsed.get("action_title")
|
||
or _parsed.get("reasoning")
|
||
or ""
|
||
)
|
||
_desc = str(_desc).strip()
|
||
# Path 2: description 本身是巢狀 JSON 含我們結構?
|
||
if _desc.startswith("{") and "narrative" in _desc:
|
||
try:
|
||
_inner = _json.loads(_desc)
|
||
if isinstance(_inner, dict) and "narrative" in _inner:
|
||
_parsed_narrative = str(_inner.get("narrative", "")).strip()
|
||
_parsed_items = _inner.get("items", []) if isinstance(_inner.get("items"), list) else None
|
||
except (_json.JSONDecodeError, ValueError):
|
||
pass
|
||
# Path 3: 只有 narrative(來自 description),items 用 Python fallback
|
||
if _parsed_narrative is None and _desc:
|
||
_parsed_narrative = _desc
|
||
_parsed_items = None # 觸發下方 fallback_items
|
||
except (_json.JSONDecodeError, ValueError) as e:
|
||
logger.warning("drift_narrator_json_parse_fail", err=str(e),
|
||
raw_prefix=_raw[:100], provider=provider)
|
||
|
||
# 驗證 + 清洗
|
||
if _parsed_narrative:
|
||
# 清洗 items (若 LLM 有給)
|
||
clean_items = []
|
||
if isinstance(_parsed_items, list):
|
||
for it in _parsed_items[:5]:
|
||
if isinstance(it, dict) and it.get("field") and it.get("summary"):
|
||
clean_items.append({
|
||
"level": it.get("level", "medium"),
|
||
"field": str(it["field"])[:60],
|
||
"summary": str(it["summary"])[:80],
|
||
})
|
||
# items 空就用 Python smart fallback (不是 str()[:30])
|
||
if not clean_items:
|
||
clean_items = self._fallback_items(report)
|
||
|
||
narrative = _parsed_narrative
|
||
items = clean_items
|
||
status = "success"
|
||
llm_accepted = True
|
||
|
||
if not llm_accepted:
|
||
logger.warning("drift_narrator_openclaw_failed", provider=provider)
|
||
|
||
except Exception as e:
|
||
logger.warning("drift_narrator_llm_error", error=str(e))
|
||
|
||
# Fallback
|
||
if not llm_accepted:
|
||
narrative = self._fallback_narrative(report, interpretation)
|
||
items = self._fallback_items(report)
|
||
status = "failed"
|
||
|
||
# ADR-090-C: 同步寫 DB 稽核（永不 propagate error,保護主流程）
|
||
duration_ms = int((time.time() - started_ms) * 1000)
|
||
try:
|
||
await self._log_ai_action_to_db(
|
||
report=report,
|
||
prompt=prompt,
|
||
raw_response=raw_response,
|
||
narrative=narrative,
|
||
items=items,
|
||
provider=provider,
|
||
status=status,
|
||
llm_accepted=llm_accepted,
|
||
duration_ms=duration_ms,
|
||
)
|
||
except Exception as e:
|
||
logger.warning("drift_narrator_audit_write_failed", error=str(e))
|
||
|
||
return narrative, items
|
||
|
||
async def _log_ai_action_to_db(
|
||
self,
|
||
report: "DriftReport",
|
||
prompt: str,
|
||
raw_response: str | None,
|
||
narrative: str,
|
||
items: list[dict],
|
||
provider: str,
|
||
status: str,
|
||
llm_accepted: bool,
|
||
duration_ms: int,
|
||
) -> None:
|
||
"""
|
||
ADR-090-C: 把 drift narrator 的 AI 動作寫入 automation_operation_log +
|
||
ai_collaboration_trace(L4 稽核 + 未來 RLHF 語料)
|
||
|
||
- op_type='notification_formatted'
|
||
- actor='drift_narrator'
|
||
- 若能找到該 drift 的 incident 關聯,設 parent_op_id
|
||
"""
|
||
import json as _json
|
||
from sqlalchemy import text as _sql
|
||
from src.db.base import get_db_context
|
||
|
||
input_json = _json.dumps({
|
||
"report_id": report.report_id,
|
||
"namespace": report.namespace,
|
||
"high_count": report.high_count,
|
||
"medium_count": report.medium_count,
|
||
"items_scanned": len(report.items),
|
||
})
|
||
output_json = _json.dumps({
|
||
"narrative": narrative,
|
||
"items": items,
|
||
"items_count": len(items),
|
||
}, ensure_ascii=False)
|
||
trace_response = _json.dumps({
|
||
"narrative": narrative,
|
||
"items": items,
|
||
"raw_prefix": (raw_response or "")[:500],
|
||
}, ensure_ascii=False)
|
||
|
||
async with get_db_context() as db:
|
||
# P2.4: 嘗試找 parent_op_id（若未來有 drift→alert_fired 鏈路）
|
||
parent_row = await db.execute(
|
||
_sql("""
|
||
SELECT op_id FROM automation_operation_log
|
||
WHERE operation_type='alert_fired'
|
||
AND (input::jsonb->>'report_id' = :rid
|
||
OR input::jsonb->>'drift_report_id' = :rid)
|
||
ORDER BY created_at DESC LIMIT 1
|
||
"""),
|
||
{"rid": report.report_id},
|
||
)
|
||
parent_op_id = parent_row.scalar() if parent_row else None
|
||
|
||
# 寫 automation_operation_log
|
||
# 2026-04-18 hotfix: tags 要傳 Python list（不是 PG array literal 字串）
|
||
# 否則 asyncpg 會報 "a sized iterable container expected"
|
||
op_row = await db.execute(
|
||
_sql("""
|
||
INSERT INTO automation_operation_log (
|
||
operation_type, actor, status,
|
||
input, output,
|
||
duration_ms, parent_op_id, tags
|
||
) VALUES (
|
||
'notification_formatted',
|
||
'drift_narrator',
|
||
:status,
|
||
CAST(:input AS jsonb),
|
||
CAST(:output AS jsonb),
|
||
:duration_ms, :parent_op_id,
|
||
:tags
|
||
)
|
||
RETURNING op_id
|
||
"""),
|
||
{
|
||
"status": status,
|
||
"input": input_json,
|
||
"output": output_json,
|
||
"duration_ms": duration_ms,
|
||
"parent_op_id": parent_op_id,
|
||
"tags": ["drift", "type4d", "llm_summary"],
|
||
},
|
||
)
|
||
op_id = op_row.scalar()
|
||
|
||
# 寫 ai_collaboration_trace
|
||
await db.execute(
|
||
_sql("""
|
||
INSERT INTO ai_collaboration_trace (
|
||
op_id, step_order, agent, model,
|
||
prompt, response, duration_ms, accepted
|
||
) VALUES (
|
||
:op_id, 1, 'drift_narrator', :model,
|
||
:prompt, CAST(:response AS jsonb), :duration_ms, :accepted
|
||
)
|
||
"""),
|
||
{
|
||
"op_id": op_id,
|
||
"model": provider,
|
||
"prompt": prompt[:8000],
|
||
"response": trace_response,
|
||
"duration_ms": duration_ms,
|
||
"accepted": llm_accepted,
|
||
},
|
||
)
|
||
# get_db_context() 在 exit 時 auto-commit（src/db/base.py:128）
|
||
logger.info(
|
||
"drift_narrator_audit_written",
|
||
op_id=str(op_id),
|
||
parent_op_id=str(parent_op_id) if parent_op_id else None,
|
||
status=status,
|
||
items_count=len(items),
|
||
)
|
||
|
||
def _format_drift_for_llm(self, report: "DriftReport") -> str:
|
||
"""
|
||
2026-04-18 ogt + Claude Opus 4.7: B 方案 — 餵 LLM 用的 JSON 序列化
|
||
保留更多原始 context 給 LLM 推理,不做 30 字元暴力截斷
|
||
"""
|
||
import json as _json
|
||
items_for_llm = []
|
||
for item in report.items[:12]:
|
||
if item.is_allowlisted or item.field_path in _HPA_ALLOWLIST_PATHS:
|
||
continue
|
||
items_for_llm.append({
|
||
"level": item.drift_level.value,
|
||
"resource": f"{item.resource_kind}/{item.resource_name}",
|
||
"field": item.field_path,
|
||
"git_value": str(item.git_value)[:200] if item.git_value is not None else None,
|
||
"actual_value": str(item.actual_value)[:200] if item.actual_value is not None else None,
|
||
})
|
||
return _json.dumps(items_for_llm, ensure_ascii=False, indent=2)
|
||
|
||
def _smart_shorten(self, val) -> str:
|
||
"""型別安全摘要 — dict/list 顯示大小,字串保留頭尾,None 轉「未設」"""
|
||
if val is None:
|
||
return "(未設)"
|
||
s = str(val)
|
||
# 嘗試判斷是不是 JSON 字串
|
||
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 _fallback_items(self, report: "DriftReport") -> list[dict]:
|
||
"""LLM 失敗時的 Python 智能摘要（取代舊 str()[:30]）"""
|
||
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)
|
||
items.append({
|
||
"level": item.drift_level.value,
|
||
"field": item.field_path[:60],
|
||
"summary": f"{from_val} → {to_val}",
|
||
})
|
||
return items
|
||
|
||
def _format_intent_summary(self, interpretation: "DriftInterpretation | None") -> str:
|
||
if not interpretation:
|
||
return "無意圖分析"
|
||
return (
|
||
f"意圖: {interpretation.intent.value} | "
|
||
f"說明: {interpretation.explanation} | "
|
||
f"信心: {interpretation.confidence:.0%}"
|
||
)
|
||
|
||
def _fallback_narrative(
|
||
self,
|
||
report: "DriftReport",
|
||
interpretation: "DriftInterpretation | None",
|
||
) -> str:
|
||
"""LLM 失敗時的結構化 fallback"""
|
||
resources = list({
|
||
f"{i.resource_kind}/{i.resource_name}"
|
||
for i in report.items[:5]
|
||
if not i.is_allowlisted
|
||
})
|
||
resource_str = "、".join(resources) if resources else "未知資源"
|
||
intent_str = interpretation.explanation if interpretation else "意圖分析不可用"
|
||
return (
|
||
f"偵測到 {resource_str} 等資源發生配置漂移。\n"
|
||
f"嚴重度：HIGH {report.high_count} 項、MEDIUM {report.medium_count} 項。\n"
|
||
f"研判原因：{intent_str}\n"
|
||
f"建議：確認是否需要 rollback 回 Git 狀態。"
|
||
)
|
||
|
||
async def _send_telegram(
|
||
self,
|
||
report: "DriftReport",
|
||
narrative: str,
|
||
items: list[dict],
|
||
) -> None:
|
||
"""
|
||
推送 TYPE-4D Config Drift 卡片（ADR-075）+ B 方案智能摘要
|
||
|
||
2026-04-18 ogt + Claude Opus 4.7: 改用 LLM 產的結構化 items,
|
||
取代 str()[:30] 暴力截斷產生的亂碼
|
||
"""
|
||
from src.services.telegram_gateway import get_telegram_gateway
|
||
|
||
diff_summary = self._render_telegram_body(report, narrative, items)
|
||
|
||
try:
|
||
tg = get_telegram_gateway()
|
||
await tg.send_drift_card(
|
||
incident_id=report.report_id,
|
||
approval_id=report.report_id,
|
||
resource_name=report.namespace,
|
||
diff_summary=diff_summary[:500],
|
||
detected_at="",
|
||
)
|
||
except Exception as e:
|
||
logger.warning("drift_narrator_telegram_error", error=str(e))
|
||
|
||
def _render_telegram_body(
|
||
self,
|
||
report: "DriftReport",
|
||
narrative: str,
|
||
items: list[dict],
|
||
) -> str:
|
||
"""
|
||
組裝 Telegram 卡片 body（B 方案格式）
|
||
|
||
範例輸出:
|
||
🤖 AI 研判
|
||
volumes 與 affinity 被手動修改...
|
||
|
||
📊 漂移明細 (HIGH: 1 | MEDIUM: 29)
|
||
🔴 spec.template.spec.volumes: 新增 2 項 repair-ssh-key 掛載
|
||
🟡 spec.template.spec.serviceAccount: (未設) → awoooi-executor
|
||
🟡 spec.template.spec.affinity.podAntiAffinity: 新增 preferred 規則
|
||
... 還有 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
|
||
shown = len(items)
|
||
if total > shown:
|
||
lines.append(f"... 還有 {total - shown} 項 (按 🔍 查看 Diff)")
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
# ============================================================
|
||
# Singleton
|
||
# ============================================================
|
||
|
||
_narrator: DriftNarratorService | None = None
|
||
|
||
|
||
def get_drift_narrator_service() -> DriftNarratorService:
|
||
global _narrator
|
||
if _narrator is None:
|
||
_narrator = DriftNarratorService()
|
||
return _narrator
|