diff --git a/scripts/check_migration_safety.py b/scripts/check_migration_safety.py index 79f370f6a..4d53c9d25 100644 --- a/scripts/check_migration_safety.py +++ b/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,58 @@ 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_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/", "") 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: 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 +175,70 @@ 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, + ) + + result = subprocess.run( + ["git", "diff", "--name-only", "--diff-filter=A", f"{diff_target}...HEAD"], + 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]: