From f8cee206c89136efd7c05951cd52407396681780 Mon Sep 17 00:00:00 2001 From: Wenjie Zhang Date: Mon, 13 Oct 2025 17:29:59 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=E7=94=A8=E6=88=B7?= =?UTF-8?q?=E8=A1=A8=E8=BD=AF=E5=88=A0=E9=99=A4=E5=AD=97=E6=AE=B5=E8=BF=81?= =?UTF-8?q?=E7=A7=BB=E8=84=9A=E6=9C=AC=EF=BC=8C=E6=9B=B4=E6=96=B0=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E5=BA=93=E8=BF=81=E7=A7=BB=E9=80=BB=E8=BE=91=E4=BB=A5?= =?UTF-8?q?=E6=94=AF=E6=8C=81=E6=96=B0=E5=AD=97=E6=AE=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pyproject.toml | 1 - scripts/migrate_user_soft_delete.py | 41 +++++++++++++++++++++++++++++ server/utils/migrate.py | 39 ++++++++++++++++++--------- 3 files changed, 68 insertions(+), 13 deletions(-) create mode 100644 scripts/migrate_user_soft_delete.py 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)