f5b19cf108
- 新增 _promote_playbook: 高評分提升信心度 +0.1 - 新增 _demote_playbook: 低評分降低信心度 -0.15 - 新增 find_by_source_incident: 按 incident_id 查詢 Playbook - 新增 adjust_confidence: 信心度調整 + 狀態自動轉換 - 新增 Playbook.failure_rate 屬性 自動狀態轉換: - ai_confidence >= 0.9 + DRAFT → 自動 APPROVED - ai_confidence < 0.3 + failure_rate > 50% → 自動 DEPRECATED 測試: 13 案例全部通過 Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
374 lines
8.7 KiB
Python
374 lines
8.7 KiB
Python
"""
|
||
Repository Interfaces - Protocol 定義
|
||
======================================
|
||
Phase 16 R3: Repository 層 Protocol 介面
|
||
|
||
設計原則:
|
||
- 使用 Python Protocol 實現 DI
|
||
- Service 層只依賴 Protocol，不依賴具體實作
|
||
- 便於測試 (可注入 Mock Repository)
|
||
|
||
版本: v1.0
|
||
建立: 2026-03-26 (台北時區)
|
||
建立者: Claude Code (Phase 16 架構重構)
|
||
"""
|
||
|
||
from typing import Protocol, runtime_checkable
|
||
from uuid import UUID
|
||
|
||
from src.models.approval import ApprovalRequest, ApprovalRequestCreate, ApprovalStatus
|
||
from src.models.incident import Incident
|
||
from src.models.playbook import (
|
||
Playbook,
|
||
PlaybookStatus,
|
||
SymptomPattern,
|
||
)
|
||
|
||
|
||
@runtime_checkable
|
||
class IApprovalRepository(Protocol):
|
||
"""
|
||
Approval Repository Protocol
|
||
|
||
職責: ApprovalRecord CRUD 操作
|
||
實作: ApprovalDBRepository (PostgreSQL)
|
||
"""
|
||
|
||
async def create(self, request: ApprovalRequestCreate) -> ApprovalRequest:
|
||
"""建立新的 Approval"""
|
||
...
|
||
|
||
async def get_by_id(self, approval_id: UUID) -> ApprovalRequest | None:
|
||
"""根據 ID 取得 Approval"""
|
||
...
|
||
|
||
async def get_pending(self) -> list[ApprovalRequest]:
|
||
"""取得所有待審核的 Approval"""
|
||
...
|
||
|
||
async def update_status(
|
||
self,
|
||
approval_id: UUID,
|
||
status: ApprovalStatus,
|
||
actor: str | None = None,
|
||
) -> ApprovalRequest | None:
|
||
"""更新 Approval 狀態"""
|
||
...
|
||
|
||
async def add_signature(
|
||
self,
|
||
approval_id: UUID,
|
||
signer: str,
|
||
decision: str,
|
||
comment: str | None = None,
|
||
) -> ApprovalRequest | None:
|
||
"""新增簽核"""
|
||
...
|
||
|
||
|
||
@runtime_checkable
|
||
class IIncidentRepository(Protocol):
|
||
"""
|
||
Incident Repository Protocol
|
||
|
||
職責: IncidentRecord CRUD 操作
|
||
實作: IncidentDBRepository (PostgreSQL)
|
||
"""
|
||
|
||
async def create(self, incident: Incident) -> Incident:
|
||
"""建立新的 Incident"""
|
||
...
|
||
|
||
async def get_by_id(self, incident_id: str) -> Incident | None:
|
||
"""根據 ID 取得 Incident"""
|
||
...
|
||
|
||
async def get_active(self) -> list[Incident]:
|
||
"""取得所有活躍的 Incident"""
|
||
...
|
||
|
||
async def update(self, incident: Incident) -> Incident | None:
|
||
"""更新 Incident"""
|
||
...
|
||
|
||
async def upsert(self, incident: Incident) -> bool:
|
||
"""Upsert Incident (存在則更新，不存在則建立)"""
|
||
...
|
||
|
||
|
||
@runtime_checkable
|
||
class ITimelineRepository(Protocol):
|
||
"""
|
||
Timeline Repository Protocol
|
||
|
||
職責: TimelineEvent CRUD 操作
|
||
實作: TimelineDBRepository (PostgreSQL)
|
||
"""
|
||
|
||
async def add_event(
|
||
self,
|
||
approval_id: UUID,
|
||
event_type: str,
|
||
actor: str,
|
||
details: dict | None = None,
|
||
) -> bool:
|
||
"""新增 Timeline 事件"""
|
||
...
|
||
|
||
async def get_events(
|
||
self,
|
||
approval_id: UUID,
|
||
limit: int = 100,
|
||
) -> list[dict]:
|
||
"""取得 Approval 的 Timeline 事件"""
|
||
...
|
||
|
||
async def get_recent_events(
|
||
self,
|
||
limit: int = 50,
|
||
hours: int = 24,
|
||
) -> list[dict]:
|
||
"""取得最近的 Timeline 事件"""
|
||
...
|
||
|
||
|
||
@runtime_checkable
|
||
class IMetricsRepository(Protocol):
|
||
"""
|
||
Metrics Repository Protocol
|
||
|
||
職責: Metrics 相關 DB 查詢 (AI Success Rate)
|
||
實作: MetricsDBRepository (PostgreSQL)
|
||
|
||
版本: v1.0
|
||
建立: 2026-03-26 (台北時區)
|
||
建立者: Claude Code (Phase 17 技術債修復)
|
||
"""
|
||
|
||
async def get_ai_success_rate(
|
||
self,
|
||
hours: int = 24,
|
||
) -> tuple[float, int, int]:
|
||
"""
|
||
計算 AI 提案成功執行率
|
||
|
||
Args:
|
||
hours: 統計時間範圍 (小時)
|
||
|
||
Returns:
|
||
(success_rate_percent, executed_count, total_count)
|
||
"""
|
||
...
|
||
|
||
async def get_ai_success_trend(
|
||
self,
|
||
hours: int = 24,
|
||
points: int = 10,
|
||
) -> list[float]:
|
||
"""
|
||
取得 AI 成功率趨勢 (Sparkline 用)
|
||
|
||
Args:
|
||
hours: 統計時間範圍 (小時)
|
||
points: 趨勢點數量
|
||
|
||
Returns:
|
||
list[float]: 每小時成功率列表 (由舊到新)
|
||
"""
|
||
...
|
||
|
||
|
||
@runtime_checkable
|
||
class IPlaybookRepository(Protocol):
|
||
"""
|
||
Playbook Repository Protocol
|
||
|
||
職責: Playbook CRUD 操作 (PostgreSQL + Redis 雙層)
|
||
實作: PlaybookRepository
|
||
|
||
版本: v1.0
|
||
建立: 2026-03-26 (台北時區)
|
||
建立者: Claude Code (#7 Playbook 萃取)
|
||
"""
|
||
|
||
async def create(self, playbook: Playbook) -> Playbook:
|
||
"""建立新的 Playbook"""
|
||
...
|
||
|
||
async def get_by_id(self, playbook_id: str) -> Playbook | None:
|
||
"""根據 ID 取得 Playbook"""
|
||
...
|
||
|
||
async def update(self, playbook: Playbook) -> Playbook | None:
|
||
"""更新 Playbook"""
|
||
...
|
||
|
||
async def delete(self, playbook_id: str) -> bool:
|
||
"""刪除 Playbook (軟刪除 → DEPRECATED)"""
|
||
...
|
||
|
||
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
|
||
|
||
Returns:
|
||
(items, total_count)
|
||
"""
|
||
...
|
||
|
||
async def find_by_symptoms(
|
||
self,
|
||
symptoms: SymptomPattern,
|
||
top_k: int = 5,
|
||
min_similarity: float = 0.5,
|
||
) -> list[tuple[Playbook, float]]:
|
||
"""
|
||
根據症狀模式找相似 Playbook
|
||
|
||
Returns:
|
||
list[(Playbook, similarity_score)]
|
||
"""
|
||
...
|
||
|
||
async def update_stats(
|
||
self,
|
||
playbook_id: str,
|
||
success: bool,
|
||
) -> bool:
|
||
"""更新執行統計"""
|
||
...
|
||
|
||
async def find_by_source_incident(
|
||
self,
|
||
incident_id: str,
|
||
) -> list[Playbook]:
|
||
"""
|
||
根據來源 Incident ID 找 Playbook
|
||
|
||
2026-03-30 Claude Code: Learning Service 信心度調整用
|
||
尋找 source_incident_ids 包含此 incident_id 的 Playbooks
|
||
"""
|
||
...
|
||
|
||
async def adjust_confidence(
|
||
self,
|
||
playbook_id: str,
|
||
delta: float,
|
||
reason: str,
|
||
) -> Playbook | None:
|
||
"""
|
||
調整 Playbook 信心度
|
||
|
||
2026-03-30 Claude Code: Learning Service 信心度調整用
|
||
|
||
Args:
|
||
playbook_id: Playbook ID
|
||
delta: 調整量 (+/- 0.0~1.0)
|
||
reason: 調整原因 (審計用)
|
||
|
||
Returns:
|
||
更新後的 Playbook，或 None (如果不存在)
|
||
"""
|
||
...
|
||
|
||
|
||
@runtime_checkable
|
||
class ILearningRepository(Protocol):
|
||
"""
|
||
Learning Repository Protocol
|
||
|
||
職責: 學習數據持久化 (Redis)
|
||
實作: LearningRepository
|
||
|
||
版本: v1.0
|
||
建立: 2026-03-29 (台北時區)
|
||
建立者: Claude Code (Phase D-G P0 修正)
|
||
|
||
設計原則:
|
||
- Service 層不直接存取 Redis
|
||
- 透過 Repository 進行資料存取
|
||
- 符合 leWOOOgo 積木化原則
|
||
"""
|
||
|
||
async def record_repair(
|
||
self,
|
||
anomaly_key: str,
|
||
repair_action: str,
|
||
success: bool,
|
||
root_cause: str | None = None,
|
||
fix_description: str | None = None,
|
||
execution_time_seconds: float | None = None,
|
||
) -> bool:
|
||
"""記錄修復結果"""
|
||
...
|
||
|
||
async def get_repair_stats(
|
||
self,
|
||
anomaly_key: str,
|
||
repair_action: str,
|
||
) -> dict:
|
||
"""取得修復統計 (成功率、執行次數)"""
|
||
...
|
||
|
||
async def get_all_repair_stats(
|
||
self,
|
||
anomaly_key: str,
|
||
) -> dict[str, dict]:
|
||
"""取得所有修復動作的統計"""
|
||
...
|
||
|
||
async def get_repair_history(
|
||
self,
|
||
anomaly_key: str,
|
||
repair_action: str,
|
||
limit: int = 20,
|
||
) -> list[dict]:
|
||
"""取得修復歷史記錄"""
|
||
...
|
||
|
||
async def get_learning_summary(
|
||
self,
|
||
anomaly_key: str,
|
||
) -> dict:
|
||
"""取得學習摘要"""
|
||
...
|
||
|
||
|
||
@runtime_checkable
|
||
class IEmbeddingCacheRepository(Protocol):
|
||
"""
|
||
Embedding Cache Repository Protocol
|
||
|
||
職責: Playbook 向量快取 (Redis)
|
||
實作: EmbeddingCacheRepository
|
||
|
||
2026-03-27 ogt: 模組化改造 (P1 違規修復)
|
||
"""
|
||
|
||
async def store(
|
||
self,
|
||
playbook_id: str,
|
||
embedding: list[float],
|
||
metadata: dict | None = None,
|
||
) -> bool:
|
||
"""儲存 Playbook 向量"""
|
||
...
|
||
|
||
async def get(self, playbook_id: str) -> list[float] | None:
|
||
"""取得 Playbook 向量"""
|
||
...
|
||
|
||
async def get_all(self) -> dict[str, list[float]]:
|
||
"""取得所有 Playbook 向量"""
|
||
...
|
||
|
||
async def remove(self, playbook_id: str) -> bool:
|
||
"""移除 Playbook 向量"""
|
||
...
|