style: black格式化check_migration_naming.py

- 新增migration文件命名规范检查（check_migration_naming.py）
- 新增downgrade -1回滚验证（双向一致性验证）
- 新增alembic check检测未生成migration的model变更（警告模式）
- 整合链完整性检查到validate_migration.sh
- 静态检查前置，快速失败节省资源

scripts/ci/check_migration_naming.py

@@ -0,0 +1,137 @@
+#!/usr/bin/env python3
+"""
+检查 Alembic migration 文件命名规范。
+
+规则:
+1. 文件名必须以数字前缀开头（3位补零），如 001_xxx.py、052_add_table.py
+2. 数字前缀必须连续递增（与 check_migration_chain.py 一致，但只看文件名）
+3. 数字前缀后必须跟有描述性后缀（不能只有数字）
+4. 文件名使用小写+下划线（snake_case）
+5. revision 变量值必须与文件名数字前缀一致（可选带描述后缀）
+
+用法:
+    python3 scripts/ci/check_migration_naming.py [alembic_versions_dir]
+
+默认目录: alembic/versions/
+
+退出码:
+    0 - 全部通过
+    1 - 有命名违规
+"""
+
+from __future__ import annotations
+
+import re
+import sys
+from pathlib import Path
+
+# 文件名格式: 3位数字_描述.py
+FILE_NAME_PATTERN = re.compile(r"^(\d{3})_[a-z][a-z0-9_]*\.py$")
+# 纯数字文件名（不允许）
+PURE_NUM_PATTERN = re.compile(r"^\d{3}\.py$")
+# revision 值的数字前缀
+REV_NUM_PATTERN = re.compile(r"^(\d{3})")
+# revision 变量行
+REV_LINE_PATTERN = re.compile(
+    r'^\s*revision\s*(?::\s*str\s*)?=\s*["\']([^"\']+)["\']',
+    re.MULTILINE,
+)
+
+
+def check_naming(versions_dir: Path) -> list[str]:
+    """检查 migration 文件命名，返回错误列表。"""
+    errors: list[str] = []
+
+    if not versions_dir.is_dir():
+        return [f"目录不存在: {versions_dir}"]
+
+    py_files = sorted(f for f in versions_dir.iterdir() if f.suffix == ".py")
+    if not py_files:
+        return [f"目录下没有 migration 文件: {versions_dir}"]
+
+    print(f"检查 migration 文件命名: {versions_dir}")
+    print(f"共 {len(py_files)} 个文件")
+    print()
+
+    # 1. 文件名格式检查
+    print("1. 文件名格式检查...")
+    file_nums: list[int] = []
+    for f in py_files:
+        name = f.name
+        if PURE_NUM_PATTERN.match(name):
+            errors.append(f"  ❌ {name}: 只有数字编号，缺少描述性后缀")
+            continue
+        m = FILE_NAME_PATTERN.match(name)
+        if not m:
+            errors.append(f"  ❌ {name}: 命名格式不规范，应为 NNN_description.py " f"（3位数字前缀+下划线+小写描述）")
+            continue
+        file_nums.append(int(m.group(1)))
+
+    if not any("命名格式不规范" in e or "缺少描述性后缀" in e for e in errors):
+        print(f"  ✅ 全部 {len(py_files)} 个文件名格式正确")
+    else:
+        for e in errors:
+            if "命名格式不规范" in e or "缺少描述性后缀" in e:
+                print(e)
+
+    # 2. 编号连续性检查（基于文件名数字前缀）
+    print()
+    print("2. 编号连续性检查...")
+    if file_nums:
+        expected = set(range(min(file_nums), max(file_nums) + 1))
+        actual = set(file_nums)
+        missing = sorted(expected - actual)
+        if missing:
+            errors.append(f"  ❌ 编号不连续，缺少: {', '.join(f'{n:03d}' for n in missing)}")
+            print(f"  ❌ 编号不连续，缺少 {len(missing)} 个: " f"{', '.join(f'{n:03d}' for n in missing)}")
+        else:
+            print(f"  ✅ 编号连续（{min(file_nums):03d} ~ {max(file_nums):03d}）")
+
+    # 3. revision 变量与文件名前缀一致性检查
+    print()
+    print("3. revision变量与文件名一致性检查...")
+    rev_mismatch = 0
+    for f in py_files:
+        m = FILE_NAME_PATTERN.match(f.name)
+        if not m:
+            continue  # 格式不对的已经报过了
+        file_num = m.group(1)
+        content = f.read_text(encoding="utf-8")
+        rev_match = REV_LINE_PATTERN.search(content)
+        if not rev_match:
+            errors.append(f"  ❌ {f.name}: 未找到 revision 变量定义")
+            rev_mismatch += 1
+            continue
+        rev_value = rev_match.group(1)
+        rev_num_match = REV_NUM_PATTERN.match(rev_value)
+        if not rev_num_match or rev_num_match.group(1) != file_num:
+            errors.append(f"  ❌ {f.name}: revision='{rev_value}' 与文件名前缀 {file_num} 不一致")
+            rev_mismatch += 1
+
+    if rev_mismatch == 0:
+        print(f"  ✅ 全部 {len(py_files)} 个文件的 revision 与文件名一致")
+
+    return errors
+
+
+def main() -> int:
+    versions_dir = Path(sys.argv[1]) if len(sys.argv) > 1 else Path("alembic/versions")
+
+    errors = check_naming(versions_dir)
+
+    print()
+    if errors:
+        print(f"❌ 发现 {len(errors)} 个命名问题")
+        print()
+        print("命名规范:")
+        print("  - 文件名格式: NNN_description.py（3位数字前缀 + 下划线 + 小写描述）")
+        print("  - 编号必须连续，不能跳号")
+        print("  - revision 变量的数字前缀必须与文件名一致")
+        return 1
+
+    print("✅ 所有 migration 文件命名规范检查通过")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/ci/validate_migration.sh

@@ -1,15 +1,63 @@
 #!/bin/bash
-# CI Validate: Alembic迁移验证（并行Job 3/3）
-# 需要PostgreSQL数据库
+# CI Validate: Alembic迁移验证（升级版）
+# 检查项:
+#   1. migration文件命名规范检查
+#   2. migration编号链完整性检查
+#   3. upgrade head 升级验证（真实PG执行）
+#   4. downgrade -1 回滚验证
+#   5. alembic check 检测未生成migration的model变更
+#
+# 需要PostgreSQL数据库（共享PG或临时容器）
+
 set -eu
-# 加载CI共享常量
+
 SCRIPT_DIR="$(dirname "${BASH_SOURCE[0]}")"
 # shellcheck source=ci_env.sh
 source "${SCRIPT_DIR}/ci_env.sh"
 
-echo "=== CI Validate: Alembic迁移验证 ==="
+echo "=== CI Validate: Alembic迁移验证（升级版）==="
+echo ""
+
+# ============================================================
+# 阶段0: 静态检查（不需要数据库，先快速失败）
+# ============================================================
+
+echo "📋 阶段0: 静态检查（命名规范 + 链完整性）"
+echo ""
+
+STATIC_FAILED=0
+
+echo "0.1 检查 migration 文件命名规范..."
+if python3 scripts/ci/check_migration_naming.py alembic/versions; then
+    echo "   ✅ 命名规范检查通过"
+else
+    echo "   ❌ 命名规范检查失败"
+    STATIC_FAILED=1
+fi
+
+echo ""
+echo "0.2 检查 migration 编号链完整性..."
+if python3 scripts/ci/check_migration_chain.py alembic/versions; then
+    echo "   ✅ 编号链完整性检查通过"
+else
+    echo "   ❌ 编号链完整性检查失败"
+    STATIC_FAILED=1
+fi
+
+if [ "$STATIC_FAILED" -ne 0 ]; then
+    echo ""
+    echo "❌ 静态检查失败，请修复上述问题后重试"
+    exit 1
+fi
+
+echo ""
+echo "✅ 静态检查全部通过"
+echo ""
+
+# ============================================================
+# DooD模式检测：确定宿主机访问地址
+# ============================================================
 
-# --- DooD模式检测：确定宿主机访问地址 ---
 detect_docker_host() {
   local test_port="${1:-${CI_LOCAL_PG_PORT}}"
 
@@ -63,20 +111,6 @@ except:
   return 1
 }
 
-# 获取宿主机IP
-if [ -S /var/run/docker.sock ]; then
-  DOCKER_HOST_IP=$(detect_docker_host "${CI_SHARED_PG_PORT}")
-  if [ "$DOCKER_HOST_IP" = "127.0.0.1" ]; then
-    DOCKER_HOST_IP=$(detect_docker_host 22)
-  fi
-  echo "检测到DooD模式，宿主机地址: $DOCKER_HOST_IP"
-else
-  DOCKER_HOST_IP="127.0.0.1"
-  echo "非DooD模式，使用 127.0.0.1"
-fi
-PG_HOST="$DOCKER_HOST_IP"
-echo "PG host: $PG_HOST"
-
 # 指数退避TCP连接检查
 wait_tcp_ready() {
   local host="$1"
@@ -96,8 +130,32 @@ wait_tcp_ready() {
   return 1
 }
 
+# 获取宿主机IP
+if [ -S /var/run/docker.sock ]; then
+  DOCKER_HOST_IP=$(detect_docker_host "${CI_SHARED_PG_PORT}")
+  if [ "$DOCKER_HOST_IP" = "127.0.0.1" ]; then
+    DOCKER_HOST_IP=$(detect_docker_host 22)
+  fi
+  echo "检测到DooD模式，宿主机地址: $DOCKER_HOST_IP"
+else
+  DOCKER_HOST_IP="127.0.0.1"
+  echo "非DooD模式，使用 127.0.0.1"
+fi
+PG_HOST="$DOCKER_HOST_IP"
+echo "PG host: $PG_HOST"
+echo ""
+
 USE_SHARED_PG="${CI_USE_SHARED_PG:-false}"
 
+# ============================================================
+# 准备数据库
+# ============================================================
+
+echo "🗄️  阶段1: 准备测试数据库"
+echo ""
+
+CI_DB_NAME="ci_migrate_${GITHUB_RUN_ID:-$$}"
+
 if [ "$USE_SHARED_PG" = "true" ]; then
   # 使用常驻共享PG实例
   echo "使用常驻共享PG实例（CI_USE_SHARED_PG=true）"
@@ -105,7 +163,6 @@ if [ "$USE_SHARED_PG" = "true" ]; then
   SHARED_PG_PORT="${CI_SHARED_PG_PORT}"
   SHARED_PG_USER="${CI_SHARED_PG_USER}"
   SHARED_PG_PASSWORD="${CI_SHARED_PG_PASSWORD}"
-  CI_DB_NAME="ci_run_${GITHUB_RUN_ID:-$$}"
 
   echo "等待共享PG连接就绪..."
   wait_tcp_ready "$SHARED_PG_HOST" "$SHARED_PG_PORT" 5
@@ -124,13 +181,10 @@ conn.close()
   export DATABASE_URL="postgresql+psycopg://${SHARED_PG_USER}:${SHARED_PG_PASSWORD}@${SHARED_PG_HOST}:${SHARED_PG_PORT}/${CI_DB_NAME}"
   echo "✅ 共享PG数据库已创建: $CI_DB_NAME"
 
-  # 执行迁移
-  PYTHONPATH="$PWD/apps/api:$PWD" python3 -m alembic upgrade head
-  echo "✅ Alembic migrations applied successfully"
-
-  # 清理数据库
-  echo "清理测试数据库: $CI_DB_NAME"
-  PGPASSWORD="$SHARED_PG_PASSWORD" python3 -c "
+  cleanup_db() {
+    echo ""
+    echo "清理测试数据库: $CI_DB_NAME"
+    PGPASSWORD="$SHARED_PG_PASSWORD" python3 -c "
 import psycopg2
 conn = psycopg2.connect(host='$SHARED_PG_HOST', port=$SHARED_PG_PORT, user='$SHARED_PG_USER', password='$SHARED_PG_PASSWORD', dbname='postgres')
 conn.autocommit = True
@@ -139,7 +193,8 @@ cur.execute(f'DROP DATABASE IF EXISTS \"$CI_DB_NAME\" WITH (FORCE)')
 cur.close()
 conn.close()
 " 2>/dev/null || echo "WARN: 数据库清理失败"
-  echo "✅ 共享PG数据库已清理"
+    echo "✅ 数据库已清理"
+  }
 else
   # 使用临时PG容器（默认模式）
   echo "使用临时PG容器模式"
@@ -176,12 +231,96 @@ else
   wait_tcp_ready "$PG_HOST" "$PG_PORT" 5
   echo "TCP connectivity to PostgreSQL confirmed on port $PG_PORT"
 
-  # 执行迁移
-  PYTHONPATH="$PWD/apps/api:$PWD" python3 -m alembic upgrade head
-  echo "✅ Alembic migrations applied successfully"
+  cleanup_db() {
+    docker rm -f "$PG_CONTAINER" 2>/dev/null || true
+  }
+fi
 
-  docker rm -f "$PG_CONTAINER" 2>/dev/null || true
+trap cleanup_db EXIT
+
+echo ""
+
+# ============================================================
+# 阶段2: upgrade head 升级验证
+# ============================================================
+
+echo "⬆️  阶段2: upgrade head 升级验证"
+echo ""
+
+PYTHONPATH="$PWD/apps/api:$PWD" python3 -m alembic upgrade head
+echo "✅ upgrade head 通过"
+echo ""
+
+# ============================================================
+# 阶段3: downgrade -1 回滚验证
+# ============================================================
+
+echo "⬇️  阶段3: downgrade -1 回滚验证"
+echo ""
+
+# 获取当前head版本号
+HEAD_REV=$(PYTHONPATH="$PWD/apps/api:$PWD" python3 -m alembic current 2>&1 | awk '{print $1}' | head -1)
+echo "当前版本 (head): $HEAD_REV"
+
+# 检查是否只有1个migration（baseline），downgrade -1会到base
+TOTAL_REVS=$(PYTHONPATH="$PWD/apps/api:$PWD" python3 -c "
+from alembic.config import Config
+from alembic.script import ScriptDirectory
+config = Config('alembic.ini')
+script = ScriptDirectory.from_config(config)
+print(len(list(script.walk_revisions())))
+")
+
+echo "总 migration 数量: $TOTAL_REVS"
+
+if [ "$TOTAL_REVS" -le 1 ]; then
+    echo "⚠️  只有1个migration，跳过 downgrade 回滚验证（没有可回滚的版本）"
+else
+    echo "执行 downgrade -1..."
+    PYTHONPATH="$PWD/apps/api:$PWD" python3 -m alembic downgrade -1
+    echo "✅ downgrade -1 通过"
+
+    # 回滚后再升级回去，确保双向都通
+    echo ""
+    echo "重新 upgrade head 验证双向一致性..."
+    PYTHONPATH="$PWD/apps/api:$PWD" python3 -m alembic upgrade head
+    echo "✅ 重新 upgrade head 通过（双向验证完成）"
 fi
 
 echo ""
-echo "=== CI Validate: Alembic迁移验证 通过 ✅ ==="
+
+# ============================================================
+# 阶段4: alembic check - 检测未生成migration的model变更
+# ============================================================
+
+echo "🔍 阶段4: 检查是否有未生成migration的model变更"
+echo ""
+
+# alembic check: 没有待生成的migration时退出码0，有变更时退出码1
+# 这里只检测，不阻断（警告模式），因为有些场景model变更不需要migration
+set +e
+CHECK_OUTPUT=$(PYTHONPATH="$PWD/apps/api:$PWD" python3 -m alembic check 2>&1)
+CHECK_EXIT=$?
+set -e
+
+if [ "$CHECK_EXIT" -eq 0 ]; then
+    echo "✅ 没有检测到未生成migration的model变更"
+else
+    if echo "$CHECK_OUTPUT" | grep -q "New upgrade operations detected"; then
+        echo "⚠️  检测到未生成migration的model变更！"
+        echo ""
+        echo "$CHECK_OUTPUT"
+        echo ""
+        echo "提示: 如果model变更是有意的且需要生成migration，请运行:"
+        echo "  alembic revision --autogenerate -m \"description\""
+        echo "如果model变更不涉及数据库schema（如仅索引/约束重命名或纯业务逻辑），请确认后忽略此警告。"
+        # 暂时不阻断，避免误报
+        echo "（当前为警告模式，不阻断CI，后续稳定后可升级为阻断）"
+    else
+        echo "⚠️  alembic check 执行出错（非阻断）"
+        echo "$CHECK_OUTPUT"
+    fi
+fi
+
+echo ""
+echo "=== CI Validate: Alembic迁移验证 全部通过 ✅ ==="