From 83b2940f7c0c2aff06339a609e08e044f30da7f9 Mon Sep 17 00:00:00 2001
From: xiaoxia <dev@xiaoxiajianji.com>
Date: Mon, 13 Jul 2026 16:29:53 +0800
Subject: [PATCH] =?UTF-8?q?fix(ci):=20check=5Fmigration=5Fsafety.py=20blac?=
 =?UTF-8?q?k=E6=A0=BC=E5=BC=8F=E5=8C=96=E4=BF=AE=E5=A4=8D?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 scripts/check_migration_safety.py | 32 ++++++++-----------------------
 1 file changed, 8 insertions(+), 24 deletions(-)

diff --git a/scripts/check_migration_safety.py b/scripts/check_migration_safety.py
index f2acd9194..2c0a08907 100644
--- a/scripts/check_migration_safety.py
+++ b/scripts/check_migration_safety.py
@@ -145,9 +145,7 @@ def check_not_null_without_default(file_path: Path, upgrade_content: str) -> Lis
             continue
 
         # 检查是否有默认值（server_default 或 default）
-        has_default = bool(
-            re.search(r"server_default\s*=", call) or re.search(r"\bdefault\s*=", call)
-        )
+        has_default = bool(re.search(r"server_default\s*=", call) or re.search(r"\bdefault\s*=", call))
 
         if not has_default:
             # 提取表名和列名
@@ -160,14 +158,10 @@ def check_not_null_without_default(file_path: Path, upgrade_content: str) -> Lis
                 table = table_col_match.group(1)
                 col = table_col_match.group(2)
                 issues.append(
-                    f"{file_path.name}: 新增 NOT NULL 约束无默认值 "
-                    f"（表: {table}, 列: {col}）- 旧数据为空时迁移失败"
+                    f"{file_path.name}: 新增 NOT NULL 约束无默认值 " f"（表: {table}, 列: {col}）- 旧数据为空时迁移失败"
                 )
             else:
-                issues.append(
-                    f"{file_path.name}: 新增 NOT NULL 约束无默认值 "
-                    "- 旧数据为空时迁移失败"
-                )
+                issues.append(f"{file_path.name}: 新增 NOT NULL 约束无默认值 " "- 旧数据为空时迁移失败")
 
     return issues
 
@@ -210,9 +204,7 @@ def check_large_table_alter(file_path: Path, upgrade_content: str) -> List[str]:
             )
 
     # 检测大表上的 drop_column
-    drop_col_pattern = (
-        r"op\.drop_column\(\s*['\"]([^'\"]+)['\"]\s*,\s*['\"]([^'\"]+)['\"]"
-    )
+    drop_col_pattern = r"op\.drop_column\(\s*['\"]([^'\"]+)['\"]\s*,\s*['\"]([^'\"]+)['\"]"
     for match in re.finditer(drop_col_pattern, upgrade_content):
         table = match.group(1)
         col = match.group(2)
@@ -245,9 +237,7 @@ def get_new_migrations_via_diff(diff_target: str) -> List[Path]:
             text=True,
             check=True,
         )
-        files = [
-            line.strip() for line in result.stdout.strip().split("\n") if line.strip()
-        ]
+        files = [line.strip() for line in result.stdout.strip().split("\n") if line.strip()]
         return [REPO_ROOT / f for f in files]
     except subprocess.CalledProcessError as e:
         print(f"⚠️  git diff 失败（{diff_target}）：{e.stderr.strip()}")
@@ -255,9 +245,7 @@ def get_new_migrations_via_diff(diff_target: str) -> List[Path]:
         return sorted(ALEMBIC_VERSIONS_DIR.glob("*.py"))
 
 
-def find_new_migrations(
-    since_revision: str | None = None, diff_against: str | None = None
-) -> List[Path]:
+def find_new_migrations(since_revision: str | None = None, diff_against: str | None = None) -> List[Path]:
     """
     找出需要检查的迁移文件。
     优先级：diff_against > since_revision > 全部
@@ -320,9 +308,7 @@ def analyze_migration(file_path: Path) -> Tuple[List[str], List[str], List[str]]
 
 
 def main() -> int:
-    parser = argparse.ArgumentParser(
-        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
-    )
+    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
     parser.add_argument(
         "--since",
         default=os.getenv("MIGRATION_SINCE_REVISION"),
@@ -388,9 +374,7 @@ def main() -> int:
         print()
 
     print("=" * 60)
-    print(
-        f"检查结果：{len(all_safe)} 项安全 / {len(all_medium)} 项中风险 / {len(all_high)} 项高风险"
-    )
+    print(f"检查结果：{len(all_safe)} 项安全 / {len(all_medium)} 项中风险 / {len(all_high)} 项高风险")
     print()
 
     if all_high: