Skill 使用实战：把团队经验沉淀成 Agent 可执行的工作方法

# skill-worthiness-checklist.yaml
criteria:
  - question: "这个任务团队多久做一次？"
    threshold: "每周至少 1 次"
    reason: "低频任务不值得沉淀成 Skill"
  
  - question: "做这个任务有没有标准流程？"
    threshold: "有，但新人经常做错"
    reason: "没有标准流程的任务不适合写成 Skill"
  
  - question: "任务中有没有需要判断的分支？"
    threshold: "有，而且判断逻辑团队已有共识"
    reason: "没有判断逻辑的任务用脚本就够了"
  
  - question: "做错这个任务的代价大吗？"
    threshold: "中等以上（需要回滚、影响用户、造成损失）"
    reason: "低成本任务不需要 Skill 的标准化保障"

---
name: database-migration-review
description: 数据库迁移前的风险评估和审查流程
version: 1.2.0
author: "@dba-team"
last_updated: "2024-06-10"

# 触发条件
triggers:
  - "db-migration"
  - "数据库迁移"
  - "/skill migration"

# 输入材料
inputs:
  - name: migration_file
    type: file
    required: true
    description: "SQL 迁移文件路径"
  - name: target_table
    type: string
    required: false
    description: "主要影响的表名"

# 输出
outputs:
  - name: risk_report
    type: json
    description: "风险评估报告"
  - name: rollback_script
    type: file
    description: "回滚 SQL 脚本"

# 依赖文件
dependencies:
  - "docs/schema-snapshot.json"
  - "docs/table-stats.json"

# 验证命令
validation:
  command: "python scripts/validate-migration.py {migration_file}"
  expected_exit_code: 0

# 不适用场景
exclusions:
  - "纯数据迁移（INSERT 不涉及 schema 变更）"
  - "测试环境的快速迭代迁移"
---

# 数据库迁移风险评估 Skill

## 适用边界
本 Skill 适用于生产环境的 schema 变更（ALTER TABLE、CREATE INDEX 等）。
不适用于测试环境或纯数据迁移。

## 工作流程

### 阶段 1：Schema Diff 分析
读取迁移文件，分析涉及的表和变更类型。

判断逻辑：
- 如果涉及 >100 万行的表 → 标记为高风险
- 如果包含 ALTER COLUMN TYPE → 标记为需要维护窗口
- 如果是 CREATE INDEX → 检查是否使用 CONCURRENTLY

### 阶段 2：风险评估
根据分析结果生成风险报告。

**人工确认点**：如果风险等级为 HIGH 或 CRITICAL，必须等待 DBA 确认后才能继续。

### 阶段 3：回滚脚本生成
为迁移脚本生成配套的回滚脚本。

要求：
- 回滚脚本必须幂等（IF EXISTS / IF NOT EXISTS）
- 回滚脚本必须经过 dry-run 验证

### 阶段 4：Dry-Run 验证
在测试数据库上执行迁移和回滚。

**失败处理**：如果 dry-run 失败，停止流程并报告具体问题。不要尝试自动修复。

## 异常处理

| 异常 | 处理 |
|------|------|
| 迁移文件语法错误 | 停止，报告具体错误位置 |
| 表统计信息不存在 | 使用默认阈值，标记"未验证" |
| DBA 未响应（超过 2 小时） | 告警，不自动继续 |
| dry-run 超时（超过 30 分钟） | 停止，建议分批执行 |

# 在 Claude Code 中调用 Skill
# 方式 1：Slash 命令
/database-migration-review migrations/20240615_add_tags.sql

# 方式 2：自然语言触发
"帮我做一次数据库迁移风险评估，迁移文件是 migrations/20240615_add_tags.sql"

# 方式 3：Agent 自动匹配
# Agent 检测到用户要执行数据库迁移时，自动建议调用 Skill

# scripts/skill-runner.py
"""
Skill 执行器。
读取 SKILL.md，解析工作流程，驱动 Agent 按步骤执行。
"""
import yaml
import re
from pathlib import Path

class SkillRunner:
    def __init__(self, skill_path: str):
        self.skill_dir = Path(skill_path)
        self.skill_md = self.skill_dir / "SKILL.md"
        self.config = self._parse_skill_md()
    
    def _parse_skill_md(self) -> dict:
        """解析 SKILL.md 的 frontmatter"""
        content = self.skill_md.read_text()
        # 提取 frontmatter（--- 之间的内容）
        match = re.match(r"^---\n(.+?)\n---", content, re.DOTALL)
        if match:
            return yaml.safe_load(match.group(1))
        return {}
    
    def validate_inputs(self, inputs: dict) -> tuple[bool, str]:
        """验证输入参数"""
        for inp in self.config.get("inputs", []):
            if inp["required"] and inp["name"] not in inputs:
                return False, f"缺少必填参数: {inp['name']}"
        return True, "OK"
    
    def run_validation(self) -> tuple[bool, str]:
        """运行验证命令"""
        validation = self.config.get("validation", {})
        if not validation:
            return True, "无验证命令"
        
        command = validation["command"]
        expected = validation.get("expected_exit_code", 0)
        
        import subprocess
        result = subprocess.run(command, shell=True, capture_output=True, text=True)
        
        if result.returncode == expected:
            return True, "验证通过"
        else:
            return False, f"验证失败: {result.stderr[:500]}"
    
    def get_workflow_steps(self) -> list:
        """提取工作流程步骤"""
        content = self.skill_md.read_text()
        # 提取 ### 标题作为步骤
        steps = re.findall(r"### (阶段 \d+.+)", content)
        return steps
    
    def has_human_checkpoint(self, step: str) -> bool:
        """检查步骤是否有人工确认点"""
        content = self.skill_md.read_text()
        # 查找步骤下的 "人工确认点" 标记
        pattern = rf"{re.escape(step)}.*?人工确认点"
        return bool(re.search(pattern, content, re.DOTALL))

# skill-versions.yaml
skill: "database-migration-review"
versions:
  - version: "1.0.0"
    date: "2024-01-15"
    author: "@dba-lead"
    changes: "初始版本"
    
  - version: "1.1.0"
    date: "2024-03-20"
    author: "@senior-dba"
    changes:
      - "增加了 PG 11+ DEFAULT 值行为判断"
      - "增加了 CONCURRENTLY 索引的检查"
    
  - version: "1.2.0"
    date: "2024-06-10"
    author: "@dba-team"
    changes:
      - "增加了 dry-run 超时处理"
      - "优化了风险等级判断规则"

# 版本策略
versioning:
  major: "不兼容的工作流变更（删除步骤、修改输入输出格式）"
  minor: "新增步骤、优化判断逻辑、增加异常处理"
  patch: "修正文案、修复 bug"

skills/
└── database-migration-review/
    ├── SKILL.md              # Skill 定义（frontmatter + 工作流）
    ├── scripts/
    │   ├── analyze-migration.py
    │   ├── generate-rollback.py
    │   └── validate-migration.py
    ├── templates/
    │   ├── risk-report.json
    │   └── rollback-template.sql
    ├── examples/
    │   ├── low-risk-example.md
    │   └── high-risk-example.md
    └── tests/
        ├── test-analyzer.py
        └── test-rollback.py