awoooi/apps/api/src/services/rule_to_playbook_migrator.py

"""
Rule → Playbook Migrator
========================
將 alert_rules.yaml 中的 25 條規則遷移為 DRAFT Playbook，讓飛輪 RAG 有料可查。

設計原則:
- status=DRAFT（不直接 APPROVED — 違反「禁寫死」鐵律）
- ai_confidence=0.3（誠實標示，非假 1.0 — 違反 feedback_confidence_truthfulness）
- source=PlaybookSource.YAML_RULE（現有 enum，不新增 RULE_MIGRATED）
- 冪等：name LIKE 'AutoMigrated: %' 已存在則跳過
- INSERT ON CONFLICT → repo.create() UPSERT（playbook_id 唯一鍵）
- 與 playbook_seed_service.py 完全解耦（不擾動既有 seed 機制）

name 格式: "AutoMigrated: {rule.id}" — 與 seed_service 用 description 作 name 的格式區隔

W1 PR-R1 — 規則 → Playbook 遷移
2026-04-28 ogt + Claude Sonnet 4.6
"""
from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

import structlog
import yaml

logger = structlog.get_logger(__name__)

# 告警 severity → risk 等級
_SEVERITY_TO_RISK: dict[str, str] = {
    "low": "LOW",
    "medium": "MEDIUM",
    "high": "HIGH",
    "critical": "CRITICAL",
}

# yaml risk 欄位允許 "high" 但 RiskLevel enum 有 HIGH；seed_service 用的 map 少了 high
_YAML_RISK_MAP: dict[str, str] = {
    "low": "LOW",
    "medium": "MEDIUM",
    "high": "HIGH",
    "critical": "CRITICAL",
}


@dataclass
class MigrationReport:
    """遷移報告"""
    total_rules: int = 0
    created: int = 0
    skipped: int = 0
    failed: int = 0
    dry_run: bool = False
    errors: list[str] = field(default_factory=list)
    created_names: list[str] = field(default_factory=list)
    skipped_names: list[str] = field(default_factory=list)

    def summary(self) -> str:
        mode = "[DRY-RUN] " if self.dry_run else ""
        return (
            f"{mode}遷移完成 — "
            f"總計 {self.total_rules} 條規則，"
            f"建立 {self.created}，跳過 {self.skipped}，失敗 {self.failed}"
        )


# =============================================================================
# 命令類型判斷（不依賴 SPF-2 action_parser，用既有 regex 守門）
# =============================================================================

def _infer_action_type(kubectl_command: str) -> str:
    """
    從指令字串推斷 ActionType（字串形式，對應 ActionType enum 值）

    規則:
    - 空字串 → "manual"
    - 以 "ssh " 開頭 → "ssh_command"
    - 其他有指令 → "kubectl"
    """
    cmd = (kubectl_command or "").strip()
    if not cmd:
        return "manual"
    if cmd.startswith("ssh "):
        return "ssh_command"
    return "kubectl"


def _infer_risk_level(risk_str: str) -> str:
    """
    YAML risk 欄位 → RiskLevel 字串

    alert_rules.yaml 的 risk 欄位值: low / medium / high / critical
    """
    return _YAML_RISK_MAP.get((risk_str or "medium").lower(), "MEDIUM")


def _build_symptom_pattern(rule: dict[str, Any]) -> dict[str, Any]:
    """
    從規則 match block 推導 SymptomPattern dict

    symptom_pattern 包含:
    - alert_names: match.alertname list
    - affected_services: 從 id/description 推導關鍵字（保守策略：留空，讓 RAG 學習）
    - severity_range: 從 risk 反推 ["P1"] / ["P2"] / ["P3"]
    - keywords: match.message list（部分匹配關鍵字）
    """
    match_block = rule.get("match", {})
    alertnames: list[str] = match_block.get("alertname", [])
    messages: list[str] = match_block.get("message", [])
    alert_types: list[str] = match_block.get("alert_type", [])

    # risk → severity_range 反推
    risk_str = (rule.get("response", {}).get("risk", "medium") or "medium").lower()
    if risk_str == "critical":
        severity_range = ["P1", "P2"]
    elif risk_str in ("high", "medium"):
        severity_range = ["P2", "P3"]
    else:
        severity_range = ["P3"]

    # keywords: message + alert_type 列表合併（最多 15 個）
    keywords = list(messages) + list(alert_types)
    # 過濾萬用符（generic_fallback 有 "*"）
    keywords = [k for k in keywords if k != "*"][:15]

    # 2026-04-29 ogt + Claude Opus 4.7: critic Major #1 修
    # 過濾 alert_names 中的 "*" wildcard（generic_fallback）— 否則進 RAG 向量化
    # 後變成「告警: *」污染語料，每筆查詢都會跟它算相似度
    raw_names = alertnames if isinstance(alertnames, list) else [alertnames]
    filtered_names = [n for n in raw_names if n and n != "*"]
    return {
        "alert_names": filtered_names,
        "affected_services": [],
        "severity_range": severity_range,
        "keywords": keywords,
        "label_patterns": {},
    }


def _build_repair_steps(rule: dict[str, Any]) -> list[dict[str, Any]]:
    """
    從規則 response block 建立 RepairStep dict list

    策略:
    - kubectl_command 存在且非空 → step 1
    - 若 optimization list 存在 → 每項追加為額外步驟
    - 若 kubectl_command 空 (NO_ACTION) → step 1 action_type=manual，command=描述文字
    """
    resp = rule.get("response", {})
    kubectl_cmd = (resp.get("kubectl_command", "") or "").strip()
    risk_level = _infer_risk_level(resp.get("risk", "medium"))
    suggested_action = resp.get("suggested_action", "NO_ACTION") or "NO_ACTION"

    steps: list[dict[str, Any]] = []

    if kubectl_cmd:
        action_type = _infer_action_type(kubectl_cmd)
        steps.append({
            "step_number": 1,
            "action_type": action_type,
            "command": kubectl_cmd,
            "expected_result": resp.get("action_title", ""),
            "risk_level": risk_level,
            "requires_approval": risk_level == "CRITICAL" or suggested_action in ("RESTART_DEPLOYMENT", "DELETE_POD", "SCALE_DEPLOYMENT"),
            # 2026-04-29 ogt + Claude Opus 4.7: critic Major #3 修
            # yaml_rule 來源的 kubectl_command 未經 SPF-2 action_parser 驗證
            # promote 流程（DRAFT → APPROVED）必須強制走 action_parser，否則危險指令直達 prod
            "metadata": {
                "unverified_command": True,
                "needs_action_parser_review": True,
                "source": "yaml_rule_migration",
            },
        })
    else:
        # NO_ACTION — 記錄診斷描述為 manual step，讓 RAG 至少有症狀可查
        description_text = resp.get("description", rule.get("description", "人工診斷"))
        steps.append({
            "step_number": 1,
            "action_type": "manual",
            "command": description_text[:500],
            "expected_result": resp.get("action_title", ""),
            "risk_level": risk_level,
            "requires_approval": True,
        })

    # 追加 optimization steps（最多 3 個，step_number 2/3/4）
    # 2026-04-29 critic Minor 修：原 `if idx >= 4: break` 寫在 append 後易誤讀
    # 改用 [:3] slice 明確限制最多 3 個
    for idx, opt in enumerate((resp.get("optimization", []) or [])[:3], start=2):
        opt_cmd = (opt.get("command", "") or "").strip()
        if not opt_cmd or opt_cmd.startswith("#"):
            continue
        steps.append({
            "step_number": idx,
            "action_type": _infer_action_type(opt_cmd),
            "command": opt_cmd,
            "expected_result": opt.get("description", ""),
            "risk_level": "LOW",
            "requires_approval": False,
        })

    return steps


def _estimated_duration(risk_level: str, suggested_action: str) -> int:
    """估算修復時間（分鐘）"""
    if suggested_action in ("NO_ACTION",):
        return 15
    if risk_level == "CRITICAL":
        return 5
    return 3


def _build_tags(rule: dict[str, Any]) -> list[str]:
    """從規則提取標籤"""
    tags: set[str] = {"yaml_rule", "auto_migrated"}

    rule_id = rule.get("id", "")
    resp = rule.get("response", {})
    responsibility = resp.get("responsibility", "")
    if responsibility:
        tags.add(responsibility.lower())

    # 從 alertname 推導類型標籤
    alertnames = rule.get("match", {}).get("alertname", [])
    for name in alertnames:
        name_lower = (name or "").lower()
        if "cpu" in name_lower:
            tags.add("cpu")
        if "memory" in name_lower or "oom" in name_lower:
            tags.add("memory")
        if "disk" in name_lower or "storage" in name_lower:
            tags.add("disk")
        if "pod" in name_lower or "k8s" in name_lower or "kube" in name_lower:
            tags.add("kubernetes")
        if "ssl" in name_lower or "cert" in name_lower:
            tags.add("ssl")
        if "backup" in name_lower:
            tags.add("backup")
        if "postgresql" in name_lower or "postgres" in name_lower:
            tags.add("database")
        if "redis" in name_lower:
            tags.add("cache")
        if "ollama" in name_lower:
            tags.add("ai")

    return list(tags)[:10]


def parse_yaml_rules(yaml_path: Path) -> list[dict[str, Any]]:
    """
    讀取並解析 alert_rules.yaml，回傳 rules list

    Raises:
        FileNotFoundError: yaml 不存在
        yaml.YAMLError: yaml 格式錯誤
    """
    data = yaml.safe_load(yaml_path.read_text(encoding="utf-8"))
    rules = data.get("rules", [])
    return [r for r in rules if isinstance(r, dict)]


def build_playbook_dict(rule: dict[str, Any]) -> dict[str, Any]:
    """
    從單條規則建立 Playbook 初始化 dict（不寫 DB）

    Returns dict 可直接傳給 Playbook(**dict)
    """
    rule_id = rule.get("id", "unknown")
    resp = rule.get("response", {})
    description = resp.get("description", rule.get("description", f"規則 {rule_id} 自動遷移"))
    risk_str = (resp.get("risk", "medium") or "medium").lower()
    suggested_action = resp.get("suggested_action", "NO_ACTION") or "NO_ACTION"

    symptom_pattern = _build_symptom_pattern(rule)
    repair_steps = _build_repair_steps(rule)
    tags = _build_tags(rule)
    risk_level = _infer_risk_level(risk_str)
    duration = _estimated_duration(risk_level, suggested_action)

    return {
        "name": f"AutoMigrated: {rule_id}",
        "description": description[:2000],
        "status": "draft",
        "source": "yaml_rule",
        "symptom_pattern": symptom_pattern,
        "repair_steps": repair_steps,
        "estimated_duration_minutes": duration,
        "ai_confidence": 0.3,
        "trust_score": 0.3,
        "tags": tags,
        "notes": f"自動從 alert_rules.yaml rule.id={rule_id} 遷移。priority={rule.get('priority', 999)}",
        "created_by_agent": "migrator",
    }


# =============================================================================
# 核心遷移函式（async，依賴 DB）
# =============================================================================

async def migrate_yaml_rules_to_playbooks(
    yaml_path: Path,
    dry_run: bool = True,
    enable_migration: bool = True,
) -> MigrationReport:
    """
    將 alert_rules.yaml 遷移為 DRAFT Playbook

    Args:
        yaml_path: alert_rules.yaml 路徑
        dry_run: True=只印計畫不寫 DB，False=真實寫入
        enable_migration: feature flag（ENABLE_RULE_MIGRATION_DRAFT），False 時直接 return

    Returns:
        MigrationReport

    設計：
    - 冪等：name LIKE 'AutoMigrated: %' 已存在任何狀態的 playbook 即跳過
    - 不依賴 seed_service（source=yaml_rule 但 name prefix 不同，互不干擾）
    - generic_fallback 規則（id=generic_fallback）也遷移，讓 RAG 能學到「兜底症狀」
    """
    report = MigrationReport(dry_run=dry_run)

    if not enable_migration:
        logger.info("rule_migration_disabled_by_flag")
        return report

    if not yaml_path.exists():
        logger.error("rule_migration_yaml_not_found", path=str(yaml_path))
        report.errors.append(f"yaml 不存在: {yaml_path}")
        return report

    # 1. 解析 yaml
    try:
        rules = parse_yaml_rules(yaml_path)
    except Exception as e:
        logger.error("rule_migration_parse_error", error=str(e))
        report.errors.append(f"yaml 解析失敗: {e}")
        return report

    report.total_rules = len(rules)

    if dry_run:
        # Dry-run：只建立 dict，不查 DB、不寫 DB
        for rule in rules:
            rule_id = rule.get("id", "unknown")
            try:
                pb_dict = build_playbook_dict(rule)
                report.created_names.append(pb_dict["name"])
                report.created += 1
                logger.info(
                    "rule_migration_dry_run_would_create",
                    rule_id=rule_id,
                    name=pb_dict["name"],
                    alert_names=pb_dict["symptom_pattern"]["alert_names"],
                )
            except Exception as e:
                report.failed += 1
                report.errors.append(f"rule_id={rule_id} 建立 dict 失敗: {e}")
                logger.warning("rule_migration_dry_run_error", rule_id=rule_id, error=str(e))
        return report

    # 2. 查詢現有 AutoMigrated Playbook（冪等去重）
    from src.db.base import get_db_context
    from sqlalchemy import text as sa_text

    async with get_db_context() as db:
        rows = await db.execute(
            sa_text("SELECT name FROM playbooks WHERE name LIKE 'AutoMigrated: %'")
        )
        existing_names: set[str] = {r[0] for r in rows.fetchall()}

    # 3. 逐條遷移
    from src.models.playbook import Playbook
    from src.repositories.playbook_repository import get_playbook_repository

    repo = get_playbook_repository()

    for rule in rules:
        rule_id = rule.get("id", "unknown")
        try:
            pb_dict = build_playbook_dict(rule)
            name = pb_dict["name"]

            if name in existing_names:
                report.skipped += 1
                report.skipped_names.append(name)
                logger.debug("rule_migration_skip_existing", rule_id=rule_id, name=name)
                continue

            playbook = Playbook(**pb_dict)
            await repo.create(playbook)
            report.created += 1
            report.created_names.append(name)
            existing_names.add(name)  # 防止同 session 重複建立

            logger.info(
                "rule_migration_created",
                rule_id=rule_id,
                playbook_id=playbook.playbook_id,
                name=name,
            )

        except Exception as e:
            report.failed += 1
            report.errors.append(f"rule_id={rule_id} 失敗: {e}")
            logger.warning("rule_migration_create_error", rule_id=rule_id, error=str(e))

    logger.info(
        "rule_migration_complete",
        total=report.total_rules,
        created=report.created,
        skipped=report.skipped,
        failed=report.failed,
    )
    return report