document-management/backend/dms/services/user_service.py

"""用户摘要、组织解析、列表查询与用户维护写操作。"""

from __future__ import annotations

from datetime import UTC, datetime, timezone
from typing import Any

from flask import current_app
from sqlalchemy import func, or_, select
from sqlalchemy.exc import IntegrityError

from dms.common.enums import (
    AllowedModule,
    AuditAction,
    AuditTarget,
    EnabledStatus,
    RoleCode,
    SecurityLevel,
)
from dms.common.errors import (
    AuthenticationError,
    ConflictError,
    ForbiddenError,
    InvalidArgumentError,
    ResourceNotFoundError,
    UserSelfProtectionError,
)
from dms.common.pagination import PageRequest, page_result
from dms.common.response import serialize_id
from dms.database.transaction import transaction
from dms.extensions import db
from dms.models import Organization, Role, User
from dms.security.auth_context import get_auth_context
from dms.security.decorators import require_reauth
from dms.security.passwords import hash_password, validate_password_policy
from dms.services.admin_permission import (
    MANAGED_ORG_SELF,
    MANAGED_ORG_SELF_ANCHOR,
    SUPER_ADMIN_USERNAME,
    can_edit_categories,
    can_edit_organizations,
    can_edit_roles,
    can_edit_users,
    can_manage_system,
    can_view_audit_log,
    effective_manage_role,
    effective_role_snapshot,
    ensure_org_in_managed_scope,
    ensure_user_scope_in_managed_scope,
    is_super_admin,
    managed_scope_ids,
    require_user_editor,
)
from dms.services.audit_service import business_audit
from dms.services.authorization_service import (
    PLAN_MANAGE_MATRIX_COLUMNS,
    VIEW_MATRIX_COLUMNS,
    has_any_matrix_permission,
    has_any_upload_permission,
)


CREATE_FIELDS = {
    "username",
    "realName",
    "initialPassword",
    "organizationId",
    "roleCode",
    "roleId",
    "securityLevel",
    "status",
}
UPDATE_FIELDS = {
    "realName",
    "organizationId",
    "roleCode",
    "roleId",
    "securityLevel",
    "status",
    "rowVersion",
}
RESET_PASSWORD_FIELDS = {"newPassword", "rowVersion"}
# 用户级管理范围（契约1.30）为可选字段：create 未提交缺省落 SELF（哨兵0），
# update 未提交保持原值，兼容不带该字段的存量调用方
OPTIONAL_MANAGED_ORG_FIELDS = {"managedOrgId"}


def _utc_now() -> datetime:
    return datetime.now(timezone.utc).replace(tzinfo=None)


def effective_organization(user: User) -> tuple[int | None, str | None]:
    if user.organization_id is not None:
        organization = db.session.scalar(
            select(Organization).where(
                Organization.id == user.organization_id,
                Organization.is_deleted.is_(False),
                Organization.status == EnabledStatus.ENABLED.value,
            )
        )
        if organization is not None:
            return organization.id, organization.org_name
    return user.organization_id, user.organization_name


def _format_utc(value: datetime | None) -> str | None:
    if value is None:
        return None
    return (
        value.replace(tzinfo=UTC)
        .isoformat(timespec="milliseconds")
        .replace("+00:00", "Z")
    )


def allowed_modules_for_user(user: User) -> list[str]:
    """按角色权限推导可见模块（契约1.35菜单口径，与用户类别无关）。

    菜单可见性完全由生效角色权限决定（纯矩阵/管理位判定，不查文档
    存在性）；页签内按钮与接口鉴权仍按 evaluate_plan_access 实时判定：
    - 方案权限矩阵含管理类列（上传/编辑/删除/授权/挂载）任一，
      或具备分类管理位：渲染「方案管理」，不再渲染「方案浏览」；
    - 否则矩阵含浏览类列（查看/下载）任一：渲染「方案浏览」；
    - 日志审计/系统管理/机构用户管理位（机构/用户/角色任一）：
      渲染对应页签；
    - 矩阵与管理位全空：兜底渲染「方案浏览」；
    - 内置 admin 恒全量四页签（无方案浏览）。
    """
    if is_super_admin(user):
        return [
            AllowedModule.BACKEND_MANAGEMENT.value,
            AllowedModule.AUDIT_LOG.value,
            AllowedModule.ORGANIZATION_USER_MANAGEMENT.value,
            AllowedModule.SYSTEM_MANAGEMENT.value,
        ]
    from dms.security.auth_context import AuthContext

    context = AuthContext(
        user_id=user.id,
        username=user.username,
        real_name=user.real_name,
        organization_id=user.organization_id,
        organization_name=user.organization_name,
        role_code=RoleCode(user.role_code),
        security_level=SecurityLevel(user.security_level),
        auth_version=user.auth_version,
    )
    modules: list[str] = []
    if has_any_matrix_permission(
        context, PLAN_MANAGE_MATRIX_COLUMNS
    ) or can_edit_categories(db.session, user):
        modules.append(AllowedModule.BACKEND_MANAGEMENT.value)
    elif has_any_matrix_permission(context, VIEW_MATRIX_COLUMNS):
        modules.append(AllowedModule.DOCUMENT_BROWSER.value)
    if can_view_audit_log(db.session, user):
        modules.append(AllowedModule.AUDIT_LOG.value)
    if (
        can_edit_organizations(db.session, user)
        or can_edit_users(db.session, user)
        or can_edit_roles(db.session, user)
    ):
        modules.append(AllowedModule.ORGANIZATION_USER_MANAGEMENT.value)
    if can_manage_system(db.session, user):
        modules.append(AllowedModule.SYSTEM_MANAGEMENT.value)
    if not modules:
        modules.append(AllowedModule.DOCUMENT_BROWSER.value)
    return modules


def effective_role(user: User) -> tuple[str | None, str | None]:
    if user.role_id is None:
        return None, None
    role = db.session.get(Role, user.role_id)
    if role is None or role.is_deleted:
        return serialize_id(user.role_id), None
    return serialize_id(role.id), role.role_name


def user_summary(
    user: User,
    *,
    allowed_modules: list[str] | None = None,
) -> dict[str, object]:
    organization_id, organization_name = effective_organization(user)
    role_id, role_name = effective_role(user)
    manage_role = effective_manage_role(db.session, user)
    # 内置 admin 恒为全权超管（契约1.17③）：perm 位恒开、锚点恒全局
    super_admin = is_super_admin(user)
    return {
        "id": serialize_id(user.id),
        "username": user.username,
        "realName": user.real_name,
        "organizationId": serialize_id(organization_id),
        "organizationName": organization_name,
        "roleCode": user.role_code,
        "roleId": role_id,
        "roleName": role_name,
        "securityLevel": user.security_level,
        "status": user.status,
        "authVersion": user.auth_version,
        "rowVersion": user.row_version,
        # 管理权限改挂角色：字段名保留，值取自生效角色
        "permCategoryEdit": (
            True
            if super_admin
            else (
                bool(manage_role.manage_category_edit)
                if manage_role is not None
                else False
            )
        ),
        "permOrgEdit": (
            True
            if super_admin
            else (
                bool(manage_role.manage_org_edit) if manage_role is not None else False
            )
        ),
        "permUserEdit": (
            True
            if super_admin
            else (
                bool(manage_role.manage_user_edit) if manage_role is not None else False
            )
        ),
        "permSystemManage": (
            True
            if super_admin
            else (
                bool(manage_role.manage_system) if manage_role is not None else False
            )
        ),
        "permAuditLog": (
            True
            if super_admin
            else (
                bool(manage_role.manage_audit_log)
                if manage_role is not None
                else False
            )
        ),
        "permRoleEdit": (
            True
            if super_admin
            else (
                bool(manage_role.manage_role_edit)
                if manage_role is not None
                else False
            )
        ),
        # 上传权限（契约1.31）：生效角色方案权限矩阵任一 OWN 行 can_upload=True
        "permUpload": (
            True
            if super_admin
            else has_any_upload_permission(
                manage_role.id if manage_role is not None else None
            )
        ),
        # 用户级管理范围（契约1.30）：null=全局、"SELF"=跟随本人所在机构（哨兵0）、
        # 机构ID=固定锚点；实际生效范围以角色锚点为准，角色为 SELF 时才回落本设置。
        # 内置 admin 恒全局。
        "managedOrgId": (
            None if super_admin else _managed_org_value(user.managed_org_id)
        ),
        "lastLoginAt": _format_utc(user.last_login_at),
        "lastLoginIp": user.last_login_ip,
        "uiTheme": user.ui_theme,
        "allowedModules": (
            allowed_modules
            if allowed_modules is not None
            else allowed_modules_for_user(user)
        ),
    }


def parse_string_id(raw_value: str | None, *, name: str) -> int | None:
    if raw_value is None or raw_value == "":
        return None
    if not raw_value.isdecimal() or int(raw_value) <= 0:
        raise InvalidArgumentError(f"{name}必须是正整数形式的字符串ID")
    return int(raw_value)


def _descendant_ids(root_id: int) -> set[int]:
    organizations = db.session.scalars(
        select(Organization).where(Organization.is_deleted.is_(False))
    ).all()
    if not any(item.id == root_id for item in organizations):
        raise ResourceNotFoundError("组织不存在")
    children: dict[int | None, list[int]] = {}
    for item in organizations:
        if item.status == EnabledStatus.ENABLED.value:
            children.setdefault(item.parent_id, []).append(item.id)
    found = {root_id}
    pending = [root_id]
    while pending:
        current = pending.pop()
        for child_id in children.get(current, []):
            if child_id not in found:
                found.add(child_id)
                pending.append(child_id)
    return found


def list_users(
    *,
    organization_id: int | None,
    include_descendants: bool,
    keyword: str | None,
    status: str | None,
    page_request: PageRequest,
) -> dict[str, object]:
    filters = [User.is_deleted.is_(False)]
    # 超级管理员登录开关（安全需求7）关闭时用户列表隐藏内置 admin
    if not current_app.config.get("DMS_ALLOW_ADMIN_LOGIN", True):
        filters.append(User.username != SUPER_ADMIN_USERNAME)
    # 锚点管理员强制收敛到管理范围子树（全局管理员不受限），
    # 与请求的 organization_id 过滤取交集。
    actor = db.session.get(User, get_auth_context().user_id)
    scope = managed_scope_ids(db.session, actor) if actor is not None else None
    if scope is not None:
        filters.append(User.organization_id.in_(scope))
    if organization_id is not None:
        organization_ids = (
            _descendant_ids(organization_id)
            if include_descendants
            else {organization_id}
        )
        # 显式请求的机构不在管理范围内：明确报无权限，
        # 不再静默交集返回空页（避免前端误显示"该机构无用户"）
        if scope is not None and not (scope & organization_ids):
            raise ForbiddenError("无权限查看该机构的用户")
        filters.append(User.organization_id.in_(organization_ids))
    if keyword:
        pattern = f"%{keyword.strip()}%"
        filters.append(or_(User.username.like(pattern), User.real_name.like(pattern)))
    if status:
        try:
            normalized_status = EnabledStatus(status).value
        except ValueError as exc:
            raise InvalidArgumentError("status必须是ENABLED或DISABLED") from exc
        filters.append(User.status == normalized_status)

    total = db.session.scalar(select(func.count(User.id)).where(*filters)) or 0
    users = db.session.scalars(
        select(User)
        .where(*filters)
        .order_by(User.sort_no.asc(), User.id.asc())
        .offset(page_request.offset)
        .limit(page_request.page_size)
    ).all()
    return page_result(
        [user_summary(user) for user in users],
        page=page_request.page,
        page_size=page_request.page_size,
        total=total,
    )


# ===== 用户维护写操作 =====


def _required_text(
    value: Any,
    *,
    field: str,
    max_length: int,
) -> str:
    if not isinstance(value, str) or not value.strip():
        raise InvalidArgumentError(f"{field}不能为空")
    normalized = value.strip()
    if len(normalized) > max_length:
        raise InvalidArgumentError(f"{field}长度不能超过{max_length}")
    return normalized


def _required_password(value: Any) -> str:
    if not isinstance(value, str) or not value:
        raise InvalidArgumentError("密码不能为空")
    try:
        validate_password_policy(value)
    except ValueError as exc:
        raise InvalidArgumentError(str(exc)) from exc
    return value


def _role_code(value: Any) -> str:
    if not isinstance(value, str):
        raise InvalidArgumentError("roleCode必须是字符串")
    try:
        return RoleCode(value).value
    except ValueError as exc:
        raise InvalidArgumentError("roleCode不是有效枚举值") from exc


def _security_level(value: Any) -> str:
    if not isinstance(value, str):
        raise InvalidArgumentError("securityLevel必须是字符串")
    try:
        return SecurityLevel(value).value
    except ValueError as exc:
        raise InvalidArgumentError("securityLevel不是有效枚举值") from exc


def _status_value(value: Any) -> str:
    if not isinstance(value, str):
        raise InvalidArgumentError("status必须是字符串")
    try:
        return EnabledStatus(value).value
    except ValueError as exc:
        raise InvalidArgumentError("status必须是ENABLED或DISABLED") from exc


def _row_version(value: Any) -> int:
    if type(value) is not int or value < 0:
        raise InvalidArgumentError("rowVersion必须是非负整数")
    return value


def _exact_payload(
    payload: Any,
    expected: set[str],
    *,
    optional: set[str] | None = None,
) -> dict[str, Any]:
    allowed = expected | (optional or set())
    if not isinstance(payload, dict) or not expected <= set(payload) <= allowed:
        raise InvalidArgumentError(
            "请求体字段必须且只能包含：" + "、".join(sorted(allowed))
        )
    return payload


def _resolve_organization(
    session, organization_id: int | None
) -> tuple[int | None, str | None]:
    if organization_id is None:
        return None, None
    organization = session.scalar(
        select(Organization)
        .where(
            Organization.id == organization_id,
            Organization.is_deleted.is_(False),
            Organization.status == EnabledStatus.ENABLED.value,
        )
        .with_for_update()
    )
    if organization is None:
        raise InvalidArgumentError("organizationId必须指向有效且已启用的机构")
    return organization.id, organization.org_name


def _resolve_role(session, role_id: int | None) -> tuple[int | None, str | None]:
    """校验并解析自定义角色；USER 与受限 ADMIN 均可关联角色。

    ADMIN 关联的角色仅约束其文档操作权限矩阵，管理类权限位不受影响；
    未关联角色（含内置 admin）由授权层回退默认角色。
    """
    if role_id is None:
        return None, None
    role = session.scalar(
        select(Role).where(Role.id == role_id, Role.is_deleted.is_(False))
    )
    if role is None:
        raise InvalidArgumentError("roleId必须指向有效角色")
    return role.id, role.role_name


def _optional_managed_org(session, value: Any) -> int | None:
    """解析用户级管理范围三态（契约1.30）。

    null/空串=全局、"SELF"=跟随本人所在机构（哨兵0）、机构ID=固定锚点；
    锚点须指向未删除机构（与角色锚点同口径，不要求启用、不限操作者范围）。
    """
    if value is None or value == "":
        return None
    if value == MANAGED_ORG_SELF:
        return MANAGED_ORG_SELF_ANCHOR
    if not isinstance(value, str) or not value.isdecimal() or int(value) <= 0:
        raise InvalidArgumentError("managedOrgId必须是SELF、机构ID或null")
    org_id = int(value)
    exists = session.scalar(
        select(Organization.id).where(
            Organization.id == org_id,
            Organization.is_deleted.is_(False),
        )
    )
    if exists is None:
        raise InvalidArgumentError("managedOrgId必须指向有效机构")
    return org_id


def _managed_org_value(value: int | None) -> str | None:
    """用户级管理范围序列化：NULL→null、哨兵0→"SELF"、其余→机构ID字符串。"""
    if value is None:
        return None
    if value == MANAGED_ORG_SELF_ANCHOR:
        return MANAGED_ORG_SELF
    return serialize_id(value)


def _ensure_not_self(actor: Any, target_user_id: int, *, action: str) -> None:
    if actor.user_id == target_user_id:
        raise UserSelfProtectionError(
            f"不能对本账号执行{action}操作",
        )


def _ensure_not_builtin_admin(user: User, *, action: str) -> None:
    """内置 admin 账号不允许被任何用户编辑、删除或重置密码。"""
    if user.username == SUPER_ADMIN_USERNAME:
        raise ForbiddenError(f"内置admin账号不允许被{action}")


def _other_active_admin_count(session, exclude_user_id: int) -> int:
    """统计除指定用户外的活跃管理员数量。"""
    return (
        session.scalar(
            select(func.count(User.id)).where(
                User.role_code == RoleCode.ADMIN.value,
                User.status == EnabledStatus.ENABLED.value,
                User.is_deleted.is_(False),
                User.id != exclude_user_id,
            )
        )
        or 0
    )


def _ensure_not_last_active_admin(session, user: User, *, reason: str) -> None:
    """禁止影响系统最后一个活跃管理员（仅在 user 当前是活跃管理员时生效）。

    统一的守卫：覆盖删除、降级、停用三种场景。
    若 user 当前不是活跃管理员（角色非 ADMIN 或状态非 ENABLED），
    则其消失不影响活跃管理员数，直接放行。
    """
    if (
        user.role_code != RoleCode.ADMIN.value
        or user.status != EnabledStatus.ENABLED.value
    ):
        return
    if _other_active_admin_count(session, user.id) == 0:
        raise ConflictError(
            "系统至少需保留一个启用的系统管理员",
            details={"roleId": user.role_code, "reason": reason},
        )


def create_user(payload: Any) -> dict[str, object]:
    body = _exact_payload(
        payload, CREATE_FIELDS, optional=OPTIONAL_MANAGED_ORG_FIELDS
    )
    username = _required_text(body["username"], field="username", max_length=64)
    real_name = _required_text(body["realName"], field="realName", max_length=64)
    initial_password = _required_password(body["initialPassword"])
    organization_id = parse_string_id(body["organizationId"], name="organizationId")
    role_code = _role_code(body["roleCode"])
    role_id = parse_string_id(body["roleId"], name="roleId")
    security_level = _security_level(body["securityLevel"])
    status = _status_value(body["status"])
    actor_id = get_auth_context().user_id
    # 超级管理员登录开关（安全需求7）关闭时拒绝创建内置 admin 用户名：
    # 报错不提示"已存在"，不泄露系统中存在该账号
    if (
        username == SUPER_ADMIN_USERNAME
        and not current_app.config.get("DMS_ALLOW_ADMIN_LOGIN", True)
    ):
        raise InvalidArgumentError("不能创建该名称的用户")

    try:
        with transaction() as session:
            actor = require_user_editor(session)
            ensure_org_in_managed_scope(session, actor, organization_id)
            if (
                session.scalar(
                    select(User.id).where(
                        User.username == username,
                        User.is_deleted.is_(False),
                    )
                )
                is not None
            ):
                raise ConflictError("用户名已存在")
            org_id, org_name = _resolve_organization(session, organization_id)
            resolved_role_id, resolved_role_name = _resolve_role(session, role_id)
            # 缺省落 SELF（哨兵0）：显式赋值不依赖数据库缺省，
            # 保证未提交该字段的新建用户与迁移回填的存量行语义一致
            managed_org_id = (
                _optional_managed_org(session, body["managedOrgId"])
                if "managedOrgId" in body
                else MANAGED_ORG_SELF_ANCHOR
            )
            # 防提权（契约1.30修订）：目标 ADMIN 的生效管理范围不得超出操作者
            ensure_user_scope_in_managed_scope(
                session,
                actor,
                role_code=role_code,
                effective_role=effective_role_snapshot(session, resolved_role_id),
                managed_org_id=managed_org_id,
                organization_id=org_id,
            )
            user = User(
                username=username,
                password_hash=hash_password(initial_password),
                real_name=real_name,
                organization_id=org_id,
                organization_name=org_name,
                role_code=role_code,
                role_id=resolved_role_id,
                security_level=security_level,
                status=status,
                managed_org_id=managed_org_id,
                created_by=actor_id,
                updated_by=actor_id,
            )
            session.add(user)
            session.flush()
            # 回读服务端缺省列（ui_theme/auth_version/row_version 等 server_default）：
            # flush 不回填且 expire_on_commit=False，不 refresh 则响应字段为 null 违反契约
            session.refresh(user)
            session.add(
                business_audit(
                    action=AuditAction.CREATE_USER,
                    target=AuditTarget.USER,
                    target_id=user.id,
                    target_name=user.real_name,
                    detail={
                        "username": user.username,
                        "organizationId": serialize_id(user.organization_id),
                        "organizationName": user.organization_name,
                        "roleCode": user.role_code,
                        "roleId": serialize_id(user.role_id),
                        "roleName": resolved_role_name,
                        "securityLevel": user.security_level,
                        "status": user.status,
                        "managedOrgId": _managed_org_value(user.managed_org_id),
                    },
                )
            )
        return user_summary(user)
    except IntegrityError as exc:
        raise ConflictError("用户名已存在") from exc


def update_user(user_id: int, payload: Any) -> dict[str, object]:
    body = _exact_payload(
        payload, UPDATE_FIELDS, optional=OPTIONAL_MANAGED_ORG_FIELDS
    )
    real_name = _required_text(body["realName"], field="realName", max_length=64)
    organization_id = parse_string_id(
        body["organizationId"], name="organizationId"
    )
    new_role_code = _role_code(body["roleCode"])
    new_role_id = parse_string_id(body["roleId"], name="roleId")
    new_security_level = _security_level(body["securityLevel"])
    new_status = _status_value(body["status"])
    expected_version = _row_version(body["rowVersion"])
    actor = get_auth_context()
    actor_id = actor.user_id
    now = _utc_now()
    _ensure_not_self(actor, user_id, action="编辑角色或状态")

    with transaction() as session:
        actor_user = require_user_editor(session)
        user = session.scalar(
            select(User)
            .where(
                User.id == user_id,
                User.is_deleted.is_(False),
            )
            .with_for_update()
        )
        if user is None:
            raise ResourceNotFoundError("用户不存在")
        _ensure_not_builtin_admin(user, action="编辑")
        ensure_org_in_managed_scope(session, actor_user, user.organization_id)
        if user.row_version != expected_version:
            raise ConflictError(
                "数据已被其他用户修改，请刷新后重试",
                details={"currentRowVersion": user.row_version},
            )

        before = {
            "realName": user.real_name,
            "organizationId": serialize_id(user.organization_id),
            "organizationName": user.organization_name,
            "roleCode": user.role_code,
            "roleId": serialize_id(user.role_id),
            "securityLevel": user.security_level,
            "status": user.status,
            "managedOrgId": _managed_org_value(user.managed_org_id),
            "rowVersion": user.row_version,
        }

        # 角色变更或停用：合并检查最后一个活跃管理员
        role_changes = new_role_code != user.role_code
        status_changes = new_status != user.status
        security_level_changes = new_security_level != user.security_level
        will_lose_active_admin = (
            role_changes
            or (status_changes and new_status == EnabledStatus.DISABLED.value)
        )
        if will_lose_active_admin:
            _ensure_not_last_active_admin(
                session, user, reason="role_or_status_change"
            )

        org_id, org_name = _resolve_organization(session, organization_id)
        # 迁入机构同样须在管理范围内：锚点管理员不可将用户移出自己管理的子树
        ensure_org_in_managed_scope(session, actor_user, org_id)
        resolved_role_id, resolved_role_name = _resolve_role(session, new_role_id)
        role_assignment_changes = resolved_role_id != user.role_id
        # 可选字段（契约1.30）：未提交保持原值；调整不改变认证身份，
        # 不触发二次认证、不失效旧 Token（与角色锚点同口径）
        new_managed_org_id = (
            _optional_managed_org(session, body["managedOrgId"])
            if "managedOrgId" in body
            else None
        )
        # 防提权（契约1.30修订）：目标 ADMIN 最终生效管理范围不得超出操作者；
        # 未提交 managedOrgId 时按保持原值判定。校验置于 reauth 之前，
        # 避免换角色场景先抛 REAUTH_REQUIRED 掩盖越权拒绝。
        final_managed_org_id = (
            new_managed_org_id if "managedOrgId" in body else user.managed_org_id
        )
        ensure_user_scope_in_managed_scope(
            session,
            actor_user,
            role_code=new_role_code,
            effective_role=effective_role_snapshot(session, resolved_role_id),
            managed_org_id=final_managed_org_id,
            organization_id=org_id,
        )

        # 敏感字段（密级/角色/状态）实际变化：强制操作人二次认证（契约1.21）
        if (
            security_level_changes
            or role_changes
            or role_assignment_changes
            or status_changes
        ):
            require_reauth()

        user.real_name = real_name
        user.organization_id = org_id
        user.organization_name = org_name
        user.role_code = new_role_code
        user.role_id = resolved_role_id
        user.security_level = new_security_level
        user.status = new_status
        if "managedOrgId" in body:
            user.managed_org_id = new_managed_org_id
        user.updated_by = actor_id
        user.updated_at = now
        user.row_version += 1

        # 内置角色、自定义角色变更或停用都需让旧 Token 失效；
        # 管理类权限与锚点调整不改变认证身份，不失效旧 Token
        auth_bump = (
            role_changes
            or role_assignment_changes
            or (status_changes and new_status == EnabledStatus.DISABLED.value)
        )
        if auth_bump:
            user.auth_version += 1

        after = {
            "realName": user.real_name,
            "organizationId": serialize_id(user.organization_id),
            "organizationName": user.organization_name,
            "roleCode": user.role_code,
            "roleId": serialize_id(user.role_id),
            "securityLevel": user.security_level,
            "status": user.status,
            "managedOrgId": _managed_org_value(user.managed_org_id),
            "rowVersion": user.row_version,
        }
        session.add(
            business_audit(
                action=AuditAction.EDIT_USER,
                target=AuditTarget.USER,
                target_id=user.id,
                target_name=user.real_name,
                detail={
                    "before": before,
                    "after": after,
                    "roleName": resolved_role_name,
                    "authVersionBumped": auth_bump,
                },
            )
        )
    return user_summary(user)


def delete_user(user_id: int, row_version: int) -> dict[str, object]:
    actor = get_auth_context()
    actor_id = actor.user_id
    now = _utc_now()
    _ensure_not_self(actor, user_id, action="删除")

    with transaction() as session:
        actor_user = require_user_editor(session)
        user = session.scalar(
            select(User)
            .where(
                User.id == user_id,
                User.is_deleted.is_(False),
            )
            .with_for_update()
        )
        if user is None:
            raise ResourceNotFoundError("用户不存在")
        _ensure_not_builtin_admin(user, action="删除")
        ensure_org_in_managed_scope(session, actor_user, user.organization_id)
        if user.row_version != row_version:
            raise ConflictError(
                "数据已被其他用户修改，请刷新后重试",
                details={"currentRowVersion": user.row_version},
            )

        # 删除：禁止删除最后一个活跃管理员
        _ensure_not_last_active_admin(session, user, reason="delete")

        target_name = user.real_name
        target_username = user.username
        user.is_deleted = True
        user.deleted_at = now
        user.updated_at = now
        user.updated_by = actor_id
        user.status = EnabledStatus.DISABLED.value
        user.auth_version += 1
        user.row_version += 1
        session.add(
            business_audit(
                action=AuditAction.DELETE_USER,
                target=AuditTarget.USER,
                target_id=user.id,
                target_name=target_name,
                detail={
                    "username": target_username,
                    "rowVersion": user.row_version,
                },
            )
        )
    return {"id": serialize_id(user.id), "deleted": True}


def reset_user_password(user_id: int, payload: Any) -> dict[str, object]:
    body = _exact_payload(payload, RESET_PASSWORD_FIELDS)
    new_password = _required_password(body["newPassword"])
    expected_version = _row_version(body["rowVersion"])
    actor = get_auth_context()
    actor_id = actor.user_id
    now = _utc_now()
    _ensure_not_self(actor, user_id, action="重置密码")

    with transaction() as session:
        actor_user = require_user_editor(session)
        user = session.scalar(
            select(User)
            .where(
                User.id == user_id,
                User.is_deleted.is_(False),
            )
            .with_for_update()
        )
        if user is None:
            raise ResourceNotFoundError("用户不存在")
        _ensure_not_builtin_admin(user, action="重置密码")
        ensure_org_in_managed_scope(session, actor_user, user.organization_id)
        if user.row_version != expected_version:
            raise ConflictError(
                "数据已被其他用户修改，请刷新后重试",
                details={"currentRowVersion": user.row_version},
            )

        user.password_hash = hash_password(new_password)
        user.auth_version += 1
        user.updated_by = actor_id
        user.updated_at = now
        user.row_version += 1
        session.add(
            business_audit(
                action=AuditAction.RESET_USER_PASSWORD,
                target=AuditTarget.USER,
                target_id=user.id,
                target_name=user.real_name,
                detail={
                    "username": user.username,
                    "rowVersion": user.row_version,
                },
            )
        )
    return {
        "id": serialize_id(user.id),
        "reset": True,
        "authVersion": user.auth_version,
    }


UI_THEMES = ("blue", "red", "green")


REORDER_FIELDS = {"orderedIds"}


def reorder_users(payload: Any) -> dict[str, object]:
    """用户列表拖动重排（契约1.27）：全体用户槽位置换，步长10重编号 sort_no。

    orderedIds 允许为当前分页视图的子集，被提交用户按原槽位次序互换
    （未提交用户相对次序不变），与分类同级重排的置换语义一致。
    """
    body = _exact_payload(payload, REORDER_FIELDS)
    ordered_ids = body["orderedIds"]
    if not isinstance(ordered_ids, list) or any(
        not isinstance(item, str) or not item.isdecimal() or int(item) <= 0
        for item in ordered_ids
    ):
        raise InvalidArgumentError("orderedIds必须是正整数形式的字符串ID数组")
    if len(set(ordered_ids)) != len(ordered_ids):
        raise InvalidArgumentError("orderedIds存在重复ID")
    if len(ordered_ids) < 2:
        raise InvalidArgumentError("orderedIds必须是至少包含2个ID的数组")
    actor_id = get_auth_context().user_id
    now = _utc_now()

    with transaction() as session:
        require_user_editor(session)
        rows = session.scalars(
            select(User)
            .where(User.is_deleted.is_(False))
            .order_by(User.sort_no.asc(), User.id.asc())
            .with_for_update()
        ).all()
        row_by_id = {row.id: row for row in rows}
        all_ids = [row.id for row in rows]
        ordered = [int(item) for item in ordered_ids]
        if any(user_id not in row_by_id for user_id in ordered):
            raise ResourceNotFoundError("用户不存在")
        # 槽位置换：被提交ID按原槽位顺序排好，依次填入目标顺序
        slot_ids = sorted(ordered, key=all_ids.index)
        new_order = list(all_ids)
        for position, user_id in enumerate(slot_ids):
            new_order[all_ids.index(user_id)] = ordered[position]
        reordered_count = 0
        for index, user_id in enumerate(new_order):
            row = row_by_id[user_id]
            target = (index + 1) * 10
            if row.sort_no == target:
                continue
            row.sort_no = target
            row.updated_by = actor_id
            row.updated_at = now
            row.row_version += 1
            reordered_count += 1
        first = row_by_id[ordered[0]]
        session.add(
            business_audit(
                action=AuditAction.REORDER_USER,
                target=AuditTarget.USER,
                target_id=first.id,
                target_name=first.real_name,
                detail={
                    "orderedIds": list(ordered_ids),
                    "reorderedCount": reordered_count,
                },
            )
        )
    return {"reorderedCount": reordered_count}


def update_ui_theme(payload: Any) -> dict[str, object]:
    """本人界面主题偏好（契约1.20）：白名单校验，仅操作本人行。

    主题是账号级漫游的纯视觉偏好：不 bump rowVersion（避免干扰并发
    编辑乐观锁）、不 bump authVersion（不下线会话）、不写审计。
    """
    body = _exact_payload(payload, {"uiTheme"})
    theme = body["uiTheme"]
    if not isinstance(theme, str) or theme not in UI_THEMES:
        raise InvalidArgumentError("uiTheme必须是blue、red或green")
    context = get_auth_context()
    with transaction() as session:
        user = session.scalar(
            select(User).where(
                User.id == context.user_id,
                User.is_deleted.is_(False),
            )
        )
        if user is None:
            raise AuthenticationError()
        user.ui_theme = theme
        session.flush()
        return user_summary(user)