迁移前自动升级 SQLite 结构

scripts/migrate_sqlite_to_mysql.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import argparse
 import json
 import os
+import subprocess
 import sys
 from collections.abc import Iterable
 from pathlib import Path
@@ -108,6 +109,19 @@ def _ensure_target_is_empty(target_engine: Engine, target_tables: list[Table]) -
         raise RuntimeError(f"目标 MySQL 已存在业务数据，拒绝迁移：{', '.join(occupied)}")
 
 
+def _upgrade_source_sqlite(source_path: Path) -> None:
+    """先将源 SQLite 升级到当前 Alembic 版本，补齐历史表字段。"""
+    print("正在升级源 SQLite 表结构...")
+    source_env = os.environ.copy()
+    source_env["DATABASE_URL"] = f"sqlite:///{source_path}"
+    subprocess.run(
+        [sys.executable, "-c", "from web.backend.database import run_migrations; run_migrations()"],
+        cwd=PROJECT_ROOT,
+        env=source_env,
+        check=True,
+    )
+
+
 def main() -> int:
     """执行迁移、计数核验并返回进程退出码。"""
     args = parse_args()
@@ -117,6 +131,8 @@ def main() -> int:
     if not source_path.is_file():
         raise FileNotFoundError(f"未找到 SQLite 数据库：{source_path}")
 
+    _upgrade_source_sqlite(source_path)
+
     # 必须在导入数据库模块前设置，Alembic 环境才会使用迁移目标库。
     if target_url:
         os.environ["DATABASE_URL"] = target_url
@@ -183,7 +199,7 @@ def main() -> int:
         source_engine.dispose()
         target_engine.dispose()
 
-    print("迁移完成，SQLite 源文件未修改。")
+    print("迁移完成，SQLite 业务数据未修改，表结构已升级到当前版本。")
     return 0