Hooks 系统：自动化工作流

#!/bin/bash
# .claude/hooks/post-write-auto-label.sh
# 当 Claude 修改特定类型文件时，自动给当前 PR 打标签
 
FILE="$1"
PR_NUMBER=$(gh pr view --json number -q '.number' 2>/dev/null || echo "")
 
if [ -z "$PR_NUMBER" ]; then
    exit 0
fi
 
# 根据文件类型打标签
if "$FILE" == *"test*" || "$FILE" == *"spec*"; then
    gh pr edit "$PR_NUMBER" --add-label "auto:tests"
fi
 
if "$FILE" == *"migration*" || "$FILE" == *"migrate*"; then
    gh pr edit "$PR_NUMBER" --add-label "auto:migration"
fi
 
if "$FILE" == *"api*" || "$FILE" == *"openapi*"; then
    gh pr edit "$PR_NUMBER" --add-label "auto:api-change"
fi
 
# 安全相关文件
if "$FILE" == *"auth*" || "$FILE" == *"security*" || "$FILE" == *"permission*"; then
    gh pr edit "$PR_NUMBER" --add-label "security-review-required"
fi
 
echo "✅ PR #$PR_NUMBER 标签已更新"

#!/bin/bash
# .claude/hooks/pre-write.sh
# 在 Claude 写入文件前自动格式化
 
FILE="$1"  # Claude 传入的文件路径
 
if "$FILE" == *.py; then
    black "$FILE"
elif "$FILE" == *.js || "$FILE" == *.ts; then
    prettier --write "$FILE"
elif "$FILE" == *.go; then
    gofmt -w "$FILE"
fi

#!/bin/bash
# .claude/hooks/pre-command.sh
# 拦截危险命令
 
COMMAND="$1"
 
# 禁止删除命令
if "* || *"*; then
    echo "ERROR: 危险命令被拦截: $COMMAND"
    exit 1
fi
 
# 禁止直接操作生产数据库
if "$COMMAND" == *"psql production"* || "$COMMAND" == *"mysql -h prod"*; then
    echo "ERROR: 生产数据库操作需要手动确认"
    exit 1
fi
 
echo "OK: 命令通过安全检查"

# .claude/hooks/slack-notify.yaml
name: slack-notify
trigger: after_file_write
method: POST
url: https://hooks.slack.com/services/T00/B00/XXX
headers:
  Content-Type: application/json
body: |
  {
    "text": "Claude 修改了文件: {{file_path}}",
    "blocks": [
      {
        "type": "section",
        "text": {
          "type": "mrkdwn",
          "text": "*文件修改通知*\n项目: {{project_name}}\n文件: `{{file_path}}`\n操作者: {{user}}"
        }
      }
    ]
  }

# .claude/hooks/audit-log.yaml
name: audit-log
trigger: before_command
method: POST
url: https://internal.company.com/api/audit
headers:
  Authorization: Bearer {{env.AUDIT_API_TOKEN}}
body: |
  {
    "action": "ai_command",
    "command": "{{command}}",
    "session_id": "{{session_id}}",
    "timestamp": "{{timestamp}}"
  }

# .claude/hooks/custom_prompt.py
 
def before_think(context):
    """在 AI 思考前修改提示词"""
    
    # 根据时间添加提醒
    import datetime
    hour = datetime.datetime.now().hour
    
    if hour >= 18:
        context.system_prompt += "\n注意：现在是下班时间，优先给出简洁的方案。"
    
    # 根据文件类型添加规范
    if context.current_file.endswith('.py'):
        context.system_prompt += "\nPython 代码必须使用类型注解。"
    
    return context

# .claude/hooks/mcp-intercept.yaml
name: mcp-audit
trigger: before_mcp_tool
condition: "tool_name == 'postgres'"
action: ask  # 数据库操作前询问确认

#!/bin/bash
# 01-check-style.sh
 
echo "🔍 运行代码风格检查..."
 
if ! black --check src/; then
    echo "❌ 代码格式不符合 Black 规范"
    echo "建议运行: black src/"
    exit 1
fi
 
if ! flake8 src/; then
    echo "❌ 存在代码风格问题"
    exit 1
fi
 
echo "✅ 代码风格检查通过"

#!/bin/bash
# 03-security-scan.sh
 
echo "🔒 运行安全扫描..."
 
# 检查是否引入了危险函数
if grep -r "eval(" src/ || grep -r "exec(" src/; then
    echo "⚠️ 发现危险函数 eval/exec，请确认安全性"
    exit 1
fi
 
# 检查密钥泄露
if grep -r "sk-" src/ || grep -r "password=" src/; then
    echo "⚠️ 可能包含敏感信息，请检查"
    exit 1
fi
 
echo "✅ 安全扫描通过"

# 查看 Hook 执行日志
claude hooks logs
 
# 测试单个 Hook
claude hooks test pre-write.sh --file src/test.py
 
# 禁用所有 Hooks
claude hooks disable
 
# 启用 Hooks
claude hooks enable

# .claude/hooks/compliance-gate.yaml
# 企业策略：所有 Claude 修改必须经过合规门禁
 
name: financial-compliance-gate
version: 1.0.0
description: 金融级代码合规检查门禁系统
 
hooks:
  - id: pre-write-license
    trigger: before_file_write
    script: scripts/check_license.py
    fail_action: block
    timeout: 5s
    description: 确保所有源文件包含公司许可证头
 
  - id: pre-write-secrets
    trigger: before_file_write
    script: scripts/scan_secrets.py
    fail_action: block
    timeout: 10s
    description: 阻止任何密钥、密码、Token 被写入代码
 
  - id: pre-command-sandbox
    trigger: before_command
    script: scripts/sandbox_check.sh
    fail_action: block
    timeout: 3s
    description: 禁止执行任何可能接触生产环境的命令
 
  - id: post-write-sonarqube
    trigger: after_file_write
    script: scripts/trigger_sonar.sh
    fail_action: warn
    timeout: 60s
    description: 触发 SonarQube 质量门禁扫描
 
  - id: post-write-audit
    trigger: after_file_write
    http:
      method: POST
      url: https://audit.company.com/api/v1/claude-events
      headers:
        Authorization: Bearer {{env.AUDIT_API_TOKEN}}
        X-Audit-Source: claude-code
      body: |
        {
          "event_type": "file_write",
          "file_path": "{{file_path}}",
          "session_id": "{{session_id}}",
          "user": "{{env.USER}}",
          "timestamp": "{{timestamp}}",
          "diff_summary": "{{diff_summary}}"
        }
    description: 写入审计日志（不可失败，失败则告警）
 
  - id: session-end-report
    trigger: session_end
    http:
      method: POST
      url: https://compliance.company.com/api/v1/session-reports
      headers:
        Authorization: Bearer {{env.COMPLIANCE_TOKEN}}
      body: |
        {
          "session_id": "{{session_id}}",
          "duration_sec": {{duration_sec}},
          "files_modified": {{files_modified}},
          "commands_executed": {{commands_executed}},
          "model": "{{model}}",
          "cost_usd": {{cost_usd}}
        }
    description: 会话结束时上报完整使用报告

# scripts/scan_secrets.py - 生产级密钥扫描
#!/usr/bin/env python3
import re
import sys
 
SECRETS_PATTERNS = [
    (r'sk-[a-zA-Z0-9]{48}', 'OpenAI/Anthropic API Key'),
    (r'AKIA[0-9A-Z]{16}', 'AWS Access Key ID'),
    (r'ghp_[a-zA-Z0-9]{36}', 'GitHub Personal Access Token'),
    (r'password\s*=\s*["\'][^"\']+["\']', 'Hardcoded Password'),
    (r'secret\s*=\s*["\'][^"\']{8,}["\']', 'Hardcoded Secret'),
    (r'-----BEGIN (RSA |EC |DSA |OPENSSH )?PRIVATE KEY-----', 'Private Key'),
]
 
WHITELIST_FILES = ['.env.example', 'docker-compose.yml', 'README.md']
 
def scan_content(content: str, filename: str) -> list:
    violations = []
    lines = content.split('\n')
    for line_num, line in enumerate(lines, 1):
        for pattern, name in SECRETS_PATTERNS:
            if re.search(pattern, line, re.IGNORECASE):
                # 检查是否在白名单注释中
                if '# nosec' in line or '# noqa' in line:
                    continue
                violations.append({
                    'line': line_num,
                    'type': name,
                    'snippet': line.strip()[:80]
                })
    return violations
 
if __name__ == '__main__':
    file_path = sys.argv[1] if len(sys.argv) > 1 else ''
    content = sys.stdin.read()
    
    if any(wl in file_path for wl in WHITELIST_FILES):
        print(f"SKIP: {file_path} is whitelisted")
        sys.exit(0)
    
    violations = scan_content(content, file_path)
    if violations:
        print(f"BLOCKED: Found {len(violations)} secrets in {file_path}")
        for v in violations:
            print(f"  Line {v['line']}: {v['type']} | {v['snippet']}")
        sys.exit(1)
    
    print(f"PASS: No secrets found in {file_path}")
    sys.exit(0)

# scripts/sandbox_check.sh - 生产环境隔离检查
#!/bin/bash
COMMAND="$1"
 
# 严格禁止的模式
DENY_PATTERNS=(
    "rm -rf /"
    "rm -rf /*"
    ":(){ :|:& };:"
    "> /dev/sda"
    "mkfs."
    "dd if=/dev/zero"
    "psql.*production"
    "mysql -h.*prod"
    "redis-cli -h.*prod"
    "kubectl.*delete.*namespace"
    "terraform destroy"
)
 
for pattern in "${DENY_PATTERNS[@]}"; do
    if "$COMMAND" =~ $pattern; then
        echo "BLOCKED: Dangerous command pattern detected: '$COMMAND'"
        echo "Matches pattern: $pattern"
        exit 1
    fi
done
 
# 需要二次确认的模式
CONFIRM_PATTERNS=(
    "pip install"
    "npm install"
    "docker run"
    "kubectl apply"
    "git push"
    "git reset --hard"
)
 
for pattern in "${CONFIRM_PATTERNS[@]}"; do
    if "$COMMAND" == *"$pattern"*; then
        echo "REQUIRE_CONFIRM: '$COMMAND' requires manual approval"
        exit 2  # 特殊退出码表示需要确认
    fi
done
 
echo "PASS: Command '$COMMAND' is allowed"
exit 0

# .claude/hooks/doc-sync-pipeline.yaml
name: doc-sync-pipeline
description: API 代码变更自动同步到知识库
 
hooks:
  - id: detect-api-change
    trigger: after_file_write
    condition: "file_path matches '.*(controller|router|handler|api).*\\.(py|ts|go|java)$'"
    script: scripts/extract_openapi.py
    description: 检测 API 文件变更并提取 OpenAPI 定义
 
  - id: generate-markdown-doc
    trigger: after_file_write
    depends_on: [detect-api-change]
    script: scripts/gen_api_doc.py
    description: 根据代码变更生成 Markdown 文档片段
 
  - id: sync-to-confluence
    trigger: after_file_write
    depends_on: [generate-markdown-doc]
    http:
      method: PUT
      url: https://company.atlassian.net/wiki/rest/api/content/{{page_id}}
      headers:
        Authorization: Basic {{env.CONFLUENCE_AUTH}}
        Content-Type: application/json
      body: |
        {
          "version": {"number": {{new_version}}},
          "title": "{{page_title}}",
          "type": "page",
          "body": {
            "storage": {
              "value": "{{generated_html}}",
              "representation": "storage"
            }
          }
        }
    description: 将生成的文档同步到 Confluence
 
  - id: notify-team
    trigger: after_file_write
    depends_on: [sync-to-confluence]
    http:
      method: POST
      url: {{env.SLACK_WEBHOOK_URL}}
      body: |
        {
          "text": "📚 API 文档已自动更新",
          "blocks": [
            {
              "type": "section",
              "text": {
                "type": "mrkdwn",
                "text": "*{{file_name}}* 的变更已同步到 Confluence\n"
                         "<{{confluence_url}}|查看更新后的文档>"
              }
            }
          ]
        }
    description: 通知团队文档已更新

# scripts/extract_openapi.py - 从代码中提取 OpenAPI 定义
#!/usr/bin/env python3
"""自动从 FastAPI/Flask/Express 代码中提取 API 变更"""
 
import ast
import json
import sys
from pathlib import Path
 
def extract_fastapi_routes(file_path: str) -> list:
    """从 FastAPI 文件中提取路由定义"""
    content = Path(file_path).read_text()
    tree = ast.parse(content)
    routes = []
    
    for node in ast.walk(tree):
        if isinstance(node, ast.Call):
            # 检测 @app.get / @app.post 等装饰器调用
            func_name = ""
            if isinstance(node.func, ast.Attribute):
                func_name = node.func.attr
            
            if func_name in ('get', 'post', 'put', 'delete', 'patch'):
                route_info = {'method': func_name.upper(), 'args': []}
                for kw in node.keywords:
                    if kw.arg == 'path' or kw.arg == 'response_model':
                        route_info[kw.arg] = ast.literal_eval(kw.value) if isinstance(kw.value, ast.Constant) else str(kw.value)
                routes.append(route_info)
    
    return routes
 
def generate_changelog(old_routes: list, new_routes: list) -> str:
    """生成 API 变更日志"""
    old_paths = {r.get('path', '') for r in old_routes}
    new_paths = {r.get('path', '') for r in new_routes}
    
    added = new_paths - old_paths
    removed = old_paths - new_paths
    
    lines = ["## API 变更"]
    if added:
        lines.append("### 新增接口")
        for path in sorted(added):
            lines.append(f"- `+ {path}`")
    if removed:
        lines.append("### 移除接口")
        for path in sorted(removed):
            lines.append(f"- `- {path}`")
    if not added and not removed:
        lines.append("- 接口路径无变更，可能有参数或返回值调整")
    
    return "\n".join(lines)
 
if __name__ == '__main__':
    file_path = sys.argv[1]
    routes = extract_fastapi_routes(file_path)
    print(json.dumps({
        "file": file_path,
        "routes": routes,
        "route_count": len(routes)
    }, indent=2))