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

"""用户摘要、组织解析与列表查询。"""

from __future__ import annotations

from sqlalchemy import func, or_, select

from dms.common.enums import AllowedModule, EnabledStatus, RoleCode
from dms.common.errors import InvalidArgumentError, ResourceNotFoundError
from dms.common.pagination import PageRequest, page_result
from dms.common.response import serialize_id
from dms.extensions import db
from dms.models import Organization, User


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


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 user_summary(user: User) -> dict[str, object]:
    organization_id, organization_name = effective_organization(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,
        "securityLevel": user.security_level,
        "status": user.status,
        "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,
    )