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

from __future__ import annotations

import json

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

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 _audit_access_control(action, resource_uid, status, safe_detail):
    db.session.execute(
        text(
            "INSERT INTO public.access_control_audit_events "
            "(uid,action,actor_uid,resource_type,resource_uid,status,safe_detail) VALUES "
            "(CAST(:uid AS uuid),:action,CAST(:actor AS uuid),'user',:resource_uid,:status,CAST(:detail AS jsonb))"
        ),
        {
            "uid": new_governance_uid(),
            "action": action,
            "actor": g.current_user["id"],
            "resource_uid": resource_uid,
            "status": status,
            "detail": json.dumps(safe_detail, ensure_ascii=False),
        },
    )
    try:
        with db.session.begin_nested():
            db.session.execute(
                text(
                    "INSERT INTO public.identity_audit_events "
                    "(uid,event_type,outcome,actor_uid,resource_type,resource_uid,safe_detail) VALUES "
                    "(CAST(:uid AS uuid),:action,:outcome,CAST(:actor AS uuid),'user',:resource_uid,CAST(:detail AS jsonb))"
                ),
                {"uid": new_governance_uid(), "action": action, "outcome": "success" if status == "success" else "failure",
                 "actor": g.current_user["id"], "resource_uid": resource_uid,
                 "detail": json.dumps(safe_detail, ensure_ascii=False)},
            )
    except ProgrammingError:
        pass


def _revoke_identity_sessions(user_id: str, reason: str) -> None:
    try:
        with db.session.begin_nested():
            db.session.execute(
                text("UPDATE public.enterprise_identity_links SET token_version=token_version+1,updated_at=CURRENT_TIMESTAMP WHERE user_uid=CAST(:id AS uuid)"),
                {"id": user_id},
            )
            db.session.execute(
                text("UPDATE public.identity_sessions SET status='revoked',revoke_reason=:reason,revoked_at=CURRENT_TIMESTAMP WHERE user_uid=CAST(:id AS uuid) AND status='active'"),
                {"id": user_id, "reason": reason},
            )
    except ProgrammingError:
        pass


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)},
        )
        _audit_access_control(
            "user_created", user_id, "success", {"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)
        and _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
    _audit_access_control(
        "user_updated",
        user_id,
        "success",
        {
            "changed_fields": sorted(
                key for key in ("status", "display_name", "password") if key in body
            ),
            "password_value_retained": False,
        },
    )
    if status == "disabled":
        _revoke_identity_sessions(user_id, "account_disabled")
    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
        and _active_admin_count(db.session) <= 1
    ):
        return jsonify(failed("不能移除最后一个有效管理员角色", code=409)), 409
    previous_roles = list(
        db.session.execute(
            text(
                "SELECT r.name FROM public.user_roles ur JOIN public.roles r ON r.id=ur.role_id "
                "WHERE ur.user_id=CAST(:id AS uuid) ORDER BY r.name"
            ),
            {"id": user_id},
        ).scalars()
    )
    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
    _audit_access_control(
        "user_roles_updated",
        user_id,
        "success",
        {"before_roles": previous_roles, "after_roles": sorted(roles)},
    )
    if sorted(previous_roles) != sorted(roles):
        _revoke_identity_sessions(user_id, "roles_changed")
    db.session.commit()
    return jsonify(success(message="角色更新成功"))