fix(ci): merge develop, 保留push event before/after对比逻辑

- 新增SAFE_HISTORICAL_MIGRATIONS白名单机制
- 已执行过的历史schema重构（如016_phase8_edit_template_plan）不再误报
- 与push event before/after对比优化叠加，双重保障

scripts/check_migration_safety.py

@@ -35,6 +35,7 @@ import os
 import re
 import subprocess
 import sys
+import time
 import urllib.error
 import urllib.request
 from pathlib import Path
@@ -78,6 +79,15 @@ SAFE_PATTERNS = [
 ]
 
 
+def _get_env(*names: str, default: str = "") -> str:
+    """按优先级尝试多个环境变量名，返回第一个非空值。"""
+    for name in names:
+        val = os.environ.get(name, "")
+        if val:
+            return val
+    return default
+
+
 def extract_upgrade_content(content: str) -> str:
     """
     从迁移文件中提取 upgrade 函数的内容。
@@ -100,25 +110,87 @@ def extract_upgrade_content(content: str) -> str:
     return content[upgrade_start:upgrade_end]
 
 
-def get_new_migrations_via_diff(diff_target: str) -> List[Path]:
+def _api_get_with_retry(url: str, token: str, max_retries: int = 3) -> dict | list:
     """
-    通过 Gitea API 对比目标分支，找出 alembic/versions/ 下新增的迁移文件。
-    不依赖本地 git，避免 CI 环境下 git 操作不稳定的问题。
+    带重试的 API 调用。
+    指数退避：1s, 2s, 4s
     """
-    api_url = os.environ.get("GITHUB_API_URL", "")
-    repo = os.environ.get("GITHUB_REPOSITORY", "")
-    token = os.environ.get("GITHUB_TOKEN", "")
+    last_error = None
+    for attempt in range(max_retries):
+        try:
+            req = urllib.request.Request(url, headers={"Authorization": f"token {token}"})
+            with urllib.request.urlopen(req, timeout=15) as resp:
+                return json.loads(resp.read().decode())
+        except urllib.error.HTTPError as e:
+            # 404 说明目录不存在或分支不存在，直接抛
+            if e.code == 404:
+                raise
+            last_error = e
+            if attempt < max_retries - 1:
+                wait = 2**attempt
+                print(f"   （API 请求失败，{wait}s 后重试 {attempt + 1}/{max_retries}：{e}）")
+                time.sleep(wait)
+        except Exception as e:
+            last_error = e
+            if attempt < max_retries - 1:
+                wait = 2**attempt
+                print(f"   （API 请求失败，{wait}s 后重试 {attempt + 1}/{max_retries}：{e}）")
+                time.sleep(wait)
+    raise last_error  # type: ignore[misc]
+
+
+
+def _get_migration_files_at_ref(ref: str, api_url: str, repo: str, token: str) -> set:
+    """通过 API 获取指定 ref（分支/commit/tag）下的迁移文件名集合"""
+    url = f"{api_url}/repos/{repo}/contents/alembic/versions?ref={ref}"
+    data = _api_get_with_retry(url, token)
+    if isinstance(data, list):
+        return {item["name"] for item in data if item["name"].endswith(".py")}
+    return set()
+
+def get_new_migrations_via_api(diff_target: str) -> List[Path] | None:
+    """
+    通过 Gitea/GitHub Contents API 对比目标分支，找出 alembic/versions/ 下新增的迁移文件。
+    返回 None 表示 API 方式不可用，调用方应尝试其他方式。
+    """
+    # 同时支持 Gitea 和 GitHub 的环境变量命名
+    api_url = _get_env("GITEA_API_URL", "GITHUB_API_URL", "CI_API_V4_URL")
+    repo = _get_env("GITEA_REPOSITORY", "GITHUB_REPOSITORY", "CI_PROJECT_PATH")
+    token = _get_env("GITEA_TOKEN", "GITHUB_TOKEN", "CI_JOB_TOKEN")
     branch = diff_target.replace("origin/", "")
 
+    # push event 下优先用 before/after 精确对比本次 push 的变更范围
+    event_name = _get_env("GITEA_EVENT_NAME", "GITHUB_EVENT_NAME", "CI_EVENT_NAME")
+    before_sha = _get_env("GITEA_BEFORE", "GITHUB_BEFORE", "CI_COMMIT_BEFORE_SHA")
+    after_sha = _get_env("GITEA_AFTER", "GITHUB_AFTER", "GITEA_SHA", "GITHUB_SHA", "CI_COMMIT_SHA")
+
     if not api_url or not repo or not token:
-        print("⚠️  CI 环境变量不完整，降级为检查所有迁移文件")
-        return sorted(ALEMBIC_VERSIONS_DIR.glob("*.py"))
+        print(
+            f"   （API 环境变量不完整：api_url={'✓' if api_url else '✗'} repo={'✓' if repo else '✗'} token={'✓' if token else '✗'}）"
+        )
+        return None
 
     try:
+        # push event: 用 before/after 精确对比本次 push 新增的迁移
+        if event_name == "push" and before_sha and after_sha and before_sha != "0" * 40:
+            before_files = _get_migration_files_at_ref(before_sha, api_url, repo, token)
+            after_files = _get_migration_files_at_ref(after_sha, api_url, repo, token)
+            new_file_names = sorted(after_files - before_files)
+            if new_file_names:
+                result = [ALEMBIC_VERSIONS_DIR / f for f in new_file_names]
+                print(f"   （push event，对比 {before_sha[:8]}..{after_sha[:8]}，发现 {len(result)} 个新增迁移）")
+                return result
+            else:
+                print(f"   （push event，对比 {before_sha[:8]}..{after_sha[:8]}，无新增迁移）")
+                return []
+
         url = f"{api_url}/repos/{repo}/contents/alembic/versions?ref={branch}"
-        req = urllib.request.Request(url, headers={"Authorization": f"token {token}"})
-        with urllib.request.urlopen(req, timeout=15) as resp:
-            data = json.loads(resp.read().decode())
+        data = _api_get_with_retry(url, token)
+
+        if isinstance(data, dict):
+            # Gitea 目录不存在时返回 404，不会到这里；如果返回 dict 可能是错误信息
+            print(f"   （API 返回异常：{str(data)[:100]}）")
+            return None
 
         remote_files = {item["name"] for item in data if item["name"].endswith(".py")}
         local_files = {f.name for f in ALEMBIC_VERSIONS_DIR.glob("*.py")}
@@ -132,9 +204,82 @@ def get_new_migrations_via_diff(diff_target: str) -> List[Path]:
             print(f"   （API 对比 {branch} 分支，无新增迁移）")
             return []
     except Exception as e:
-        print(f"⚠️  API 获取迁移列表失败：{e}")
-        print("   降级为检查所有迁移文件")
-        return sorted(ALEMBIC_VERSIONS_DIR.glob("*.py"))
+        print(f"   （API 获取迁移列表失败：{e}）")
+        return None
+
+
+def get_new_migrations_via_git(diff_target: str) -> List[Path] | None:
+    """
+    Fallback：通过本地 git diff 找出新增的迁移文件。
+    CI 环境中 git 可用时作为 API 失败后的兜底方案。
+    """
+    try:
+        # 确保目标分支存在
+        subprocess.run(
+            ["git", "fetch", "origin", diff_target.replace("origin/", ""), "--depth=50"],
+            capture_output=True,
+            cwd=str(REPO_ROOT),
+            timeout=30,
+        )
+
+        # 优先使用三点diff（找合并基线），失败时回退到两点diff（兼容tar.gz checkout + git init的CI环境）
+        diff_args = ["git", "diff", "--name-only", "--diff-filter=A", f"{diff_target}...HEAD"]
+        result = subprocess.run(
+            diff_args,
+            capture_output=True,
+            text=True,
+            cwd=str(REPO_ROOT),
+            timeout=10,
+        )
+        if result.returncode != 0:
+            # fallback: 两点diff（无需共同祖先）
+            diff_args_2 = ["git", "diff", "--name-only", "--diff-filter=A", diff_target, "HEAD"]
+            result = subprocess.run(
+                diff_args_2,
+                capture_output=True,
+                text=True,
+                cwd=str(REPO_ROOT),
+                timeout=10,
+            )
+            if result.returncode != 0:
+                print(f"   （git diff 失败：{result.stderr.strip()}）")
+                return None
+
+        new_migrations = []
+        for line in result.stdout.strip().split("\n"):
+            line = line.strip()
+            if line.startswith("alembic/versions/") and line.endswith(".py"):
+                new_migrations.append(REPO_ROOT / line)
+
+        new_migrations.sort()
+        print(f"   （git diff 对比 {diff_target}，发现 {len(new_migrations)} 个新增迁移）")
+        return new_migrations
+    except Exception as e:
+        print(f"   （git diff 方式失败：{e}）")
+        return None
+
+
+def get_new_migrations_via_diff(diff_target: str) -> List[Path]:
+    """
+    找出相对目标分支新增的迁移文件，按优先级尝试多种方式：
+    1. Gitea/GitHub Contents API（最可靠，不受本地 checkout 深度影响）
+    2. git diff（API 失败时的兜底）
+    3. 全量扫描（以上都失败时的最后兜底，会输出警告）
+    """
+    print("🔍 尝试通过 API 获取新增迁移列表...")
+    result = get_new_migrations_via_api(diff_target)
+    if result is not None:
+        return result
+
+    print("🔍 API 不可用，尝试 git diff 方式...")
+    result = get_new_migrations_via_git(diff_target)
+    if result is not None:
+        return result
+
+    print("⚠️  所有增量方式均失败，降级为检查所有迁移文件")
+    print("   这可能导致历史迁移中的破坏性操作被误报")
+    print("   建议检查 CI 环境变量配置（GITHUB_API_URL / GITHUB_REPOSITORY / GITHUB_TOKEN）")
+    return sorted(ALEMBIC_VERSIONS_DIR.glob("*.py"))
 
 
 def find_new_migrations(since_revision: str | None = None, diff_against: str | None = None) -> List[Path]: