wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

refactor(phase-s): Phase S 技術債清理 - 五項架構改善

Browse Source 
S-01: generate_alert_fingerprint() 移至 alert_analyzer_service (Router→Service)
S-02: 移除廢棄 USE_NEW_ENGINE config (Phase R 已完成歷史使命)
S-03: github_webhook.py linter 清理 (Field unused + delivery_id noqa)
S-04: Pydantic v2 遷移 - approval/incident models (class Config → ConfigDict)
S-05: Skill 09 v1.1 更新 (USE_NEW_ENGINE 廢棄說明)

測試: 393 passed, 零失敗

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
OG T
2026-04-01 13:12:02 +08:00
parent d02efd4998
commit 22de22c989
30 changed files with 2914 additions and 982 deletions
Download Patch File Download Diff File Expand all files Collapse all files

911
apps/api/src/api/v1/github_webhook.py
View File

@@ -25,10 +25,10 @@ Phase 13.1: GitHub PR/Push/CI → OpenClaw AI 整合
🔴 HARD RULE: 時間顯示使用 Asia/Taipei (UTC+8)
版本: v2.0
最後修改: 2026-03-26 16:30 (台北時區)
版本: v2.1
最後修改: 2026-04-01 11:00 (台北時區)
修改者: Claude Code
變更: Phase 13.1 #76 CI 失敗診斷
變更: 協調函數移至 Service 層 (leWOOOgo ADR-024)
"""
import hashlib
@@ -38,22 +38,11 @@ import uuid
from typing import Literal
from fastapi import APIRouter, BackgroundTasks, Header, HTTPException, Request, status
from pydantic import BaseModel, Field
from pydantic import BaseModel
from src.core.config import settings
from src.core.logging import get_logger
from src.models.approval import (
ApprovalRequestCreate,
BlastRadius,
DataImpact,
RiskLevel,
)
from src.services.approval_db import get_approval_service
from src.services.github_api_service import get_github_api_service
from src.services.github_webhook_service import get_github_webhook_service
from src.services.openclaw_http_service import get_openclaw_http_service
from src.services.telegram_gateway import get_telegram_gateway
from src.utils.timezone import now_taipei_iso
logger = get_logger("awoooi.github_webhook")
@@ -147,20 +136,6 @@ class GitHubWorkflowJob(BaseModel):
steps: list[dict] = []
class CIFailureDiagnosis(BaseModel):
"""CI 失敗診斷結果 (Phase 13.1 #76)"""
summary: str = Field(..., description="失敗摘要")
root_cause: str = Field(..., description="根本原因分析")
failed_step: str | None = Field(None, description="失敗的步驟名稱")
error_type: str = Field(..., description="錯誤類型 (build/test/lint/deploy/timeout)")
suggestions: list[str] = Field(default=[], description="修復建議")
auto_fixable: bool = Field(False, description="是否可自動修復")
fix_command: str | None = Field(None, description="自動修復指令 (如可自動修復)")
risk_level: str = Field("medium", description="風險等級 (low/medium/high/critical)")
analyzed_by: str = Field(..., description="分析模型")
confidence: float = Field(..., ge=0, le=1, description="信心度")
class GitHubWebhookPayload(BaseModel):
"""GitHub Webhook Payload (通用)"""
action: str | None = None # PR: opened, synchronize, etc.
@@ -179,17 +154,6 @@ class GitHubWebhookPayload(BaseModel):
workflow_job: GitHubWorkflowJob | None = None
class CodeReviewResult(BaseModel):
"""AI 代碼審查結果"""
summary: str = Field(..., description="審查摘要")
issues: list[dict] = Field(default=[], description="發現的問題列表")
suggestions: list[dict] = Field(default=[], description="改進建議")
security_concerns: list[str] = Field(default=[], description="安全疑慮")
quality_score: float = Field(..., ge=0, le=100, description="代碼品質分數 0-100")
analyzed_by: str = Field(..., description="分析模型 (ollama/claude)")
confidence: float = Field(..., ge=0, le=1, description="分析信心度 0-1")
class GitHubWebhookResponse(BaseModel):
"""Webhook 回應"""
status: Literal["accepted", "ignored", "error"]
@@ -359,7 +323,7 @@ async def handle_github_webhook(
1. 驗證簽章
2. 驗證倉庫白名單
3. 解析事件類型
4. 背景執行 AI 審查
4. 背景執行 AI 審查 (委派給 GitHubWebhookService)
"""
try:
# 1. 驗證 HMAC 簽章
@@ -445,13 +409,13 @@ async def handle_github_webhook(
# =============================================================================
# Event Handlers
# Event Handlers (HTTP 層: 解析、驗證、回應 — 業務邏輯在 Service 層)
# =============================================================================
async def handle_pull_request(
payload: GitHubWebhookPayload,
background_tasks: BackgroundTasks,
delivery_id: str | None,
delivery_id: str | None, # noqa: ARG001 — reserved for idempotency (future use)
) -> GitHubWebhookResponse:
"""
處理 Pull Request 事件
@@ -481,9 +445,10 @@ async def handle_pull_request(
# 生成審查 ID
review_id = f"gh-pr-{payload.repository.id}-{pr.number}-{uuid.uuid4().hex[:8]}"
# 背景執行審查
# 背景執行審查 (委派給 Service)
service = get_github_webhook_service()
background_tasks.add_task(
review_pull_request,
service.review_pull_request,
repo=payload.repository,
pr=pr,
sender=payload.sender,
@@ -511,7 +476,7 @@ async def handle_pull_request(
async def handle_push(
payload: GitHubWebhookPayload,
background_tasks: BackgroundTasks,
delivery_id: str | None,
delivery_id: str | None, # noqa: ARG001 — reserved for idempotency (future use)
) -> GitHubWebhookResponse:
"""
處理 Push 事件
@@ -539,9 +504,10 @@ async def handle_push(
# 生成審查 ID
review_id = f"gh-push-{payload.repository.id}-{payload.after[:8]}-{uuid.uuid4().hex[:8]}"
# 背景執行審查
# 背景執行審查 (委派給 Service)
service = get_github_webhook_service()
background_tasks.add_task(
review_push,
service.review_push,
repo=payload.repository,
commits=commits,
sender=payload.sender,
@@ -571,7 +537,7 @@ async def handle_push(
async def handle_workflow_run(
payload: GitHubWebhookPayload,
background_tasks: BackgroundTasks,
delivery_id: str | None,
delivery_id: str | None, # noqa: ARG001 — reserved for idempotency (future use)
) -> GitHubWebhookResponse:
"""
處理 Workflow Run 事件 (Phase 13.1 #76 CI 失敗診斷)
@@ -605,9 +571,10 @@ async def handle_workflow_run(
# 生成診斷 ID
diagnosis_id = f"gh-ci-{payload.repository.id}-{workflow_run.id}-{uuid.uuid4().hex[:8]}"
# 背景執行 CI 失敗診斷
# 背景執行 CI 失敗診斷 (委派給 Service)
service = get_github_webhook_service()
background_tasks.add_task(
diagnose_ci_failure,
service.diagnose_ci_failure,
repo=payload.repository,
workflow_run=workflow_run,
sender=payload.sender,
@@ -632,848 +599,6 @@ async def handle_workflow_run(
)
# =============================================================================
# Background Tasks: AI Review
# =============================================================================
async def review_pull_request(
repo: GitHubRepository,
pr: GitHubPullRequest,
sender: GitHubUser,
review_id: str,
action: str,
):
"""
背景任務: PR 代碼審查
1. 取得 PR diff
2. 呼叫 OpenClaw 分析
3. 儲存結果到 Redis
4. 發送 Telegram 通知
5. 建立 Approval (可選)
"""
try:
logger.info(
"github_pr_review_started",
review_id=review_id,
repo=repo.full_name,
pr_number=pr.number,
)
# 1. 取得 PR diff
diff_content = await fetch_pr_diff(pr.diff_url)
# 2. 呼叫 OpenClaw 進行代碼審查
analysis = await call_openclaw_code_review(
repo_name=repo.full_name,
pr_title=pr.title,
pr_body=pr.body or "",
diff_content=diff_content,
changed_files=pr.changed_files,
additions=pr.additions,
deletions=pr.deletions,
)
# 3. 儲存結果到 Redis
await save_review_result(
review_id=review_id,
event_type="pull_request",
repo=repo.full_name,
target=f"PR #{pr.number}",
analysis=analysis,
metadata={
"pr_number": pr.number,
"pr_title": pr.title,
"pr_url": pr.html_url,
"author": pr.user.login,
"action": action,
"changed_files": pr.changed_files,
"additions": pr.additions,
"deletions": pr.deletions,
},
)
# 4. 發送 Telegram 通知
await send_github_telegram_alert(
review_id=review_id,
event_type="pull_request",
repo=repo.full_name,
target=f"PR #{pr.number}: {pr.title[:50]}",
url=pr.html_url,
author=pr.user.login,
analysis=analysis,
)
# 5. 如果有安全疑慮，建立 Approval
if analysis and analysis.security_concerns:
await create_github_approval(
review_id=review_id,
repo=repo.full_name,
target=f"PR #{pr.number}",
url=pr.html_url,
analysis=analysis,
)
logger.info(
"github_pr_review_completed",
review_id=review_id,
quality_score=analysis.quality_score if analysis else None,
has_security_concerns=bool(analysis and analysis.security_concerns),
)
except Exception as e:
logger.exception(
"github_pr_review_failed",
review_id=review_id,
error=str(e),
)
async def review_push(
repo: GitHubRepository,
commits: list[GitHubCommit],
sender: GitHubUser,
review_id: str,
ref: str,
before_sha: str | None,
after_sha: str | None,
):
"""
背景任務: Push 代碼審查
1. 整理 commit 資訊
2. 呼叫 OpenClaw 分析
3. 儲存結果到 Redis
4. 發送 Telegram 通知
"""
try:
logger.info(
"github_push_review_started",
review_id=review_id,
repo=repo.full_name,
commit_count=len(commits),
)
# 1. 整理 commit 資訊
commit_summary = []
all_files = {"added": [], "modified": [], "removed": []}
for commit in commits:
commit_summary.append({
"sha": commit.id[:8],
"message": commit.message[:100],
"author": commit.author.get("name", "unknown"),
})
all_files["added"].extend(commit.added)
all_files["modified"].extend(commit.modified)
all_files["removed"].extend(commit.removed)
# 2. 呼叫 OpenClaw 進行代碼審查 (Push 版)
analysis = await call_openclaw_push_review(
repo_name=repo.full_name,
ref=ref,
commits=commit_summary,
files_changed=all_files,
)
# 3. 儲存結果到 Redis
await save_review_result(
review_id=review_id,
event_type="push",
repo=repo.full_name,
target=f"push to {ref.split('/')[-1]}",
analysis=analysis,
metadata={
"ref": ref,
"before_sha": before_sha,
"after_sha": after_sha,
"commit_count": len(commits),
"pusher": sender.login,
"files": all_files,
},
)
# 4. 發送 Telegram 通知 (只有發現問題時才通知)
if analysis and (analysis.issues or analysis.security_concerns or analysis.quality_score < 70):
await send_github_telegram_alert(
review_id=review_id,
event_type="push",
repo=repo.full_name,
target=f"push to {ref.split('/')[-1]} ({len(commits)} commits)",
url=repo.html_url,
author=sender.login,
analysis=analysis,
)
logger.info(
"github_push_review_completed",
review_id=review_id,
quality_score=analysis.quality_score if analysis else None,
)
except Exception as e:
logger.exception(
"github_push_review_failed",
review_id=review_id,
error=str(e),
)
async def diagnose_ci_failure(
repo: GitHubRepository,
workflow_run: GitHubWorkflowRun,
sender: GitHubUser,
diagnosis_id: str,
):
"""
背景任務: CI 失敗診斷 (Phase 13.1 #76)
1. 收集 workflow 失敗資訊
2. 呼叫 OpenClaw 進行根因分析
3. 評估風險等級與自動修復可行性
4. 儲存結果到 Redis
5. 發送 Telegram 通知
6. (可選) 建立 Approval 等待人工確認
"""
try:
logger.info(
"github_ci_failure_diagnosis_started",
diagnosis_id=diagnosis_id,
repo=repo.full_name,
workflow_name=workflow_run.name,
workflow_id=workflow_run.id,
)
# 1. 收集失敗資訊
failure_context = {
"workflow_name": workflow_run.name,
"workflow_id": workflow_run.id,
"run_number": workflow_run.run_number,
"run_attempt": workflow_run.run_attempt,
"conclusion": workflow_run.conclusion,
"head_sha": workflow_run.head_sha,
"head_branch": workflow_run.head_branch,
"event_trigger": workflow_run.event,
"html_url": workflow_run.html_url,
"created_at": workflow_run.created_at,
"updated_at": workflow_run.updated_at,
}
# 2. 呼叫 OpenClaw 進行 CI 失敗診斷
diagnosis = await call_openclaw_ci_diagnosis(
repo_name=repo.full_name,
failure_context=failure_context,
)
# 3. 評估自動修復策略 (Phase 13.1 #78)
repair_decision = None
if diagnosis:
from src.services.ci_auto_repair import get_ci_auto_repair_service
repair_service = get_ci_auto_repair_service()
repair_decision = await repair_service.evaluate_repair(
error_type=diagnosis.error_type,
workflow_name=workflow_run.name,
repo=repo.full_name,
failure_context=failure_context,
diagnosis_summary=diagnosis.summary,
)
# 4. 儲存結果到 Redis (含修復決策)
service = get_github_webhook_service()
await service.save_review_result(
review_id=diagnosis_id,
result={
"event_type": "workflow_run",
"repo": repo.full_name,
"target": f"CI: {workflow_run.name}",
"diagnosis": diagnosis.model_dump() if diagnosis else None,
"repair_decision": {
"should_repair": repair_decision.should_repair,
"execution_decision": repair_decision.execution_decision.value,
"risk_level": repair_decision.risk_level.value,
"reason": repair_decision.reason,
"recommendations": [
{"action": r.action.value, "command": r.command, "confidence": r.confidence}
for r in repair_decision.recommendations[:3]
],
} if repair_decision else None,
"failure_context": failure_context,
"reviewed_at": now_taipei_iso(),
},
ttl=GITHUB_REVIEW_TTL_SECONDS,
)
# 5. 發送 Telegram 通知 (含修復建議)
await send_ci_failure_telegram_alert(
diagnosis_id=diagnosis_id,
repo=repo.full_name,
workflow_name=workflow_run.name,
workflow_url=workflow_run.html_url,
sender=sender.login,
diagnosis=diagnosis,
repair_decision=repair_decision,
)
# 6. 根據修復決策建立 Approval 或自動執行
if repair_decision:
from src.services.ci_auto_repair import ExecutionDecision
if repair_decision.execution_decision == ExecutionDecision.APPROVAL_REQUIRED:
await create_ci_failure_approval(
diagnosis_id=diagnosis_id,
repo=repo.full_name,
workflow_run=workflow_run,
diagnosis=diagnosis,
)
elif repair_decision.execution_decision == ExecutionDecision.AUTO_EXECUTE:
logger.info(
"ci_auto_repair_eligible",
diagnosis_id=diagnosis_id,
action=repair_decision.recommendations[0].action.value if repair_decision.recommendations else None,
# TODO: 實際執行修復指令 (Phase 13.1 後續迭代)
)
elif diagnosis and diagnosis.risk_level in ("high", "critical"):
await create_ci_failure_approval(
diagnosis_id=diagnosis_id,
repo=repo.full_name,
workflow_run=workflow_run,
diagnosis=diagnosis,
)
logger.info(
"github_ci_failure_diagnosis_completed",
diagnosis_id=diagnosis_id,
root_cause=diagnosis.root_cause if diagnosis else None,
auto_fixable=diagnosis.auto_fixable if diagnosis else False,
risk_level=diagnosis.risk_level if diagnosis else None,
repair_decision=repair_decision.execution_decision.value if repair_decision else None,
)
except Exception as e:
logger.exception(
"github_ci_failure_diagnosis_failed",
diagnosis_id=diagnosis_id,
error=str(e),
)
# =============================================================================
# Helper Functions
# =============================================================================
async def fetch_pr_diff(diff_url: str) -> str:
"""
取得 PR diff 內容
Args:
diff_url: GitHub diff URL
Returns:
str: diff 內容
Phase 22 P0 修復: 使用 GitHubApiService (2026-03-31)
"""
service = get_github_api_service()
return await service.fetch_pr_diff(diff_url)
async def call_openclaw_code_review(
repo_name: str,
pr_title: str,
pr_body: str,
diff_content: str,
changed_files: int,
additions: int,
deletions: int,
) -> CodeReviewResult | None:
"""
呼叫 OpenClaw 進行 PR 代碼審查
優先使用 Ollama (本地，零成本)
Fallback: Claude (大型 PR)
Phase 22 P0 修復: 使用 OpenClawHttpService (2026-03-31)
"""
try:
service = get_openclaw_http_service()
data = await service.code_review(
repo_name=repo_name,
pr_title=pr_title,
pr_body=pr_body,
diff_content=diff_content,
changed_files=changed_files,
additions=additions,
deletions=deletions,
prefer_local=True,
timeout=120.0,
)
if data:
return CodeReviewResult(**data)
return None
except Exception as e:
logger.exception("openclaw_code_review_error", error=str(e))
return None
async def call_openclaw_push_review(
repo_name: str,
ref: str,
commits: list[dict],
files_changed: dict,
) -> CodeReviewResult | None:
"""
呼叫 OpenClaw 進行 Push 代碼審查
Phase 22 P0 修復: 使用 OpenClawHttpService (2026-03-31)
"""
try:
service = get_openclaw_http_service()
data = await service.push_review(
repo_name=repo_name,
ref=ref,
commits=commits,
files_changed=files_changed,
prefer_local=True,
timeout=120.0,
)
if data:
return CodeReviewResult(**data)
return None
except Exception as e:
logger.exception("openclaw_push_review_error", error=str(e))
return None
async def call_openclaw_ci_diagnosis(
repo_name: str,
failure_context: dict,
) -> CIFailureDiagnosis | None:
"""
呼叫 OpenClaw 進行 CI 失敗診斷 (Phase 13.1 #76)
分析 CI/CD pipeline 失敗原因，提供根因分析和修復建議
Phase 22 P0 修復: 使用 OpenClawHttpService (2026-03-31)
"""
try:
service = get_openclaw_http_service()
data = await service.ci_diagnosis(
repo_name=repo_name,
failure_context=failure_context,
prefer_local=True,
timeout=120.0,
)
if data:
return CIFailureDiagnosis(**data)
else:
# 返回基本診斷結果 (API 失敗時的 fallback)
return CIFailureDiagnosis(
summary=f"CI workflow '{failure_context.get('workflow_name')}' failed",
root_cause="OpenClaw API unavailable, manual investigation required",
error_type="unknown",
suggestions=["Check workflow logs manually", "Verify runner status"],
auto_fixable=False,
risk_level="medium",
analyzed_by="fallback",
confidence=0.0, # 🔴 Fallback 不是 AI 分析
)
except Exception as e:
logger.exception("openclaw_ci_diagnosis_error", error=str(e))
return CIFailureDiagnosis(
summary="CI diagnosis error",
root_cause=f"Exception: {str(e)}",
error_type="error",
suggestions=["Check OpenClaw service status"],
auto_fixable=False,
risk_level="low",
analyzed_by="fallback",
confidence=0.0,
)
async def send_ci_failure_telegram_alert(
diagnosis_id: str,
repo: str,
workflow_name: str,
workflow_url: str,
sender: str,
diagnosis: CIFailureDiagnosis | None,
repair_decision=None, # Phase 13.1 #78: CIRepairDecision
):
"""
發送 CI 失敗診斷 Telegram 通知 (Phase 13.1 #76-78)
"""
try:
telegram = get_telegram_gateway()
# 構建訊息
risk_emoji = {
"low": "🟢",
"medium": "🟡",
"high": "🟠",
"critical": "🔴",
}
emoji = risk_emoji.get(diagnosis.risk_level if diagnosis else "medium", "🟡")
# 修復決策狀態
decision_text = "❓ 待評估"
if repair_decision:
decision_map = {
"auto_execute": "🤖 自動修復中",
"telegram_confirm": "📱 等待確認",
"approval_required": "📋 需人工審核",
"blocked": "🚫 禁止自動修復",
}
decision_text = decision_map.get(repair_decision.execution_decision.value, "❓ 未知")
message_lines = [
f"{emoji} **CI 失敗診斷** | {repo}",
"",
f"📋 **Workflow**: {workflow_name}",
f"👤 **觸發者**: {sender}",
f"🔗 [查看 Workflow]({workflow_url})",
"",
]
if diagnosis:
message_lines.extend([
f"**📝 摘要**: {diagnosis.summary}",
f"**🔍 根因**: {diagnosis.root_cause}",
f"**⚠️ 錯誤類型**: {diagnosis.error_type}",
f"**🎯 風險等級**: {diagnosis.risk_level.upper()}",
f"**🔧 修復決策**: {decision_text}",
"",
])
if diagnosis.suggestions:
message_lines.append("**💡 AI 建議**:")
for i, suggestion in enumerate(diagnosis.suggestions[:3], 1):
message_lines.append(f" {i}. {suggestion}")
# 顯示修復建議 (Phase 13.1 #78)
if repair_decision and repair_decision.recommendations:
message_lines.extend(["", "**🔨 修復選項**:"])
for i, rec in enumerate(repair_decision.recommendations[:2], 1):
confidence_pct = int(rec.confidence * 100)
message_lines.append(
f" {i}. `{rec.action.value}` ({confidence_pct}% 信心)"
)
if rec.command:
message_lines.append(f" `{rec.command[:50]}...`" if len(rec.command) > 50 else f" `{rec.command}`")
message_lines.extend([
"",
f"🆔 `{diagnosis_id}`",
])
message = "\n".join(message_lines)
await telegram.send_message(
message=message,
parse_mode="Markdown",
)
logger.info(
"ci_failure_telegram_alert_sent",
diagnosis_id=diagnosis_id,
repo=repo,
repair_decision=repair_decision.execution_decision.value if repair_decision else None,
)
except Exception as e:
logger.exception(
"ci_failure_telegram_alert_failed",
diagnosis_id=diagnosis_id,
error=str(e),
)
async def create_ci_failure_approval(
diagnosis_id: str,
repo: str,
workflow_run: GitHubWorkflowRun,
diagnosis: CIFailureDiagnosis,
) -> str:
"""
為需要人工審核的 CI 修復建立 Approval 記錄 (Phase 13.1 #76)
Returns:
str: Approval ID
"""
try:
approval_service = get_approval_service()
# 映射風險等級
risk_map = {
"low": RiskLevel.LOW,
"medium": RiskLevel.MEDIUM,
"high": RiskLevel.HIGH,
"critical": RiskLevel.CRITICAL,
}
risk_level = risk_map.get(diagnosis.risk_level, RiskLevel.MEDIUM)
# P1-2 修正: 欄位對齊 ApprovalRequestBase (2026-03-29)
suggestion = diagnosis.fix_command or "; ".join(diagnosis.suggestions[:2])
approval_request = ApprovalRequestCreate(
action=f"CI Failure Repair: {repo}",
description=f"Root Cause: {diagnosis.root_cause}\nSuggestion: {suggestion}",
risk_level=risk_level,
blast_radius=BlastRadius(
affected_pods=1 if diagnosis.auto_fixable else 2,
estimated_downtime="~5min",
related_services=[repo],
data_impact=DataImpact.NONE,
),
dry_run_checks=[],
requested_by="github-webhook",
metadata={
"source": "github",
"alert_type": "ci_failure_repair",
"target_resource": repo,
"namespace": "github-actions",
"ci_diagnosis_id": diagnosis_id,
"workflow_name": workflow_run.name,
"workflow_id": workflow_run.id,
"workflow_url": workflow_run.html_url,
"head_sha": workflow_run.head_sha,
"error_type": diagnosis.error_type,
"auto_fixable": diagnosis.auto_fixable,
"fix_command": diagnosis.fix_command,
"llm_provider": diagnosis.analyzed_by,
"llm_confidence": diagnosis.confidence,
},
)
# 創建 Approval
approval_id = str(uuid.uuid4())
await approval_service.create_approval(
approval_id=approval_id,
request=approval_request,
)
logger.info(
"ci_failure_approval_created",
approval_id=approval_id,
diagnosis_id=diagnosis_id,
risk_level=risk_level.value,
)
return approval_id
except Exception as e:
logger.exception("ci_failure_approval_creation_failed", error=str(e))
return f"temp-{uuid.uuid4().hex[:8]}"
async def save_review_result(
review_id: str,
event_type: str,
repo: str,
target: str,
analysis: CodeReviewResult | None,
metadata: dict,
):
"""
儲存審查結果到 Redis (透過 Service)
Key: github_review:{review_id}
TTL: 7 天
"""
result = {
"review_id": review_id,
"event_type": event_type,
"repo": repo,
"target": target,
"created_at": now_taipei_iso(),
"analysis": analysis.model_dump() if analysis else None,
"metadata": metadata,
}
service = get_github_webhook_service()
success = await service.save_review_result(review_id, result)
if success:
logger.info(
"github_review_saved",
review_id=review_id,
ttl_days=7,
)
else:
logger.error("github_review_save_failed", review_id=review_id)
async def send_github_telegram_alert(
review_id: str,
event_type: str,
repo: str,
target: str,
url: str,
author: str,
analysis: CodeReviewResult | None,
):
"""
發送 GitHub 審查告警到 Telegram
格式:
═══════════════════════════
🔍 GITHUB CODE REVIEW
═══════════════════════════
📦 repo/name
🔀 PR #123: Feature title
👤 @author
───────────────────────────
📊 品質分數: 85/100
⚠️ 發現 2 個問題
🔐 1 個安全疑慮
───────────────────────────
🧠 AI 摘要:
「代碼品質良好，但建議...」
───────────────────────────
[ 🔗 查看 PR ] [ 📋 詳情 ]
"""
try:
telegram = get_telegram_gateway()
# 檢查是否有設定 Bot Token
if not settings.OPENCLAW_TG_BOT_TOKEN:
logger.debug(
"github_telegram_skipped",
reason="Bot token not configured",
)
return
await telegram.initialize()
# 構建訊息
quality_emoji = "🟢" if analysis and analysis.quality_score >= 80 else "🟡" if analysis and analysis.quality_score >= 60 else "🔴"
message_lines = [
"═══════════════════════════",
"🔍 GITHUB CODE REVIEW",
"═══════════════════════════",
f"📦 {repo}",
f"🔀 {target}",
f"👤 @{author}",
"───────────────────────────",
]
if analysis:
message_lines.extend([
f"{quality_emoji} 品質分數: {analysis.quality_score:.0f}/100",
])
if analysis.issues:
message_lines.append(f"⚠️ 發現 {len(analysis.issues)} 個問題")
if analysis.security_concerns:
message_lines.append(f"🔐 {len(analysis.security_concerns)} 個安全疑慮")
message_lines.extend([
"───────────────────────────",
"🧠 AI 摘要:",
f"「{analysis.summary[:150]}」",
])
else:
message_lines.append("❌ AI 分析失敗")
message_lines.extend([
"───────────────────────────",
f"🔗 {url}",
f"📋 Review ID: {review_id}",
])
message = "\n".join(message_lines)
# 發送訊息 (使用 send_notification 而非 send_message)
await telegram.send_notification(message)
logger.info(
"github_telegram_sent",
review_id=review_id,
repo=repo,
)
except Exception as e:
logger.exception("github_telegram_failed", error=str(e))
async def create_github_approval(
review_id: str,
repo: str,
target: str,
url: str,
analysis: CodeReviewResult,
) -> str:
"""
為有安全疑慮的 PR 建立 Approval 記錄
Returns:
str: Approval ID
"""
try:
approval_service = get_approval_service()
# 決定風險等級
if len(analysis.security_concerns) > 2 or analysis.quality_score < 50:
risk_level = RiskLevel.CRITICAL
elif analysis.security_concerns or analysis.quality_score < 70:
risk_level = RiskLevel.HIGH
else:
risk_level = RiskLevel.MEDIUM
# P1-2 修正: 欄位對齊 ApprovalRequestBase (2026-03-29)
root_cause = f"Code review found security concerns in {target}"
suggestion = f"Review {len(analysis.security_concerns)} security concern(s): {', '.join(analysis.security_concerns[:3])}"
approval_request = ApprovalRequestCreate(
action=f"Code Review Security: {repo}",
description=f"Root Cause: {root_cause}\nSuggestion: {suggestion}",
risk_level=risk_level,
blast_radius=BlastRadius(
affected_pods=1,
estimated_downtime="0",
related_services=[repo],
data_impact=DataImpact.READ_ONLY,
),
dry_run_checks=[],
requested_by="github-webhook",
metadata={
"source": "github",
"alert_type": "code_review_security",
"target_resource": repo,
"namespace": "github",
"github_review_id": review_id,
"target": target,
"url": url,
"quality_score": analysis.quality_score,
"security_concerns": analysis.security_concerns,
"issues_count": len(analysis.issues),
"llm_provider": analysis.analyzed_by,
"llm_confidence": analysis.confidence,
},
)
# 創建 Approval
approval_id = str(uuid.uuid4())
await approval_service.create_approval(
approval_id=approval_id,
request=approval_request,
)
logger.info(
"github_approval_created",
approval_id=approval_id,
review_id=review_id,
risk_level=risk_level.value,
)
return approval_id
except Exception as e:
logger.exception("github_approval_creation_failed", error=str(e))
return f"temp-{uuid.uuid4().hex[:8]}"
# =============================================================================
# Query Endpoints
# =============================================================================

30
apps/api/src/api/v1/webhooks.py
View File

@@ -44,8 +44,9 @@ from src.models.approval import (
)
from src.models.incident import Incident, IncidentStatus, Severity, Signal
# R4 #129 (2026-04-01 ogt): AlertPayload/AlertResponse 移至 models 層，AlertAnalyzer 移至 services 層
# ogt 更新 v1.1 2026-04-01 台北時間: generate_alert_fingerprint 移至 alert_analyzer_service (ADR-024)
from src.models.webhook import AlertPayload, AlertResponse
from src.services.alert_analyzer_service import AlertAnalyzer
from src.services.alert_analyzer_service import AlertAnalyzer, generate_alert_fingerprint
from src.services.approval_db import get_approval_service
# Phase 17 P0: Service 層 (消除 Router 直接存取 Redis)
@@ -337,32 +338,7 @@ async def verify_webhook_signature(
return True
# =============================================================================
# 戰略 B: 告警指紋生成
# =============================================================================
def generate_alert_fingerprint(alert: "AlertPayload") -> str:
"""
生成告警唯一指紋 (SHA256 Hash)
指紋組成: namespace:deployment:alert_type:target_resource
同一個告警模式（相同位置、相同類型）會產生相同指紋，
用於識別重複告警並進行聚合。
"""
# 從 labels 取得 deployment，如果沒有則用 target_resource
deployment = ""
if alert.labels:
deployment = alert.labels.get("deployment", alert.labels.get("app", ""))
if not deployment:
deployment = alert.target_resource
# 組合指紋來源
fingerprint_source = f"{alert.namespace}:{deployment}:{alert.alert_type}:{alert.target_resource}"
# SHA256 Hash
return hashlib.sha256(fingerprint_source.encode()).hexdigest()[:32]
# generate_alert_fingerprint 已移至 src/services/alert_analyzer_service.py (ogt v1.1 2026-04-01 台北時間)
# 戰略 B: 滑動時間窗 (5 分鐘)
DEBOUNCE_WINDOW_MINUTES = 5

11
apps/api/src/core/config.py
View File

@@ -51,17 +51,6 @@ class Settings(BaseSettings):
description="Enable mock mode for external services (Redis, Ollama, OpenClaw, PostgreSQL, SigNoz)",
)
# ==========================================================================
# Phase 16: leWOOOgo 積木化絞殺者模式 (Strangler Fig Pattern)
# 2026-03-26 統帥批准立即執行
# 2026-04-01 ogt: Phase R-R2 完成，內嵌版本已移除，此開關已失效 (ADR-046 P2-03)
# 回滾方式: git revert c7b3f8f d17b67c + kubectl rollout restart deployment/awoooi-api
# ==========================================================================
USE_NEW_ENGINE: bool = Field(
default=True,
description="[已失效] Phase R-R2 後內嵌版本已移除，此開關無消費者，僅保留供環境相容性",
)
# ==========================================================================
# Phase 22: OpenClaw + Nemotron 協作 (ADR-044)
# 2026-03-31 Claude Code: 統帥批准實作

14
apps/api/src/models/approval.py
View File

@@ -14,7 +14,7 @@ from datetime import UTC, datetime
from enum import Enum
from uuid import UUID, uuid4
from pydantic import BaseModel, Field
from pydantic import BaseModel, ConfigDict, Field
# =============================================================================
# Enums
@@ -123,11 +123,13 @@ class Signature(BaseModel):
description="Telegram 訊息 ID",
)
class Config:
json_encoders = {
# Claude 遷移 Pydantic v1→v2 2026-04-01 Asia/Taipei
model_config = ConfigDict(
json_encoders={
datetime: lambda v: v.isoformat(),
UUID: lambda v: str(v),
}
)
# =============================================================================
@@ -185,11 +187,13 @@ class ApprovalRequest(ApprovalRequestBase):
"""檢查某人是否已簽核"""
return any(s.signer_id == signer_id for s in self.signatures)
class Config:
json_encoders = {
# Claude 遷移 Pydantic v1→v2 2026-04-01 Asia/Taipei
model_config = ConfigDict(
json_encoders={
datetime: lambda v: v.isoformat(),
UUID: lambda v: str(v),
}
)
# =============================================================================

32
apps/api/src/models/incident.py
View File

@@ -25,7 +25,7 @@ from enum import Enum
from typing import Literal
from uuid import UUID, uuid4
from pydantic import BaseModel, Field
from pydantic import BaseModel, ConfigDict, Field
# 復用現有模型 (避免重複定義)
from src.models.approval import BlastRadius
@@ -107,10 +107,10 @@ class Signal(BaseModel):
description="告警指紋 Hash，用於去重與聚合",
)
class Config:
json_encoders = {
datetime: lambda v: v.isoformat(),
}
# Claude 遷移 Pydantic v1→v2 2026-04-01 Asia/Taipei
model_config = ConfigDict(
json_encoders={datetime: lambda v: v.isoformat()}
)
# =============================================================================
@@ -181,10 +181,10 @@ class AIDecisionChain(BaseModel):
inference_completed_at: datetime = Field(..., description="推論完成時間")
latency_ms: int = Field(..., description="推論延遲 (毫秒)")
class Config:
json_encoders = {
datetime: lambda v: v.isoformat(),
}
# Claude 遷移 Pydantic v1→v2 2026-04-01 Asia/Taipei
model_config = ConfigDict(
json_encoders={datetime: lambda v: v.isoformat()}
)
# =============================================================================
@@ -423,11 +423,13 @@ class Incident(BaseModel):
description="是否已向量化到 Vector DB (Semantic Memory)",
)
class Config:
json_encoders = {
# Claude 遷移 Pydantic v1→v2 2026-04-01 Asia/Taipei
model_config = ConfigDict(
json_encoders={
datetime: lambda v: v.isoformat(),
UUID: lambda v: str(v),
}
)
# =============================================================================
@@ -487,7 +489,7 @@ class IncidentResponse(BaseModel):
closed_at=incident.closed_at,
)
class Config:
json_encoders = {
datetime: lambda v: v.isoformat(),
}
# Claude 遷移 Pydantic v1→v2 2026-04-01 Asia/Taipei
model_config = ConfigDict(
json_encoders={datetime: lambda v: v.isoformat()}
)

30
apps/api/src/services/alert_analyzer_service.py
View File

@@ -19,6 +19,8 @@ Alert Analyzer Service - 告警分析大腦
建立者: Claude Code (R4 Router 瘦身 #129)
"""
import hashlib
from src.models.approval import (
ApprovalRequestCreate,
BlastRadius,
@@ -30,6 +32,34 @@ from src.models.webhook import AlertPayload
from src.utils.k8s_naming import normalize_resource_name
# =============================================================================
# 戰略 B: 告警指紋生成
# ogt 移至 Service 層 v1.1 2026-04-01 台北時間 (ADR-024 R4 #129)
# =============================================================================
def generate_alert_fingerprint(alert: AlertPayload) -> str:
"""
生成告警唯一指紋 (SHA256 Hash)
指紋組成: namespace:deployment:alert_type:target_resource
同一個告警模式（相同位置、相同類型）會產生相同指紋，
用於識別重複告警並進行聚合。
"""
# 從 labels 取得 deployment，如果沒有則用 target_resource
deployment = ""
if alert.labels:
deployment = alert.labels.get("deployment", alert.labels.get("app", ""))
if not deployment:
deployment = alert.target_resource
# 組合指紋來源
fingerprint_source = f"{alert.namespace}:{deployment}:{alert.alert_type}:{alert.target_resource}"
# SHA256 Hash
return hashlib.sha256(fingerprint_source.encode()).hexdigest()[:32]
class AlertAnalyzer:
"""
告警分析器 - AWOOOI 核心大腦

945
apps/api/src/services/github_webhook_service.py
View File

@@ -1,19 +1,30 @@
"""
GitHub Webhook Service - Phase 13.1
====================================
封裝 GitHub Webhook 相關的 Redis 操作
封裝 GitHub Webhook 相關的業務邏輯與 Redis 操作
遵循 leWOOOgo 積木化原則:
- Router 層不直接存取 Redis
- 透過 Service 層封裝資料存取邏輯
- Router 層不包含業務邏輯 (orchestration)
- 透過 Service 層封裝資料存取邏輯與協調流程
# Claude Code 移動協調函數至 Service 層 v2.1 2026-04-01 11:00 (台北)
# 移動: review_pull_request, review_push, diagnose_ci_failure,
# fetch_pr_diff, call_openclaw_*, send_*_telegram_alert,
# create_*_approval, save_review_result
# 移動: 結果模型 CodeReviewResult, CIFailureDiagnosis
"""
import json
import uuid
from typing import Protocol
import structlog
from pydantic import BaseModel, Field
from src.core.config import settings
from src.core.redis_client import get_redis
from src.utils.timezone import now_taipei_iso
logger = structlog.get_logger(__name__)
@@ -22,6 +33,39 @@ logger = structlog.get_logger(__name__)
GITHUB_REVIEW_TTL_SECONDS = 7 * 24 * 60 * 60
# =============================================================================
# Result Models (Service 層數據契約)
# =============================================================================
class CodeReviewResult(BaseModel):
"""AI 代碼審查結果"""
summary: str = Field(..., description="審查摘要")
issues: list[dict] = Field(default=[], description="發現的問題列表")
suggestions: list[dict] = Field(default=[], description="改進建議")
security_concerns: list[str] = Field(default=[], description="安全疑慮")
quality_score: float = Field(..., ge=0, le=100, description="代碼品質分數 0-100")
analyzed_by: str = Field(..., description="分析模型 (ollama/claude)")
confidence: float = Field(..., ge=0, le=1, description="分析信心度 0-1")
class CIFailureDiagnosis(BaseModel):
"""CI 失敗診斷結果 (Phase 13.1 #76)"""
summary: str = Field(..., description="失敗摘要")
root_cause: str = Field(..., description="根本原因分析")
failed_step: str | None = Field(None, description="失敗的步驟名稱")
error_type: str = Field(..., description="錯誤類型 (build/test/lint/deploy/timeout)")
suggestions: list[str] = Field(default=[], description="修復建議")
auto_fixable: bool = Field(False, description="是否可自動修復")
fix_command: str | None = Field(None, description="自動修復指令 (如可自動修復)")
risk_level: str = Field("medium", description="風險等級 (low/medium/high/critical)")
analyzed_by: str = Field(..., description="分析模型")
confidence: float = Field(..., ge=0, le=1, description="信心度")
# =============================================================================
# Repository Interface & Implementation
# =============================================================================
class IGitHubReviewRepository(Protocol):
"""GitHub Review Repository Interface"""
@@ -86,28 +130,921 @@ class GitHubReviewRedisRepository:
return None
# =============================================================================
# Service
# =============================================================================
class GitHubWebhookService:
"""
GitHub Webhook 服務
封裝審查結果的儲存與查詢
封裝審查結果的儲存、查詢以及全部業務協調流程:
- PR 代碼審查 (review_pull_request)
- Push 代碼審查 (review_push)
- CI 失敗診斷 (diagnose_ci_failure)
- OpenClaw 呼叫封裝
- Telegram 通知
- Approval 建立
"""
def __init__(self, repository: IGitHubReviewRepository | None = None):
self._repository = repository or GitHubReviewRedisRepository()
# ------------------------------------------------------------------
# Redis CRUD
# ------------------------------------------------------------------
async def save_review_result(
self,
review_id: str,
review_data: dict,
ttl: int | None = None,
) -> bool:
"""儲存審查結果"""
"""儲存審查結果 (支援自訂 TTL)"""
if ttl is not None and ttl != GITHUB_REVIEW_TTL_SECONDS:
# 直接寫 Redis 以使用自訂 TTL
try:
redis_client = get_redis()
key = f"{self._repository.KEY_PREFIX}{review_id}" # type: ignore[attr-defined]
await redis_client.set(
key,
json.dumps(review_data, ensure_ascii=False),
ex=ttl,
)
logger.debug("github_review_saved_custom_ttl", review_id=review_id, ttl=ttl)
return True
except Exception as e:
logger.error("github_review_save_failed", review_id=review_id, error=str(e))
return False
return await self._repository.save_review(review_id, review_data)
async def get_review_result(self, review_id: str) -> dict | None:
"""取得審查結果"""
return await self._repository.get_review(review_id)
# ------------------------------------------------------------------
# Internal helpers: OpenClaw
# ------------------------------------------------------------------
async def _fetch_pr_diff(self, diff_url: str) -> str:
"""取得 PR diff 內容 (委派給 GitHubApiService)"""
from src.services.github_api_service import get_github_api_service
service = get_github_api_service()
return await service.fetch_pr_diff(diff_url)
async def _call_openclaw_code_review(
self,
repo_name: str,
pr_title: str,
pr_body: str,
diff_content: str,
changed_files: int,
additions: int,
deletions: int,
) -> CodeReviewResult | None:
"""
呼叫 OpenClaw 進行 PR 代碼審查
優先使用 Ollama (本地，零成本)
Fallback: Claude (大型 PR)
Phase 22 P0 修復: 使用 OpenClawHttpService (2026-03-31)
"""
try:
from src.services.openclaw_http_service import get_openclaw_http_service
service = get_openclaw_http_service()
data = await service.code_review(
repo_name=repo_name,
pr_title=pr_title,
pr_body=pr_body,
diff_content=diff_content,
changed_files=changed_files,
additions=additions,
deletions=deletions,
prefer_local=True,
timeout=120.0,
)
if data:
return CodeReviewResult(**data)
return None
except Exception as e:
logger.exception("openclaw_code_review_error", error=str(e))
return None
async def _call_openclaw_push_review(
self,
repo_name: str,
ref: str,
commits: list[dict],
files_changed: dict,
) -> CodeReviewResult | None:
"""
呼叫 OpenClaw 進行 Push 代碼審查
Phase 22 P0 修復: 使用 OpenClawHttpService (2026-03-31)
"""
try:
from src.services.openclaw_http_service import get_openclaw_http_service
service = get_openclaw_http_service()
data = await service.push_review(
repo_name=repo_name,
ref=ref,
commits=commits,
files_changed=files_changed,
prefer_local=True,
timeout=120.0,
)
if data:
return CodeReviewResult(**data)
return None
except Exception as e:
logger.exception("openclaw_push_review_error", error=str(e))
return None
async def _call_openclaw_ci_diagnosis(
self,
repo_name: str,
failure_context: dict,
) -> CIFailureDiagnosis | None:
"""
呼叫 OpenClaw 進行 CI 失敗診斷 (Phase 13.1 #76)
分析 CI/CD pipeline 失敗原因，提供根因分析和修復建議
Phase 22 P0 修復: 使用 OpenClawHttpService (2026-03-31)
"""
try:
from src.services.openclaw_http_service import get_openclaw_http_service
service = get_openclaw_http_service()
data = await service.ci_diagnosis(
repo_name=repo_name,
failure_context=failure_context,
prefer_local=True,
timeout=120.0,
)
if data:
return CIFailureDiagnosis(**data)
else:
# 返回基本診斷結果 (API 失敗時的 fallback)
return CIFailureDiagnosis(
summary=f"CI workflow '{failure_context.get('workflow_name')}' failed",
root_cause="OpenClaw API unavailable, manual investigation required",
error_type="unknown",
suggestions=["Check workflow logs manually", "Verify runner status"],
auto_fixable=False,
risk_level="medium",
analyzed_by="fallback",
confidence=0.0, # 🔴 Fallback 不是 AI 分析
)
except Exception as e:
logger.exception("openclaw_ci_diagnosis_error", error=str(e))
return CIFailureDiagnosis(
summary="CI diagnosis error",
root_cause=f"Exception: {str(e)}",
error_type="error",
suggestions=["Check OpenClaw service status"],
auto_fixable=False,
risk_level="low",
analyzed_by="fallback",
confidence=0.0,
)
# ------------------------------------------------------------------
# Internal helpers: persist
# ------------------------------------------------------------------
async def _save_review_with_analysis(
self,
review_id: str,
event_type: str,
repo: str,
target: str,
analysis: CodeReviewResult | None,
metadata: dict,
) -> None:
"""
組裝並儲存代碼審查結果到 Redis (透過 Service)
Key: github_review:{review_id}
TTL: 7 天
"""
result = {
"review_id": review_id,
"event_type": event_type,
"repo": repo,
"target": target,
"created_at": now_taipei_iso(),
"analysis": analysis.model_dump() if analysis else None,
"metadata": metadata,
}
success = await self._repository.save_review(review_id, result)
if success:
logger.info("github_review_saved", review_id=review_id, ttl_days=7)
else:
logger.error("github_review_save_failed", review_id=review_id)
# ------------------------------------------------------------------
# Internal helpers: Telegram
# ------------------------------------------------------------------
async def _send_github_telegram_alert(
self,
review_id: str,
event_type: str,
repo: str,
target: str,
url: str,
author: str,
analysis: CodeReviewResult | None,
) -> None:
"""
發送 GitHub 審查告警到 Telegram
格式:
═══════════════════════════
🔍 GITHUB CODE REVIEW
═══════════════════════════
📦 repo/name
🔀 PR #123: Feature title
👤 @author
───────────────────────────
📊 品質分數: 85/100
⚠️ 發現 2 個問題
🔐 1 個安全疑慮
───────────────────────────
🧠 AI 摘要:
「代碼品質良好，但建議...」
───────────────────────────
[ 🔗 查看 PR ] [ 📋 詳情 ]
"""
try:
from src.services.telegram_gateway import get_telegram_gateway
telegram = get_telegram_gateway()
# 檢查是否有設定 Bot Token
if not settings.OPENCLAW_TG_BOT_TOKEN:
logger.debug("github_telegram_skipped", reason="Bot token not configured")
return
await telegram.initialize()
# 構建訊息
quality_emoji = (
"🟢" if analysis and analysis.quality_score >= 80
else "🟡" if analysis and analysis.quality_score >= 60
else "🔴"
)
message_lines = [
"═══════════════════════════",
"🔍 GITHUB CODE REVIEW",
"═══════════════════════════",
f"📦 {repo}",
f"🔀 {target}",
f"👤 @{author}",
"───────────────────────────",
]
if analysis:
message_lines.extend([
f"{quality_emoji} 品質分數: {analysis.quality_score:.0f}/100",
])
if analysis.issues:
message_lines.append(f"⚠️ 發現 {len(analysis.issues)} 個問題")
if analysis.security_concerns:
message_lines.append(f"🔐 {len(analysis.security_concerns)} 個安全疑慮")
message_lines.extend([
"───────────────────────────",
"🧠 AI 摘要:",
f"「{analysis.summary[:150]}」",
])
else:
message_lines.append("❌ AI 分析失敗")
message_lines.extend([
"───────────────────────────",
f"🔗 {url}",
f"📋 Review ID: {review_id}",
])
message = "\n".join(message_lines)
# 發送訊息 (使用 send_notification 而非 send_message)
await telegram.send_notification(message)
logger.info("github_telegram_sent", review_id=review_id, repo=repo, event_type=event_type)
except Exception as e:
logger.exception("github_telegram_failed", error=str(e))
async def _send_ci_failure_telegram_alert(
self,
diagnosis_id: str,
repo: str,
workflow_name: str,
workflow_url: str,
sender: str,
diagnosis: CIFailureDiagnosis | None,
repair_decision=None, # Phase 13.1 #78: CIRepairDecision
) -> None:
"""
發送 CI 失敗診斷 Telegram 通知 (Phase 13.1 #76-78)
"""
try:
from src.services.telegram_gateway import get_telegram_gateway
telegram = get_telegram_gateway()
# 構建訊息
risk_emoji = {
"low": "🟢",
"medium": "🟡",
"high": "🟠",
"critical": "🔴",
}
emoji = risk_emoji.get(diagnosis.risk_level if diagnosis else "medium", "🟡")
# 修復決策狀態
decision_text = "❓ 待評估"
if repair_decision:
decision_map = {
"auto_execute": "🤖 自動修復中",
"telegram_confirm": "📱 等待確認",
"approval_required": "📋 需人工審核",
"blocked": "🚫 禁止自動修復",
}
decision_text = decision_map.get(repair_decision.execution_decision.value, "❓ 未知")
message_lines = [
f"{emoji} **CI 失敗診斷** | {repo}",
"",
f"📋 **Workflow**: {workflow_name}",
f"👤 **觸發者**: {sender}",
f"🔗 [查看 Workflow]({workflow_url})",
"",
]
if diagnosis:
message_lines.extend([
f"**📝 摘要**: {diagnosis.summary}",
f"**🔍 根因**: {diagnosis.root_cause}",
f"**⚠️ 錯誤類型**: {diagnosis.error_type}",
f"**🎯 風險等級**: {diagnosis.risk_level.upper()}",
f"**🔧 修復決策**: {decision_text}",
"",
])
if diagnosis.suggestions:
message_lines.append("**💡 AI 建議**:")
for i, suggestion in enumerate(diagnosis.suggestions[:3], 1):
message_lines.append(f" {i}. {suggestion}")
# 顯示修復建議 (Phase 13.1 #78)
if repair_decision and repair_decision.recommendations:
message_lines.extend(["", "**🔨 修復選項**:"])
for i, rec in enumerate(repair_decision.recommendations[:2], 1):
confidence_pct = int(rec.confidence * 100)
message_lines.append(
f" {i}. `{rec.action.value}` ({confidence_pct}% 信心)"
)
if rec.command:
message_lines.append(
f" `{rec.command[:50]}...`" if len(rec.command) > 50
else f" `{rec.command}`"
)
message_lines.extend([
"",
f"🆔 `{diagnosis_id}`",
])
message = "\n".join(message_lines)
await telegram.send_message(message=message, parse_mode="Markdown")
logger.info(
"ci_failure_telegram_alert_sent",
diagnosis_id=diagnosis_id,
repo=repo,
repair_decision=repair_decision.execution_decision.value if repair_decision else None,
)
except Exception as e:
logger.exception(
"ci_failure_telegram_alert_failed",
diagnosis_id=diagnosis_id,
error=str(e),
)
# ------------------------------------------------------------------
# Internal helpers: Approval
# ------------------------------------------------------------------
async def _create_github_approval(
self,
review_id: str,
repo: str,
target: str,
url: str,
analysis: CodeReviewResult,
) -> str:
"""
為有安全疑慮的 PR 建立 Approval 記錄
Returns:
str: Approval ID
"""
try:
from src.models.approval import (
ApprovalRequestCreate,
BlastRadius,
DataImpact,
RiskLevel,
)
from src.services.approval_db import get_approval_service
approval_service = get_approval_service()
# 決定風險等級
if len(analysis.security_concerns) > 2 or analysis.quality_score < 50:
risk_level = RiskLevel.CRITICAL
elif analysis.security_concerns or analysis.quality_score < 70:
risk_level = RiskLevel.HIGH
else:
risk_level = RiskLevel.MEDIUM
# P1-2 修正: 欄位對齊 ApprovalRequestBase (2026-03-29)
root_cause = f"Code review found security concerns in {target}"
suggestion = f"Review {len(analysis.security_concerns)} security concern(s): {', '.join(analysis.security_concerns[:3])}"
approval_request = ApprovalRequestCreate(
action=f"Code Review Security: {repo}",
description=f"Root Cause: {root_cause}\nSuggestion: {suggestion}",
risk_level=risk_level,
blast_radius=BlastRadius(
affected_pods=1,
estimated_downtime="0",
related_services=[repo],
data_impact=DataImpact.READ_ONLY,
),
dry_run_checks=[],
requested_by="github-webhook",
metadata={
"source": "github",
"alert_type": "code_review_security",
"target_resource": repo,
"namespace": "github",
"github_review_id": review_id,
"target": target,
"url": url,
"quality_score": analysis.quality_score,
"security_concerns": analysis.security_concerns,
"issues_count": len(analysis.issues),
"llm_provider": analysis.analyzed_by,
"llm_confidence": analysis.confidence,
},
)
# 創建 Approval
approval_id = str(uuid.uuid4())
await approval_service.create_approval(
approval_id=approval_id,
request=approval_request,
)
logger.info(
"github_approval_created",
approval_id=approval_id,
review_id=review_id,
risk_level=risk_level.value,
)
return approval_id
except Exception as e:
logger.exception("github_approval_creation_failed", error=str(e))
return f"temp-{uuid.uuid4().hex[:8]}"
async def _create_ci_failure_approval(
self,
diagnosis_id: str,
repo: str,
workflow_run, # GitHubWorkflowRun — 避免循環 import，用 Any
diagnosis: CIFailureDiagnosis,
) -> str:
"""
為需要人工審核的 CI 修復建立 Approval 記錄 (Phase 13.1 #76)
Returns:
str: Approval ID
"""
try:
from src.models.approval import (
ApprovalRequestCreate,
BlastRadius,
DataImpact,
RiskLevel,
)
from src.services.approval_db import get_approval_service
approval_service = get_approval_service()
# 映射風險等級
risk_map = {
"low": RiskLevel.LOW,
"medium": RiskLevel.MEDIUM,
"high": RiskLevel.HIGH,
"critical": RiskLevel.CRITICAL,
}
risk_level = risk_map.get(diagnosis.risk_level, RiskLevel.MEDIUM)
# P1-2 修正: 欄位對齊 ApprovalRequestBase (2026-03-29)
suggestion = diagnosis.fix_command or "; ".join(diagnosis.suggestions[:2])
approval_request = ApprovalRequestCreate(
action=f"CI Failure Repair: {repo}",
description=f"Root Cause: {diagnosis.root_cause}\nSuggestion: {suggestion}",
risk_level=risk_level,
blast_radius=BlastRadius(
affected_pods=1 if diagnosis.auto_fixable else 2,
estimated_downtime="~5min",
related_services=[repo],
data_impact=DataImpact.NONE,
),
dry_run_checks=[],
requested_by="github-webhook",
metadata={
"source": "github",
"alert_type": "ci_failure_repair",
"target_resource": repo,
"namespace": "github-actions",
"ci_diagnosis_id": diagnosis_id,
"workflow_name": workflow_run.name,
"workflow_id": workflow_run.id,
"workflow_url": workflow_run.html_url,
"head_sha": workflow_run.head_sha,
"error_type": diagnosis.error_type,
"auto_fixable": diagnosis.auto_fixable,
"fix_command": diagnosis.fix_command,
"llm_provider": diagnosis.analyzed_by,
"llm_confidence": diagnosis.confidence,
},
)
# 創建 Approval
approval_id = str(uuid.uuid4())
await approval_service.create_approval(
approval_id=approval_id,
request=approval_request,
)
logger.info(
"ci_failure_approval_created",
approval_id=approval_id,
diagnosis_id=diagnosis_id,
risk_level=risk_level.value,
)
return approval_id
except Exception as e:
logger.exception("ci_failure_approval_creation_failed", error=str(e))
return f"temp-{uuid.uuid4().hex[:8]}"
# ------------------------------------------------------------------
# Public: Orchestration (Background Tasks)
# ------------------------------------------------------------------
async def review_pull_request(
self,
repo, # GitHubRepository
pr, # GitHubPullRequest
sender, # GitHubUser
review_id: str,
action: str,
) -> None:
"""
背景任務: PR 代碼審查
1. 取得 PR diff
2. 呼叫 OpenClaw 分析
3. 儲存結果到 Redis
4. 發送 Telegram 通知
5. 建立 Approval (可選)
"""
try:
logger.info(
"github_pr_review_started",
review_id=review_id,
repo=repo.full_name,
pr_number=pr.number,
sender=sender.login,
)
# 1. 取得 PR diff
diff_content = await self._fetch_pr_diff(pr.diff_url)
# 2. 呼叫 OpenClaw 進行代碼審查
analysis = await self._call_openclaw_code_review(
repo_name=repo.full_name,
pr_title=pr.title,
pr_body=pr.body or "",
diff_content=diff_content,
changed_files=pr.changed_files,
additions=pr.additions,
deletions=pr.deletions,
)
# 3. 儲存結果到 Redis
await self._save_review_with_analysis(
review_id=review_id,
event_type="pull_request",
repo=repo.full_name,
target=f"PR #{pr.number}",
analysis=analysis,
metadata={
"pr_number": pr.number,
"pr_title": pr.title,
"pr_url": pr.html_url,
"author": pr.user.login,
"action": action,
"changed_files": pr.changed_files,
"additions": pr.additions,
"deletions": pr.deletions,
},
)
# 4. 發送 Telegram 通知
await self._send_github_telegram_alert(
review_id=review_id,
event_type="pull_request",
repo=repo.full_name,
target=f"PR #{pr.number}: {pr.title[:50]}",
url=pr.html_url,
author=pr.user.login,
analysis=analysis,
)
# 5. 如果有安全疑慮，建立 Approval
if analysis and analysis.security_concerns:
await self._create_github_approval(
review_id=review_id,
repo=repo.full_name,
target=f"PR #{pr.number}",
url=pr.html_url,
analysis=analysis,
)
logger.info(
"github_pr_review_completed",
review_id=review_id,
quality_score=analysis.quality_score if analysis else None,
has_security_concerns=bool(analysis and analysis.security_concerns),
)
except Exception as e:
logger.exception(
"github_pr_review_failed",
review_id=review_id,
error=str(e),
)
async def review_push(
self,
repo, # GitHubRepository
commits: list, # list[GitHubCommit]
sender, # GitHubUser
review_id: str,
ref: str,
before_sha: str | None,
after_sha: str | None,
) -> None:
"""
背景任務: Push 代碼審查
1. 整理 commit 資訊
2. 呼叫 OpenClaw 分析
3. 儲存結果到 Redis
4. 發送 Telegram 通知 (只有發現問題時才通知)
"""
try:
logger.info(
"github_push_review_started",
review_id=review_id,
repo=repo.full_name,
commit_count=len(commits),
)
# 1. 整理 commit 資訊
commit_summary = []
all_files: dict[str, list] = {"added": [], "modified": [], "removed": []}
for commit in commits:
commit_summary.append({
"sha": commit.id[:8],
"message": commit.message[:100],
"author": commit.author.get("name", "unknown"),
})
all_files["added"].extend(commit.added)
all_files["modified"].extend(commit.modified)
all_files["removed"].extend(commit.removed)
# 2. 呼叫 OpenClaw 進行代碼審查 (Push 版)
analysis = await self._call_openclaw_push_review(
repo_name=repo.full_name,
ref=ref,
commits=commit_summary,
files_changed=all_files,
)
# 3. 儲存結果到 Redis
await self._save_review_with_analysis(
review_id=review_id,
event_type="push",
repo=repo.full_name,
target=f"push to {ref.split('/')[-1]}",
analysis=analysis,
metadata={
"ref": ref,
"before_sha": before_sha,
"after_sha": after_sha,
"commit_count": len(commits),
"pusher": sender.login,
"files": all_files,
},
)
# 4. 發送 Telegram 通知 (只有發現問題時才通知)
if analysis and (
analysis.issues
or analysis.security_concerns
or analysis.quality_score < 70
):
await self._send_github_telegram_alert(
review_id=review_id,
event_type="push",
repo=repo.full_name,
target=f"push to {ref.split('/')[-1]} ({len(commits)} commits)",
url=repo.html_url,
author=sender.login,
analysis=analysis,
)
logger.info(
"github_push_review_completed",
review_id=review_id,
quality_score=analysis.quality_score if analysis else None,
)
except Exception as e:
logger.exception(
"github_push_review_failed",
review_id=review_id,
error=str(e),
)
async def diagnose_ci_failure(
self,
repo, # GitHubRepository
workflow_run, # GitHubWorkflowRun
sender, # GitHubUser
diagnosis_id: str,
) -> None:
"""
背景任務: CI 失敗診斷 (Phase 13.1 #76)
1. 收集 workflow 失敗資訊
2. 呼叫 OpenClaw 進行根因分析
3. 評估風險等級與自動修復可行性
4. 儲存結果到 Redis
5. 發送 Telegram 通知
6. (可選) 建立 Approval 等待人工確認
"""
try:
logger.info(
"github_ci_failure_diagnosis_started",
diagnosis_id=diagnosis_id,
repo=repo.full_name,
workflow_name=workflow_run.name,
workflow_id=workflow_run.id,
)
# 1. 收集失敗資訊
failure_context = {
"workflow_name": workflow_run.name,
"workflow_id": workflow_run.id,
"run_number": workflow_run.run_number,
"run_attempt": workflow_run.run_attempt,
"conclusion": workflow_run.conclusion,
"head_sha": workflow_run.head_sha,
"head_branch": workflow_run.head_branch,
"event_trigger": workflow_run.event,
"html_url": workflow_run.html_url,
"created_at": workflow_run.created_at,
"updated_at": workflow_run.updated_at,
}
# 2. 呼叫 OpenClaw 進行 CI 失敗診斷
diagnosis = await self._call_openclaw_ci_diagnosis(
repo_name=repo.full_name,
failure_context=failure_context,
)
# 3. 評估自動修復策略 (Phase 13.1 #78)
repair_decision = None
if diagnosis:
from src.services.ci_auto_repair import get_ci_auto_repair_service
repair_service = get_ci_auto_repair_service()
repair_decision = await repair_service.evaluate_repair(
error_type=diagnosis.error_type,
workflow_name=workflow_run.name,
repo=repo.full_name,
failure_context=failure_context,
diagnosis_summary=diagnosis.summary,
)
# 4. 儲存結果到 Redis (含修復決策)
await self.save_review_result(
review_id=diagnosis_id,
review_data={
"event_type": "workflow_run",
"repo": repo.full_name,
"target": f"CI: {workflow_run.name}",
"diagnosis": diagnosis.model_dump() if diagnosis else None,
"repair_decision": {
"should_repair": repair_decision.should_repair,
"execution_decision": repair_decision.execution_decision.value,
"risk_level": repair_decision.risk_level.value,
"reason": repair_decision.reason,
"recommendations": [
{"action": r.action.value, "command": r.command, "confidence": r.confidence}
for r in repair_decision.recommendations[:3]
],
} if repair_decision else None,
"failure_context": failure_context,
"reviewed_at": now_taipei_iso(),
},
ttl=GITHUB_REVIEW_TTL_SECONDS,
)
# 5. 發送 Telegram 通知 (含修復建議)
await self._send_ci_failure_telegram_alert(
diagnosis_id=diagnosis_id,
repo=repo.full_name,
workflow_name=workflow_run.name,
workflow_url=workflow_run.html_url,
sender=sender.login,
diagnosis=diagnosis,
repair_decision=repair_decision,
)
# 6. 根據修復決策建立 Approval 或自動執行
if repair_decision:
from src.services.ci_auto_repair import ExecutionDecision
if repair_decision.execution_decision == ExecutionDecision.APPROVAL_REQUIRED:
await self._create_ci_failure_approval(
diagnosis_id=diagnosis_id,
repo=repo.full_name,
workflow_run=workflow_run,
diagnosis=diagnosis,
)
elif repair_decision.execution_decision == ExecutionDecision.AUTO_EXECUTE:
logger.info(
"ci_auto_repair_eligible",
diagnosis_id=diagnosis_id,
action=repair_decision.recommendations[0].action.value if repair_decision.recommendations else None,
# TODO: 實際執行修復指令 (Phase 13.1 後續迭代)
)
elif diagnosis and diagnosis.risk_level in ("high", "critical"):
await self._create_ci_failure_approval(
diagnosis_id=diagnosis_id,
repo=repo.full_name,
workflow_run=workflow_run,
diagnosis=diagnosis,
)
logger.info(
"github_ci_failure_diagnosis_completed",
diagnosis_id=diagnosis_id,
root_cause=diagnosis.root_cause if diagnosis else None,
auto_fixable=diagnosis.auto_fixable if diagnosis else False,
risk_level=diagnosis.risk_level if diagnosis else None,
repair_decision=repair_decision.execution_decision.value if repair_decision else None,
)
except Exception as e:
logger.exception(
"github_ci_failure_diagnosis_failed",
diagnosis_id=diagnosis_id,
error=str(e),
)
# =============================================================================
# Singleton
# =============================================================================
# 單例
_service: GitHubWebhookService | None = None