MaxKB/apps/users/views/user.py

# coding=utf-8
"""
    @project: qabot
    @Author：虎
    @file： user.py
    @date：2023/9/4 10:57
    @desc:
"""
from django.core import cache
from django.utils.translation import gettext_lazy as _
from drf_yasg import openapi
from drf_yasg.utils import swagger_auto_schema
from rest_framework.decorators import action
from rest_framework.decorators import permission_classes
from rest_framework.permissions import AllowAny
from rest_framework.views import APIView
from rest_framework.views import Request

from common.auth.authenticate import TokenAuth
from common.auth.authentication import has_permissions
from common.constants.permission_constants import PermissionConstants, CompareConstants, ViewPermission, RoleConstants
from common.log.log import log
from common.response import result
from common.util.common import encryption
from smartdoc.settings import JWT_AUTH
from users.serializers.user_serializers import RegisterSerializer, LoginSerializer, CheckCodeSerializer, \
    RePasswordSerializer, \
    SendEmailSerializer, UserProfile, UserSerializer, UserManageSerializer, UserInstanceSerializer, SystemSerializer, \
    SwitchLanguageSerializer, CaptchaSerializer
from users.views.common import get_user_operation_object, get_re_password_details

user_cache = cache.caches['user_cache']
token_cache = cache.caches['token_cache']


class Profile(APIView):
    @action(methods=['GET'], detail=False)
    @swagger_auto_schema(operation_summary=_("Get MaxKB related information"),
                         operation_id=_("Get MaxKB related information"),
                         responses=result.get_api_response(SystemSerializer.get_response_body_api()),
                         tags=[_('System parameters')])
    def get(self, request: Request):
        return result.success(SystemSerializer.get_profile())


class User(APIView):
    authentication_classes = [TokenAuth]

    @action(methods=['GET'], detail=False)
    @swagger_auto_schema(operation_summary=_("Get current user information"),
                         operation_id=_("Get current user information"),
                         responses=result.get_api_response(UserProfile.get_response_body_api()),
                         tags=[])
    @has_permissions(PermissionConstants.USER_READ)
    def get(self, request: Request):
        return result.success(UserProfile.get_user_profile(request.user))

    class Query(APIView):
        authentication_classes = [TokenAuth]

        @action(methods=['GET'], detail=False)
        @swagger_auto_schema(operation_summary=_("Get user list"),
                             operation_id=_("Get user list"),
                             manual_parameters=UserSerializer.Query.get_request_params_api(),
                             responses=result.get_api_array_response(UserSerializer.Query.get_response_body_api()),
                             tags=[_("User management")])
        @has_permissions(PermissionConstants.USER_READ)
        def get(self, request: Request):
            return result.success(
                UserSerializer.Query(data={'email_or_username': request.query_params.get('email_or_username')}).list())


class SwitchUserLanguageView(APIView):
    authentication_classes = [TokenAuth]

    @action(methods=['POST'], detail=False)
    @swagger_auto_schema(operation_summary=_("Switch Language"),
                         operation_id=_("Switch Language"),
                         request_body=openapi.Schema(
                             type=openapi.TYPE_OBJECT,
                             required=['language'],
                             properties={
                                 'language': openapi.Schema(type=openapi.TYPE_STRING, title=_("language"),
                                                            description=_("language")),
                             }
                         ),
                         responses=RePasswordSerializer().get_response_body_api(),
                         tags=[_("User management")])
    @log(menu='User management', operate='Switch Language',
         get_operation_object=lambda r, k: {'name': r.user.username})
    def post(self, request: Request):
        data = {**request.data, 'user_id': request.user.id}
        return result.success(SwitchLanguageSerializer(data=data).switch())


class ResetCurrentUserPasswordView(APIView):
    authentication_classes = [TokenAuth]

    @action(methods=['POST'], detail=False)
    @swagger_auto_schema(operation_summary=_("Modify current user password"),
                         operation_id=_("Modify current user password"),
                         request_body=openapi.Schema(
                             type=openapi.TYPE_OBJECT,
                             required=['email', 'code', "password", 're_password'],
                             properties={
                                 'code': openapi.Schema(type=openapi.TYPE_STRING, title=_("Verification code"),
                                                        description=_("Verification code")),
                                 'password': openapi.Schema(type=openapi.TYPE_STRING, title=_("Password"),
                                                            description=_("Password")),
                                 're_password': openapi.Schema(type=openapi.TYPE_STRING, title=_("Password"),
                                                               description=_("Password"))
                             }
                         ),
                         responses=RePasswordSerializer().get_response_body_api(),
                         tags=[_("User management")])
    @log(menu='User management', operate='Modify current user password',
         get_operation_object=lambda r, k: {'name': r.user.username},
         get_details=get_re_password_details)
    def post(self, request: Request):
        data = {'email': request.user.email}
        data.update(request.data)
        serializer_obj = RePasswordSerializer(data=data)
        if serializer_obj.reset_password():
            token_cache.delete(request.META.get('HTTP_AUTHORIZATION'))
            return result.success(True)
        return result.error(_("Failed to change password"))


class SendEmailToCurrentUserView(APIView):
    authentication_classes = [TokenAuth]

    @action(methods=['POST'], detail=False)
    @permission_classes((AllowAny,))
    @swagger_auto_schema(operation_summary=_("Send email to current user"),
                         operation_id=_("Send email to current user"),
                         responses=SendEmailSerializer().get_response_body_api(),
                         tags=[_("User management")])
    @log(menu='User management', operate='Send email to current user',
         get_operation_object=lambda r, k: {'name': r.user.username})
    def post(self, request: Request):
        serializer_obj = SendEmailSerializer(data={'email': request.user.email, 'type': "reset_password"})
        if serializer_obj.is_valid(raise_exception=True):
            return result.success(serializer_obj.send())


class Logout(APIView):
    authentication_classes = [TokenAuth]

    @action(methods=['POST'], detail=False)
    @permission_classes((AllowAny,))
    @swagger_auto_schema(operation_summary=_("Sign out"),
                         operation_id=_("Sign out"),
                         responses=SendEmailSerializer().get_response_body_api(),
                         tags=[_("User management")])
    @log(menu='User management', operate='Sign out',
         get_operation_object=lambda r, k: {'name': r.user.username})
    def post(self, request: Request):
        token_cache.delete(request.META.get('HTTP_AUTHORIZATION'))
        return result.success(True)


def _get_details(request):
    path = request.path
    body = request.data
    query = request.query_params
    return {
        'path': path,
        'body': {**body, 'password': encryption(body.get('password', ''))},
        'query': query
    }


class CaptchaView(APIView):

    @action(methods=['GET'], detail=False)
    @swagger_auto_schema(operation_summary=_("Obtain graphical captcha"),
                         operation_id=_("Obtain graphical captcha"),
                         responses=CaptchaSerializer().get_response_body_api(),
                         security=[],
                         tags=[_("User management")])
    def get(self, request: Request):
        return result.success(CaptchaSerializer().generate())


class Login(APIView):

    @action(methods=['POST'], detail=False)
    @swagger_auto_schema(operation_summary=_("Log in"),
                         operation_id=_("Log in"),
                         request_body=LoginSerializer().get_request_body_api(),
                         responses=LoginSerializer().get_response_body_api(),
                         security=[],
                         tags=[_("User management")])
    @log(menu='User management', operate='Log in', get_user=lambda r: {'username': r.data.get('username', None)},
         get_details=_get_details,
         get_operation_object=lambda r, k: {'name': r.data.get('username')})
    def post(self, request: Request):
        login_request = LoginSerializer(data=request.data)
        # 校验请求参数
        user = login_request.is_valid(raise_exception=True)
        token = login_request.get_user_token()
        token_cache.set(token, user, timeout=JWT_AUTH['JWT_EXPIRATION_DELTA'])
        return result.success(token)


class Register(APIView):

    @action(methods=['POST'], detail=False)
    @permission_classes((AllowAny,))
    @swagger_auto_schema(operation_summary=_("User registration"),
                         operation_id=_("User registration"),
                         request_body=RegisterSerializer().get_request_body_api(),
                         responses=RegisterSerializer().get_response_body_api(),
                         security=[],
                         tags=[_("User management")])
    @log(menu='User management', operate='User registration',
         get_operation_object=lambda r, k: {'name': r.data.get('username', None)},
         get_user=lambda r: {'user_name': r.data.get('username', None)})
    def post(self, request: Request):
        serializer_obj = RegisterSerializer(data=request.data)
        if serializer_obj.is_valid(raise_exception=True):
            serializer_obj.save()
            return result.success(_("Registration successful"))


class RePasswordView(APIView):

    @action(methods=['POST'], detail=False)
    @permission_classes((AllowAny,))
    @swagger_auto_schema(operation_summary=_("Change password"),
                         operation_id=_("Change password"),
                         request_body=RePasswordSerializer().get_request_body_api(),
                         responses=RePasswordSerializer().get_response_body_api(),
                         security=[],
                         tags=[_("User management")])
    @log(menu='User management', operate='Change password',
         get_operation_object=lambda r, k: {'name': r.data.get('email', None)},
         get_user=lambda r: {'user_name': None, 'email': r.data.get('email', None)},
         get_details=get_re_password_details)
    def post(self, request: Request):
        serializer_obj = RePasswordSerializer(data=request.data)
        return result.success(serializer_obj.reset_password())


class CheckCode(APIView):

    @action(methods=['POST'], detail=False)
    @permission_classes((AllowAny,))
    @swagger_auto_schema(operation_summary=_("Check whether the verification code is correct"),
                         operation_id=_("Check whether the verification code is correct"),
                         request_body=CheckCodeSerializer().get_request_body_api(),
                         responses=CheckCodeSerializer().get_response_body_api(),
                         security=[],
                         tags=[_("User management")])
    @log(menu='User management', operate='Check whether the verification code is correct',
         get_operation_object=lambda r, k: {'name': r.data.get('email', None)},
         get_user=lambda r: {'user_name': None, 'email': r.data.get('email', None)})
    def post(self, request: Request):
        return result.success(CheckCodeSerializer(data=request.data).is_valid(raise_exception=True))


class SendEmail(APIView):

    @action(methods=['POST'], detail=False)
    @swagger_auto_schema(operation_summary=_("Send email"),
                         operation_id=_("Send email"),
                         request_body=SendEmailSerializer().get_request_body_api(),
                         responses=SendEmailSerializer().get_response_body_api(),
                         security=[],
                         tags=[_("User management")])
    @log(menu='User management', operate='Send email',
         get_operation_object=lambda r, k: {'name': r.data.get('email', None)},
         get_user=lambda r: {'user_name': None, 'email': r.data.get('email', None)})
    def post(self, request: Request):
        serializer_obj = SendEmailSerializer(data=request.data)
        if serializer_obj.is_valid(raise_exception=True):
            return result.success(serializer_obj.send())


class UserManage(APIView):
    authentication_classes = [TokenAuth]

    @action(methods=['POST'], detail=False)
    @swagger_auto_schema(operation_summary=_("Add user"),
                         operation_id=_("Add user"),
                         request_body=UserManageSerializer.UserInstance.get_request_body_api(),
                         responses=result.get_api_response(UserInstanceSerializer.get_response_body_api()),
                         tags=[_("User management")]
                         )
    @has_permissions(ViewPermission(
        [RoleConstants.ADMIN],
        [PermissionConstants.USER_READ],
        compare=CompareConstants.AND))
    @log(menu='User management', operate='Add user',
         get_operation_object=lambda r, k: {'name': r.data.get('username', None)})
    def post(self, request: Request):
        return result.success(UserManageSerializer().save(request.data))

    class Page(APIView):
        authentication_classes = [TokenAuth]

        @action(methods=['GET'], detail=False)
        @swagger_auto_schema(operation_summary=_("Get user paginated list"),
                             operation_id=_("Get user paginated list"),
                             tags=[_("User management")],
                             manual_parameters=UserManageSerializer.Query.get_request_params_api(),
                             responses=result.get_page_api_response(UserInstanceSerializer.get_response_body_api()),
                             )
        @has_permissions(ViewPermission(
            [RoleConstants.ADMIN],
            [PermissionConstants.USER_READ],
            compare=CompareConstants.AND))
        def get(self, request: Request, current_page, page_size):
            d = UserManageSerializer.Query(
                data={'email_or_username': request.query_params.get('email_or_username', None),
                      'user_id': str(request.user.id)})
            return result.success(d.page(current_page, page_size))

    class RePassword(APIView):
        authentication_classes = [TokenAuth]

        @action(methods=['PUT'], detail=False)
        @swagger_auto_schema(operation_summary=_("Change password"),
                             operation_id=_("Change password"),
                             manual_parameters=UserInstanceSerializer.get_request_params_api(),
                             request_body=UserManageSerializer.RePasswordInstance.get_request_body_api(),
                             responses=result.get_default_response(),
                             tags=[_("User management")])
        @has_permissions(ViewPermission(
            [RoleConstants.ADMIN],
            [PermissionConstants.USER_READ],
            compare=CompareConstants.AND))
        @log(menu='User management', operate='Change password',
             get_operation_object=lambda r, k: get_user_operation_object(k.get('user_id')),
             get_details=get_re_password_details)
        def put(self, request: Request, user_id):
            return result.success(
                UserManageSerializer.Operate(data={'id': user_id}).re_password(request.data, with_valid=True))

    class Operate(APIView):
        authentication_classes = [TokenAuth]

        @action(methods=['DELETE'], detail=False)
        @swagger_auto_schema(operation_summary=_("Delete user"),
                             operation_id=_("Delete user"),
                             manual_parameters=UserInstanceSerializer.get_request_params_api(),
                             responses=result.get_default_response(),
                             tags=[_("User management")])
        @has_permissions(ViewPermission(
            [RoleConstants.ADMIN],
            [PermissionConstants.USER_READ],
            compare=CompareConstants.AND))
        @log(menu='User management', operate='Delete user',
             get_operation_object=lambda r, k: get_user_operation_object(k.get('user_id')))
        def delete(self, request: Request, user_id):
            return result.success(UserManageSerializer.Operate(data={'id': user_id}).delete(with_valid=True))

        @action(methods=['GET'], detail=False)
        @swagger_auto_schema(operation_summary=_("Get user information"),
                             operation_id=_("Get user information"),
                             manual_parameters=UserInstanceSerializer.get_request_params_api(),
                             responses=result.get_api_response(UserInstanceSerializer.get_response_body_api()),
                             tags=[_("User management")]
                             )
        @has_permissions(ViewPermission(
            [RoleConstants.ADMIN],
            [PermissionConstants.USER_READ],
            compare=CompareConstants.AND))
        def get(self, request: Request, user_id):
            return result.success(UserManageSerializer.Operate(data={'id': user_id}).one(with_valid=True))

        @action(methods=['PUT'], detail=False)
        @swagger_auto_schema(operation_summary=_("Update user information"),
                             operation_id=_("Update user information"),
                             manual_parameters=UserInstanceSerializer.get_request_params_api(),
                             request_body=UserManageSerializer.UserEditInstance.get_request_body_api(),
                             responses=result.get_api_response(UserInstanceSerializer.get_response_body_api()),
                             tags=[_("User management")]
                             )
        @has_permissions(ViewPermission(
            [RoleConstants.ADMIN],
            [PermissionConstants.USER_READ],
            compare=CompareConstants.AND))
        @log(menu='User management', operate='Update user information',
             get_operation_object=lambda r, k: get_user_operation_object(k.get('user_id')))
        def put(self, request: Request, user_id):
            return result.success(
                UserManageSerializer.Operate(data={'id': user_id}).edit(request.data, with_valid=True))


class UserListView(APIView):
    authentication_classes = [TokenAuth]

    @swagger_auto_schema(operation_summary=_("Get user list by type"),
                         operation_id=_("Get user list by type"),
                         manual_parameters=UserSerializer.Query.get_request_params_api(),
                         responses=result.get_api_array_response(UserSerializer.Query.get_response_body_api()),
                         tags=[_("User management")])
    @has_permissions(PermissionConstants.USER_READ)
    def get(self, request: Request, type):
        return result.success(UserSerializer().listByType(type, request.user.id))