DataOps-platform/deployment/app/api/system/users.py

from __future__ import annotations

from flask import g, jsonify, request
from sqlalchemy import text
from sqlalchemy.exc import IntegrityError

from app import db
from app.api.system import bp
from app.commands.bootstrap_admin import validate_password
from app.core.common.identifiers import new_governance_uid
from app.core.system.auth import hash_password
from app.core.system.permissions import MANAGE_USERS, require_permissions
from app.models.result import failed, success


VALID_ROLES = {"admin", "editor", "viewer"}


def _active_admin_count(session) -> int:
    return int(
        session.execute(
            text(
                "SELECT COUNT(DISTINCT u.id) FROM public.users u "
                "JOIN public.user_roles ur ON ur.user_id = u.id "
                "JOIN public.roles r ON r.id = ur.role_id "
                "WHERE u.status = 'active' AND r.name = 'admin'"
            )
        ).scalar_one()
    )


def _is_active_admin(session, user_id: str) -> bool:
    return bool(
        session.execute(
            text(
                "SELECT 1 FROM public.users u "
                "JOIN public.user_roles ur ON ur.user_id = u.id "
                "JOIN public.roles r ON r.id = ur.role_id "
                "WHERE u.id = CAST(:id AS uuid) AND u.status = 'active' "
                "AND r.name = 'admin'"
            ),
            {"id": user_id},
        ).scalar()
    )


def _serialize_users(session) -> list[dict]:
    rows = session.execute(
        text(
            "SELECT u.id::text, u.username, u.display_name, u.status, "
            "u.created_at, u.last_login_at, "
            "COALESCE(array_agg(r.name ORDER BY r.name) FILTER "
            "(WHERE r.name IS NOT NULL), ARRAY[]::varchar[]) "
            "FROM public.users u "
            "LEFT JOIN public.user_roles ur ON ur.user_id = u.id "
            "LEFT JOIN public.roles r ON r.id = ur.role_id "
            "GROUP BY u.id ORDER BY u.created_at, u.username"
        )
    )
    return [
        {
            "id": row[0],
            "username": row[1],
            "display_name": row[2],
            "status": row[3],
            "created_at": row[4].isoformat(),
            "last_login_at": row[5].isoformat() if row[5] else None,
            "roles": list(row[6]),
        }
        for row in rows
    ]


@bp.route("/users", methods=["GET"])
@require_permissions(MANAGE_USERS)
def list_users():
    return jsonify(success(_serialize_users(db.session)))


@bp.route("/users", methods=["POST"])
@require_permissions(MANAGE_USERS)
def create_user():
    body = request.get_json(silent=True) or {}
    username = str(body.get("username") or "").strip()
    password = str(body.get("password") or "")
    roles = set(body.get("roles") or ["viewer"])
    if not username or not roles or not roles <= VALID_ROLES:
        return jsonify(failed("用户名或角色无效", code=400)), 400
    try:
        validate_password(password)
        user_id = new_governance_uid()
        db.session.execute(
            text(
                "INSERT INTO public.users (id, username, display_name, password_hash) "
                "VALUES (CAST(:id AS uuid), :username, :display_name, :password_hash)"
            ),
            {
                "id": user_id,
                "username": username,
                "display_name": body.get("display_name") or username,
                "password_hash": hash_password(password),
            },
        )
        db.session.execute(
            text(
                "INSERT INTO public.user_roles (user_id, role_id, assigned_by) "
                "SELECT CAST(:user_id AS uuid), id, CAST(:assigned_by AS uuid) "
                "FROM public.roles WHERE name = ANY(:roles)"
            ),
            {"user_id": user_id, "assigned_by": g.current_user["id"], "roles": sorted(roles)},
        )
        db.session.commit()
        return jsonify(success({"id": user_id}, "用户创建成功", code=201)), 201
    except (ValueError, IntegrityError) as exc:
        db.session.rollback()
        return jsonify(failed(str(exc), code=400)), 400


@bp.route("/users/<user_id>", methods=["PUT"])
@require_permissions(MANAGE_USERS)
def update_user(user_id: str):
    body = request.get_json(silent=True) or {}
    status = body.get("status")
    if status not in (None, "active", "disabled"):
        return jsonify(failed("用户状态无效", code=400)), 400
    if status == "disabled" and _is_active_admin(db.session, user_id):
        if _active_admin_count(db.session) <= 1:
            return jsonify(failed("不能停用最后一个有效管理员", code=409)), 409
    values = {"id": user_id}
    assignments = []
    if status is not None:
        assignments.append("status = :status")
        values["status"] = status
    if "display_name" in body:
        assignments.append("display_name = :display_name")
        values["display_name"] = str(body.get("display_name") or "")[:100] or None
    if "password" in body:
        validate_password(str(body["password"]))
        assignments.append("password_hash = :password_hash")
        values["password_hash"] = hash_password(str(body["password"]))
    if not assignments:
        return jsonify(failed("没有可更新字段", code=400)), 400
    result = db.session.execute(
        text(
            "UPDATE public.users SET " + ", ".join(assignments) +
            ", updated_at = CURRENT_TIMESTAMP WHERE id = CAST(:id AS uuid)"
        ),
        values,
    )
    if not result.rowcount:
        db.session.rollback()
        return jsonify(failed("用户不存在", code=404)), 404
    db.session.commit()
    return jsonify(success(message="用户更新成功"))


@bp.route("/users/<user_id>/roles", methods=["PUT"])
@require_permissions(MANAGE_USERS)
def update_user_roles(user_id: str):
    body = request.get_json(silent=True) or {}
    roles = set(body.get("roles") or [])
    if not roles or not roles <= VALID_ROLES:
        return jsonify(failed("角色无效", code=400)), 400
    if _is_active_admin(db.session, user_id) and "admin" not in roles:
        if _active_admin_count(db.session) <= 1:
            return jsonify(failed("不能移除最后一个有效管理员角色", code=409)), 409
    db.session.execute(
        text("DELETE FROM public.user_roles WHERE user_id = CAST(:id AS uuid)"),
        {"id": user_id},
    )
    result = db.session.execute(
        text(
            "INSERT INTO public.user_roles (user_id, role_id, assigned_by) "
            "SELECT CAST(:user_id AS uuid), id, CAST(:assigned_by AS uuid) "
            "FROM public.roles WHERE name = ANY(:roles)"
        ),
        {"user_id": user_id, "assigned_by": g.current_user["id"], "roles": sorted(roles)},
    )
    if result.rowcount != len(roles):
        db.session.rollback()
        return jsonify(failed("用户或角色不存在", code=404)), 404
    db.session.commit()
    return jsonify(success(message="角色更新成功"))