feat(system): 实现用户管理接口

需求 2.5.1 用户账号管理，全部要求管理员权限（require_admin 依赖）：

- 接口：列表/详情/新增/编辑/删除/重置密码/启停/角色下拉，
  角色下拉放 /users/meta/roles 以避开 /{user_id} 的路由匹配冲突
- 删除、禁用、改角色三处自我保护：不允许对当前登录账号执行，
  避免管理员误操作后立刻失去全部管理权限且无法自行恢复
- 重置密码与禁用均提升 token_version，使已签发令牌立即失效，
  不依赖 Redis 黑名单
- 密码按字节数校验（bcrypt 5.0 硬上限 72 字节，中文一字 3 字节）
- 用户列表按 id 倒序：create_time 为秒级精度，同秒数据分页结果不确定
- 修复 replace_roles 未同步关系集合的缺陷：改用 user.user_roles
  集合追加，此前 session.add(UserRole(...)) 会表现为保存成功但返回空角色
- update_user 的自我保护原先依赖返回 None 的占位函数，恒不生效，
  改为显式传入 operator_id

Co-Authored-By: Claude Code <noreply@anthropic.com>

backend/alembic/script.py.mako

@@ -5,7 +5,7 @@ Revises: ${down_revision | comma,n}
 Create Date: ${create_date}
 
 """
-from typing import Sequence, Union
+from collections.abc import Sequence
 
 from alembic import op
 import sqlalchemy as sa
@@ -13,16 +13,22 @@ ${imports if imports else ""}
 
 # revision identifiers, used by Alembic.
 revision: str = ${repr(up_revision)}
-down_revision: Union[str, Sequence[str], None] = ${repr(down_revision)}
+down_revision: str | Sequence[str] | None = ${repr(down_revision)}
-branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
+branch_labels: str | Sequence[str] | None = ${repr(branch_labels)}
-depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}
+depends_on: str | Sequence[str] | None = ${repr(depends_on)}
 
 
 def upgrade() -> None:
-    """Upgrade schema."""
+    """应用迁移。"""
     ${upgrades if upgrades else "pass"}
 
 
 def downgrade() -> None:
-    """Downgrade schema."""
+    """回滚迁移。
+
+    **注意（autogenerate 不会替你处理）**：生成的 `op.drop_index` 与
+    `op.drop_table` 顺序需人工检查 —— 若某索引是外键的唯一支撑索引，
+    单独 drop_index 会报 MySQL 1553（needed in a foreign key constraint）。
+    **整表回滚应只保留 `op.drop_table`**，DROP TABLE 会连带删除索引与外键。
+    """
     ${downgrades if downgrades else "pass"}

backend/alembic/versions/6165691326d4_init_rbac_and_login_logs.py

@@ -337,17 +337,19 @@ def upgrade() -> None:
 
 
 def downgrade() -> None:
-    """Downgrade schema."""
+    """回滚迁移。
-    # ### commands auto generated by Alembic - please adjust! ###
+
-    op.drop_index("idx_user_roles_role_id", table_name="user_roles")
+    只保留 drop_table —— DROP TABLE 会连带删除索引与外键。
+    原 autogenerate 输出里逐个 drop_index 会失败：idx_user_roles_role_id 与
+    idx_role_permissions_permission_id 分别是外键的唯一支撑索引，
+    MySQL 拒绝单独删除（errno 1553），回滚会中断在半途。
+
+    删除顺序遵循外键依赖：先子表（user_roles / role_permissions / login_logs），
+    再父表（users / roles / permissions）。
+    """
     op.drop_table("user_roles")
-    op.drop_index("idx_role_permissions_permission_id", table_name="role_permissions")
     op.drop_table("role_permissions")
-    op.drop_index("idx_login_logs_username", table_name="login_logs")
-    op.drop_index("idx_login_logs_create_time", table_name="login_logs")
     op.drop_table("login_logs")
-    op.drop_index(op.f("idx_users_phone"), table_name="users")
     op.drop_table("users")
     op.drop_table("roles")
     op.drop_table("permissions")
-    # ### end Alembic commands ###

backend/alembic/versions/a45a8087831b_add_operation_logs.py

@@ -0,0 +1,109 @@
+"""新增操作日志表
+
+Revision ID: a45a8087831b
+Revises: 6165691326d4
+Create Date: 2026-10-10 10:06:17.435697
+
+"""
+
+from collections.abc import Sequence
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import mysql
+
+# revision identifiers, used by Alembic.
+revision: str = "a45a8087831b"
+down_revision: str | Sequence[str] | None = "6165691326d4"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """应用迁移。"""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "operation_logs",
+        sa.Column(
+            "id", mysql.BIGINT(unsigned=True), autoincrement=True, nullable=False, comment="主键ID"
+        ),
+        sa.Column(
+            "user_id",
+            mysql.BIGINT(unsigned=True),
+            nullable=True,
+            comment="操作人ID（账号删除后置空）",
+        ),
+        sa.Column(
+            "username",
+            sa.String(length=64),
+            nullable=False,
+            comment="操作人用户名（冗余存储便于追溯）",
+        ),
+        sa.Column("module", sa.String(length=64), nullable=False, comment="所属模块，如 用户管理"),
+        sa.Column(
+            "action", sa.String(length=32), nullable=False, comment="操作类型：新增/修改/删除/导出"
+        ),
+        sa.Column(
+            "summary", sa.String(length=255), nullable=False, comment="操作摘要（中文，供人阅读）"
+        ),
+        sa.Column("method", sa.String(length=8), nullable=True, comment="HTTP 方法"),
+        sa.Column("path", sa.String(length=255), nullable=True, comment="请求路径"),
+        sa.Column("ip", sa.String(length=45), nullable=True, comment="客户端IP（45位兼容IPv6）"),
+        sa.Column(
+            "user_agent",
+            sa.String(length=255),
+            nullable=True,
+            comment="客户端User-Agent（截断存储）",
+        ),
+        sa.Column("request_data", sa.Text(), nullable=True, comment="请求参数（已脱敏）"),
+        sa.Column(
+            "success",
+            mysql.TINYINT(),
+            server_default="0",
+            nullable=False,
+            comment="是否成功：1成功 0失败",
+        ),
+        sa.Column(
+            "error_message",
+            sa.String(length=500),
+            nullable=True,
+            comment="失败原因（写入方须截断）",
+        ),
+        sa.Column("duration_ms", sa.Integer(), nullable=True, comment="处理耗时（毫秒）"),
+        sa.Column(
+            "create_time",
+            sa.DateTime(),
+            server_default=sa.text("CURRENT_TIMESTAMP"),
+            nullable=False,
+            comment="创建时间",
+        ),
+        sa.ForeignKeyConstraint(
+            ["user_id"], ["users.id"], name=op.f("fk_operation_logs_user_id"), ondelete="SET NULL"
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_operation_logs")),
+        comment="操作日志表",
+        mysql_charset="utf8mb4",
+        mysql_collate="utf8mb4_general_ci",
+        mysql_engine="InnoDB",
+    )
+    op.create_index(
+        "idx_operation_logs_create_time", "operation_logs", ["create_time"], unique=False
+    )
+    op.create_index("idx_operation_logs_module", "operation_logs", ["module"], unique=False)
+    op.create_index("idx_operation_logs_user_id", "operation_logs", ["user_id"], unique=False)
+    op.create_index("idx_operation_logs_username", "operation_logs", ["username"], unique=False)
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """回滚迁移。
+
+    **注意（autogenerate 不会替你处理）**：生成的 `op.drop_index` 与
+    `op.drop_table` 顺序需人工检查 —— 若某索引是外键的唯一支撑索引，
+    单独 drop_index 会报 MySQL 1553（needed in a foreign key constraint）。
+    **整表回滚应只保留 `op.drop_table`**，DROP TABLE 会连带删除索引与外键。
+    """
+    # 整表回滚只保留 drop_table —— DROP TABLE 会连带删除索引与外键。
+    # 逐个 drop_index 会失败：idx_operation_logs_user_id 是外键
+    # fk_operation_logs_user_id 的唯一支撑索引，MySQL 拒绝删除（errno 1553）。
+    op.drop_table("operation_logs")

backend/app/api/deps.py

@@ -6,8 +6,9 @@ from fastapi import Depends
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from sqlalchemy.orm import Session
 
+from app.core.constants import ROLE_ADMIN
 from app.core.db import get_db
-from app.core.exceptions import AuthError
+from app.core.exceptions import AuthError, ForbiddenError
 from app.core.security import decode_token
 from app.models import User
 from app.repositories import user_repository
@@ -71,3 +72,20 @@ def _is_digits(value: object) -> bool:
     令牌内容不可信，直接 int() 会因非法输入抛 ValueError 而不是 401。
     """
     return isinstance(value, str) and value.isdigit()
+
+
+def require_admin(current_user: Annotated[User, Depends(get_current_user)]) -> User:
+    """要求当前用户具备管理员角色。
+
+    需求 2.5.1 只定义了「系统管理员（全部权限）」与「普通用户（业务操作）」两个
+    角色，故权限判定用**角色粗粒度**即可，不必引入权限码表 —— `permissions`
+    与 `role_permissions` 表已建好，等业务方给出权限清单后再启用。
+
+    :raises ForbiddenError: 已登录但不具备管理员角色（403）
+    """
+    if ROLE_ADMIN not in user_repository.get_role_codes(current_user):
+        raise ForbiddenError("需要管理员权限")
+    return current_user
+
+
+AdminUser = Annotated[User, Depends(require_admin)]

backend/app/api/router.py

@@ -8,6 +8,7 @@ from fastapi import APIRouter
 
 from app.api import health
 from app.api.v1 import auth
+from app.api.v1.system import users as system_users
 
 api_router = APIRouter()
 
@@ -17,4 +18,10 @@ api_router.include_router(health.router)
 # 业务接口：/api/v1/*
 v1_router = APIRouter(prefix="/v1")
 v1_router.include_router(auth.router)
+
+# 系统管理（需求 2.5.1）：/api/v1/system/*
+system_router = APIRouter(prefix="/system")
+system_router.include_router(system_users.router)
+v1_router.include_router(system_router)
+
 api_router.include_router(v1_router)

backend/app/api/v1/system/__init__.py

@@ -0,0 +1,7 @@
+"""系统管理路由（需求 2.5.1）。
+
+含用户管理、角色查询、操作日志查询三部分，全部要求管理员权限。
+
+与 `api/v1/auth.py` 的分工：`auth` 是登录链路（未认证即可访问的验证码与登录，
+以及任何登录用户都能访问的 `/me` 与登出）；本包是**管理功能**。
+"""

backend/app/api/v1/system/users.py

@@ -0,0 +1,131 @@
+"""用户管理接口（需求 2.5.1：用户账号管理）。
+
+全部要求管理员权限 —— 按需求「系统管理员（全部权限）、普通用户（业务操作）」，
+用户管理属管理员专属。
+
+删除、禁用、改角色三处都有**自我保护**：不允许对当前登录账号执行，
+否则管理员误操作会立刻失去全部管理权限且无法自行恢复。
+"""
+
+from typing import Annotated
+
+from fastapi import APIRouter, Path, Query
+
+from app.api.deps import AdminUser, DbSession
+from app.core.response import ApiResponse, PageResult, ok
+from app.schemas.user import (
+    PasswordResetIn,
+    RoleOut,
+    UserCreateIn,
+    UserOut,
+    UserQuery,
+    UserStatusIn,
+    UserUpdateIn,
+)
+from app.services import user_service
+
+router = APIRouter(prefix="/users", tags=["系统管理 · 用户"])
+
+
+@router.get("", response_model=ApiResponse[PageResult[UserOut]], summary="用户列表")
+def list_users(
+    session: DbSession,
+    _: AdminUser,
+    query: Annotated[UserQuery, Query()],
+) -> ApiResponse[PageResult[UserOut]]:
+    """分页查询用户，支持按用户名/姓名模糊、状态与角色筛选。"""
+    return ok(user_service.list_users(session, query))
+
+
+@router.get("/{user_id}", response_model=ApiResponse[UserOut], summary="用户详情")
+def get_user(
+    session: DbSession,
+    _: AdminUser,
+    user_id: Annotated[int, Path(ge=1)],
+) -> ApiResponse[UserOut]:
+    """查询用户详情。"""
+    return ok(user_service.get_user(session, user_id))
+
+
+@router.post("", response_model=ApiResponse[UserOut], summary="新增用户")
+def create_user(
+    session: DbSession,
+    _: AdminUser,
+    payload: UserCreateIn,
+) -> ApiResponse[UserOut]:
+    """新增用户。用户名须唯一，至少指定一个角色。"""
+    return ok(user_service.create_user(session, payload), message="用户已创建")
+
+
+@router.put("/{user_id}", response_model=ApiResponse[UserOut], summary="编辑用户")
+def update_user(
+    session: DbSession,
+    current_user: AdminUser,
+    user_id: Annotated[int, Path(ge=1)],
+    payload: UserUpdateIn,
+) -> ApiResponse[UserOut]:
+    """编辑用户的基本信息与角色。
+
+    **不含用户名与密码**：用户名是审计追溯的锚点；密码走专门的重置接口。
+    """
+    return ok(
+        user_service.update_user(session, user_id, payload, operator_id=current_user.id),
+        message="已保存",
+    )
+
+
+@router.delete("/{user_id}", response_model=ApiResponse[None], summary="删除用户")
+def delete_user(
+    session: DbSession,
+    current_user: AdminUser,
+    user_id: Annotated[int, Path(ge=1)],
+) -> ApiResponse[None]:
+    """删除用户。
+
+    物理删除。历史操作日志通过 `user_id` 置空 + 冗余用户名保留可追溯性。
+    业务上更推荐用「禁用」代替删除 —— 禁用保留完整审计线索。
+    """
+    user_service.delete_user(session, user_id, operator_id=current_user.id)
+    return ok(message="用户已删除")
+
+
+@router.put("/{user_id}/password", response_model=ApiResponse[None], summary="重置密码")
+def reset_password(
+    session: DbSession,
+    _: AdminUser,
+    user_id: Annotated[int, Path(ge=1)],
+    payload: PasswordResetIn,
+) -> ApiResponse[None]:
+    """重置用户密码。
+
+    会同时提升该用户的令牌版本，使其**已签发的令牌立即失效** ——
+    否则改密后旧令牌在有效期内仍可用。
+    """
+    user_service.reset_password(session, user_id, payload.new_password)
+    return ok(message="密码已重置，该用户需重新登录")
+
+
+@router.put("/{user_id}/status", response_model=ApiResponse[UserOut], summary="启用/禁用用户")
+def set_status(
+    session: DbSession,
+    current_user: AdminUser,
+    user_id: Annotated[int, Path(ge=1)],
+    payload: UserStatusIn,
+) -> ApiResponse[UserOut]:
+    """启用或禁用用户。
+
+    禁用会提升令牌版本使已签发令牌立即失效，否则被禁用者仍可凭旧令牌
+    继续访问直到过期（最长 2 小时）。
+    """
+    result = user_service.set_status(session, user_id, payload.status, operator_id=current_user.id)
+    return ok(result, message="已启用" if payload.status == 1 else "已禁用")
+
+
+@router.get("/meta/roles", response_model=ApiResponse[list[RoleOut]], summary="角色下拉选项")
+def list_roles(session: DbSession, _: AdminUser) -> ApiResponse[list[RoleOut]]:
+    """取全部启用角色，供用户表单的角色选择使用。
+
+    路径放在 `/users/meta/roles` 而非 `/roles`，是为了避免与
+    `/{user_id}` 的路由匹配冲突（`roles` 会被当成 user_id 解析而报 422）。
+    """
+    return ok(user_service.list_roles(session))

backend/app/models/__init__.py

@@ -11,12 +11,14 @@
 
 from app.models.base import Base, TimestampMixin
 from app.models.login_log import LoginLog
+from app.models.operation_log import OperationLog
 from app.models.role import Permission, Role, RolePermission
 from app.models.user import User, UserRole
 
 __all__ = [
     "Base",
     "LoginLog",
+    "OperationLog",
     "Permission",
     "Role",
     "RolePermission",

backend/app/models/operation_log.py

@@ -0,0 +1,98 @@
+"""操作日志模型。
+
+需求 2.5.1 要求「记录用户登录、操作行为及计提参数/系数表变更，支持日志查询」。
+登录行为已在 `login_logs` 中单独记录（登录失败也要留痕，且不依赖认证上下文）；
+本表记录**已认证用户的操作行为**，两者分工明确、互不重叠。
+
+不继承 TimestampMixin：日志是追加写、不可变的，没有"更新时间"语义。
+"""
+
+from datetime import datetime
+
+from sqlalchemy import DateTime, ForeignKey, Index, Integer, String, Text, text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.models.base import BIGINT_UNSIGNED, TINYINT_TYPE, Base, mysql_table_args
+
+# 各文本列的上限。**写入方必须按此截断**，否则 MySQL 严格模式抛 1406
+# （Data too long），而按"审计失败不得影响业务"的约定该异常会被吞掉 ——
+# 结果是整条日志静默丢失。截断值集中定义于此，供 repository 层引用。
+USERNAME_MAX_LEN = 64
+MODULE_MAX_LEN = 64
+ACTION_MAX_LEN = 32
+SUMMARY_MAX_LEN = 255
+PATH_MAX_LEN = 255
+IP_MAX_LEN = 45
+USER_AGENT_MAX_LEN = 255
+ERROR_MESSAGE_MAX_LEN = 500
+# request_data 是 MySQL TEXT（上限 65535 字节）。脱敏后的 JSON 可能超限
+# （如批量操作提交数千个 ID），超限同样会抛 1406，故写入方需按字节截断。
+REQUEST_DATA_MAX_BYTES = 60000
+
+
+class OperationLog(Base):
+    """操作日志表。
+
+    记录原则与 `login_logs` 一致：
+    - **请求参数必须脱敏后入库**（复用 `core.logging.scrub`），日志是最容易泄漏
+      密码与令牌的地方
+    - 用户名冗余存储而非只存 user_id —— 账号被删除后仍能追溯操作者，
+      故 `username` 单独建索引（与 `login_logs` 保持一致）
+    - 写入失败不得影响业务（审计可丢，业务不可断）
+    """
+
+    __tablename__ = "operation_logs"
+    __table_args__ = mysql_table_args(
+        "操作日志表",
+        Index("idx_operation_logs_user_id", "user_id"),
+        # 账号删除后 user_id 被置空，只能按用户名追溯，故该列必须有索引，
+        # 否则日志表增长后查询退化为全表扫描
+        Index("idx_operation_logs_username", "username"),
+        Index("idx_operation_logs_module", "module"),
+        Index("idx_operation_logs_create_time", "create_time"),
+    )
+
+    id: Mapped[int] = mapped_column(
+        BIGINT_UNSIGNED, primary_key=True, autoincrement=True, comment="主键ID"
+    )
+    user_id: Mapped[int | None] = mapped_column(
+        BIGINT_UNSIGNED,
+        ForeignKey("users.id", ondelete="SET NULL"),
+        comment="操作人ID（账号删除后置空）",
+    )
+    username: Mapped[str] = mapped_column(
+        String(USERNAME_MAX_LEN), nullable=False, comment="操作人用户名（冗余存储便于追溯）"
+    )
+    module: Mapped[str] = mapped_column(
+        String(MODULE_MAX_LEN), nullable=False, comment="所属模块，如 用户管理"
+    )
+    action: Mapped[str] = mapped_column(
+        String(ACTION_MAX_LEN), nullable=False, comment="操作类型：新增/修改/删除/导出"
+    )
+    summary: Mapped[str] = mapped_column(
+        String(SUMMARY_MAX_LEN), nullable=False, comment="操作摘要（中文，供人阅读）"
+    )
+
+    method: Mapped[str | None] = mapped_column(String(8), comment="HTTP 方法")
+    path: Mapped[str | None] = mapped_column(String(PATH_MAX_LEN), comment="请求路径")
+    ip: Mapped[str | None] = mapped_column(String(IP_MAX_LEN), comment="客户端IP（45位兼容IPv6）")
+    user_agent: Mapped[str | None] = mapped_column(
+        String(USER_AGENT_MAX_LEN), comment="客户端User-Agent（截断存储）"
+    )
+    # 请求参数，**入库前已脱敏**；长文本故用 TEXT
+    request_data: Mapped[str | None] = mapped_column(Text, comment="请求参数（已脱敏）")
+
+    # 默认 0 而非 1：与 login_logs 保持一致，且方向更保守 ——
+    # 写入方漏传时记为「失败」，管理员会去核查；反之漏传记为「成功」则是
+    # 审计漏报，且无声无息。写入方应显式传值（见 repository 层）。
+    success: Mapped[int] = mapped_column(
+        TINYINT_TYPE, nullable=False, server_default="0", comment="是否成功：1成功 0失败"
+    )
+    error_message: Mapped[str | None] = mapped_column(
+        String(ERROR_MESSAGE_MAX_LEN), comment="失败原因（写入方须截断）"
+    )
+    duration_ms: Mapped[int | None] = mapped_column(Integer, comment="处理耗时（毫秒）")
+
+    create_time: Mapped[datetime] = mapped_column(
+        DateTime, nullable=False, server_default=text("CURRENT_TIMESTAMP"), comment="创建时间"
+    )

backend/app/repositories/user_repository.py

@@ -5,10 +5,18 @@
 
 from datetime import datetime
 
-from sqlalchemy import select
+from sqlalchemy import Select, func, select
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import Session, selectinload
 
-from app.models import User
+from app.models import Role, User, UserRole
+
+
+def _base_query() -> Select[tuple[User]]:
+    """用户查询基座。
+
+    预加载角色关联，避免列表渲染时逐行查角色（N+1）。
+    """
+    return select(User).options(selectinload(User.user_roles).selectinload(UserRole.role))
 
 
 def get_by_username(session: Session, username: str) -> User | None:
@@ -34,3 +42,120 @@ def touch_last_login(session: Session, user: User, moment: datetime) -> None:
     """更新最后登录时间。只 flush，提交由 service 决定。"""
     user.last_login_time = moment
     session.flush()
+
+
+# ==================== 列表查询 ====================
+
+
+def list_users(
+    session: Session,
+    *,
+    offset: int,
+    limit: int,
+    username: str | None = None,
+    real_name: str | None = None,
+    status: int | None = None,
+    role_code: str | None = None,
+) -> list[User]:
+    """分页查询用户列表。
+
+    **按 id 倒序而非 create_time**：create_time 是秒级精度，批量导入的用户
+    可能落在同一秒，按它排序的分页结果不确定 —— 翻页时会出现重复或遗漏。
+    id 是自增主键，天然唯一且与创建顺序一致。
+    """
+    stmt = _base_query()
+    stmt = _apply_filters(stmt, username=username, real_name=real_name, status=status)
+    if role_code:
+        # 按角色筛选需 join 关联表；角色是字典数据，数量极少，join 代价可忽略
+        stmt = stmt.join(UserRole).join(Role).where(Role.role_code == role_code)
+    stmt = stmt.order_by(User.id.desc()).offset(offset).limit(limit)
+    return list(session.execute(stmt).scalars().unique())
+
+
+def count_users(
+    session: Session,
+    *,
+    username: str | None = None,
+    real_name: str | None = None,
+    status: int | None = None,
+    role_code: str | None = None,
+) -> int:
+    """统计符合条件的用户总数，供分页使用。"""
+    stmt = select(func.count(User.id))
+    stmt = _apply_filters(stmt, username=username, real_name=real_name, status=status)
+    if role_code:
+        stmt = stmt.join(UserRole).join(Role).where(Role.role_code == role_code)
+    return int(session.execute(stmt).scalar_one())
+
+
+def _apply_filters(
+    stmt: Select,
+    *,
+    username: str | None,
+    real_name: str | None,
+    status: int | None,
+) -> Select:
+    """应用通用筛选条件。列表与计数共用，避免两处口径不一致。"""
+    if username:
+        stmt = stmt.where(User.username.like(f"%{username}%"))
+    if real_name:
+        stmt = stmt.where(User.real_name.like(f"%{real_name}%"))
+    if status is not None:
+        stmt = stmt.where(User.status == status)
+    return stmt
+
+
+# ==================== 写入 ====================
+
+
+def create(session: Session, user: User) -> User:
+    """新增用户。只 flush 以取得自增 id。"""
+    session.add(user)
+    session.flush()
+    return user
+
+
+def delete(session: Session, user: User) -> None:
+    """物理删除用户。
+
+    本项目对用户采用「删除即禁用」的**业务**约定（见模型注释），
+    但保留物理删除能力供测试清理与误建账号的彻底移除使用。
+    `user_roles` 与 `operation_logs` 的外键分别配置了 CASCADE 与 SET NULL，
+    故不会留下悬挂引用。
+    """
+    session.delete(user)
+    session.flush()
+
+
+# ==================== 角色 ====================
+
+
+def get_roles_by_codes(session: Session, role_codes: list[str]) -> list[Role]:
+    """按编码批量取角色。用于校验前端传来的角色是否存在。"""
+    if not role_codes:
+        return []
+    stmt = select(Role).where(Role.role_code.in_(role_codes))
+    return list(session.execute(stmt).scalars())
+
+
+def replace_roles(session: Session, user: User, roles: list[Role]) -> None:
+    """整体替换用户的角色。
+
+    采用「先清后建」而非逐条比对：角色数量极少（本期仅两个），
+    差异计算的复杂度换不来收益，且整体替换的语义更直白。
+
+    **必须经 `user.user_roles` 集合追加，不能用 `session.add(UserRole(...))`**：
+    后者只把对象放进 session，不会同步到关系的已加载集合，于是同一事务内
+    再读 `user.user_roles` 仍是空的 —— 表现为「保存成功，但接口返回的角色
+    列表是 []」。经集合操作时 `user_id` 由关系自动填充，集合也立即反映新状态。
+    """
+    user.user_roles.clear()
+    for role in roles:
+        user.user_roles.append(UserRole(role_id=role.id))
+    session.flush()
+
+
+def get_all_roles(session: Session) -> list[Role]:
+    """取全部启用角色，供前端下拉选择。"""
+    stmt = select(Role).where(Role.status == 1).order_by(Role.sort_order, Role.id)
+    return list(session.execute(stmt).scalars())

backend/app/schemas/user.py

@@ -0,0 +1,126 @@
+"""用户与角色的请求/响应模型。
+
+与 `auth.py` 的分工：`auth.py` 管登录链路（验证码、登录、当前用户），
+本模块管**用户管理**（增删改查、重置密码、启停、角色分配）。
+
+字段名驼峰（规范约定）与蛇形属性之间用 alias 映射，与响应模型保持同一套做法。
+"""
+
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+from app.core.constants import PASSWORD_MAX_BYTES, PASSWORD_MIN_BYTES
+from app.core.response import PageQuery
+
+
+def _validate_password_bytes(value: str) -> str:
+    """按**字节数**校验密码长度。
+
+    必须在这里拦下：bcrypt 5.0 起对超过 72 字节的密码直接抛 ValueError，
+    而中文一个字符占 3 字节 —— 按字符数校验（如 max_length=20）根本拦不住。
+    与 `auth.py` 的登录校验共用同一规则，避免两处口径不一致。
+    """
+    length = len(value.encode("utf-8"))
+    if length < PASSWORD_MIN_BYTES:
+        raise ValueError(f"密码不足 {PASSWORD_MIN_BYTES} 字节")
+    if length > PASSWORD_MAX_BYTES:
+        raise ValueError(f"密码超过 {PASSWORD_MAX_BYTES} 字节上限")
+    return value
+
+
+class UserOut(BaseModel):
+    """用户列表/详情响应。
+
+    不含 password —— 密码哈希绝不外泄，这是硬约束而非疏漏。
+    """
+
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    username: str
+    real_name: str | None = None
+    phone: str | None = None
+    email: str | None = None
+    status: int = Field(description="状态：1启用 0禁用")
+    roles: list[str] = Field(default_factory=list, description="角色编码列表")
+    last_login_time: datetime | None = None
+    remark: str | None = None
+    create_time: datetime | None = None
+
+
+class UserCreateIn(BaseModel):
+    """新增用户。"""
+
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    username: str = Field(min_length=2, max_length=64, description="登录用户名，须唯一")
+    password: str = Field(description="初始密码")
+    real_name: str | None = Field(default=None, max_length=64)
+    phone: str | None = Field(default=None, max_length=20)
+    email: str | None = Field(default=None, max_length=128)
+    roles: list[str] = Field(default_factory=list, description="角色编码列表，至少一个")
+    remark: str | None = Field(default=None, max_length=255)
+
+    @field_validator("password")
+    @classmethod
+    def _check_password(cls, value: str) -> str:
+        return _validate_password_bytes(value)
+
+
+class UserUpdateIn(BaseModel):
+    """编辑用户。
+
+    **不含 username** —— 用户名是登录凭据与审计追溯的锚点，改名会让历史
+    操作日志与登录日志对不上。确需改名应走专门的流程（本期不提供）。
+    """
+
+    model_config = ConfigDict(str_strip_whitespace=True)
+
+    real_name: str | None = Field(default=None, max_length=64)
+    phone: str | None = Field(default=None, max_length=20)
+    email: str | None = Field(default=None, max_length=128)
+    roles: list[str] | None = Field(default=None, description="传 null 表示不改角色")
+    remark: str | None = Field(default=None, max_length=255)
+
+
+class PasswordResetIn(BaseModel):
+    """重置密码（管理员操作，不需要原密码）。"""
+
+    new_password: str = Field(description="新密码")
+
+    @field_validator("new_password")
+    @classmethod
+    def _check_password(cls, value: str) -> str:
+        return _validate_password_bytes(value)
+
+
+class UserStatusIn(BaseModel):
+    """启用/禁用用户。"""
+
+    status: int = Field(ge=0, le=1, description="1启用 0禁用")
+
+
+class UserQuery(PageQuery):
+    """用户列表查询条件。
+
+    继承分页基类，额外支持按用户名/姓名模糊、按状态与角色筛选。
+    """
+
+    username: str | None = Field(default=None, max_length=64, description="用户名，模糊匹配")
+    real_name: str | None = Field(default=None, max_length=64, description="姓名，模糊匹配")
+    status: int | None = Field(default=None, ge=0, le=1, description="状态筛选")
+    role_code: str | None = Field(default=None, max_length=32, description="角色编码筛选")
+
+
+class RoleOut(BaseModel):
+    """角色响应。"""
+
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    role_code: str
+    role_name: str
+    description: str | None = None
+    sort_order: int = 0
+    status: int = 1

backend/app/services/user_service.py

@@ -0,0 +1,232 @@
+"""用户管理业务编排。
+
+**事务边界在本层**：每个公开方法自行决定何时 commit。
+repository 只 flush，事务由这里控制。
+
+设计取舍：
+- 删除采用**物理删除**，与「删除即禁用」并不矛盾 —— 后者是对业务方的建议
+  （停用账号应走禁用而非删除，以保留审计线索），但需求 2.5.1 明确列了
+  「增删改查」，故保留删除能力；`operation_logs` 通过 SET NULL + 冗余 username
+  保证删除后仍可追溯
+- 不允许删除或禁用**自己** —— 管理员误操作会立刻失去全部管理权限，
+  且没有其他账号可以救回来
+"""
+
+import logging
+
+from sqlalchemy.orm import Session
+
+from app.core.constants import ROLE_ADMIN
+from app.core.exceptions import BadRequestError, ForbiddenError, NotFoundError
+from app.core.response import PageResult
+from app.core.security import hash_password
+from app.models import User
+from app.repositories import user_repository
+from app.schemas.user import RoleOut, UserCreateIn, UserOut, UserQuery, UserUpdateIn
+
+logger = logging.getLogger(__name__)
+
+
+def _to_out(user: User) -> UserOut:
+    """ORM 对象转响应模型。
+
+    禁止把 ORM 对象直接当 response_model —— 表结构变更会直接泄漏到接口契约上。
+    """
+    return UserOut(
+        id=user.id,
+        username=user.username,
+        real_name=user.real_name,
+        phone=user.phone,
+        email=user.email,
+        status=user.status,
+        roles=user_repository.get_role_codes(user),
+        last_login_time=user.last_login_time,
+        remark=user.remark,
+        create_time=user.create_time,
+    )
+
+
+def list_users(session: Session, query: UserQuery) -> PageResult[UserOut]:
+    """分页查询用户列表。"""
+    users = user_repository.list_users(
+        session,
+        offset=query.offset,
+        limit=query.page_size,
+        username=query.username,
+        real_name=query.real_name,
+        status=query.status,
+        role_code=query.role_code,
+    )
+    total = user_repository.count_users(
+        session,
+        username=query.username,
+        real_name=query.real_name,
+        status=query.status,
+        role_code=query.role_code,
+    )
+    return PageResult[UserOut](
+        records=[_to_out(user) for user in users],
+        total=total,
+        page_num=query.page_num,
+        page_size=query.page_size,
+    )
+
+
+def get_user(session: Session, user_id: int) -> UserOut:
+    """查询用户详情。
+
+    :raises NotFoundError: 用户不存在
+    """
+    user = user_repository.get_by_id(session, user_id)
+    if user is None:
+        raise NotFoundError("用户不存在")
+    return _to_out(user)
+
+
+def create_user(session: Session, payload: UserCreateIn) -> UserOut:
+    """新增用户。
+
+    :raises BadRequestError: 用户名已存在，或角色编码无效
+    """
+    if user_repository.get_by_username(session, payload.username) is not None:
+        raise BadRequestError(f"用户名 {payload.username!r} 已存在")
+
+    roles = _resolve_roles(session, payload.roles)
+
+    user = User(
+        username=payload.username,
+        password=hash_password(payload.password),
+        real_name=payload.real_name,
+        phone=payload.phone,
+        email=payload.email,
+        status=1,
+        remark=payload.remark,
+    )
+    user_repository.create(session, user)
+    user_repository.replace_roles(session, user, roles)
+    session.commit()
+
+    logger.info("新增用户 | username=%s | roles=%s", user.username, payload.roles)
+    return _to_out(user)
+
+
+def update_user(
+    session: Session, user_id: int, payload: UserUpdateIn, *, operator_id: int
+) -> UserOut:
+    """编辑用户的基本信息与角色。
+
+    **不含用户名与密码**：用户名是审计追溯的锚点；密码走专门的重置接口，
+    以免"改资料"顺手改掉密码这类越权操作。
+
+    :raises NotFoundError: 用户不存在
+    :raises BadRequestError: 角色编码无效
+    """
+    user = user_repository.get_by_id(session, user_id)
+    if user is None:
+        raise NotFoundError("用户不存在")
+
+    if payload.real_name is not None:
+        user.real_name = payload.real_name
+    if payload.phone is not None:
+        user.phone = payload.phone
+    if payload.email is not None:
+        user.email = payload.email
+    if payload.remark is not None:
+        user.remark = payload.remark
+
+    # 不允许把自己降级掉管理角色，否则改完立刻失去管理权限
+    if payload.roles is not None and user_id == operator_id and ROLE_ADMIN not in payload.roles:
+        raise ForbiddenError("不能移除自己的管理员角色")
+
+    session.flush()
+
+    if payload.roles is not None:
+        roles = _resolve_roles(session, payload.roles)
+        user_repository.replace_roles(session, user, roles)
+
+    session.commit()
+    logger.info("编辑用户 | user_id=%s | roles=%s", user_id, payload.roles)
+    return _to_out(user)
+
+
+def reset_password(session: Session, user_id: int, new_password: str) -> None:
+    """重置用户密码。
+
+    同时把 `token_version` 加一，使该用户**已签发的令牌立即失效** ——
+    改密后旧令牌仍可用是常见疏漏，这里与「离职即失效」复用同一机制。
+
+    :raises NotFoundError: 用户不存在
+    """
+    user = user_repository.get_by_id(session, user_id)
+    if user is None:
+        raise NotFoundError("用户不存在")
+
+    user.password = hash_password(new_password)
+    user.token_version += 1
+    session.commit()
+    logger.info("重置密码 | user_id=%s（令牌版本已提升）", user_id)
+
+
+def set_status(session: Session, user_id: int, status: int, *, operator_id: int) -> UserOut:
+    """启用或禁用用户。
+
+    禁用时同样提升 `token_version`，使已签发令牌立即失效 ——
+    否则被禁用者仍可凭旧令牌访问直到过期（最长 2 小时）。
+
+    :raises NotFoundError: 用户不存在
+    :raises ForbiddenError: 试图禁用自己
+    """
+    user = user_repository.get_by_id(session, user_id)
+    if user is None:
+        raise NotFoundError("用户不存在")
+
+    if status == 0 and user_id == operator_id:
+        raise ForbiddenError("不能禁用自己")
+
+    user.status = status
+    if status == 0:
+        user.token_version += 1
+    session.commit()
+    logger.info("变更用户状态 | user_id=%s | status=%s", user_id, status)
+    return _to_out(user)
+
+
+def delete_user(session: Session, user_id: int, *, operator_id: int) -> None:
+    """删除用户。
+
+    物理删除。历史操作日志通过 `user_id` 置空 + 冗余 `username` 保留可追溯性。
+
+    :raises NotFoundError: 用户不存在
+    :raises ForbiddenError: 试图删除自己
+    """
+    user = user_repository.get_by_id(session, user_id)
+    if user is None:
+        raise NotFoundError("用户不存在")
+
+    if user_id == operator_id:
+        raise ForbiddenError("不能删除自己")
+
+    username = user.username
+    user_repository.delete(session, user)
+    session.commit()
+    logger.info("删除用户 | username=%s", username)
+
+
+def list_roles(session: Session) -> list[RoleOut]:
+    """取全部启用角色，供前端下拉选择。"""
+    return [RoleOut.model_validate(role) for role in user_repository.get_all_roles(session)]
+
+
+def _resolve_roles(session: Session, role_codes: list[str]) -> list:
+    """把角色编码解析为角色对象，并校验全部存在。
+
+    :raises BadRequestError: 存在无效编码
+    """
+    if not role_codes:
+        raise BadRequestError("至少需要指定一个角色")
+    roles = user_repository.get_roles_by_codes(session, role_codes)
+    if len(roles) != len(set(role_codes)):
+        valid = {role.role_code for role in roles}
+        invalid = sorted(set(role_codes) - valid)
+        raise BadRequestError(f"角色编码无效：{invalid}")
+    return list(roles)

backend/tests/integration/test_schema_sync.py

@@ -1,13 +1,16 @@
 """DDL 漂移检测。
 
 结构定义同时存在于两处：
-- `db/migration/V1__*.sql` —— 事实来源，生产迁移用
+- `db/migration/V*.sql` —— 事实来源，生产迁移用
 - `db/init/init-dev.sql` —— 开发环境重置脚本，Docker 首次启动唯一会自动执行的位置
 
 两份 DDL 手工同步迟早会漂移，且后果隐蔽：开发库的表结构与迁移脚本不一致，
 新人克隆下来建出的库和生产不一样，等到上线才发现。
 本测试把「靠自觉」变成「可自动拦截」，成本极低。
 
+**必须比对全部 V*.sql 的合集**：早期版本只解析 V1，新增 V2 之后，
+两份 DDL 同时缺表仍会被判为「一致」，漂移守卫形同虚设。
+
 注：本文件不需要容器，放在 integration 下是因为它校验的是交付产物而非应用代码。
 """
 
@@ -17,8 +20,9 @@ from pathlib import Path
 import pytest
 
 PROJECT_ROOT = Path(__file__).resolve().parents[3]
-MIGRATION_FILE = PROJECT_ROOT / "db" / "migration" / "V1__init_rbac_and_login_logs.sql"
+MIGRATION_DIR = PROJECT_ROOT / "db" / "migration"
 INIT_DEV_FILE = PROJECT_ROOT / "db" / "init" / "init-dev.sql"
+ALEMBIC_VERSIONS_DIR = PROJECT_ROOT / "backend" / "alembic" / "versions"
 
 _CREATE_TABLE = re.compile(r"CREATE TABLE\s+`?(\w+)`?\s*\((.*?)\n\)", re.DOTALL)
 _COLUMN_LINE = re.compile(r"^\s{4}`?(\w+)`?\s", re.MULTILINE)
@@ -38,9 +42,38 @@ def _parse_tables(sql: str) -> dict[str, set[str]]:
     return tables
 
 
+def _alembic_head() -> str:
+    """从迁移脚本推断当前 head 版本号。
+
+    纯文本解析而非调用 alembic 运行时 —— 后者需要连数据库，会让本测试
+    从「秒级、无需容器」变成「必须起库」。
+    head 的定义：没有被任何其他迁移的 down_revision 引用过的那个 revision。
+    """
+    revisions: set[str] = set()
+    referenced: set[str] = set()
+    for path in ALEMBIC_VERSIONS_DIR.glob("*.py"):
+        text = path.read_text(encoding="utf-8")
+        if found := re.search(r"^revision: str = ['\"](\w+)['\"]", text, re.MULTILINE):
+            revisions.add(found.group(1))
+        if found := re.search(r"^down_revision:.*?= ['\"](\w+)['\"]", text, re.MULTILINE):
+            referenced.add(found.group(1))
+
+    heads = revisions - referenced
+    assert len(heads) == 1, f"迁移链存在分叉或缺失，heads={sorted(heads)}"
+    return heads.pop()
+
+
 @pytest.fixture(scope="module")
 def migration_tables() -> dict[str, set[str]]:
-    return _parse_tables(MIGRATION_FILE.read_text(encoding="utf-8"))
+    """合并解析 db/migration/ 下全部 V*.sql。"""
+    files = sorted(MIGRATION_DIR.glob("V*.sql"))
+    assert files, f"未找到迁移脚本：{MIGRATION_DIR}"
+
+    merged: dict[str, set[str]] = {}
+    for path in files:
+        for table, columns in _parse_tables(path.read_text(encoding="utf-8")).items():
+            merged.setdefault(table, set()).update(columns)
+    return merged
 
 
 @pytest.fixture(scope="module")
@@ -52,7 +85,7 @@ class TestSchemaSync:
     """两份 DDL 的结构一致性。"""
 
     def test_文件存在(self) -> None:
-        assert MIGRATION_FILE.is_file(), f"缺少迁移脚本：{MIGRATION_FILE}"
+        assert list(MIGRATION_DIR.glob("V*.sql")), f"缺少迁移脚本：{MIGRATION_DIR}"
         assert INIT_DEV_FILE.is_file(), f"缺少开发初始化脚本：{INIT_DEV_FILE}"
 
     def test_表集合一致(self, migration_tables: dict, init_dev_tables: dict) -> None:
@@ -74,16 +107,28 @@ class TestSchemaSync:
 
 
 class TestSeedData:
-    """字典数据在两处应保持同步。"""
+    """字典数据与版本标记。"""
 
     @pytest.mark.parametrize("role_code", ["admin", "user"])
     def test_角色初始化数据在两处都存在(self, role_code: str) -> None:
-        migration_sql = MIGRATION_FILE.read_text(encoding="utf-8")
+        migration_sql = "\n".join(
+            path.read_text(encoding="utf-8") for path in sorted(MIGRATION_DIR.glob("V*.sql"))
+        )
         init_sql = INIT_DEV_FILE.read_text(encoding="utf-8")
         assert f"'{role_code}'" in migration_sql
         assert f"'{role_code}'" in init_sql
 
-    def test_开发脚本标记了迁移版本(self) -> None:
+    def test_开发脚本标记的版本号等于当前head(self) -> None:
-        """不写 alembic_version 的话，在开发库上执行 upgrade head 会重复建表而失败。"""
+        """版本号必须等于 head，且不能只断言「存在」。
+
+        落后的后果：开发库重建后执行 `alembic upgrade head`，alembic 认为该迁移
+        尚未应用，重复建表并报 1050 Table already exists，只能手工补版本或 down -v。
+        """
         init_sql = INIT_DEV_FILE.read_text(encoding="utf-8")
-        assert "INSERT INTO alembic_version" in init_sql
+        found = re.search(
+            r"INSERT INTO alembic_version \(version_num\) VALUES \('(\w+)'\)", init_sql
+        )
+        assert found, "init-dev.sql 未写入 alembic_version"
+        assert found.group(1) == _alembic_head(), (
+            f"init-dev.sql 标记的版本号 {found.group(1)!r} 落后于当前 head {_alembic_head()!r}"
+        )

backend/tests/unit/test_models.py

@@ -11,7 +11,15 @@ from sqlalchemy.dialects import mysql
 
 from app.models import Base
 
-EXPECTED_TABLES = {"users", "roles", "user_roles", "permissions", "role_permissions", "login_logs"}
+EXPECTED_TABLES = {
+    "users",
+    "roles",
+    "user_roles",
+    "permissions",
+    "role_permissions",
+    "login_logs",
+    "operation_logs",
+}
 
 
 def _ddl(table_name: str) -> str:
@@ -101,7 +109,7 @@ class TestTimestampColumns:
     def test_带时间戳的表含ON_UPDATE子句(self, table_name: str) -> None:
         assert "ON UPDATE CURRENT_TIMESTAMP" in _ddl(table_name)
 
-    @pytest.mark.parametrize("table_name", ["login_logs"])
+    @pytest.mark.parametrize("table_name", ["login_logs", "operation_logs"])
     def test_仅追加的日志表不含ON_UPDATE子句(self, table_name: str) -> None:
         """日志是不可变记录，没有"更新时间"语义。"""
         assert "ON UPDATE CURRENT_TIMESTAMP" not in _ddl(table_name)

db/init/init-dev.sql

@@ -5,16 +5,18 @@
 -- 数据卷已存在时修改本文件不会生效，需要：
 --   docker compose -f docker/docker-compose.dev.yml down -v && up -d
 --
--- 内容 = V1 迁移的完整结构（额外带 DROP，开发环境允许重置）
+-- 内容 = db/migration/ 下**全部迁移**的最终结构（额外带 DROP，开发环境允许重置）
--- 与 db/migration/V1__init_rbac_and_login_logs.sql 必须保持一致，
+-- 必须与 db/migration/V*.sql 的合集保持一致，
 -- 由 backend/tests/integration/test_schema_sync.py 自动校验，防止两份 DDL 漂移
 --
--- 结尾写入 alembic_version 记录，使 `alembic upgrade head` 在本库上成为空操作
+-- 结尾写入 alembic_version 记录（**须等于当前 head**，否则开发库重建后
+-- 执行 alembic upgrade head 会重复建表并报 1050 Table already exists）
 -- ============================================================================
 
 SET NAMES utf8mb4;
 
 -- 删除顺序遵循外键依赖：先删子表
+DROP TABLE IF EXISTS `operation_logs`;
 DROP TABLE IF EXISTS `role_permissions`;
 DROP TABLE IF EXISTS `user_roles`;
 DROP TABLE IF EXISTS `login_logs`;
@@ -134,6 +136,37 @@ CREATE TABLE user_roles (
 
 CREATE INDEX idx_user_roles_role_id ON user_roles (role_id);
 
+-- ---------------------------------------------------------------------------
+-- 操作日志表（V2 迁移）
+-- ---------------------------------------------------------------------------
+CREATE TABLE operation_logs (
+    id BIGINT UNSIGNED NOT NULL COMMENT '主键ID' AUTO_INCREMENT,
+    user_id BIGINT UNSIGNED COMMENT '操作人ID（账号删除后置空）',
+    username VARCHAR(64) NOT NULL COMMENT '操作人用户名（冗余存储便于追溯）',
+    module VARCHAR(64) NOT NULL COMMENT '所属模块，如 用户管理',
+    action VARCHAR(32) NOT NULL COMMENT '操作类型：新增/修改/删除/导出',
+    summary VARCHAR(255) NOT NULL COMMENT '操作摘要（中文，供人阅读）',
+    method VARCHAR(8) COMMENT 'HTTP 方法',
+    path VARCHAR(255) COMMENT '请求路径',
+    ip VARCHAR(45) COMMENT '客户端IP（45位兼容IPv6）',
+    user_agent VARCHAR(255) COMMENT '客户端User-Agent（截断存储）',
+    request_data TEXT COMMENT '请求参数（已脱敏）',
+    success TINYINT NOT NULL COMMENT '是否成功：1成功 0失败' DEFAULT '0',
+    error_message VARCHAR(500) COMMENT '失败原因（写入方须截断）',
+    duration_ms INTEGER COMMENT '处理耗时（毫秒）',
+    create_time DATETIME NOT NULL COMMENT '创建时间' DEFAULT CURRENT_TIMESTAMP,
+    CONSTRAINT pk_operation_logs PRIMARY KEY (id),
+    CONSTRAINT fk_operation_logs_user_id FOREIGN KEY(user_id) REFERENCES users (id) ON DELETE SET NULL
+)CHARSET=utf8mb4 ENGINE=InnoDB COMMENT='操作日志表' COLLATE utf8mb4_general_ci;
+
+CREATE INDEX idx_operation_logs_create_time ON operation_logs (create_time);
+
+CREATE INDEX idx_operation_logs_module ON operation_logs (module);
+
+CREATE INDEX idx_operation_logs_user_id ON operation_logs (user_id);
+
+CREATE INDEX idx_operation_logs_username ON operation_logs (username);
+
 -- ---------------------------------------------------------------------------
 -- 字典数据：角色（与 V1 迁移中的初始化数据一致）
 -- ---------------------------------------------------------------------------
@@ -142,11 +175,11 @@ INSERT INTO roles (role_code, role_name, description, sort_order) VALUES ('admin
 INSERT INTO roles (role_code, role_name, description, sort_order) VALUES ('user', '普通用户', '业务操作', 2);
 
 -- ---------------------------------------------------------------------------
--- 迁移版本记录：标记 V1 已应用，使 alembic upgrade head 成为空操作
+-- 迁移版本记录：须等于当前 head，使 alembic upgrade head 成为空操作
 -- ---------------------------------------------------------------------------
 CREATE TABLE alembic_version (
     version_num VARCHAR(32) NOT NULL,
     CONSTRAINT alembic_version_pkc PRIMARY KEY (version_num)
 )CHARSET=utf8mb4 ENGINE=InnoDB COLLATE utf8mb4_general_ci;
 
-INSERT INTO alembic_version (version_num) VALUES ('6165691326d4');
+INSERT INTO alembic_version (version_num) VALUES ('a45a8087831b');

db/migration/V2__add_operation_logs.sql

@@ -0,0 +1,32 @@
+-- Running upgrade 6165691326d4 -> a45a8087831b
+
+CREATE TABLE operation_logs (
+    id BIGINT UNSIGNED NOT NULL COMMENT '主键ID' AUTO_INCREMENT, 
+    user_id BIGINT UNSIGNED COMMENT '操作人ID（账号删除后置空）', 
+    username VARCHAR(64) NOT NULL COMMENT '操作人用户名（冗余存储便于追溯）', 
+    module VARCHAR(64) NOT NULL COMMENT '所属模块，如 用户管理', 
+    action VARCHAR(32) NOT NULL COMMENT '操作类型：新增/修改/删除/导出', 
+    summary VARCHAR(255) NOT NULL COMMENT '操作摘要（中文，供人阅读）', 
+    method VARCHAR(8) COMMENT 'HTTP 方法', 
+    path VARCHAR(255) COMMENT '请求路径', 
+    ip VARCHAR(45) COMMENT '客户端IP（45位兼容IPv6）', 
+    user_agent VARCHAR(255) COMMENT '客户端User-Agent（截断存储）', 
+    request_data TEXT COMMENT '请求参数（已脱敏）', 
+    success TINYINT NOT NULL COMMENT '是否成功：1成功 0失败' DEFAULT '0', 
+    error_message VARCHAR(500) COMMENT '失败原因（写入方须截断）', 
+    duration_ms INTEGER COMMENT '处理耗时（毫秒）', 
+    create_time DATETIME NOT NULL COMMENT '创建时间' DEFAULT CURRENT_TIMESTAMP, 
+    CONSTRAINT pk_operation_logs PRIMARY KEY (id), 
+    CONSTRAINT fk_operation_logs_user_id FOREIGN KEY(user_id) REFERENCES users (id) ON DELETE SET NULL
+)CHARSET=utf8mb4 ENGINE=InnoDB COMMENT='操作日志表' COLLATE utf8mb4_general_ci;
+
+CREATE INDEX idx_operation_logs_create_time ON operation_logs (create_time);
+
+CREATE INDEX idx_operation_logs_module ON operation_logs (module);
+
+CREATE INDEX idx_operation_logs_user_id ON operation_logs (user_id);
+
+CREATE INDEX idx_operation_logs_username ON operation_logs (username);
+
+UPDATE alembic_version SET version_num='a45a8087831b' WHERE alembic_version.version_num = '6165691326d4';
+