document-management/backend/dms/api/v1/users.py

"""当前用户、用户分页查询与用户维护写操作。"""

from __future__ import annotations

from flask import request
from sqlalchemy import select

from dms.api.v1.blueprint import api_v1
from dms.common.errors import AuthenticationError, InvalidArgumentError
from dms.common.pagination import PageRequest
from dms.common.response import success_response
from dms.extensions import db
from dms.models import User
from dms.security.auth_context import get_auth_context
from dms.security.decorators import admin_required, bearer_auth_required
from dms.services.user_service import (
    allowed_modules_for_user,
    create_user,
    delete_user,
    list_users,
    parse_string_id,
    reorder_users,
    reset_user_password,
    update_ui_theme,
    update_user,
    user_summary,
)


def _integer_query(name: str, default: int) -> int:
    raw_value = request.args.get(name)
    if raw_value is None or raw_value == "":
        return default
    try:
        return int(raw_value)
    except ValueError as exc:
        raise InvalidArgumentError(f"{name}必须是整数") from exc


def _boolean_query(name: str, default: bool) -> bool:
    raw_value = request.args.get(name)
    if raw_value is None or raw_value == "":
        return default
    normalized = raw_value.lower()
    if normalized not in {"true", "false"}:
        raise InvalidArgumentError(f"{name}必须是true或false")
    return normalized == "true"


def _json_body():
    if not request.is_json:
        raise InvalidArgumentError("请求体必须是JSON对象")
    body = request.get_json(silent=True)
    if not isinstance(body, dict):
        raise InvalidArgumentError("请求体必须是JSON对象")
    return body


def _parse_user_id(value: str) -> int:
    if not value.isdecimal() or int(value) <= 0:
        raise InvalidArgumentError("id必须是正整数形式的字符串ID")
    return int(value)


def _parse_delete_row_version(value: str | None) -> int:
    if value is None or not value.isdecimal():
        raise InvalidArgumentError("rowVersion必须是非负整数")
    return int(value)


@api_v1.get("/users/me")
@bearer_auth_required
def current_user():
    context = get_auth_context()
    user = db.session.scalar(
        select(User).where(User.id == context.user_id, User.is_deleted.is_(False))
    )
    if user is None:
        raise AuthenticationError()
    # 会话自描述接口：模块可见性按该用户对文档的实际权限动态计算。
    return success_response(
        user_summary(user, allowed_modules=allowed_modules_for_user(user))
    )


@api_v1.put("/users/me/ui-theme")
@bearer_auth_required
def update_ui_theme_route():
    # 本人主题偏好（契约1.20）：无需管理权限，任意登录用户可改自己。
    return success_response(update_ui_theme(_json_body()))


@api_v1.get("/users")
@bearer_auth_required
@admin_required
def users():
    page_request = PageRequest(
        page=_integer_query("page", 1),
        page_size=_integer_query("pageSize", 20),
    )
    data = list_users(
        organization_id=parse_string_id(
            request.args.get("organizationId"),
            name="organizationId",
        ),
        include_descendants=_boolean_query("includeDescendants", False),
        keyword=request.args.get("keyword"),
        status=request.args.get("status"),
        page_request=page_request,
    )
    return success_response(data)


@api_v1.post("/users")
@bearer_auth_required
@admin_required
def create_user_route():
    return success_response(create_user(_json_body()), status=201)


@api_v1.put("/users/reorder")
@bearer_auth_required
@admin_required
def reorder_users_route():
    return success_response(reorder_users(_json_body()))


@api_v1.put("/users/<id>")
@bearer_auth_required
@admin_required
def update_user_route(id: str):
    return success_response(update_user(_parse_user_id(id), _json_body()))


@api_v1.delete("/users/<id>")
@bearer_auth_required
@admin_required
def delete_user_route(id: str):
    return success_response(
        delete_user(
            _parse_user_id(id),
            _parse_delete_row_version(request.args.get("rowVersion")),
        )
    )


@api_v1.post("/users/<id>/reset-password")
@bearer_auth_required
@admin_required
def reset_user_password_route(id: str):
    return success_response(
        reset_user_password(_parse_user_id(id), _json_body())
    )