feat(api): #7 Playbook 萃取功能 (Phase 7.1-7.4)
實作內容:
- models/playbook.py: Playbook 資料模型 + Request/Response
- repositories/playbook_repository.py: Redis 雙層儲存
- repositories/interfaces.py: IPlaybookRepository Protocol
- services/playbook_service.py: 業務邏輯 (萃取/推薦/核准)
- api/v1/playbooks.py: REST API 端點
API 端點:
- POST /playbooks/extract/{incident_id} - 從成功案例萃取
- POST /playbooks/recommend - 症狀匹配推薦
- POST /playbooks/{id}/approve - 人工核准
- GET/PATCH/DELETE /playbooks/{id} - CRUD
遵循 leWOOOgo 積木化: Router → Service → Repository
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
OG T
474
apps/api/src/services/playbook_service.py
Normal file
474
apps/api/src/services/playbook_service.py
Normal file
@@ -0,0 +1,474 @@
|
||||
"""
|
||||
Playbook Service - #7 Playbook 萃取
|
||||
===================================
|
||||
Playbook 業務邏輯層
|
||||
|
||||
Phase 7.3: Service 實作
|
||||
建立時間: 2026-03-26 (台北時區)
|
||||
建立者: Claude Code (#7 Playbook 萃取)
|
||||
|
||||
遵循 leWOOOgo 積木化原則:
|
||||
- Service 層只依賴 Repository Interface
|
||||
- 不直接存取 Redis/DB
|
||||
- 封裝所有業務邏輯
|
||||
"""
|
||||
|
||||
from datetime import UTC, datetime
|
||||
from typing import Protocol
|
||||
|
||||
import structlog
|
||||
|
||||
from src.models.incident import Incident, IncidentStatus
|
||||
from src.models.playbook import (
|
||||
ActionType,
|
||||
Playbook,
|
||||
PlaybookRecommendation,
|
||||
PlaybookSource,
|
||||
PlaybookStatus,
|
||||
RepairStep,
|
||||
RiskLevel,
|
||||
SymptomPattern,
|
||||
)
|
||||
from src.repositories.interfaces import IPlaybookRepository
|
||||
from src.repositories.playbook_repository import get_playbook_repository
|
||||
|
||||
logger = structlog.get_logger(__name__)
|
||||
|
||||
|
||||
class IPlaybookService(Protocol):
|
||||
"""Playbook Service Interface"""
|
||||
|
||||
async def extract_from_incident(
|
||||
self,
|
||||
incident: Incident,
|
||||
auto_approve: bool = False,
|
||||
) -> Playbook | None:
|
||||
"""從成功案例萃取 Playbook"""
|
||||
...
|
||||
|
||||
async def get_recommendations(
|
||||
self,
|
||||
symptoms: SymptomPattern,
|
||||
top_k: int = 3,
|
||||
) -> list[PlaybookRecommendation]:
|
||||
"""取得 Playbook 推薦"""
|
||||
...
|
||||
|
||||
async def approve(
|
||||
self,
|
||||
playbook_id: str,
|
||||
approved_by: str,
|
||||
notes: str | None = None,
|
||||
) -> Playbook | None:
|
||||
"""核准 Playbook"""
|
||||
...
|
||||
|
||||
async def record_execution(
|
||||
self,
|
||||
playbook_id: str,
|
||||
success: bool,
|
||||
) -> bool:
|
||||
"""記錄 Playbook 執行結果"""
|
||||
...
|
||||
|
||||
|
||||
class PlaybookService:
|
||||
"""
|
||||
Playbook Service 實作
|
||||
|
||||
職責:
|
||||
- 從 Incident 萃取 Playbook
|
||||
- 提供 Playbook 推薦
|
||||
- 管理 Playbook 生命週期
|
||||
"""
|
||||
|
||||
def __init__(self, repository: IPlaybookRepository | None = None):
|
||||
self._repository = repository or get_playbook_repository()
|
||||
|
||||
# === Core Operations ===
|
||||
|
||||
async def extract_from_incident(
|
||||
self,
|
||||
incident: Incident,
|
||||
auto_approve: bool = False,
|
||||
) -> Playbook | None:
|
||||
"""
|
||||
從成功案例萃取 Playbook
|
||||
|
||||
前置條件:
|
||||
- Incident 狀態為 RESOLVED 或 CLOSED
|
||||
- outcome.execution_success == True
|
||||
- outcome.effectiveness_score >= 4
|
||||
|
||||
Args:
|
||||
incident: 來源 Incident
|
||||
auto_approve: 是否自動核准 (僅限高信心度)
|
||||
|
||||
Returns:
|
||||
Playbook | None
|
||||
"""
|
||||
# 1. 驗證前置條件
|
||||
if incident.status not in [IncidentStatus.RESOLVED, IncidentStatus.CLOSED]:
|
||||
logger.warning(
|
||||
"playbook_extract_invalid_status",
|
||||
incident_id=incident.incident_id,
|
||||
status=incident.status,
|
||||
)
|
||||
return None
|
||||
|
||||
if not incident.outcome or not incident.outcome.execution_success:
|
||||
logger.warning(
|
||||
"playbook_extract_no_successful_outcome",
|
||||
incident_id=incident.incident_id,
|
||||
)
|
||||
return None
|
||||
|
||||
effectiveness = incident.outcome.effectiveness_score or 0
|
||||
if effectiveness < 4:
|
||||
logger.info(
|
||||
"playbook_extract_low_effectiveness",
|
||||
incident_id=incident.incident_id,
|
||||
effectiveness=effectiveness,
|
||||
)
|
||||
return None
|
||||
|
||||
# 2. 萃取症狀模式
|
||||
symptom_pattern = self._extract_symptom_pattern(incident)
|
||||
|
||||
# 3. 萃取修復步驟
|
||||
repair_steps = self._extract_repair_steps(incident)
|
||||
|
||||
# 4. 計算信心度
|
||||
confidence = self._calculate_confidence(incident, effectiveness)
|
||||
|
||||
# 5. 生成名稱和描述
|
||||
name = self._generate_name(incident)
|
||||
description = self._generate_description(incident)
|
||||
|
||||
# 6. 建立 Playbook
|
||||
playbook = Playbook(
|
||||
name=name,
|
||||
description=description,
|
||||
status=PlaybookStatus.APPROVED if auto_approve and confidence >= 0.9 else PlaybookStatus.DRAFT,
|
||||
source=PlaybookSource.EXTRACTED,
|
||||
symptom_pattern=symptom_pattern,
|
||||
repair_steps=repair_steps,
|
||||
source_incident_ids=[incident.incident_id],
|
||||
ai_confidence=confidence,
|
||||
tags=self._extract_tags(incident),
|
||||
)
|
||||
|
||||
# 7. 儲存
|
||||
playbook = await self._repository.create(playbook)
|
||||
|
||||
logger.info(
|
||||
"playbook_extracted",
|
||||
playbook_id=playbook.playbook_id,
|
||||
incident_id=incident.incident_id,
|
||||
confidence=confidence,
|
||||
auto_approved=playbook.status == PlaybookStatus.APPROVED,
|
||||
)
|
||||
|
||||
return playbook
|
||||
|
||||
async def get_recommendations(
|
||||
self,
|
||||
symptoms: SymptomPattern,
|
||||
top_k: int = 3,
|
||||
) -> list[PlaybookRecommendation]:
|
||||
"""
|
||||
取得 Playbook 推薦
|
||||
|
||||
策略:
|
||||
1. 從 Repository 找相似症狀的 Playbook
|
||||
2. 按 similarity_score * success_rate 排序
|
||||
3. 返回 Top K 推薦
|
||||
"""
|
||||
# 查詢相似 Playbook
|
||||
similar_playbooks = await self._repository.find_by_symptoms(
|
||||
symptoms=symptoms,
|
||||
top_k=top_k * 2, # 多取一些用於後續過濾
|
||||
min_similarity=0.4,
|
||||
)
|
||||
|
||||
if not similar_playbooks:
|
||||
return []
|
||||
|
||||
# 建立推薦列表
|
||||
recommendations: list[PlaybookRecommendation] = []
|
||||
|
||||
for playbook, similarity in similar_playbooks:
|
||||
# 找出匹配的症狀
|
||||
matched_symptoms = self._find_matched_symptoms(symptoms, playbook.symptom_pattern)
|
||||
|
||||
# 生成推薦原因
|
||||
reason = self._generate_recommendation_reason(
|
||||
playbook,
|
||||
similarity,
|
||||
matched_symptoms,
|
||||
)
|
||||
|
||||
recommendations.append(
|
||||
PlaybookRecommendation(
|
||||
playbook=playbook,
|
||||
similarity_score=similarity,
|
||||
matched_symptoms=matched_symptoms,
|
||||
reason=reason,
|
||||
)
|
||||
)
|
||||
|
||||
# 按綜合分數排序
|
||||
recommendations.sort(
|
||||
key=lambda r: r.similarity_score * (0.5 + 0.5 * r.playbook.success_rate),
|
||||
reverse=True,
|
||||
)
|
||||
|
||||
return recommendations[:top_k]
|
||||
|
||||
async def approve(
|
||||
self,
|
||||
playbook_id: str,
|
||||
approved_by: str,
|
||||
notes: str | None = None,
|
||||
) -> Playbook | None:
|
||||
"""核准 Playbook"""
|
||||
playbook = await self._repository.get_by_id(playbook_id)
|
||||
if not playbook:
|
||||
return None
|
||||
|
||||
if playbook.status != PlaybookStatus.DRAFT:
|
||||
logger.warning(
|
||||
"playbook_approve_invalid_status",
|
||||
playbook_id=playbook_id,
|
||||
current_status=playbook.status,
|
||||
)
|
||||
return None
|
||||
|
||||
playbook.status = PlaybookStatus.APPROVED
|
||||
playbook.approved_by = approved_by
|
||||
playbook.approved_at = datetime.now(UTC)
|
||||
if notes:
|
||||
playbook.notes = notes
|
||||
|
||||
updated = await self._repository.update(playbook)
|
||||
|
||||
if updated:
|
||||
logger.info(
|
||||
"playbook_approved",
|
||||
playbook_id=playbook_id,
|
||||
approved_by=approved_by,
|
||||
)
|
||||
|
||||
return updated
|
||||
|
||||
async def record_execution(
|
||||
self,
|
||||
playbook_id: str,
|
||||
success: bool,
|
||||
) -> bool:
|
||||
"""記錄 Playbook 執行結果"""
|
||||
return await self._repository.update_stats(playbook_id, success)
|
||||
|
||||
# === CRUD Proxies ===
|
||||
|
||||
async def get_by_id(self, playbook_id: str) -> Playbook | None:
|
||||
"""取得 Playbook"""
|
||||
return await self._repository.get_by_id(playbook_id)
|
||||
|
||||
async def list_playbooks(
|
||||
self,
|
||||
status: PlaybookStatus | None = None,
|
||||
tags: list[str] | None = None,
|
||||
limit: int = 20,
|
||||
offset: int = 0,
|
||||
) -> tuple[list[Playbook], int]:
|
||||
"""列出 Playbooks"""
|
||||
return await self._repository.list_playbooks(
|
||||
status=status,
|
||||
tags=tags,
|
||||
limit=limit,
|
||||
offset=offset,
|
||||
)
|
||||
|
||||
async def update(self, playbook: Playbook) -> Playbook | None:
|
||||
"""更新 Playbook"""
|
||||
return await self._repository.update(playbook)
|
||||
|
||||
async def delete(self, playbook_id: str) -> bool:
|
||||
"""刪除 Playbook (軟刪除)"""
|
||||
return await self._repository.delete(playbook_id)
|
||||
|
||||
# === Private Helpers ===
|
||||
|
||||
def _extract_symptom_pattern(self, incident: Incident) -> SymptomPattern:
|
||||
"""從 Incident 萃取症狀模式"""
|
||||
alert_names = [s.alert_name for s in incident.signals] if incident.signals else []
|
||||
keywords = []
|
||||
|
||||
# 從 annotations 提取關鍵字
|
||||
for signal in incident.signals or []:
|
||||
if signal.annotations:
|
||||
for value in signal.annotations.values():
|
||||
if isinstance(value, str) and len(value) < 50:
|
||||
keywords.append(value)
|
||||
|
||||
return SymptomPattern(
|
||||
alert_names=alert_names,
|
||||
affected_services=incident.affected_services or [],
|
||||
severity_range=[incident.severity.value] if incident.severity else ["P2"],
|
||||
keywords=keywords[:10], # 最多 10 個關鍵字
|
||||
)
|
||||
|
||||
def _extract_repair_steps(self, incident: Incident) -> list[RepairStep]:
|
||||
"""從 Incident 萃取修復步驟"""
|
||||
steps: list[RepairStep] = []
|
||||
|
||||
# 從 decision_chain 提取
|
||||
if incident.decision_chain:
|
||||
for i, step in enumerate(incident.decision_chain.steps, 1):
|
||||
if step.executed_action:
|
||||
steps.append(
|
||||
RepairStep(
|
||||
step_number=i,
|
||||
action_type=ActionType.KUBECTL,
|
||||
command=step.executed_action,
|
||||
expected_result=step.result or None,
|
||||
risk_level=RiskLevel.MEDIUM,
|
||||
)
|
||||
)
|
||||
|
||||
# 如果沒有從 decision_chain 取得，嘗試從 outcome 取得
|
||||
if not steps and incident.outcome and incident.outcome.repair_action:
|
||||
steps.append(
|
||||
RepairStep(
|
||||
step_number=1,
|
||||
action_type=ActionType.KUBECTL,
|
||||
command=incident.outcome.repair_action,
|
||||
risk_level=RiskLevel.MEDIUM,
|
||||
)
|
||||
)
|
||||
|
||||
return steps
|
||||
|
||||
def _calculate_confidence(self, incident: Incident, effectiveness: int) -> float:
|
||||
"""計算 AI 萃取信心度"""
|
||||
base_score = 0.5
|
||||
|
||||
# effectiveness 貢獻 (4-5 → 0.2-0.4)
|
||||
effectiveness_bonus = (effectiveness - 3) * 0.2
|
||||
|
||||
# 有 decision_chain 加分
|
||||
if incident.decision_chain and incident.decision_chain.steps:
|
||||
base_score += 0.1
|
||||
|
||||
# 有多個 signals 加分 (更多資料)
|
||||
if incident.signals and len(incident.signals) >= 2:
|
||||
base_score += 0.05
|
||||
|
||||
return min(base_score + effectiveness_bonus, 1.0)
|
||||
|
||||
def _generate_name(self, incident: Incident) -> str:
|
||||
"""生成 Playbook 名稱"""
|
||||
alert_name = incident.signals[0].alert_name if incident.signals else "Unknown"
|
||||
services = incident.affected_services[:2] if incident.affected_services else []
|
||||
service_str = "/".join(services) if services else "system"
|
||||
|
||||
return f"{alert_name} - {service_str} 修復劇本"
|
||||
|
||||
def _generate_description(self, incident: Incident) -> str:
|
||||
"""生成 Playbook 描述"""
|
||||
parts = []
|
||||
|
||||
if incident.signals:
|
||||
parts.append(f"觸發告警: {incident.signals[0].alert_name}")
|
||||
|
||||
if incident.affected_services:
|
||||
parts.append(f"影響服務: {', '.join(incident.affected_services)}")
|
||||
|
||||
if incident.outcome and incident.outcome.repair_action:
|
||||
parts.append(f"修復動作: {incident.outcome.repair_action[:100]}")
|
||||
|
||||
return ". ".join(parts) if parts else "從成功案例自動萃取的修復劇本"
|
||||
|
||||
def _extract_tags(self, incident: Incident) -> list[str]:
|
||||
"""萃取標籤"""
|
||||
tags: set[str] = set()
|
||||
|
||||
# 從服務名稱提取
|
||||
for service in incident.affected_services or []:
|
||||
tags.add(service.lower())
|
||||
|
||||
# 從告警名稱提取類型
|
||||
if incident.signals:
|
||||
for signal in incident.signals:
|
||||
if "cpu" in signal.alert_name.lower():
|
||||
tags.add("cpu")
|
||||
if "memory" in signal.alert_name.lower():
|
||||
tags.add("memory")
|
||||
if "pod" in signal.alert_name.lower():
|
||||
tags.add("kubernetes")
|
||||
if "network" in signal.alert_name.lower():
|
||||
tags.add("network")
|
||||
|
||||
return list(tags)[:10]
|
||||
|
||||
def _find_matched_symptoms(
|
||||
self,
|
||||
query: SymptomPattern,
|
||||
playbook_pattern: SymptomPattern,
|
||||
) -> list[str]:
|
||||
"""找出匹配的症狀"""
|
||||
matched = []
|
||||
|
||||
# 匹配的告警
|
||||
alert_matches = set(query.alert_names) & set(playbook_pattern.alert_names)
|
||||
for alert in alert_matches:
|
||||
matched.append(f"Alert: {alert}")
|
||||
|
||||
# 匹配的服務
|
||||
service_matches = set(query.affected_services) & set(playbook_pattern.affected_services)
|
||||
for service in service_matches:
|
||||
matched.append(f"Service: {service}")
|
||||
|
||||
# 匹配的嚴重度
|
||||
if set(query.severity_range) & set(playbook_pattern.severity_range):
|
||||
matched.append(f"Severity: {query.severity_range[0]}")
|
||||
|
||||
return matched
|
||||
|
||||
def _generate_recommendation_reason(
|
||||
self,
|
||||
playbook: Playbook,
|
||||
similarity: float,
|
||||
matched_symptoms: list[str],
|
||||
) -> str:
|
||||
"""生成推薦原因"""
|
||||
parts = []
|
||||
|
||||
parts.append(f"相似度 {similarity:.0%}")
|
||||
|
||||
if playbook.success_rate > 0:
|
||||
parts.append(f"成功率 {playbook.success_rate:.0%}")
|
||||
|
||||
if playbook.total_executions > 0:
|
||||
parts.append(f"已執行 {playbook.total_executions} 次")
|
||||
|
||||
if matched_symptoms:
|
||||
parts.append(f"匹配: {', '.join(matched_symptoms[:3])}")
|
||||
|
||||
return ". ".join(parts)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Singleton
|
||||
# =============================================================================
|
||||
|
||||
_service: PlaybookService | None = None
|
||||
|
||||
|
||||
def get_playbook_service() -> IPlaybookService:
|
||||
"""取得 PlaybookService 單例"""
|
||||
global _service
|
||||
if _service is None:
|
||||
_service = PlaybookService()
|
||||
return _service
|
||||
Reference in New Issue
Block a user