用户管理

用户管理是系统的核心模块，属于复杂业务模块。包含用户 CRUD、角色关联、密码加密、行政区划、Excel 导入导出等功能。与简单 CRUD 模块不同，用户管理使用自定义 service 函数而非继承 BaseService 的类方法。

模块结构

src/modules/system/user/
├── models.py     # 用户模型
├── schemas.py    # 表单验证
├── repository.py # 数据访问层
└── service.py    # 业务逻辑层（自定义函数）

src/modules/system/user_role/
├── models.py     # 用户-角色关联模型
├── repository.py # 关联数据访问
└── service.py    # 关联业务逻辑

用户模型

# src/modules/system/user/models.py
# ============================================================
# 用户模型
# ============================================================
class User(base_model, base_db):
    """用户模型类"""
    # ============================================================
    # 表名配置
    # ============================================================
    __tablename__ = DB_PREFIX + "user"
    __table_comment__ = "用户表"

    # ============================================================
    # 字段定义
    # ============================================================
    # 用户编码
    code = Column(String(100), nullable=True, default='', server_default=text("''"), index=True, comment="用户编码")
    # 用户姓名
    realname = Column(String(150), nullable=False, index=True, comment="用户姓名")
    # 性别：1-男 2-女 3-保密
    gender = Column(Integer, default=1, server_default=text('1'), comment="性别：1-男 2-女 3-保密")
    # 用户头像
    avatar = Column(String(255), nullable=False, comment="用户头像")
    # 手机号
    mobile = Column(String(30), nullable=False, index=True, comment="手机号")
    # 邮箱
    email = Column(String(30), nullable=False, index=True, comment="邮箱")
    # 部门ID
    dept_id = Column(Integer, default=0, server_default=text('0'), index=True, comment="部门ID")
    # 职级ID
    level_id = Column(Integer, default=0, server_default=text('0'), comment="职级ID")
    # 岗位ID
    position_id = Column(Integer, default=0, server_default=text('0'), comment="岗位ID")
    # 省份编码
    province_code = Column(String(30), nullable=False, comment="省份编码")
    # 城市编码
    city_code = Column(String(30), nullable=False, comment="城市编码")
    # 县区编码
    district_code = Column(String(30), nullable=False, comment="县区编码")
    # 街道编码
    street_code = Column(String(30), nullable=False, comment="街道编码")
    # 省市区信息
    city_info = Column(String(255), nullable=True, comment="省市区信息")
    # 详细地址
    address = Column(String(255), nullable=False, comment="详细地址")
    # 用户名
    username = Column(String(30), nullable=True, index=True, comment="用户名")
    # 密码
    password = Column(String(255), nullable=True, comment="密码")
    # 加密盐
    salt = Column(String(30), nullable=True, comment="加密盐")
    # 个人简介
    intro = Column(String(255), nullable=True, comment="个人简介")
    # 状态：1-正常 2-禁用
    status = Column(Integer, default=1, server_default=text('1'), index=True, comment="状态：1-正常 2-禁用")
    # 排序
    sort = Column(Integer, default=0, server_default=text('0'), comment="排序")
    # 个人备注
    note = Column(String(255), nullable=True, comment="个人备注")

    # ============================================================
    # 内置方法
    # ============================================================
    def __str__(self):
        """返回用户ID作为字符串表示"""
        return "用户{}".format(self.id)

密码加密

使用 bcrypt 双重加盐方案，由 src/core/password.py 提供：

# src/core/password.py
# +======================================================================
# | 模块: 密码加密工具
# | 说明: 基于 bcrypt 的双重加盐方案
# +======================================================================

"""密码加密工具：基于 bcrypt 的双重加盐方案。"""

import secrets

import bcrypt

# bcrypt 工作因子，值越大计算越慢越安全，默认12（约250ms）
BCRYPT_ROUNDS = 12


# ============================================================
# 密码工具函数
# ============================================================
def generate_salt(length: int = 10) -> str:
    """生成随机盐值"""
    return secrets.token_urlsafe(length)[:length]


def encrypt_password(password: str, salt: str) -> str:
    """使用 bcrypt 加密密码（外部 salt 双重加盐）"""
    combined = (password + salt).encode('utf-8')
    hashed = bcrypt.hashpw(combined, bcrypt.gensalt(rounds=BCRYPT_ROUNDS))
    return hashed.decode('utf-8')


def verify_password(password: str, salt: str, hashed: str) -> bool:
    """验证密码是否正确"""
    combined = (password + salt).encode('utf-8')
    return bcrypt.checkpw(combined, hashed.encode('utf-8'))

双重加盐

密码加密采用双重加盐：外部 salt（存储在 user 表）+ bcrypt 内部 salt。即使数据库泄露，攻击者也无法通过彩虹表破解密码。工作因子 12，单次加密约 250ms。

角色关联

用户与角色通过 fastapi_user_role 关联表实现多对多关系：

# src/modules/system/user_role/models.py
# ============================================================
# 用户角色关联模型
# ============================================================
class UserRole(base_model, base_db):
    """用户角色关联模型类"""
    # ============================================================
    # 表名配置
    # ============================================================
    __tablename__ = DB_PREFIX + "user_role"
    __table_comment__ = "用户角色关联表"

    # ============================================================
    # 字段定义
    # ============================================================
    # 用户ID
    user_id = Column(Integer, default=0, server_default=text('0'), index=True, comment="用户ID")
    # 角色ID
    role_id = Column(Integer, default=0, server_default=text('0'), index=True, comment="角色ID")

    # ============================================================
    # 内置方法
    # ============================================================
    def __str__(self):
        """返回用户ID作为字符串表示"""
        return "用户角色表{}".format(self.user_id)

新增用户时同时创建角色关联，删除用户时硬删关联数据：

# src/modules/system/user/service.py 关键片段
async def add_user(request, data: UserForm):
    """新增用户"""
    def _add():
        # 唯一性校验
        if user_repo.get_one(username=data.username):
            return R.failed("登录账号不能重复"), None

        formData = data.model_dump(exclude={'id', 'city', 'roles'})

        # 密码加密：未传密码时使用默认密码
        pwd = data.password or DEFAULT_PASSWORD
        salt = password.generate_salt()
        formData['password'] = password.encrypt_password(pwd, salt)
        formData['salt'] = salt

        # 行政区划处理
        citys = data.city or []
        formData['provinceCode'] = str(citys[0]) if len(citys) > 0 and citys[0] else ''
        # ... 其他字段处理

        user = user_repo.create(**formData, create_user=get_realname(request))
        user.code = f"{user.id:06d}"  # 根据自增ID生成6位用户编码
        return None, user.id

    err, user_id = await asyncio.to_thread(_add)
    if err:
        return err

    # 创建用户角色数据 + 清理权限缓存
    if data.roles is not None:
        await add_user_role(user_id, data.roles)

    return R.ok(msg="添加成功")

关联数据富化

用户列表和详情需要富化部门、职级、岗位、角色等关联数据：

# src/modules/system/user/service.py
def _enrich_user_records(records):
    """为用户记录补充关联数据"""
    dept_map = {d.id: d.name for d in dept_repo.get_all()}
    level_map = {l.id: l.name for l in level_repo.get_all()}
    position_map = {p.id: p.name for p in position_repo.get_all()}
    gender_map = get_dict_map('gender')

    user_ids = [data.get("id") for data in records]
    role_map = get_user_role_map(user_ids)

    for data in records:
        data["genderName"] = gender_map.get(str(data.get("gender")))
        data["deptName"] = dept_map.get(data.get("deptId"))
        data["levelName"] = level_map.get(data.get("levelId"))
        data["positionName"] = position_map.get(data.get("positionId"))
        data["roles"] = role_map.get(data.get("id"), [])

    return records

权限缓存清理

用户增删改、角色变更、状态变更时需清理 Redis 中的权限和有效状态缓存：

from utils.perm_cache import invalidate_user_perms, invalidate_user_active

# 角色变更后清理缓存
await invalidate_user_perms(user_id)
await invalidate_user_active(user_id)

# 状态变更后清理缓存（禁用即时踢出、启用即时放行）
await invalidate_user_perms(id)
await invalidate_user_active(id)

Excel 导入导出

用户模块支持完整的 Excel 导入导出，导入列定义和导出列定义通过数组声明：

# 导入列定义（顺序即 Excel 列序）
USER_IMPORT_COLUMNS = [
    {"title": "登录账号", "key": "username", "required": True},
    {"title": "用户姓名", "key": "realname", "required": True},
    {"title": "性别", "key": "gender", "parser": _parse_gender},
    {"title": "手机号", "key": "mobile", "required": True},
    {"title": "邮箱", "key": "email", "required": True},
    {"title": "部门", "key": "dept_id", "parser": _parse_dept},
    {"title": "职级", "key": "level_id", "parser": _parse_level},
    {"title": "岗位", "key": "position_id", "parser": _parse_position},
    {"title": "用户角色", "key": "roles", "parser": _parse_roles},
    {"title": "状态", "key": "status", "parser": _parse_status},
    {"title": "城市信息", "key": "city_codes", "parser": _parse_city_info},
    {"title": "详细地址", "key": "address", "required": True},
    # ...
]

导入解析器

每个导入列可声明 parser 函数，负责将 Excel 单元格值转换为目标类型。例如 _parse_dept 按部门名称查询部门 ID，_parse_roles 按角色名称（逗号分隔）查询角色 ID 列表。解析失败时返回行级错误信息，不影响其他行。

API 接口

接口	方法	权限节点	说明
/api/v1/user/page	GET	sys:user:page	用户分页列表
/api/v1/user/detail/{id}	GET	sys:user:detail	用户详情
/api/v1/user/add	POST	sys:user:add	新增用户
/api/v1/user/update	PUT	sys:user:update	编辑用户
/api/v1/user/delete/{id}	DELETE	sys:user:delete	删除用户
/api/v1/user/batchDelete	DELETE	sys:user:batchDelete	批量删除
/api/v1/user/status	PUT	sys:user:status	设置状态
/api/v1/user/resetPwd	PUT	sys:user:resetPwd	重置密码
/api/v1/user/updatePassword	PUT	-	修改密码（本人）
/api/v1/user/updateInfo	PUT	-	更新个人信息
/api/v1/user/import	POST	sys:user:import	导入用户
/api/v1/user/export	GET	sys:user:export	导出用户

内置超管保护

用户 ID 为 1 的内置超管账号（SUPERADMIN_ID）禁止通过管理接口修改、删除、禁用、重置密码。这些操作会在 service 层提前拦截并返回失败提示。

总结

用户管理模块具备以下特点：

1. 复杂业务：自定义 service 函数，异步操作用 asyncio.to_thread 包裹
2. 密码安全：bcrypt 双重加盐，工作因子12（约250ms）
3. 角色关联：多对多关系，通过 user_role 关联表，删除时硬删关联
4. 权限缓存：角色/状态变更后即时清理 Redis 缓存
5. 关联富化：列表/详情补充部门、职级、岗位、角色、性别等展示名称
6. 行政区划：省/市/区/街道四级编码，自动拼接城市信息
7. Excel 导入导出：列定义式导入解析，逐行校验，部分成功返回明细
8. 超管保护：ID=1 账号禁止修改/删除/禁用/重置密码