用户管理

用户管理是系统的核心模块，属于复杂业务模块。包含用户 CRUD、角色关联、密码加密、行政区划、Excel 导入导出等功能。用户管理继承 BaseService 获得标准 CRUD 能力，并通过覆写钩子方法实现头像保存、密码加密、角色关联等定制逻辑。

模块结构

src/modules/system/user/
├── models.py     # 用户模型
├── schema.py     # 表单验证（WTForms）
├── repository.py # 数据访问层
├── service.py    # 业务逻辑层（继承 BaseService）
└── router.py     # 路由控制器

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

用户模型

# src/modules/system/user/models.py
# ============================================================
# 用户模型
# ============================================================
class User(BaseModel, BaseDB):
    """用户模型类"""
    # ============================================================
    # 表名配置
    # ============================================================
    __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。

角色关联

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

# src/modules/system/user_role/models.py
# ============================================================
# 用户角色关联模型
# ============================================================
class UserRole(BaseModel, BaseDB):
    """用户角色关联模型类"""
    # ============================================================
    # 表名配置
    # ============================================================
    __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 关键片段
from src.common.base_service import BaseService
from src.core.password import encrypt_password, generate_salt
from src.core.security import get_realname
from src.core.exceptions import BizError
from src.common.constants.common import DEFAULT_PASSWORD
from src.modules.association.user_role.repository import UserRoleRepository


class UserService(BaseService):
    """用户服务类，继承 BaseService 获得标准 CRUD 能力"""
    repository = UserRepository
    form = UserForm
    status_form = UserStatusForm
    model_name = "用户"

    @classmethod
    def add(cls, data: dict) -> None:
        """新增用户"""
        # 1. 表单校验
        form = cls._validate_form(data)

        # 2. 处理密码：未传密码时使用默认密码
        pwd = form.data.get("password") or DEFAULT_PASSWORD
        salt = generate_salt()

        # 3. 校验行政区划
        city = form.data.get("city", [])
        if len(city) != 4:
            raise BizError("请选择完整的所在城市（省/市/区/街道）")
        roles = form.data.get("roles")

        # 4. 清理表单数据，准备入库
        create_data = cls._clean_form_data(form)
        create_data["password"] = encrypt_password(pwd, salt)
        create_data["salt"] = salt
        create_data["create_user"] = get_realname()
        _apply_city(create_data, city)

        # 5. 创建用户记录
        user = cls.repository.create(**create_data)
        user.code = f"{user.id:06d}"  # 根据自增ID生成6位用户编码

        # 6. 关联用户角色关系
        _add_user_role(user.id, roles)
        # 7. 清除权限缓存
        invalidate_user_cache(user.id)

关联数据富化

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

# src/modules/system/user/service.py
def _enrich_user_records(records):
    """为用户记录补充关联数据"""
    dept_map = _get_dept_map()      # {dept.id: dept.name}
    level_map = _get_level_map()    # {level.id: level.name}
    position_map = _get_position_map()  # {position.id: position.name}
    gender_map = dict_utils.get_value_map("gender")

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

    for data in records:
        data["genderText"] = 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 src.modules.system.menu.service import invalidate_user_cache

# 角色变更后清理缓存
invalidate_user_cache(user_id)

# 状态变更后清理缓存
invalidate_user_cache(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. 复杂业务：继承 BaseService 并覆写 add/update 钩子，支持头像/密码/行政区划/角色
2. 密码安全：bcrypt 双重加盐，工作因子12（约250ms）
3. 角色关联：多对多关系，通过 flask_user_role 关联表，先删后增策略
4. 权限缓存：角色/状态变更后即时清理 Redis 缓存
5. 关联富化：列表/详情补充部门、职级、岗位、角色、性别等展示名称
6. 行政区划：省/市/区/街道四级编码，自动拆分到独立字段
7. Excel 导入导出：列定义式导入解析，逐行校验，部分成功返回明细
8. 超管保护：ID=1 账号禁止修改/删除/禁用/重置密码