xiaoxia-saas/scripts/check_migration_safety.py

#!/usr/bin/env python3
"""
数据库迁移破坏性变更安全检查

只检查 Alembic 迁移文件的 upgrade 函数中是否包含破坏性操作：
- DROP TABLE
- ALTER TABLE ... DROP COLUMN
- 列类型变更（可能导致数据丢失）
- NOT NULL 约束新增（无默认值时）
- RENAME TABLE / RENAME COLUMN

忽略 downgrade 函数中的操作（那是回滚逻辑，正常的）。

退出码：
  0 - 安全 / 只有非破坏性变更
  1 - 检测到高风险破坏性变更
  2 - 检测到中风险变更，需人工确认
"""

from __future__ import annotations

import argparse
import os
import re
import sys
from pathlib import Path
from typing import List, Tuple

REPO_ROOT = Path(__file__).resolve().parents[1]
ALEMBIC_VERSIONS_DIR = REPO_ROOT / "alembic" / "versions"

# 高风险模式：直接导致数据丢失（只在 upgrade 中检查）
HIGH_RISK_PATTERNS = [
    (r"\bop\.drop_table\(", "op.drop_table() - 删除表，数据永久丢失"),
    (r"\bop\.drop_column\(", "op.drop_column() - 删除列，数据永久丢失"),
]

# 中风险模式：可能导致数据丢失或兼容性问题
MEDIUM_RISK_PATTERNS = [
    (r"op\.alter_column\([^)]*nullable\s*=\s*False", "新增 NOT NULL 约束 - 旧数据可能为空导致迁移失败"),
    (r"op\.alter_column\([^)]*type_\s*=", "列类型变更 - 可能导致数据截断或转换失败"),
    (r"\bop\.rename_table\(", "op.rename_table() - 重命名表，可能导致依赖该表的代码报错"),
    (r"\bop\.rename_column\(", "op.rename_column() - 重命名列，可能导致依赖该列的代码报错"),
    (r"\bop\.drop_index\(", "op.drop_index() - 删除索引，可能影响查询性能"),
    (r"\bop\.drop_constraint\(", "op.drop_constraint() - 删除约束，可能影响数据完整性"),
]

# 安全模式：这些是安全的新增操作
SAFE_PATTERNS = [
    (r"\bop\.create_table\(", "新建表"),
    (r"\bop\.add_column\(", "新增列"),
    (r"\bop\.create_index\(", "新建索引"),
    (r"\bop\.create_unique_constraint\(", "新建唯一约束"),
    (r"\bop\.create_foreign_key\(", "新建外键约束"),
]


def extract_upgrade_content(content: str) -> str:
    """
    从迁移文件中提取 upgrade 函数的内容。
    只检查 upgrade 中的操作，忽略 downgrade。
    """
    # 匹配 def upgrade(): 或 def upgrade() -> None: 等格式
    upgrade_match = re.search(r"def upgrade\b[^:]*:", content)
    if not upgrade_match:
        return ""

    upgrade_start = upgrade_match.end()

    # 找到下一个顶层 def（通常是 def downgrade）作为结束位置
    rest = content[upgrade_start:]
    downgrade_match = re.search(r"\n\ndef\s+\w+\b", rest)
    if downgrade_match:
        upgrade_end = upgrade_start + downgrade_match.start()
    else:
        upgrade_end = len(content)

    return content[upgrade_start:upgrade_end]


def find_new_migrations(since_revision: str | None = None) -> List[Path]:
    """
    找出新增的迁移文件。
    如果指定了 since_revision，则找出该版本之后的所有迁移；
    否则找出所有迁移文件。
    """
    all_migrations = sorted(ALEMBIC_VERSIONS_DIR.glob("*.py"))
    if not since_revision:
        return all_migrations

    result = []
    found = False
    for m in all_migrations:
        if since_revision in m.name or since_revision in m.stem:
            found = True
            continue
        if found:
            result.append(m)

    return result if found else all_migrations


def analyze_migration(file_path: Path) -> Tuple[List[str], List[str], List[str]]:
    """分析单个迁移文件 upgrade 部分的风险等级"""
    content = file_path.read_text()
    upgrade_content = extract_upgrade_content(content)

    if not upgrade_content:
        return [], [], [f"{file_path.name}: 未找到 upgrade 函数"]

    high_risks = []
    medium_risks = []
    safes = []

    for pattern, desc in HIGH_RISK_PATTERNS:
        if re.search(pattern, upgrade_content):
            high_risks.append(f"{file_path.name}: {desc}")

    for pattern, desc in MEDIUM_RISK_PATTERNS:
        if re.search(pattern, upgrade_content):
            medium_risks.append(f"{file_path.name}: {desc}")

    for pattern, desc in SAFE_PATTERNS:
        if re.search(pattern, upgrade_content):
            safes.append(f"{file_path.name}: {desc}")

    return high_risks, medium_risks, safes


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument(
        "--since",
        default=os.getenv("MIGRATION_SINCE_REVISION"),
        help="只检查指定版本之后的迁移（如：030_xxx），不传则检查所有迁移",
    )
    parser.add_argument(
        "--warn-only",
        action="store_true",
        help="只警告不失败（用于非强制门禁场景）",
    )
    parser.add_argument(
        "--allow-medium-risk",
        action="store_true",
        help="允许中风险变更（只拦截高风险）",
    )
    args = parser.parse_args()

    migrations = find_new_migrations(args.since)

    if not migrations:
        print("✅ 未找到需要检查的迁移文件")
        return 0

    print(f"🔍 正在检查 {len(migrations)} 个迁移文件的 upgrade 操作...")
    print()

    all_high = []
    all_medium = []
    all_safe = []

    for m in migrations:
        high, medium, safe = analyze_migration(m)
        all_high.extend(high)
        all_medium.extend(medium)
        all_safe.extend(safe)

    if all_safe:
        print("✅ 安全变更：")
        for s in all_safe:
            print(f"   - {s}")
        print()

    if all_medium:
        print("⚠️  中风险变更（需人工确认）：")
        for m_item in all_medium:
            print(f"   - {m_item}")
        print()

    if all_high:
        print("❌ 高风险破坏性变更（禁止自动部署）：")
        for h in all_high:
            print(f"   - {h}")
        print()

    print("=" * 60)
    print(f"检查结果：{len(all_safe)} 项安全 / {len(all_medium)} 项中风险 / {len(all_high)} 项高风险")
    print()

    if all_high:
        print("❌ 检测到高风险破坏性变更，CI 检查失败！")
        print("   如果确认这是预期操作，请在 MR/PR 中说明原因并获得审批。")
        if args.warn_only:
            return 0
        return 1

    if all_medium and not args.allow_medium_risk:
        print("⚠️  检测到中风险变更，请人工确认后再部署。")
        if args.warn_only:
            return 0
        print("（如需仅拦截高风险，可使用 --allow-medium-risk 参数）")
        return 2

    print("✅ 未检测到破坏性变更")
    return 0


if __name__ == "__main__":
    sys.exit(main())