refactor(backend): 调整模块结构、调度器逻辑与健康检查路由

backend/app/modules/system/user/__init__.py

@@ -0,0 +1 @@
+

backend/app/modules/system/user/controller.py

@@ -0,0 +1,201 @@
+import urllib.parse
+from typing import Annotated
+
+from fastapi import APIRouter, Body, Depends, File, Path, Query, Security, UploadFile, status
+from fastapi.responses import JSONResponse, StreamingResponse
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.common.response import ResponseSchema, StreamResponse, SuccessResponse
+from app.core.base_schema import AuthSchema, BatchSetAvailable, PageResultSchema, PaginationQueryParam
+from app.core.dependencies import AuthPermission, db_getter, get_current_user
+from app.core.logger import logger
+from app.core.router_class import OperationLogRoute
+from app.utils.common_util import bytes2file_response
+
+from .schema import (
+    CurrentUserOutSchema,
+    CurrentUserUpdateSchema,
+    ResetPasswordSchema,
+    UserChangePasswordSchema,
+    UserCreateSchema,
+    UserForgetPasswordSchema,
+    UserOutSchema,
+    UserQueryParam,
+    UserRegisterSchema,
+    UserUpdateSchema,
+)
+from .service import UserService
+
+UserRouter = APIRouter(route_class=OperationLogRoute, prefix="/user", tags=["用户管理"])
+
+
+@UserRouter.get("/current/info", summary="查询当前用户信息", response_model=ResponseSchema[CurrentUserOutSchema])
+async def get_current_user_info_controller(
+    auth: Annotated[AuthSchema, Security(get_current_user)],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+    check_data_scope: Annotated[bool, Query(description="是否加载完整数据（含部门/岗位/角色/OAuth），True-加载全部(默认)，False-仅菜单/权限")] = True,
+) -> JSONResponse:
+    user_dict: CurrentUserOutSchema = await UserService(auth, db).current_info(check_data_scope=check_data_scope)
+    return SuccessResponse(data=user_dict, msg="获取当前用户信息成功")
+
+
+@UserRouter.put("/current/info/update", summary="更新当前用户基本信息", response_model=ResponseSchema[UserOutSchema])
+async def update_current_user_info_controller(
+    auth: Annotated[AuthSchema, Security(get_current_user)],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+    data: Annotated[CurrentUserUpdateSchema, Body(description="更新用户基本信息参数")],
+) -> JSONResponse:
+    result_dict: UserOutSchema = await UserService(auth, db).update_current_info(data=data)
+    return SuccessResponse(data=result_dict, msg="更新当前用户基本信息成功")
+
+
+@UserRouter.put("/password/change", summary="修改当前用户密码", response_model=ResponseSchema[UserOutSchema])
+async def change_current_user_password_controller(
+    auth: Annotated[AuthSchema, Security(get_current_user)],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+    data: Annotated[UserChangePasswordSchema, Body(description="修改用户密码参数")],
+) -> JSONResponse:
+    result_dict: UserOutSchema = await UserService(auth, db).change_password(data=data)
+    return SuccessResponse(data=result_dict, msg="修改密码成功, 请重新登录")
+
+
+@UserRouter.put("/password/reset/{id}", summary="重置用户密码", response_model=ResponseSchema[UserOutSchema])
+async def reset_password_controller(
+    auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:user:update"]))],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+    id: Annotated[int, Path(description="用户ID", ge=1)],
+    data: Annotated[ResetPasswordSchema, Body(description="重置用户密码参数")],
+) -> JSONResponse:
+    data.id = id
+    result_dict: UserOutSchema = await UserService(auth, db).reset_password(data=data)
+    return SuccessResponse(data=result_dict, msg="重置密码成功")
+
+
+@UserRouter.post("/password/forget", summary="忘记密码", response_model=ResponseSchema[UserOutSchema])
+async def forget_password_controller(
+    db: Annotated[AsyncSession, Depends(db_getter)],
+    data: Annotated[UserForgetPasswordSchema, Body(description="忘记密码参数")],
+) -> JSONResponse:
+    auth = AuthSchema()
+    user_forget_password_result: UserOutSchema = await UserService(auth, db).forget_password(data=data)
+    logger.info(f"{data.username} 重置密码成功")
+    return SuccessResponse(data=user_forget_password_result, msg="重置密码成功")
+
+
+@UserRouter.post("/register", summary="用户注册", response_model=ResponseSchema[UserOutSchema])
+async def register_controller(
+    db: Annotated[AsyncSession, Depends(db_getter)],
+    data: Annotated[UserRegisterSchema, Body(description="用户注册参数")],
+) -> JSONResponse:
+    auth = AuthSchema()
+    register_result: UserOutSchema = await UserService(auth, db).register(data=data)
+    logger.info(f"新用户注册成功: {data.username}")
+    return SuccessResponse(data=register_result, msg="注册成功")
+
+
+@UserRouter.get("/list", summary="查询用户", response_model=ResponseSchema[PageResultSchema[UserOutSchema]])
+async def get_user_list_controller(
+    auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:user:query"]))],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+    page: Annotated[PaginationQueryParam, Depends()],
+    search: Annotated[UserQueryParam, Query()],
+) -> JSONResponse:
+    result_dict: PageResultSchema[UserOutSchema] = await UserService(auth, db).page(
+        page_no=page.page_no,
+        page_size=page.page_size,
+        search=search,
+        order_by=page.order_by,
+    )
+    return SuccessResponse(data=result_dict, msg="查询用户成功")
+
+
+@UserRouter.get("/detail/{id}", summary="查询用户详情", response_model=ResponseSchema[UserOutSchema])
+async def get_user_detail_controller(
+    auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:user:detail"]))],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+    id: Annotated[int, Path(description="用户ID", ge=1)],
+) -> JSONResponse:
+    result_dict: UserOutSchema = await UserService(auth, db).detail(id=id)
+    return SuccessResponse(data=result_dict, msg="获取用户详情成功")
+
+
+@UserRouter.post("/create", status_code=status.HTTP_201_CREATED, summary="创建用户", response_model=ResponseSchema[UserOutSchema])
+async def create_user_controller(
+    auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:user:create"]))],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+    data: Annotated[UserCreateSchema, Body(description="创建用户参数")],
+) -> JSONResponse:
+    result_dict: UserOutSchema = await UserService(auth, db).create(data=data)
+    return SuccessResponse(data=result_dict, msg="创建用户成功")
+
+
+@UserRouter.put("/update/{id}", summary="修改用户", response_model=ResponseSchema[UserOutSchema])
+async def update_user_controller(
+    auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:user:update"]))],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+    id: Annotated[int, Path(description="用户ID", ge=1)],
+    data: Annotated[UserUpdateSchema, Body(description="修改用户参数")],
+) -> JSONResponse:
+    result_dict: UserOutSchema = await UserService(auth, db).update(id=id, data=data)
+    return SuccessResponse(data=result_dict, msg="修改用户成功")
+
+
+@UserRouter.delete("/delete", summary="删除用户", response_model=ResponseSchema[None])
+async def delete_user_controller(
+    auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:user:delete"]))],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+    ids: Annotated[list[int], Body(description="ID列表")],
+) -> JSONResponse:
+    await UserService(auth, db).delete(ids=ids)
+    return SuccessResponse(msg="删除用户成功")
+
+
+@UserRouter.patch("/status/batch", summary="批量修改用户状态", response_model=ResponseSchema[None])
+async def batch_set_available_user_controller(
+    auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:user:patch"]))],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+    data: Annotated[BatchSetAvailable, Body(description="状态设置")],
+) -> JSONResponse:
+    await UserService(auth, db).set_available(data=data)
+    return SuccessResponse(msg="批量修改用户状态成功")
+
+
+@UserRouter.get("/import/template", summary="获取用户导入模板", dependencies=[Security(AuthPermission(["module_system:user:download"]))])
+async def export_user_import_template_controller() -> StreamingResponse:
+    user_import_template_result = UserService.get_import_template()
+
+    return StreamResponse(
+        data=bytes2file_response(user_import_template_result),
+        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+        headers={
+            "Content-Disposition": f"attachment; filename={urllib.parse.quote('用户导入模板.xlsx')}",
+            "Access-Control-Expose-Headers": "Content-Disposition",
+        },
+    )
+
+
+@UserRouter.post("/export", summary="导出用户")
+async def export_user_list_controller(
+    auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:user:export"]))],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+    page: Annotated[PaginationQueryParam, Depends()],
+    search: Annotated[UserQueryParam, Body()],
+) -> StreamingResponse:
+    user_list: list[UserOutSchema] = await UserService(auth, db).get_list(search=search, order_by=page.order_by)
+    user_export_result: bytes = await UserService.export_list(user_list=[item.model_dump() for item in user_list])
+
+    return StreamResponse(
+        data=bytes2file_response(user_export_result),
+        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+        headers={"Content-Disposition": "attachment; filename=user.xlsx"},
+    )
+
+
+@UserRouter.post("/import/data", summary="导入用户", response_model=ResponseSchema[None])
+async def import_user_list_controller(
+    file: Annotated[UploadFile, File(description="用户导入文件")],
+    auth: Annotated[AuthSchema, Security(AuthPermission(["module_system:user:import"]))],
+    db: Annotated[AsyncSession, Depends(db_getter)],
+) -> JSONResponse:
+    batch_import_result: str = await UserService(auth, db).batch_import(file=file, update_support=True)
+    return SuccessResponse(data=batch_import_result, msg="导入用户成功")

backend/app/modules/system/user/crud.py

@@ -0,0 +1,72 @@
+from collections.abc import Sequence
+from datetime import datetime
+from typing import Any
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.base_crud import CRUDBase
+from app.core.base_schema import AuthSchema
+
+from .model import UserModel
+from .schema import UserCreateSchema, UserUpdateSchema
+
+
+class UserCRUD(CRUDBase[UserModel, UserCreateSchema, UserUpdateSchema]):
+    """用户模块数据层"""
+
+    def __init__(self, auth: AuthSchema, db: AsyncSession) -> None:
+        super().__init__(model=UserModel, auth=auth, db=db)
+
+    async def create_obj_crud(self, data: UserCreateSchema) -> UserModel | None:
+        """创建用户
+
+        参数:
+        - data (UserCreateSchema): 创建模型。
+
+        返回:
+        - UserModel | None: 新建实体。
+        """
+        return await self.create(data=data)
+
+    async def update_last_login(self, id: int) -> UserModel | None:
+        """更新用户最后登录时间并返回最新用户；用户不存在时返回 None。
+
+        直接更新实例属性并 flush，保证返回对象内存值与数据库一致
+        （set 走 UPDATE 不同步已加载实例，故此处不使用 set）。
+        """
+        user = await self.db.get(UserModel, id)
+        if user is None:
+            return None
+        user.last_login = datetime.now()
+        await self.db.flush()
+        return user
+
+    async def set_user_roles(self, user_objs: Sequence[UserModel], role_objs: Sequence[Any]) -> None:
+        """替换用户的角色关联（纯数据操作；目标对象加载与校验由 Service 完成）"""
+        for obj in user_objs:
+            obj.roles.clear()
+            obj.roles.extend(role_objs)
+        await self.db.flush()
+
+    async def set_user_positions(self, user_objs: Sequence[UserModel], position_objs: Sequence[Any]) -> None:
+        """替换用户的岗位关联（纯数据操作；目标对象加载与校验由 Service 完成）"""
+        for obj in user_objs:
+            obj.positions.clear()
+            obj.positions.extend(position_objs)
+        await self.db.flush()
+
+    async def change_password(self, id: int, password_hash: str) -> UserModel:
+        """修改用户密码
+
+        参数:
+        - id (int): 用户ID
+        - password_hash (str): 密码哈希值
+
+        返回:
+        - UserModel: 更新后的用户信息
+        """
+        return await self.update(id=id, data=UserUpdateSchema(password=password_hash))
+
+    async def forget_password(self, id: int, password_hash: str) -> UserModel:
+        """重置密码（与 change_password 逻辑相同）"""
+        return await self.change_password(id=id, password_hash=password_hash)

backend/app/modules/system/user/model.py

@@ -0,0 +1,90 @@
+from datetime import datetime
+from typing import TYPE_CHECKING
+
+from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, Text
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from app.core.base_model import MappedBase, ModelMixin, UserMixin
+
+if TYPE_CHECKING:
+    from app.modules.system.dept.model import DeptModel
+    from app.modules.system.position.model import PositionModel
+    from app.modules.system.role.model import RoleModel
+
+
+class UserRolesModel(MappedBase):
+    """用户角色关联表
+
+    定义用户与角色的多对多关系
+    """
+
+    __tablename__: str = "sys_user_roles"
+    __table_args__: dict[str, str] = {"comment": "用户角色关联表"}
+
+    user_id: Mapped[int] = mapped_column(
+        Integer,
+        ForeignKey("sys_user.id", ondelete="CASCADE", onupdate="CASCADE"),
+        primary_key=True,
+        comment="用户ID",
+    )
+    role_id: Mapped[int] = mapped_column(
+        Integer,
+        ForeignKey("sys_role.id", ondelete="CASCADE", onupdate="CASCADE"),
+        primary_key=True,
+        comment="角色ID",
+    )
+
+
+class UserPositionsModel(MappedBase):
+    """用户岗位关联表
+
+    定义用户与岗位的多对多关系
+    """
+
+    __tablename__: str = "sys_user_positions"
+    __table_args__: dict[str, str] = {"comment": "用户岗位关联表"}
+
+    user_id: Mapped[int] = mapped_column(
+        Integer,
+        ForeignKey("sys_user.id", ondelete="CASCADE", onupdate="CASCADE"),
+        primary_key=True,
+        comment="用户ID",
+    )
+    position_id: Mapped[int] = mapped_column(
+        Integer,
+        ForeignKey("sys_position.id", ondelete="CASCADE", onupdate="CASCADE"),
+        primary_key=True,
+        comment="岗位ID",
+    )
+
+
+class UserModel(ModelMixin, UserMixin):
+    """用户模型
+    """
+
+    __tablename__: str = "sys_user"
+    __table_args__: dict[str, str] = {"comment": "用户表"}
+
+    username: Mapped[str] = mapped_column(String(64), unique=True, nullable=False, comment="用户名/登录账号")
+    password: Mapped[str] = mapped_column(String(255), nullable=False, comment="密码哈希")
+    name: Mapped[str] = mapped_column(String(32), nullable=False, comment="昵称")
+    mobile: Mapped[str | None] = mapped_column(String(11), nullable=True, index=True, comment="手机号")
+    email: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True, comment="邮箱")
+    gender: Mapped[str | None] = mapped_column(String(1), default="2", nullable=True, comment="性别(0:男 1:女 2:未知)")
+    avatar: Mapped[str | None] = mapped_column(String(255), nullable=True, comment="头像URL地址")
+    is_superuser: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False, comment="是否超管")
+    last_login: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, comment="最后登录时间")
+    gitee_login: Mapped[str | None] = mapped_column(String(32), nullable=True, comment="Gitee登录")
+    github_login: Mapped[str | None] = mapped_column(String(32), nullable=True, comment="Github登录")
+    wx_login: Mapped[str | None] = mapped_column(String(32), nullable=True, comment="微信登录")
+    qq_login: Mapped[str | None] = mapped_column(String(32), nullable=True, comment="QQ登录")
+    status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态(0:启动 1:停用)")
+    description: Mapped[str | None] = mapped_column(Text, default=None, nullable=True, comment="备注")
+
+    dept_id: Mapped[int | None] = mapped_column(Integer, ForeignKey("sys_dept.id", ondelete="SET NULL", onupdate="CASCADE"), nullable=True, index=True, comment="部门ID")
+    dept: Mapped["DeptModel | None"] = relationship(back_populates="users", foreign_keys=[dept_id])
+    roles: Mapped[list["RoleModel"]] = relationship(secondary="sys_user_roles", back_populates="users")
+    positions: Mapped[list["PositionModel"]] = relationship(secondary="sys_user_positions", back_populates="users")
+    created_by: Mapped["UserModel | None"] = relationship("UserModel", foreign_keys="UserModel.created_id", remote_side="UserModel.id", uselist=False, viewonly=True)
+    updated_by: Mapped["UserModel | None"] = relationship("UserModel", foreign_keys="UserModel.updated_id", remote_side="UserModel.id", uselist=False, viewonly=True)
+    deleted_by: Mapped["UserModel | None"] = relationship("UserModel", foreign_keys="UserModel.deleted_id", remote_side="UserModel.id", uselist=False, viewonly=True)

backend/app/modules/system/user/schema.py

@@ -0,0 +1,277 @@
+import re
+from urllib.parse import urlparse
+
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    EmailStr,
+    Field,
+    field_validator,
+    model_validator,
+)
+
+from app.config.setting import settings
+from app.core.base_schema import BaseQueryParam, BaseSchema, CommonSchema, CoreUserSchema, UserByQueryParam, UserBySchema
+from app.core.validator import DateTimeStr, email_validator, mobile_validator, password_validator
+from app.modules.system.menu.schema import MenuTreeOutSchema
+from app.modules.system.role.schema import RoleOutSchema
+
+PASSWORD_FIELD_DESC = (
+    f"密码（{settings.PASSWORD_MIN_LENGTH}-{settings.PASSWORD_MAX_LENGTH} 位，"
+    "需包含字母、数字、符号中的至少两类）"
+)
+
+
+class CurrentUserUpdateSchema(BaseModel):
+    """基础用户信息"""
+
+    name: str | None = Field(default=None, max_length=32, description="名称")
+    mobile: str | None = Field(default=None, max_length=11, description="手机号")
+    email: EmailStr | None = Field(default=None, description="邮箱")
+    gender: str | None = Field(default=None, max_length=1, description="性别(0:男 1:女 2:未知)")
+    avatar: str | None = Field(default=None, max_length=255, description="头像")
+    description: str | None = Field(default=None, max_length=500, description="描述")
+
+    @field_validator("mobile")
+    @classmethod
+    def validate_mobile(cls, value: str | None):
+        """校验手机号格式"""
+        return mobile_validator(value)
+
+    @field_validator("email")
+    @classmethod
+    def validate_email(cls, value: str | None):
+        """校验邮箱格式"""
+        if not value:
+            return value
+        return email_validator(value)
+
+    @field_validator("gender")
+    @classmethod
+    def validate_gender(cls, value: str | None):
+        """校验性别：仅支持 0(男)、1(女)、2(未知)"""
+        if value and value not in {"0", "1", "2"}:
+            raise ValueError("性别仅支持 0(男)、1(女)、2(未知)")
+        return value
+
+    @field_validator("avatar")
+    @classmethod
+    def validate_avatar(cls, value: str | None):
+        """校验头像地址为合法的 HTTP/HTTPS URL"""
+        if not value:
+            return value
+        parsed = urlparse(str(value))
+        if parsed.scheme in ("http", "https") and parsed.netloc:
+            return value
+        raise ValueError("头像地址需为有效的 HTTP/HTTPS URL")
+
+    @model_validator(mode="after")
+    def check_model(self):
+        """校验基础用户信息长度约束"""
+        if self.name and len(self.name) > 32:
+            raise ValueError("名称长度不能超过 32 个字符")
+        return self
+
+
+class UserForgetPasswordSchema(BaseModel):
+    """忘记密码"""
+
+    username: str = Field(..., min_length=3, max_length=32, description="用户名")
+    new_password: str = Field(..., description=PASSWORD_FIELD_DESC)
+
+    @field_validator("username")
+    @classmethod
+    def validate_username(cls, value: str):
+        """校验账号：字母开头，3-32 位"""
+        v = value.strip()
+        if not v:
+            raise ValueError("账号不能为空")
+
+        if not re.match(r"^[A-Za-z][A-Za-z0-9_.-]{2,31}$", v):
+            raise ValueError("账号需以字母开头，3-32 位，仅允许字母、数字、_ . -")
+        return v
+
+    @field_validator("new_password")
+    @classmethod
+    def validate_new_password(cls, value: str):
+        """校验新密码：长度与复杂度"""
+        return password_validator(value)
+
+
+class UserChangePasswordSchema(BaseModel):
+    """修改密码"""
+
+    old_password: str = Field(..., description="旧密码")
+    new_password: str = Field(..., description=PASSWORD_FIELD_DESC)
+
+    @field_validator("old_password")
+    @classmethod
+    def validate_old_password(cls, value: str):
+        """校验旧密码：只校验长度，避免历史弱口令用户无法主动换掉弱口令"""
+        return password_validator(value, check_strength=False)
+
+    @field_validator("new_password")
+    @classmethod
+    def validate_new_password(cls, value: str):
+        """校验新密码：长度与复杂度"""
+        return password_validator(value, label="新密码")
+
+
+class ResetPasswordSchema(BaseModel):
+    """重置密码"""
+
+    id: int = Field(default=0, description="主键ID（已弃用，由路径参数传入）")
+    password: str = Field(..., description=PASSWORD_FIELD_DESC)
+
+    @field_validator("password")
+    @classmethod
+    def validate_password(cls, value: str):
+        """校验新密码：长度与复杂度"""
+        return password_validator(value, label="新密码")
+
+
+class UserCreateSchema(CurrentUserUpdateSchema):
+    """新增用户
+    """
+
+    username: str | None = Field(default=None, max_length=32, description="用户名")
+    password: str | None = Field(default=None, description=PASSWORD_FIELD_DESC)
+    status: int = Field(default=0, ge=0, le=1, description="状态(0:启动 1:停用)")
+    description: str | None = Field(default=None, max_length=255, description="备注")
+    is_superuser: bool | None = Field(default=False, description="是否超管")
+    dept_id: int | None = Field(default=None, description="部门ID")
+    role_ids: list[int] | None = Field(default=[], description="角色ID列表")
+    position_ids: list[int] | None = Field(default=[], description="岗位ID列表")
+    @field_validator("username")
+    @classmethod
+    def validate_username(cls, value: str | None):
+        """校验账号：字母开头，2-32 位"""
+        if not value:
+            return value
+        v = value.strip()
+
+        if not re.match(r"^[A-Za-z][A-Za-z0-9_.-]{1,31}$", v):
+            raise ValueError("账号需以字母开头，2-32 位，仅允许字母、数字、_ . -")
+        return v
+
+    @field_validator("password")
+    @classmethod
+    def validate_password(cls, value: str | None):
+        """校验密码：长度与复杂度（未填写则跳过）"""
+        return password_validator(value)
+
+
+class UserRegisterSchema(BaseModel):
+    """用户注册"""
+
+    username: str = Field(..., min_length=3, max_length=32, description="用户名")
+    password: str = Field(..., description=PASSWORD_FIELD_DESC)
+    email: EmailStr | None = Field(default=None, description="邮箱")
+    name: str | None = Field(default=None, max_length=32, description="名称")
+
+    @field_validator("username")
+    @classmethod
+    def validate_username(cls, value: str):
+        """校验账号：字母开头，3-32 位"""
+        v = value.strip()
+        if not v:
+            raise ValueError("账号不能为空")
+
+        if not re.match(r"^[A-Za-z][A-Za-z0-9_.-]{2,31}$", v):
+            raise ValueError("账号需以字母开头，3-32 位，仅允许字母、数字、_ . -")
+        return v
+
+    @field_validator("password")
+    @classmethod
+    def validate_password(cls, value: str):
+        """校验密码：长度与复杂度"""
+        return password_validator(value)
+
+
+class UserUpdateSchema(CurrentUserUpdateSchema):
+    """更新"""
+
+    model_config = ConfigDict(from_attributes=True)
+
+    username: str | None = Field(default=None, max_length=32, description="用户名")
+    password: str | None = Field(default=None, description=PASSWORD_FIELD_DESC)
+    status: int | None = Field(default=None, ge=0, le=1, description="状态(0:启动 1:停用)")
+    description: str | None = Field(default=None, max_length=255, description="备注")
+    dept_id: int | None = Field(default=None, description="部门ID")
+    role_ids: list[int] | None = Field(default=[], description="角色ID列表")
+    position_ids: list[int] | None = Field(default=[], description="岗位ID列表")
+
+    @field_validator("password")
+    @classmethod
+    def validate_password(cls, value: str | None):
+        """校验密码：长度与复杂度（未填写表示不改密码）"""
+        return password_validator(value)
+
+    @field_validator("username")
+    @classmethod
+    def validate_username(cls, value: str | None):
+        """校验账号：字母开头，2-32 位"""
+        if not value:
+            return value
+        v = value.strip()
+
+        if not re.match(r"^[A-Za-z][A-Za-z0-9_.-]{1,31}$", v):
+            raise ValueError("账号需以字母开头，2-32 位，仅允许字母、数字、_ . -")
+        return v
+
+
+class UserOutSchema(CoreUserSchema, BaseSchema, UserBySchema):
+    """用户管理列表/详情响应（精简版，不含大字段嵌套）"""
+
+    model_config = ConfigDict(arbitrary_types_allowed=True, from_attributes=True)
+
+    id: int = Field(default=0, description="主键ID")
+    username: str | None = Field(default=None, max_length=32, description="用户名")
+    name: str | None = Field(default=None, max_length=32, description="名称")
+    mobile: str | None = Field(default=None, max_length=11, description="手机号")
+    email: EmailStr | None = Field(default=None, description="邮箱")
+    gender: str | None = Field(default=None, max_length=1, description="性别(0:男 1:女 2:未知)")
+    avatar: str | None = Field(default=None, max_length=255, description="头像")
+    status: int | None = Field(default=0, ge=0, le=1, description="状态(0:启动 1:停用)")
+    description: str | None = Field(default=None, max_length=255, description="备注")
+    dept_id: int | None = Field(default=None, description="部门ID")
+    role_ids: list[int] | None = Field(default=[], description="角色ID列表")
+    position_ids: list[int] | None = Field(default=[], description="岗位ID列表")
+    dept_name: str | None = Field(default=None, description="部门名称")
+    is_superuser: bool = Field(default=False, description="是否超管")
+    last_login: DateTimeStr | None = Field(default=None, description="最后登录时间")
+
+
+class CurrentUserOutSchema(UserOutSchema):
+    """当前用户信息响应（含完整菜单/角色/岗位等嵌套数据）"""
+
+    dept: CommonSchema | None = Field(default=None, description="部门")
+    positions: list[CommonSchema] | None = Field(default=[], description="岗位")
+    roles: list[RoleOutSchema] | None = Field(default=[], description="角色")
+    menus: list[MenuTreeOutSchema] | None = Field(default=[], description="菜单")
+    gitee_login: str | None = Field(default=None, max_length=32, description="Gitee登录")
+    github_login: str | None = Field(default=None, max_length=32, description="Github登录")
+    wx_login: str | None = Field(default=None, max_length=32, description="微信登录")
+    qq_login: str | None = Field(default=None, max_length=32, description="QQ登录")
+
+
+class UserQueryParam(BaseQueryParam, UserByQueryParam):
+    """用户管理查询参数（继承标准 Mixin）
+
+    支持：
+    - 时间范围（BaseQueryParam）
+    - 创建人/更新人筛选（UserByQueryParam）
+    - 业务字段：用户名、名称、手机号、邮箱、部门、状态
+    """
+
+    username: str | None = Field(None, description="用户名", json_schema_extra={"q": "like"})
+    name: str | None = Field(None, description="名称", json_schema_extra={"q": "like"})
+    mobile: str | None = Field(None, description="手机号", pattern=r"^1[3-9]\d{9}$", json_schema_extra={"q": "eq"})
+    email: str | None = Field(
+        None,
+        description="邮箱",
+        pattern=r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$",
+        json_schema_extra={"q": "eq"},
+    )
+    dept_id: int | None = Field(None, description="部门ID", json_schema_extra={"q": "eq"})
+    status: int | None = Field(None, description="是否可用", json_schema_extra={"q": "eq"})

backend/app/modules/system/user/service.py

@@ -0,0 +1,503 @@
+from collections.abc import Sequence
+from typing import Any
+
+from fastapi import UploadFile
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.config.setting import settings
+from app.core.base_schema import AuthSchema, BatchSetAvailable, PageResultSchema
+from app.core.exceptions import CustomException
+from app.core.logger import logger
+from app.modules.system.dept.crud import DeptCRUD
+from app.modules.system.menu.crud import MenuCRUD
+from app.modules.system.menu.schema import MenuOutSchema, MenuTreeOutSchema
+from app.modules.system.position.crud import PositionCRUD
+from app.modules.system.role.crud import RoleCRUD
+from app.utils.common_util import search_to_dict, traversal_to_tree
+from app.utils.excel_util import ExcelUtil
+from app.utils.password_util import PwdUtil
+
+from .crud import UserCRUD
+from .model import UserModel
+from .schema import (
+    CurrentUserOutSchema,
+    CurrentUserUpdateSchema,
+    ResetPasswordSchema,
+    UserChangePasswordSchema,
+    UserCreateSchema,
+    UserForgetPasswordSchema,
+    UserOutSchema,
+    UserQueryParam,
+    UserRegisterSchema,
+    UserUpdateSchema,
+)
+
+# 用户管理列表/详情预加载
+_USER_PRELOAD = ["dept", "roles", "positions"]
+
+# 当前用户信息预加载：需完整嵌套关联
+_USER_CURRENT_PRELOAD = ["dept", "positions", "roles.menus", "roles.depts"]
+
+
+class UserService:
+    """用户管理服务"""
+
+    def __init__(self, auth: AuthSchema, db: AsyncSession) -> None:
+        self.auth = auth
+        self.db = db
+
+    async def detail(self, id: int) -> UserOutSchema:
+        user = await UserCRUD(self.auth, self.db).get_or_404(id=id, preload=_USER_PRELOAD)
+        result = UserOutSchema.model_validate(user)
+        if user.dept:
+            result.dept_name = user.dept.name
+        result.role_ids = [r.id for r in user.roles]
+        result.position_ids = [p.id for p in user.positions]
+        return result
+
+    async def get_list(
+        self,
+        search: UserQueryParam | None = None,
+        order_by: list[dict[str, str]] | None = None,
+    ) -> list[UserOutSchema]:
+        user_list = await UserCRUD(self.auth, self.db).get_list(search=search_to_dict(search), order_by=order_by, preload=_USER_PRELOAD)
+        result = [UserOutSchema.model_validate(user) for user in user_list]
+        for user, item in zip(user_list, result, strict=True):
+            if user.dept:
+                item.dept_name = user.dept.name
+            item.role_ids = [r.id for r in user.roles]
+            item.position_ids = [p.id for p in user.positions]
+        return result
+
+    async def page(
+        self,
+        page_no: int,
+        page_size: int,
+        search: UserQueryParam | None = None,
+        order_by: list[dict[str, str]] | None = None,
+    ) -> PageResultSchema[UserOutSchema]:
+        offset = (page_no - 1) * page_size
+        page_result = await UserCRUD(self.auth, self.db).page(
+            offset=offset,
+            limit=page_size,
+            order_by=order_by or [{"id": "asc"}],
+            search=search_to_dict(search),
+            preload=_USER_PRELOAD,
+        )
+        items: list[UserOutSchema] = []
+        for user in page_result.items:
+            item = UserOutSchema.model_validate(user)
+            if user.dept:
+                item.dept_name = user.dept.name
+            item.role_ids = [r.id for r in user.roles]
+            item.position_ids = [p.id for p in user.positions]
+            items.append(item)
+        page_result.items = items
+        return page_result  # type: ignore[return-value]
+
+    async def create(self, data: UserCreateSchema) -> UserOutSchema:
+        if data.is_superuser:
+            raise CustomException(msg="不允许创建超级管理员")
+        if await UserCRUD(self.auth, self.db).get(username=data.username):
+            raise CustomException(msg="已存在相同用户名称的账号")
+
+        if data.dept_id and not await DeptCRUD(self.auth, self.db).get(id=data.dept_id):
+            raise CustomException(msg="该数据不存在")
+
+        if data.password:
+            data.password = await PwdUtil.ahash_password(password=data.password)
+
+        create_data = data.model_dump(exclude_none=True, exclude={"role_ids", "position_ids"})
+        new_user = await UserCRUD(self.auth, self.db).create(data=create_data)
+        await self._set_user_roles(user_ids=[new_user.id], role_ids=data.role_ids or [])
+        await self._set_user_positions(user_ids=[new_user.id], position_ids=data.position_ids or [])
+        return await self.detail(id=new_user.id)
+
+    async def update(self, id: int, data: UserUpdateSchema) -> UserOutSchema:
+        if data.username:
+            if exist_user := await UserCRUD(self.auth, self.db).get(username=data.username):
+                if exist_user.id != id:
+                    raise CustomException(msg="更新失败，账号已存在")
+
+        if data.mobile:
+            if exist_mobile := await UserCRUD(self.auth, self.db).get(mobile=data.mobile):
+                if exist_mobile.id != id:
+                    raise CustomException(msg="该数据已存在")
+        if data.email:
+            if exist_email := await UserCRUD(self.auth, self.db).get(email=data.email):
+                if exist_email.id != id:
+                    raise CustomException(msg="该数据已存在")
+
+        if data.dept_id:
+            dept = await DeptCRUD(self.auth, self.db).get(id=data.dept_id)
+            if not dept:
+                raise CustomException(msg="该数据不存在")
+            if dept.status == 1:
+                raise CustomException(msg="部门已被禁用")
+
+        update_data = data.model_dump(exclude_unset=True, exclude_none=True, exclude={"role_ids", "position_ids"})
+        await UserCRUD(self.auth, self.db).update(id=id, data=update_data)
+        await self._set_user_roles(user_ids=[id], role_ids=data.role_ids or [])
+        await self._set_user_positions(user_ids=[id], position_ids=data.position_ids or [])
+
+        return await self.detail(id=id)
+
+    async def delete(self, ids: list[int]) -> None:
+        if not ids:
+            raise CustomException(msg="删除失败，删除对象不能为空")
+        users = await UserCRUD(self.auth, self.db).get_list(search={"id": ("in", ids)}, preload=["roles", "positions"])
+        user_map = {u.id: u for u in users}
+        errors: list[str] = []
+        for uid in ids:
+            user = user_map.get(uid)
+            if not user:
+                errors.append(f"用户[{uid}]不存在")
+                continue
+            if user.is_superuser:
+                errors.append(f"用户[{uid}]是超级管理员，不能删除")
+                continue
+            if self.auth.user.id == uid:
+                errors.append("不能删除当前登陆用户")
+                continue
+        if errors:
+            raise CustomException(msg="; ".join(errors))
+
+        # 先解除角色/岗位关联，再删除用户
+        await UserCRUD(self.auth, self.db).set_user_roles(user_objs=users, role_objs=[])
+        await UserCRUD(self.auth, self.db).set_user_positions(user_objs=users, position_objs=[])
+        await UserCRUD(self.auth, self.db).delete(ids=ids)
+
+    async def _load_users(self, user_ids: list[int], preload: list[str]) -> Sequence[UserModel]:
+        """加载用户并校验存在性。"""
+        if not user_ids:
+            raise CustomException(msg="用户ID列表不能为空")
+        users = await UserCRUD(self.auth, self.db).get_list(search={"id": ("in", user_ids)}, preload=preload)
+        if len(users) != len(set(user_ids)):
+            missing = sorted(set(user_ids) - {u.id for u in users})
+            raise CustomException(msg=f"用户不存在: {missing}")
+        return users
+
+    async def _set_user_roles(self, user_ids: list[int], role_ids: list[int]) -> None:
+        """替换用户角色关联：service 校验角色存在性与启用状态，CRUD 仅负责持久化。"""
+        if not role_ids:
+            return
+        users = await self._load_users(user_ids, preload=["roles"])
+        roles = await RoleCRUD(self.auth, self.db).get_list(search={"id": ("in", role_ids)})
+        missing = set(role_ids) - {r.id for r in roles}
+        if missing:
+            raise CustomException(msg=f"角色不存在: {sorted(missing)}")
+        if any(role.status != 0 for role in roles):
+            raise CustomException(msg="部分角色已被禁用")
+        await UserCRUD(self.auth, self.db).set_user_roles(user_objs=users, role_objs=roles)
+
+    async def _set_user_positions(self, user_ids: list[int], position_ids: list[int]) -> None:
+        """替换用户岗位关联：service 校验岗位存在性与启用状态，CRUD 仅负责持久化。"""
+        if not position_ids:
+            return
+        users = await self._load_users(user_ids, preload=["positions"])
+        positions = await PositionCRUD(self.auth, self.db).get_list(search={"id": ("in", position_ids)})
+        missing = set(position_ids) - {p.id for p in positions}
+        if missing:
+            raise CustomException(msg=f"岗位不存在: {sorted(missing)}")
+        if any(position.status != 0 for position in positions):
+            raise CustomException(msg="部分岗位已被禁用")
+        await UserCRUD(self.auth, self.db).set_user_positions(user_objs=users, position_objs=positions)
+
+    async def current_info(self, check_data_scope: bool = True) -> CurrentUserOutSchema:
+        user_id = self.auth.user.id
+        if not user_id:
+            raise CustomException(msg="该数据不存在")
+
+        if not check_data_scope:
+            # 轻量模式：只刷新菜单/权限，不加载用户嵌套数据
+            menus_raw = await self._load_menus()
+            menu_tree = [MenuTreeOutSchema(**item) for item in traversal_to_tree([menu.model_dump(mode="json") for menu in menus_raw])]
+            return CurrentUserOutSchema(menus=menu_tree)
+
+        user = await UserCRUD(self.auth, self.db).get(id=user_id, preload=_USER_CURRENT_PRELOAD)
+        if user is None:
+            raise CustomException(msg="该数据不存在")
+        user_dict = CurrentUserOutSchema.model_validate(user)
+        if user.dept:
+            user_dict.dept_name = user.dept.name
+        user_dict.is_superuser = user.is_superuser
+
+        menu_tree = [MenuTreeOutSchema(**item) for item in traversal_to_tree([menu.model_dump(mode="json") for menu in await self._load_menus()])]
+        user_dict.menus = menu_tree
+        return user_dict
+
+    async def _load_menus(self) -> list[MenuOutSchema]:
+        """加载当前用户的菜单列表（不含树形转换）"""
+        _pc_only = {"scope": "web"}
+        if self.auth.user.is_superuser:
+            menu_all = await MenuCRUD(self.auth, self.db).get_list(
+                search={"type": ("in", [1, 2, 3, 4]), "status": 0, **_pc_only},
+                order_by=[{"order": "asc"}],
+            )
+            return [MenuOutSchema.model_validate(menu) for menu in menu_all]
+        else:
+            menu_ids = set(self.auth.menu_ids)
+            if not menu_ids:
+                return []
+            return [
+                MenuOutSchema.model_validate(menu)
+                for menu in await MenuCRUD(self.auth, self.db).get_list(
+                    search={"id": ("in", list(menu_ids)), **_pc_only},
+                    order_by=[{"order": "asc"}],
+                )
+            ]
+
+    async def update_current_info(self, data: CurrentUserUpdateSchema) -> UserOutSchema:
+        user_id = self.auth.user.id
+        if not user_id:
+            raise CustomException(msg="该数据不存在")
+
+        if data.mobile:
+            if exist_mobile := await UserCRUD(self.auth, self.db).get(mobile=data.mobile):
+                if exist_mobile.id != user_id:
+                    raise CustomException(msg="该数据已存在")
+        if data.email:
+            if exist_email := await UserCRUD(self.auth, self.db).get(email=data.email):
+                if exist_email.id != user_id:
+                    raise CustomException(msg="该数据已存在")
+
+        user_update_data = UserUpdateSchema(**data.model_dump())
+        await UserCRUD(self.auth, self.db).update(id=user_id, data=user_update_data)
+        return await self.detail(id=user_id)
+
+    async def set_available(self, data: BatchSetAvailable) -> None:
+        users = await UserCRUD(self.auth, self.db).get_list(search={"id": ("in", data.ids)})
+        for user in users:
+            if user.is_superuser:
+                raise CustomException(msg="超级管理员状态不能修改")
+        await UserCRUD(self.auth, self.db).set(ids=data.ids, status=data.status)
+
+    async def change_password(self, data: UserChangePasswordSchema) -> UserOutSchema:
+        user_id = self.auth.user.id
+        if not user_id:
+            raise CustomException(msg="该数据不存在")
+
+        user = await UserCRUD(self.auth, self.db).get_or_404(id=user_id)
+        if not await PwdUtil.averify_password(plain_password=data.old_password, password_hash=user.password):
+            raise CustomException(msg="原密码输入错误")
+
+        new_password_hash = await PwdUtil.ahash_password(password=data.new_password)
+        await UserCRUD(self.auth, self.db).change_password(id=user_id, password_hash=new_password_hash)
+        return await self.detail(id=user_id)
+
+    async def reset_password(self, data: ResetPasswordSchema) -> UserOutSchema:
+        user = await UserCRUD(self.auth, self.db).get_or_404(id=data.id)
+        if user.is_superuser:
+            raise CustomException(msg="超级管理员密码不能重置")
+
+        new_password_hash = await PwdUtil.ahash_password(password=data.password)
+        await UserCRUD(self.auth, self.db).change_password(id=data.id, password_hash=new_password_hash)
+        return await self.detail(id=data.id)
+
+    async def forget_password(self, data: UserForgetPasswordSchema) -> UserOutSchema:
+        user = await UserCRUD(self.auth, self.db).get_or_404(username=data.username)
+        if user.status == 1:
+            raise CustomException(msg="用户已停用")
+        if user.is_superuser:
+            raise CustomException(msg="超级管理员密码不能重置")
+
+        new_password_hash = await PwdUtil.ahash_password(password=data.new_password)
+        await UserCRUD(self.auth, self.db).change_password(id=user.id, password_hash=new_password_hash)
+        return await self.detail(id=user.id)
+
+    async def register(self, data: UserRegisterSchema) -> UserOutSchema:
+        """用户注册"""
+        exists_user = await UserCRUD(self.auth, self.db).get(username=data.username)
+        if exists_user:
+            raise CustomException(msg="已存在相同用户名称的账号")
+
+        create_data = UserCreateSchema(
+            username=data.username,
+            password=await PwdUtil.ahash_password(password=data.password),
+            name=data.name or data.username,
+            status=0,
+        )
+        create_data_dict = create_data.model_dump(exclude_none=True, exclude={"role_ids", "position_ids"})
+        new_user = await UserCRUD(self.auth, self.db).create(data=create_data_dict)
+        if not new_user:
+            raise CustomException(msg="注册失败")
+        # 注册时 auth 无用户，created_id/updated_id 未设置，补充为自身ID
+        await UserCRUD(self.auth, self.db).set([new_user.id], created_id=new_user.id, updated_id=new_user.id)
+        logger.info(f"新用户注册成功: {data.username}")
+        return await self.detail(id=new_user.id)
+
+    async def batch_import(self, file: UploadFile, update_support: bool = False) -> str:
+        header_dict = {
+            "部门编号": "dept_id",
+            "账号": "username",
+            "昵称": "name",
+            "邮箱": "email",
+            "手机号": "mobile",
+            "性别": "gender",
+            "状态": "status",
+        }
+
+        try:
+            contents = await file.read()
+            rows = await ExcelUtil.aread_excel_to_dicts(contents)
+            await file.close()
+
+            if not rows:
+                raise CustomException(msg="导入文件为空")
+
+            missing_headers = [h for h in header_dict if h not in rows[0]]
+            if missing_headers:
+                raise CustomException(msg=f"导入文件缺少必要的列: {', '.join(missing_headers)}")
+
+            # 将中文字段名映射为英文字段
+            mapped_rows = []
+            for row in rows:
+                mapped_rows.append({en: row.get(ch) for ch, en in header_dict.items()})
+
+            required_fields = ["username", "name", "dept_id"]
+            errors = []
+            for field in required_fields:
+                missing_count = sum(1 for r in mapped_rows if r.get(field) is None)
+                if missing_count:
+                    errors.append(f"字段'{field}'有{missing_count}行缺少数据")
+
+            if errors:
+                raise CustomException(msg="\n".join(errors))
+
+            success_count = 0
+            error_msgs = []
+
+            for i, row in enumerate(mapped_rows, start=2):
+                count_delta, err = await self._process_import_row(i, row, update_support)
+                if err:
+                    error_msgs.append(err)
+                else:
+                    success_count += count_delta
+
+            result = f"成功导入 {success_count} 条数据"
+            if error_msgs:
+                result += "\n错误信息:\n" + "\n".join(error_msgs)
+            return result
+
+        except Exception as e:
+            logger.error(f"批量导入用户失败: {e!s}")
+            raise CustomException(msg=f"导入失败: {e!s}") from e
+
+    async def _process_import_row(
+        self,
+        row_num: int,
+        row: dict,
+        update_support: bool,
+    ) -> tuple[int, str | None]:
+        """处理单行导入数据
+
+        验证字段合法性，执行创建或更新操作。
+
+        参数:
+        - row_num (int): Excel 行号（用于错误提示）
+        - row (dict): 经过字段映射后的用户数据行
+        - update_support (bool): 是否支持更新已存在用户
+
+        返回:
+        - tuple[int, str | None]: (成功计数增量, 错误信息或 None)
+        """
+        try:
+            username = (str(row["username"]) if row["username"] is not None else "").strip()
+            name = (str(row["name"]) if row["name"] is not None else "").strip()
+            if not username:
+                return 0, f"第{row_num}行: 账号不能为空"
+            if not name:
+                return 0, f"第{row_num}行: 昵称不能为空"
+
+            dept_id = int(row["dept_id"])
+            dept = await DeptCRUD(self.auth, self.db).get(id=dept_id)
+            if not dept:
+                return 0, f"第{row_num}行: 部门ID {dept_id} 不存在"
+
+            user_data = {
+                "username": username,
+                "name": name,
+                "email": str(row["email"]).strip() if row.get("email") is not None else None,
+                "mobile": str(row["mobile"]).strip() if row.get("mobile") is not None else None,
+                "gender": str(row["gender"]).strip() if row.get("gender") is not None else "1",
+                "status": 0 if str(row["status"]).strip() == "正常" else 1,
+                "dept_id": dept_id,
+                "password": await PwdUtil.ahash_password(password=settings.PASSWORD_IMPORT_DEFAULT),
+            }
+
+            exists_user = await UserCRUD(self.auth, self.db).get(username=user_data["username"])
+            if exists_user:
+                if exists_user.is_superuser:
+                    return 0, f"第{row_num}行: 超级管理员不允许修改"
+                if update_support:
+                    user_update_data = UserUpdateSchema(**user_data)
+                    await UserCRUD(self.auth, self.db).update(id=exists_user.id, data=user_update_data)
+                    return 1, None
+                else:
+                    return 0, f"第{row_num}行: 用户 {user_data['username']} 已存在"
+            else:
+                user_create_schema = UserCreateSchema(**user_data)
+                new_user = await UserCRUD(self.auth, self.db).create(
+                    data=user_create_schema.model_dump(exclude_none=True, exclude={"role_ids", "position_ids"})  # type: ignore[arg-type]
+                )
+                await self._set_user_roles(user_ids=[new_user.id], role_ids=user_create_schema.role_ids or [])
+                await self._set_user_positions(user_ids=[new_user.id], position_ids=user_create_schema.position_ids or [])
+                return 1, None
+
+        except Exception as e:
+            return 0, f"第{row_num}行: 异常{e!s}"
+
+    @staticmethod
+    def get_import_template() -> bytes:
+        header_list = [
+            "部门编号",
+            "账号",
+            "昵称",
+            "邮箱",
+            "手机号",
+            "性别",
+            "状态",
+        ]
+        selector_header_list = ["性别", "状态"]
+        option_list = [
+            {"性别": ["男", "女", "未知"]},
+            {"状态": ["正常", "停用"]},
+        ]
+        return ExcelUtil.get_excel_template(
+            header_list=header_list,
+            selector_header_list=selector_header_list,
+            option_list=option_list,
+        )
+
+    @staticmethod
+    async def export_list(user_list: list[dict[str, Any]]) -> bytes:
+        if not user_list:
+            raise CustomException(msg="没有数据可导出")
+
+        mapping_dict = {
+            "id": "用户编号",
+            "avatar": "头像",
+            "username": "用户名称",
+            "name": "用户昵称",
+            "dept_name": "部门",
+            "email": "邮箱",
+            "mobile": "手机号",
+            "gender": "性别",
+            "status": "状态",
+            "is_superuser": "是否超级管理员",
+            "last_login": "最后登录时间",
+            "description": "备注",
+            "created_time": "创建时间",
+            "updated_time": "更新时间",
+            "updated_id": "更新者ID",
+        }
+
+        data = user_list.copy()
+        for item in data:
+            item["status"] = "启用" if item.get("status") == 0 else "停用"
+            gender = item.get("gender")
+            item["gender"] = "男" if gender == "1" else ("女" if gender == "2" else "未知")
+            item["is_superuser"] = "是" if item.get("is_superuser") else "否"
+            item["creator"] = item.get("created_by", {}).get("name", "未知") if isinstance(item.get("created_by"), dict) else "未知"
+
+        return await ExcelUtil.aexport_list2excel(list_data=data, mapping_dict=mapping_dict)