from django.conf import settings from django.core.exceptions import ObjectDoesNotExist from rest_framework import status from rest_framework.generics import CreateAPIView from rest_framework.permissions import AllowAny, IsAuthenticated from rest_framework.response import Response from rest_framework.views import APIView from rest_framework_simplejwt.exceptions import TokenError from rest_framework_simplejwt.tokens import RefreshToken from apps.account.filters.profile_filters import ProfileListFilter from apps.account.models import User, UserCode from apps.account.serializers import account_serializers from base.views import BaseModelViewSet from account.utils.send_email import sending_email from account.utils.send_otp import send_fast2sms_otp class LoginOrRegistrationView(CreateAPIView): """ This is the LoginOrRegistrationView. It handles user login or registration through a POST request. Attributes: permission_classes (tuple): Permissions required to access this view (AllowAny for public access). serializer_class (UserLoginOrRegistrationSerializer): The serializer used for input/output data. http_method_names (list): The allowed HTTP methods, in this case, only POST is allowed. Methods: post(request): Handles user login or registration logic. """ permission_classes = (AllowAny,) serializer_class = account_serializers.UserLoginOrRegistrationSerializer http_method_names = ["post"] def post(self, request, *args, **kwargs): """ Handles user login or registration logic. Args: request (HttpRequest): The HTTP request containing user data. Returns: Response: The serialized user data or error response. """ serializer = account_serializers.UserLoginOrRegistrationSerializer( data=request.data ) serializer.is_valid(raise_exception=True) mobile_number = request.data.get("mobile_number") mobile_country_code = request.data.get("mobile_country_code") email = request.data.get("email") if mobile_number and mobile_country_code: try: # Check user already exist OR not user = User.objects.get( mobile_number=mobile_number, mobile_country_code=mobile_country_code ) except ObjectDoesNotExist: # User does not exist then create one user = User.objects.create( mobile_number=mobile_number, mobile_country_code=mobile_country_code, email=email) if email: try: # Check user already exist OR not user = User.objects.get(email=email) except ObjectDoesNotExist: # User does not exist then create one user = User.objects.create_user(email=email, mobile_number=mobile_number, mobile_country_code=mobile_country_code) # Create new otp code for user user_code = UserCode.objects.create_or_update_code(user=user) try: if email: sending_email( subject="Your EVChargy Login OTP", otp_code=user_code.code, from_email=settings.FROM_EMAIL, recipient_list=[email], ) if mobile_number and mobile_country_code: send_fast2sms_otp( mobile_number=mobile_number, otp=user_code.code, ) except Exception as e: return Response({"custom_error": str(e)}, status=status.HTTP_400_BAD_REQUEST) serializer = account_serializers.UserLoginOrRegistrationSerializer(user) response = { "message": "OTP sent successfully", **serializer.data } return Response(response, status=status.HTTP_200_OK) class VerifyLoginOrRegistrationOtpView(CreateAPIView): """ This view handles OTP verification for user login or registration. Attributes: permission_classes (tuple): Permissions required to access this view (AllowAny for public access). http_method_names (list): The allowed HTTP methods (only POST is allowed). Methods: post(request): Handles user OTP verification logic. """ permission_classes = (AllowAny,) http_method_names = ["post"] serializer_class = account_serializers.VerifyOtpSerializer def post(self, request, *args, **kwargs): """ Handles OTP verification logic for user login or registration. Args: request (HttpRequest): The HTTP request containing user data. Returns: Response: The serialized user data or an error response. """ serializer = account_serializers.VerifyOtpSerializer(data=request.data) serializer.is_valid(raise_exception=True) mobile_number = request.data.get("mobile_number") mobile_country_code = request.data.get("mobile_country_code") email= request.data.get("email") otp = request.data.get("otp") # Check if a user exists with the provided mobile number and country code if mobile_number and mobile_country_code: try: user = User.objects.get( mobile_number=mobile_number, mobile_country_code=mobile_country_code ) except ObjectDoesNotExist: # Return an error if the user does not exist return Response( { "message": "User Not Found: The mobile number you provided does not exist in our records." }, status=status.HTTP_400_BAD_REQUEST, ) if email: try: user = User.objects.get(email=email) except ObjectDoesNotExist: return Response( { "message": "User Not Found: The email you provided does not exist in our records." }, status=status.HTTP_400_BAD_REQUEST, ) # Get a UserCode object with the user and OTP fields try: user_code = UserCode.objects.get(user=user, code=otp) if user_code.is_expired(): return Response( {"message": "OTP has been expired please try again!"}, status=status.HTTP_400_BAD_REQUEST, ) except ObjectDoesNotExist: # Return an error if the object does not exist return Response( {"message": "Invalid OTP! Please enter the correct OTP."}, status=status.HTTP_400_BAD_REQUEST, ) # Generate token for authentication refresh = RefreshToken.for_user(user) response_data = { "id": user.id, "is_new_user": user.name is None, "mobile_number": user.mobile_number, "mobile_country_code": user.mobile_country_code, "email": user.email, "token": str(refresh.access_token), "refresh": str(refresh), } return Response(response_data) class ResendOtpView(CreateAPIView): """ This view handles user login or registration and OTP resending for authentication. Attributes: permission_classes (tuple): Permissions required to access this view (AllowAny for public access). serializer_class (UserLoginOrRegistrationSerializer): The serializer used for input/output data. http_method_names (list): The allowed HTTP methods, in this case, only POST is allowed. Methods: post(request): Handles user login or registration and OTP resending logic. This view provides functionality for user login or registration, as well as resending OTP codes for authentication purposes. """ permission_classes = (AllowAny,) serializer_class = account_serializers.UserLoginOrRegistrationSerializer http_method_names = ["post"] def post(self, request, *args, **kwargs): """ Handles user login or registration and OTP resending logic. Args: request (HttpRequest): The HTTP request containing user data. Returns: Response: The serialized user data or error response. This method processes user login or registration and resending OTP logic based on the provided mobile number and mobile country code. """ serializer = account_serializers.UserLoginOrRegistrationSerializer( data=request.data ) serializer.is_valid(raise_exception=True) mobile_number = request.data.get("mobile_number") mobile_country_code = request.data.get("mobile_country_code") email = request.data.get("email") if mobile_number and mobile_country_code: try: # Check user exist OR not user = User.objects.get( mobile_number=mobile_number, mobile_country_code=mobile_country_code ) except ObjectDoesNotExist: return Response( {"message": "The mobile number you entered is invalid"}, status=status.HTTP_400_BAD_REQUEST, ) if email: try: # Check user exist OR not user = User.objects.get(email=email) except ObjectDoesNotExist: return Response( {"message": "The email you entered is invalid"}, status=status.HTTP_400_BAD_REQUEST, ) # Create new otp code for user user_code = UserCode.objects.create_or_update_code(user=user) try: if email: sending_email( subject="Resend mail for Login/Registration OTP", otp_code=user_code.code, from_email=settings.FROM_EMAIL, recipient_list=[email], ) if mobile_number and mobile_country_code: send_fast2sms_otp( mobile_number=mobile_number, otp=user_code.code, ) except Exception as e: return Response({"custom_error": str(e)}, status=status.HTTP_400_BAD_REQUEST) serializer = account_serializers.UserLoginOrRegistrationSerializer(user) response_data = serializer.data response_data["message"] = "OTP Sent Successfully to your email address" return Response(response_data) class ProfileModelViewSet(BaseModelViewSet): """ This view handles OTP verification for user login or registration and user profile retrieval. Attributes: permission_classes (tuple): Permissions required to access this view (IsAuthenticated for authenticated users). http_method_names (list): The allowed HTTP methods (PUT and GET are allowed). Methods: list(request): Retrieves a list of user profiles (Not allowed, returns an error response). retrieve(request): Retrieves a specific user's profile details, with permission checks. """ serializer_class = account_serializers.ProfileSerializer queryset = User.objects.all() filterset_class = ProfileListFilter search_fields = ["name", "email"] http_method_names = ["patch", "get", "delete"] def list(self, request, *args, **kwargs): """ Retrieves a list of user data based on the request parameters. Args: request (HttpRequest): The HTTP request object. *args: Variable-length argument list. **kwargs: Arbitrary keyword arguments. Returns: Response: A response containing a list of serialized user data. This method retrieves a list of users records based on the provided request parameters. The list of stations is paginated for efficient retrieval. """ if request.user.is_superuser: queryset = self.filter_queryset(self.get_queryset()) page = self.paginate_queryset(queryset) if page is not None: serializer = self.get_serializer(page, many=True) return self.get_paginated_response(serializer.data) else: serializer = self.get_serializer(queryset, many=True) return Response(serializer.data, status=status.HTTP_200_OK) else: return Response([], status=status.HTTP_200_OK) def retrieve(self, request, *args, **kwargs): """ Retrieve a specific user's profile details, with permission checks. Args: request (HttpRequest): The HTTP request for retrieving a user's profile. Returns: Response: The user's profile details or an error response if permission is denied. This method retrieves the profile details of a specific user, with permission checks to ensure that the requesting user has access to the requested profile. """ instance = self.get_object() if instance != request.user and not request.user.is_superuser: return Response( {"error": "You do not have permission to access this user's profile."}, status=status.HTTP_403_FORBIDDEN, ) serializer = self.serializer_class(instance, context={"request": request}) return Response(serializer.data) def destroy(self, request, *args, **kwargs): """ Handles the removal of account. Args: request (HttpRequest): The HTTP request object containing information about the account to delete. *args: Variable-length argument list. **kwargs: Arbitrary keyword arguments. Returns: Response: A response indicating the success or failure of the account deletion operation. """ instance = self.get_object() if instance != request.user and not request.user.is_superuser: return Response( {"message": "You can't delete other user account"}, status=status.HTTP_400_BAD_REQUEST, ) self.perform_destroy(instance) return Response( {"message": "Account deleted successfully"}, status=status.HTTP_204_NO_CONTENT, ) class UserLogoutView(APIView): """ This is the UserLogoutView class responsible for handling user logout through a POST request. It allows users to log out and blacklist the provided refresh token, ensuring the token can no longer be used. Attributes: permission_classes (tuple): Permissions required to access this view (IsAuthenticated for authenticated users). Methods: post(request): Handles user logout and token blacklisting logic. """ permission_classes = (IsAuthenticated,) def post(self, request): """ Handles user logout and token blacklisting logic. Args: request (HttpRequest): The HTTP request containing user data, including the refresh token. Returns: Response: A response indicating the success or failure of the logout operation. """ try: refresh_token = request.data.get("refresh") if refresh_token: token = RefreshToken(refresh_token) token.blacklist() return Response(status=status.HTTP_200_OK) except TokenError: return Response( data={ "message": "Invalid token", }, status=status.HTTP_400_BAD_REQUEST, ) except Exception as e: import traceback print(traceback.format_exc()) return Response( data={ "message": "There was some issue while logging out", "traceback": str(traceback.format_exc()), }, status=status.HTTP_400_BAD_REQUEST, )