import logging import requests import json from rest_framework.generics import CreateAPIView, RetrieveUpdateAPIView, GenericAPIView, RetrieveAPIView, UpdateAPIView, ListAPIView from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status from rest_framework.permissions import AllowAny, IsAuthenticated from rest_framework.authtoken.models import Token from rest_framework.exceptions import AuthenticationFailed from django.utils.translation import gettext_lazy as _ from django.shortcuts import get_object_or_404 from rest_framework.authtoken.models import Token from django.utils import timezone from rest_framework.authentication import TokenAuthentication from django.contrib.auth import authenticate from phonenumbers import parse, region_code_for_number from drf_yasg.utils import swagger_auto_schema from drf_yasg import openapi from rest_framework.exceptions import ValidationError from utils.exceptions import InvaliedCodeVrify, ExpiredCodeException, ServiceUnavailableException from apps.account.models import User from apps.account.serializers import UserRegisterSerializer, UserProfileSerializer, UserVerifySerializer, UserLoginSerializer, UserRecoverPasswordSerializer, UserResetPasswordSerializer from utils.redis import RedisManager from utils.exceptions import AppAPIException from utils import send_email, is_valid_email from config.settings import base as settings from apps.account.permissions import IsActiveUser from apps.account.doc import * logger = logging.getLogger(__name__) class UserRegisterView(CreateAPIView): permission_classes = [AllowAny] serializer_class = UserRegisterSerializer @swagger_auto_schema( operation_description=doc_register(), request_body=UserRegisterSerializer, ) def post(self, request): serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) data = serializer.data code = RedisManager.generate_otp_code() logger.info(f"phone= {data['email']}") print(f' send {code}/{data["email"]}') phone_number = RedisManager().add_to_redis(code, **data) send_email([data['email']], code) password = data.pop('password') return Response( data= { "user": data, "message": "The otp code was sent to the user's email" }, status=status.HTTP_202_ACCEPTED, ) class UserVerifyView(CreateAPIView): permission_classes = [AllowAny] serializer_class = UserVerifySerializer @swagger_auto_schema( operation_description=doc_verify(), request_body=UserVerifySerializer, ) def post(self, request, *args, **kwargs): return super().post(request, *args, **kwargs) def create(self, request, *args, **kwargs): serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) data = serializer.data try: verify_data = RedisManager().get_by_redis(data['email']) if not verify_data: raise ValidationError({"code": "Verification data not found or expired."}) # raise ExpiredCodeException("Verification data not found or expired.") except (ServiceUnavailableException) as e: return AppAPIException({"message": str(e)}, status_code=e.status_code) except ExpiredCodeException: # raise ExpiredCodeException("The verification code has expired.") raise ValidationError({"code": "The verification code has expired."}) code = self.valied_code(data['code'], verify_data['code']) del verify_data['code'] user = self.perform_create( email=serializer.data['email'],**verify_data ) Token.objects.filter(user=user).delete() token = Token.objects.create(user=user) return Response(data={ 'token': str(token), 'user_id': user.id, 'phone_number': str(user.phone_number), 'email': str(user.email), 'fullname': str(user.fullname), 'avatar': str(user.avatar) if user.avatar else None }, status=status.HTTP_201_CREATED) def valied_code(self, current_code, save_code): if (current_code and save_code) and ( current_code != save_code): # raise InvaliedCodeVrify() raise ValidationError({"code": "code notfound"}) return current_code def perform_create(self, *args, **kwargs): email = kwargs.get('email') user = User.objects.filter(email=email).first() if user: if kwargs['password']: user.is_active = True user.deletion_date = None user.last_login = timezone.now() user.set_password(kwargs['password']) user.save() else: user = User.objects.create(**kwargs) user.set_password(kwargs['password']) user.last_login = timezone.now() user.is_active = True user.save() return user class UserLoginView(CreateAPIView): permission_classes = [AllowAny] serializer_class = UserLoginSerializer @swagger_auto_schema( operation_description=doc_login(), request_body=UserLoginSerializer, ) def post(self, request, *args, **kwargs): return super().post(request, *args, **kwargs) def create(self, request, *args, **kwargs): serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) data = serializer.data user = authenticate(request, username=request.data['email'], password=data['password']) if not user: raise ValidationError({"email": "Unable to log in with provided credentials."}) user.last_login = timezone.now() user.is_active = True user.save token, created = Token.objects.get_or_create(user=user) serializer_data = serializer.data serializer_data['token'] = token.key return Response({ "id": user.id, "fullname": user.fullname, "email": user.email, "token": token.key, "user_type": user.user_type_based_on_groups, "avatar": request.build_absolute_uri(user.avatar.url) if user.avatar else None, }, status=status.HTTP_201_CREATED) class UserProfileView(RetrieveAPIView): serializer_class = UserProfileSerializer permission_classes = [IsAuthenticated, IsActiveUser] queryset = User.objects.all() def get_object(self): return self.request.user class UserUpdateView(UpdateAPIView): permission_classes = [IsAuthenticated, IsActiveUser] serializer_class = UserProfileSerializer def get_object(self): return self.request.user class UserRecoverPassword(CreateAPIView): serializer_class = UserRecoverPasswordSerializer @swagger_auto_schema( operation_description=doc_recover(), request_body=UserRecoverPasswordSerializer, ) def post(self, request): serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) data = serializer.data user = get_object_or_404(User, email=data['email']) code = RedisManager.generate_otp_code() print(f' send {code}') phone_number = RedisManager().add_to_redis(code, fullname=str(user.fullname), password='', email=data['email']) send_email([data['email']], code) return Response( data= { "id": user.id, "fullname": user.fullname, "phone_number": str(user.phone_number), "email": user.email if user.email else None, "avatar": user.avatar if user.avatar else None, "message": "Forgot password code sent" }, status=status.HTTP_202_ACCEPTED, ) class UserResetPassword(CreateAPIView): serializer_class = UserResetPasswordSerializer permission_classes = [IsAuthenticated] @swagger_auto_schema( operation_description=doc_reset(), request_body=UserResetPasswordSerializer, ) def post(self, request, *args, **kwargs): # Get the logged-in user user = request.user # Use the serializer to validate data serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) # Set the new password user.set_password(serializer.validated_data['password']) user.save() # Return a success response return Response({"message": "Your password has been changed successfully."}, status=status.HTTP_200_OK) class UserDeleteView(APIView): permission_classes = [IsAuthenticated] def delete(self, request, *args, **kwargs): try: user = request.user if user.email == "admin@gmail.com": raise AppAPIException({"message": "Unable to log in with provided credentials."}, status_code=status.HTTP_204_NO_CONTENT) user.soft_delete() if t := Token.objects.filter(user=user).first(): t.delete() return Response({"detail": "Your account has been deleted."}, status=status.HTTP_204_NO_CONTENT) except Exception: # پیام خطای ثابت برای سایر خطاهای غیرمنتظره return Response({"detail": "User does not exist."}, status=status.HTTP_404_NOT_FOUND)