model_train_dm/backend/apps/iam/views/person.py

import base64
from datetime import datetime

from django.conf import settings
from django.contrib.auth.hashers import make_password
from django.db.models import Max
from django.shortcuts import get_object_or_404
from gmssl import sm4
from rest_framework import status
from rest_framework.decorators import api_view, permission_classes
from rest_framework.pagination import PageNumberPagination
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response

from apps.common.serializers import AuthPersonSerializer
from apps.core.models import AIAnnotateTeamMember, AuthPerson


@api_view(["GET"])
@permission_classes([IsAuthenticated])
def get_person_list(request):
    organization_id = request.GET.get("organization_id", None)
    team_id = request.GET.get("team_id", None)
    username = request.GET.get("username", None)
    last_name = request.GET.get("last_name", None)
    page_size = request.GET.get("page_size", None)

    team_members = AIAnnotateTeamMember.objects.filter(team_id=team_id)
    member_accounts = [team_member.member_account for team_member in team_members]

    filter_kwargs = {}
    if organization_id:
        filter_kwargs["organization_id"] = organization_id
    if username:
        filter_kwargs["username__icontains"] = username
    if last_name:
        filter_kwargs["last_name__icontains"] = last_name

    persons = AuthPerson.objects.filter(**filter_kwargs).order_by("id")
    if member_accounts:
        persons = persons.exclude(username__in=member_accounts)

    paginator = PageNumberPagination()
    paginator.page_size = page_size or 10

    result_page = paginator.paginate_queryset(persons, request)
    serializer = AuthPersonSerializer(result_page, many=True)
    return paginator.get_paginated_response(serializer.data)


@api_view(["GET"])
@permission_classes([IsAuthenticated])
def read_person(request):
    person_id = request.GET.get("id", None)
    person = get_object_or_404(AuthPerson, id=person_id)
    serializer = AuthPersonSerializer(person)
    return Response(serializer.data)


@api_view(["POST"])
@permission_classes([IsAuthenticated])
def add_person(request):
    data = request.data
    max_id = AuthPerson.objects.aggregate(Max("id"))["id__max"]
    data["id"] = (max_id + 1) if max_id is not None else 1
    data["date_joined"] = datetime.now()
    data["password"] = make_password("123456")
    serializer = AuthPersonSerializer(data=data)

    if serializer.is_valid():
        serializer.save()
        return Response({"message": "Person added successfully"})

    return Response(
        {"error": "Invalid data", "details": serializer.errors},
        status=status.HTTP_400_BAD_REQUEST,
    )


@api_view(["PUT"])
@permission_classes([IsAuthenticated])
def update_person(request):
    data = request.data
    person = get_object_or_404(AuthPerson, id=data.get("id"))
    serializer = AuthPersonSerializer(instance=person, data=request.data, partial=True)

    if serializer.is_valid():
        serializer.save()
        return Response({"message": "Person updated successfully"})

    return Response(
        {"error": "Invalid data", "details": serializer.errors},
        status=status.HTTP_400_BAD_REQUEST,
    )


@api_view(["DELETE"])
@permission_classes([IsAuthenticated])
def delete_person(request):
    person_id = request.GET.get("id", None)
    person = get_object_or_404(AuthPerson, id=person_id)
    person.delete()
    return Response({"message": "Person deleted successfully"})


@api_view(["POST"])
@permission_classes([IsAuthenticated])
def update_password(request):
    username = request.data.get("username")
    encrypted_password = request.data.get("password")

    try:
        password = decrypt_password(
            encrypted_password, getattr(settings, "private_key", "")
        )
        AuthPerson.objects.filter(username=username).update(password=make_password(password))
        return Response({"code": "0", "success": "用户密码修改成功！"})
    except Exception:
        return Response({"code": "1", "error": "用户密码更新失败！"})


def decrypt_password(encrypted_text, private_key):
    cipher = sm4.CryptSM4()
    cipher.set_key(private_key.encode(), True)
    encrypted_bytes = base64.b64decode(encrypted_text)
    decrypted_bytes = cipher.crypt_ecb(encrypted_bytes)
    return decrypted_bytes.decode()