feat: 添加用户表软删除字段迁移脚本，更新数据库迁移逻辑以支持新字段

pyproject.toml

@@ -81,7 +81,6 @@ asyncio_default_fixture_loop_scope = "function"
 
 [dependency-groups]
 dev = [
-    "ipykernel>=6.30.0",
     "ruff>=0.12.1",
 ]
 test = [

scripts/migrate_user_soft_delete.py

@@ -0,0 +1,41 @@
+#!/usr/bin/env python3
+"""
+用户表软删除字段迁移脚本
+
+该脚本用于为历史数据库添加 `is_deleted` 与 `deleted_at` 字段，
+同时会执行现有的数据库迁移逻辑，确保用户表结构与最新模型保持一致。
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+# 将项目根目录加入到 Python 路径，便于脚本在容器中执行
+PROJECT_ROOT = Path(__file__).resolve().parent.parent
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+from server.utils.migrate import DatabaseMigrator  # noqa: E402
+from src import config  # noqa: E402
+
+
+def main() -> None:
+    db_path = Path(config.save_dir) / "database" / "server.db"
+    migrator = DatabaseMigrator(str(db_path))
+
+    print(f"检测数据库: {db_path}")
+    current_version = migrator.get_current_version()
+    latest_version = migrator.get_latest_migration_version()
+    print(f"当前迁移版本: v{current_version}, 最新版本: v{latest_version}")
+
+    try:
+        migrator.run_migrations()
+        print("✅ 迁移完成，数据库结构已更新")
+    except Exception as exc:
+        print(f"❌ 迁移失败: {exc}")
+        raise
+
+
+if __name__ == "__main__":
+    main()

server/utils/migrate.py

@@ -164,11 +164,14 @@ class DatabaseMigrator:
         # 如果数据库已存在但没有版本表，创建版本表并设置为最新版本
         if current_version == 0 and latest_version > 0 and os.path.exists(self.db_path):
             # 检查users表是否已有新字段，如果有，说明是通过SQLAlchemy创建的
-            if (
+            required_columns = [
-                self.check_column_exists("users", "login_failed_count")
+                "login_failed_count",
-                and self.check_column_exists("users", "last_failed_login")
+                "last_failed_login",
-                and self.check_column_exists("users", "login_locked_until")
+                "login_locked_until",
-            ):
+                "is_deleted",
+                "deleted_at",
+            ]
+            if all(self.check_column_exists("users", column) for column in required_columns):
                 # 字段已存在，直接设置为最新版本
                 logger.info(f"检测到现有数据库已包含最新字段，设置版本为 v{latest_version}")
                 self.set_version(latest_version)
@@ -217,9 +220,8 @@ class DatabaseMigrator:
 
     def get_latest_migration_version(self) -> int:
         """获取最新迁移版本号"""
-        # 这里返回硬编码的最新版本号，不依赖迁移定义
+        migrations = self.get_migrations()
-        # 因为迁移定义可能为空（字段已存在）
+        return max((version for version, _, _ in migrations), default=0)
-        return 1  # 当前最新版本是 v1
 
     def get_migrations(self) -> list[tuple[int, str, list[str]]]:
         """获取所有迁移定义
@@ -243,9 +245,18 @@ class DatabaseMigrator:
         if not self.check_column_exists("users", "login_locked_until"):
             v1_commands.append("ALTER TABLE users ADD COLUMN login_locked_until DATETIME")
 
-        # 如果有命令需要执行，才添加迁移
+        migrations.append((1, "为用户表添加登录失败限制字段", v1_commands))
-        if v1_commands:
+
-            migrations.append((1, "为用户表添加登录失败限制字段", v1_commands))
+        # 迁移 v2: 为 users 表添加软删除字段
+        v2_commands: list[str] = []
+
+        if not self.check_column_exists("users", "is_deleted"):
+            v2_commands.append("ALTER TABLE users ADD COLUMN is_deleted INTEGER NOT NULL DEFAULT 0")
+
+        if not self.check_column_exists("users", "deleted_at"):
+            v2_commands.append("ALTER TABLE users ADD COLUMN deleted_at DATETIME")
+
+        migrations.append((2, "为用户表添加软删除字段", v2_commands))
 
         # 未来的迁移可以在这里添加
         # migrations.append((
@@ -290,6 +301,8 @@ def validate_database_schema(db_path: str) -> tuple[bool, list[str]]:
                 "login_failed_count",
                 "last_failed_login",
                 "login_locked_until",
+                "is_deleted",
+                "deleted_at",
             ],
             "operation_logs": ["id", "user_id", "operation", "details", "ip_address", "timestamp"],
         }
@@ -337,7 +350,9 @@ def check_and_migrate(db_path: str):
             logger.warning(f"  - {issue}")
 
         if os.path.exists(db_path):
-            logger.info("建议运行迁移脚本: docker exec api-dev python /app/scripts/migrate_user_fields.py")
+            logger.info(
+                "建议运行迁移脚本: docker exec api-dev python /app/scripts/migrate_user_soft_delete.py"
+            )
 
     migrator = DatabaseMigrator(db_path)