From 911ff798fa9950a978dc437bbcf9c5019ae798ea Mon Sep 17 00:00:00 2001 From: CI Bot Date: Thu, 16 Jul 2026 08:59:07 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=A2=9E=E5=8A=A0=E5=8E=86=E5=8F=B2migr?= =?UTF-8?q?ation=E7=99=BD=E5=90=8D=E5=8D=95=EF=BC=8C=E9=81=BF=E5=85=8D?= =?UTF-8?q?=E5=B7=B2=E7=9F=A5schema=E9=87=8D=E6=9E=84=E8=AF=AF=E6=8A=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增SAFE_HISTORICAL_MIGRATIONS白名单机制 - 已执行过的历史schema重构（如016_phase8_edit_template_plan）不再误报 - 与push event before/after对比优化叠加，双重保障 --- scripts/check_migration_safety.py | 53 +++++++++++++++++++++++++++---- 1 file changed, 46 insertions(+), 7 deletions(-) mode change 100644 => 100755 scripts/check_migration_safety.py diff --git a/scripts/check_migration_safety.py b/scripts/check_migration_safety.py old mode 100644 new mode 100755 index 79f370f6a..0e2d57e6b --- a/scripts/check_migration_safety.py +++ b/scripts/check_migration_safety.py @@ -49,6 +49,15 @@ HIGH_RISK_PATTERNS = [ (r"\bop\.drop_column\(", "op.drop_column() - 删除列，数据永久丢失"), ] +# 已确认安全的历史 migration 白名单 +# 这些 migration 已在生产环境执行过，其中的高风险操作（如 drop_column） +# 是经过确认的 schema 重构，不是意外的破坏性变更。 +# 格式：migration 文件名（不含 .py 后缀） +SAFE_HISTORICAL_MIGRATIONS = { + # Phase 8 模板/计划表重构，edit_templates 表 schema 替换 + "016_phase8_edit_template_plan", +} + # 中风险模式：可能导致数据丢失或兼容性问题 MEDIUM_RISK_PATTERNS = [ ( @@ -100,27 +109,51 @@ def extract_upgrade_content(content: str) -> str: return content[upgrade_start:upgrade_end] +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}" + 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()) + return {item["name"] for item in data if item["name"].endswith(".py")} + + def get_new_migrations_via_diff(diff_target: str) -> List[Path]: """ - 通过 Gitea API 对比目标分支，找出 alembic/versions/ 下新增的迁移文件。 + 通过 Gitea API 对比目标引用，找出 alembic/versions/ 下新增的迁移文件。 不依赖本地 git，避免 CI 环境下 git 操作不稳定的问题。 + + push event 下优先使用 GITHUB_BEFORE/GITHUB_AFTER 对比本次 push 的变更范围； + 否则使用 diff_target 分支与当前本地文件对比。 """ api_url = os.environ.get("GITHUB_API_URL", "") repo = os.environ.get("GITHUB_REPOSITORY", "") token = os.environ.get("GITHUB_TOKEN", "") - branch = diff_target.replace("origin/", "") + event_name = os.environ.get("GITHUB_EVENT_NAME", "") + before_sha = os.environ.get("GITHUB_BEFORE", "") + after_sha = os.environ.get("GITHUB_AFTER", os.environ.get("GITHUB_SHA", "")) if not api_url or not repo or not token: print("⚠️ CI 环境变量不完整，降级为检查所有迁移文件") return sorted(ALEMBIC_VERSIONS_DIR.glob("*.py")) 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()) + # 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 [] - remote_files = {item["name"] for item in data if item["name"].endswith(".py")} + # PR/其他场景: 对比目标分支与当前本地文件 + branch = diff_target.replace("origin/", "") + remote_files = _get_migration_files_at_ref(branch, api_url, repo, token) local_files = {f.name for f in ALEMBIC_VERSIONS_DIR.glob("*.py")} new_file_names = sorted(local_files - remote_files) @@ -169,6 +202,12 @@ def analyze_migration(file_path: Path) -> Tuple[List[str], List[str], List[str]] if not upgrade_content: return [], [], [f"{file_path.name}: 未找到 upgrade 函数"] + # 已确认安全的历史 migration 直接跳过，避免历史已知的 schema 重构误报 + migration_stem = file_path.stem + if migration_stem in SAFE_HISTORICAL_MIGRATIONS: + safes = [f"{file_path.name}: 已确认安全的历史迁移（白名单）"] + return [], [], safes + high_risks = [] medium_risks = [] safes = []