diff --git a/pyproject.toml b/pyproject.toml index a366d7bc..2003b2f9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -81,7 +81,6 @@ asyncio_default_fixture_loop_scope = "function" [dependency-groups] dev = [ - "ipykernel>=6.30.0", "ruff>=0.12.1", ] test = [ diff --git a/scripts/migrate_user_soft_delete.py b/scripts/migrate_user_soft_delete.py new file mode 100644 index 00000000..304c8525 --- /dev/null +++ b/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() diff --git a/server/utils/migrate.py b/server/utils/migrate.py index e970ddc2..271f38fc 100644 --- a/server/utils/migrate.py +++ b/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 ( - self.check_column_exists("users", "login_failed_count") - and self.check_column_exists("users", "last_failed_login") - and self.check_column_exists("users", "login_locked_until") - ): + required_columns = [ + "login_failed_count", + "last_failed_login", + "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: """获取最新迁移版本号""" - # 这里返回硬编码的最新版本号,不依赖迁移定义 - # 因为迁移定义可能为空(字段已存在) - return 1 # 当前最新版本是 v1 + migrations = self.get_migrations() + return max((version for version, _, _ in migrations), default=0) 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") - # 如果有命令需要执行,才添加迁移 - if v1_commands: - migrations.append((1, "为用户表添加登录失败限制字段", 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)