feat(api): 新增 SRE 戰情室 digest preview
All checks were successful
Code Review / ai-code-review (push) Successful in 19s
CD Pipeline / tests (push) Successful in 1m43s
CD Pipeline / build-and-deploy (push) Successful in 7m20s
CD Pipeline / post-deploy-checks (push) Successful in 3m8s

This commit is contained in:
Your Name
2026-06-18 20:18:28 +08:00
parent f8c290be63
commit 7e03b9231b
4 changed files with 166 additions and 0 deletions

View File

@@ -406,6 +406,20 @@ class MonthlyReportPreviewResponse(BaseModel):
formatted_preview: str = Field(default="", description="Telegram HTML no-send preview")
class SreDigestPreviewResponse(BaseModel):
"""AwoooI SRE 戰情室 digest no-send preview 回應"""
report_date: str = Field(description="報告日期時間")
source_ok_count: int = Field(default=0, description="報表資料源可讀數")
source_total_count: int = Field(default=0, description="報表資料源總數")
source_confidence_percent: int = Field(default=0, description="報表資料源可信度")
source_gap_ids: list[str] = Field(default_factory=list, description="報表資料源缺口工作項")
no_send_preview_count: int = Field(default=0, description="日 / 週 / 月 no-send preview 數量")
live_send_allowed_count: int = Field(default=0, description="允許實發數")
runtime_gate_count: int = Field(default=0, description="runtime gate 數")
formatted_preview: str = Field(default="", description="Telegram HTML no-send preview")
def _report_source_preview_fields(source_health: dict[str, Any] | None) -> dict[str, Any]:
source_health = source_health or {}
rollups = source_health.get("rollups") or {}
@@ -419,6 +433,8 @@ def _report_source_preview_fields(source_health: dict[str, Any] | None) -> dict[
if source.get("work_item_id")
][:5],
"no_send_preview_count": int(rollups.get("no_send_preview_count") or 0),
"live_send_allowed_count": int(rollups.get("live_send_allowed_count") or 0),
"runtime_gate_count": int(rollups.get("runtime_gate_count") or 0),
}
@@ -485,6 +501,40 @@ async def preview_monthly_report(
)
@router.get(
"/sre-digest/preview",
response_model=SreDigestPreviewResponse,
summary="預覽 AwoooI SRE 戰情室 digest",
)
async def preview_sre_digest(
service: DailyReportDep = None,
) -> SreDigestPreviewResponse:
"""
預覽 AwoooI SRE 戰情室 digest (不發送)
收斂日報 / 週報 / 月報 source health、資產沉澱與工作項不寫 Gateway queue。
"""
from src.utils.timezone import now_taipei
source_health = await service.collect_report_source_health(days=30)
preview_fields = _report_source_preview_fields(source_health)
now = now_taipei()
return SreDigestPreviewResponse(
report_date=now.strftime("%Y-%m-%d %H:%M"),
source_ok_count=preview_fields["source_ok_count"],
source_total_count=preview_fields["source_total_count"],
source_confidence_percent=preview_fields["source_confidence_percent"],
source_gap_ids=preview_fields["source_gap_ids"],
no_send_preview_count=preview_fields["no_send_preview_count"],
live_send_allowed_count=preview_fields["live_send_allowed_count"],
runtime_gate_count=preview_fields["runtime_gate_count"],
formatted_preview=service.format_sre_digest_preview(
source_health,
generated_at=now,
),
)
@router.get(
"/weekly/preview",
response_model=WeeklyReportResponse,

View File

@@ -375,6 +375,54 @@ class ReportGenerationService:
]
return "\n".join(lines)
def format_sre_digest_preview(
self,
source_health: dict[str, Any] | None,
*,
generated_at: datetime | None = None,
) -> str:
"""Format an AwoooI SRE war-room no-send digest preview."""
now = generated_at or now_taipei()
source_health = source_health or {}
rollups = source_health.get("rollups") or {}
ok_count = int(rollups.get("source_ok_count") or 0)
total_count = int(rollups.get("source_count") or 0)
gap_count = int(rollups.get("source_gap_count") or 0)
confidence = int(rollups.get("confidence_percent") or 0)
preview_count = int(rollups.get("no_send_preview_count") or 0)
work_items = source_health.get("work_items") or []
gap_ids = [
str(item.get("work_item_id"))
for item in work_items
if item.get("work_item_id")
][:5]
gap_text = ", ".join(gap_ids) if gap_ids else ""
lines = [
"<b>🛡️ AwoooI SRE 戰情室 digest no-send preview</b>",
f"<b>{now.strftime('%Y-%m-%d %H:%M')}</b> 台北時間",
"",
"<b>📡 資料源健康</b>",
f" 來源: <code>{ok_count}/{total_count}</code> | 缺口: {gap_count} | 信心: <b>{confidence}%</b>",
f" 工作項: {html.escape(gap_text)}",
"",
"<b>🗓️ 日 / 週 / 月報狀態</b>",
f" no-send preview: {preview_count}",
" live Telegram send: 0",
" Gateway queue write: 0",
]
lines.extend(self._format_report_source_health_block(source_health))
lines += [
"",
"<b>🧭 下一步</b>",
" 1. 補齊 report-source-gap 專屬 PlayBook 與 Verifier readback。",
" 2. Owner review 後才允許 SRE digest 實發批准包。",
" 3. 任何中低風險自動處理仍需 rollback / post-check / audit receipt。",
"",
"<i>只讀草案:不發 Telegram、不寫 Gateway queue、不改排程、不啟動 runtime gate。</i>",
]
return "\n".join(lines)
def format_daily_report(
self,
kpi: DailyKpi,

View File

@@ -318,6 +318,46 @@ class TestFormatDailyReport:
assert "Verifier: source_health_ready 1/2" in report
assert "不代表已授權發送或自動修復" in report
def test_sre_digest_preview_contains_assets_and_boundaries(self):
"""SRE 戰情室 digest 應收斂缺口、資產與 no-send 邊界"""
source_health = {
"rollups": {
"source_ok_count": 2,
"source_count": 5,
"source_gap_count": 3,
"confidence_percent": 40,
"no_send_preview_count": 3,
},
"source_health": [
{"work_item_id": "report-source-gap:incident_summary"},
],
"work_items": [
{"work_item_id": "report-source-gap:incident_summary"},
{"work_item_id": "report-source-gap:ai_performance"},
],
"automation_assets": [
{"label": "KM", "state": "draft_ready", "done_count": 3, "blocked_count": 2},
{"label": "PlayBook", "state": "draft_required", "done_count": 0, "blocked_count": 2},
{"label": "Verifier", "state": "source_health_ready", "done_count": 1, "blocked_count": 2},
],
}
svc = ReportGenerationService()
report = svc.format_sre_digest_preview(
source_health,
generated_at=datetime(2026, 6, 18, 10, 0, tzinfo=_TZ_TAIPEI),
)
assert "AwoooI SRE 戰情室 digest no-send preview" in report
assert "來源: <code>2/5</code>" in report
assert "report-source-gap:incident_summary" in report
assert "live Telegram send: 0" in report
assert "Gateway queue write: 0" in report
assert "KM: draft_ready 3/5" in report
assert "PlayBook: draft_required 0/2" in report
assert "Verifier: source_health_ready 1/3" in report
assert "不發 Telegram" in report
assert "不啟動 runtime gate" in report
# =============================================================================
# format_postmortem

View File

@@ -70,3 +70,31 @@ def test_monthly_report_preview_exposes_source_health_no_send_preview():
assert f"來源: <code>{data['source_ok_count']}/{data['source_total_count']}</code>" in preview
assert "實發: 0" in preview
assert "不代表已授權發送或自動修復" in preview
def test_sre_digest_preview_exposes_source_health_no_send_preview():
client = TestClient(app)
response = client.get("/api/v1/stats/sre-digest/preview")
assert response.status_code == 200
data = response.json()
assert "report_date" in data
assert "source_ok_count" in data
assert "source_total_count" in data
assert "source_confidence_percent" in data
assert "source_gap_ids" in data
assert "no_send_preview_count" in data
assert "live_send_allowed_count" in data
assert "runtime_gate_count" in data
assert "formatted_preview" in data
assert data["live_send_allowed_count"] == 0
assert data["runtime_gate_count"] == 0
preview = data["formatted_preview"]
assert "AwoooI SRE 戰情室 digest no-send preview" in preview
assert "報表資料源 / 沉澱" in preview
assert f"來源: <code>{data['source_ok_count']}/{data['source_total_count']}</code>" in preview
assert "live Telegram send: 0" in preview
assert "Gateway queue write: 0" in preview
assert "不發 Telegram" in preview
assert "不啟動 runtime gate" in preview