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

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

from __future__ import annotations

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

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 (
    ConflictError,
    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.passwords import hash_password
from dms.services.audit_service import business_audit


ALLOWED_MODULES = {
    RoleCode.USER: [AllowedModule.DOCUMENT_BROWSER.value],
    RoleCode.ADMIN: [
        AllowedModule.DOCUMENT_BROWSER.value,
        AllowedModule.BACKEND_MANAGEMENT.value,
        AllowedModule.AUDIT_LOG.value,
        AllowedModule.ORGANIZATION_USER_MANAGEMENT.value,
    ],
}


CREATE_FIELDS = {
    "username",
    "realName",
    "initialPassword",
    "organizationId",
    "roleCode",
    "roleId",
    "securityLevel",
    "status",
}
UPDATE_FIELDS = {
    "realName",
    "organizationId",
    "roleCode",
    "roleId",
    "securityLevel",
    "status",
    "rowVersion",
}
RESET_PASSWORD_FIELDS = {"newPassword", "rowVersion"}


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 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) -> dict[str, object]:
    organization_id, organization_name = effective_organization(user)
    role_id, role_name = effective_role(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,
        "lastLoginAt": _format_utc(user.last_login_at),
        "lastLoginIp": user.last_login_ip,
        "allowedModules": ALLOWED_MODULES[RoleCode(user.role_code)],
    }


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)]
    if organization_id is not None:
        organization_ids = (
            _descendant_ids(organization_id)
            if include_descendants
            else {organization_id}
        )
        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.real_name.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("密码不能为空")
    if len(value) < 8 or len(value) > 64:
        raise InvalidArgumentError("密码长度必须在8~64个字符之间")
    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]) -> dict[str, Any]:
    if not isinstance(payload, dict) or set(payload) != expected:
        raise InvalidArgumentError(
            "请求体字段必须且只能包含：" + "、".join(sorted(expected))
        )
    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, role_code: str
) -> tuple[int | None, str | None]:
    """校验并解析自定义角色；ADMIN 恒为超管，不挂自定义角色。"""
    if role_id is None:
        return None, None
    if role_code == RoleCode.ADMIN.value:
        raise InvalidArgumentError("ADMIN账号不能关联自定义角色，roleId必须为空")
    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 _ensure_not_self(actor: Any, target_user_id: int, *, action: str) -> None:
    if actor.user_id == target_user_id:
        raise UserSelfProtectionError(
            f"不能对本账号执行{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)
    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

    try:
        with transaction() as session:
            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, role_code
            )
            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,
                created_by=actor_id,
                updated_by=actor_id,
            )
            session.add(user)
            session.flush()
            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,
                    },
                )
            )
        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)
    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:
        user = session.scalar(
            select(User)
            .where(
                User.id == user_id,
                User.is_deleted.is_(False),
            )
            .with_for_update()
        )
        if user is None:
            raise ResourceNotFoundError("用户不存在")
        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,
            "rowVersion": user.row_version,
        }

        # 角色变更或停用：合并检查最后一个活跃管理员
        role_changes = new_role_code != user.role_code
        status_changes = new_status != user.status
        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)
        resolved_role_id, resolved_role_name = _resolve_role(
            session, new_role_id, new_role_code
        )
        role_assignment_changes = resolved_role_id != user.role_id

        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
        user.updated_by = actor_id
        user.updated_at = now
        user.row_version += 1

        # 内置角色、自定义角色变更或停用都需让旧 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,
            "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:
        user = session.scalar(
            select(User)
            .where(
                User.id == user_id,
                User.is_deleted.is_(False),
            )
            .with_for_update()
        )
        if user is None:
            raise ResourceNotFoundError("用户不存在")
        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:
        user = session.scalar(
            select(User)
            .where(
                User.id == user_id,
                User.is_deleted.is_(False),
            )
            .with_for_update()
        )
        if user is None:
            raise ResourceNotFoundError("用户不存在")
        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,
    }