"""Account view """ """Django import""" from datetime import datetime, timedelta from rest_framework import viewsets, status, views from rest_framework.decorators import action import random import logging from django.views.decorators.csrf import csrf_exempt from django.utils import timezone import jwt from django.contrib.auth import logout """App Import""" from guardian.utils import upload_image_to_alibaba from django.contrib.auth import authenticate, login from guardian.models import Guardian from junior.models import Junior from account.models import UserProfile, UserPhoneOtp, UserEmailOtp, DefaultTaskImages, UserNotification from django.contrib.auth.models import User """Account serializer""" from .serializers import (SuperUserSerializer, GuardianSerializer, JuniorSerializer, EmailVerificationSerializer, ForgotPasswordSerializer, ResetPasswordSerializer, ChangePasswordSerializer, GoogleLoginSerializer, UpdateGuardianImageSerializer, UpdateJuniorProfileImageSerializer, DefaultTaskImagesSerializer, DefaultTaskImagesDetailsSerializer, UserDeleteSerializer, UserNotificationSerializer, UpdateUserNotificationSerializer, UserPhoneOtpSerializer) from rest_framework_simplejwt.tokens import RefreshToken from base.messages import ERROR_CODE, SUCCESS_CODE from guardian.tasks import generate_otp from account.utils import send_otp_email, send_support_email, custom_response, custom_error_response from rest_framework.permissions import IsAuthenticated from templated_email import send_templated_mail import google.oauth2.credentials import google.auth.transport.requests from rest_framework import status import requests from rest_framework.response import Response from django.conf import settings from junior.serializers import JuniorProfileSerializer from guardian.serializers import GuardianProfileSerializer class GoogleLoginMixin: """google login mixin""" def google_login(self, request): """google login function""" access_token = request.data.get('access_token') user_type = request.data.get('user_type') if not access_token: return Response({'error': 'Access token is required.'}, status=status.HTTP_400_BAD_REQUEST) try: # Validate the access token and obtain the user's email and name credentials = google.oauth2.credentials.Credentials.from_authorized_user_info( info={ 'access_token': access_token, 'token_uri': 'https://oauth2.googleapis.com/token', 'client_id': settings.GOOGLE_CLIENT_ID, 'client_secret': settings.GOOGLE_CLIENT_SECRET, 'refresh_token': None, } ) user_info_endpoint = f'https://www.googleapis.com/oauth2/v3/userinfo?access_token={access_token}' headers = {'Authorization': f'Bearer {credentials.token}'} response = requests.get(user_info_endpoint, headers=headers) response.raise_for_status() user_info = response.json() email = user_info['email'] first_name = user_info['given_name'] last_name = user_info['family_name'] profile_picture = user_info['picture'] except Exception as e: return custom_error_response(str(e), response_status=status.HTTP_400_BAD_REQUEST) # Check if the user exists in your database or create a new user # ... user_data = User.objects.filter(email__iexact=email) if user_data.exists(): if str(user_type) == '1': junior_query = Junior.objects.filter(auth=user_data.last()).last() serializer = JuniorSerializer(junior_query) if str(user_type) == '2': guardian_query = Guardian.objects.filter(user=user_data.last()).last() serializer = GuardianSerializer(guardian_query) return custom_response(SUCCESS_CODE['3003'], serializer.data, response_status=status.HTTP_200_OK) if not User.objects.filter(email__iexact=email).exists(): user_obj = User.objects.create(username=email, email=email, first_name=first_name, last_name=last_name) if str(user_type) == '1': junior_query = Junior.objects.create(auth=user_obj, is_verified=True, is_active=True, image=profile_picture, signup_method='2') serializer = JuniorSerializer(junior_query) if str(user_type) == '2': guardian_query = Guardian.objects.create(user=user_obj, is_verified=True, is_active=True, image=profile_picture,signup_method='2') serializer = GuardianSerializer(guardian_query) # Return a JSON response with the user's email and name return custom_response(SUCCESS_CODE['3003'], serializer.data, response_status=status.HTTP_200_OK) class GoogleLoginViewSet(GoogleLoginMixin, viewsets.GenericViewSet): """Google login viewset""" serializer_class = GoogleLoginSerializer def create(self, request): """create method""" serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) return self.google_login(request) class SigninWithApple(views.APIView): """This API is for sign in with Apple for app.""" def post(self, request): token = request.data.get("access_token") user_type = request.data.get("user_type") try: decoded_data = jwt.decode(token, options={"verify_signature": False}) user_data = {"email": decoded_data.get('email'), "username": decoded_data.get('email'), "is_active": True} if decoded_data.get("email"): try: user = User.objects.get(email=decoded_data.get("email")) if str(user_type) == '1': junior_query = Junior.objects.filter(auth=user).last() serializer = JuniorSerializer(junior_query) if str(user_type) == '2': guardian_query = Guardian.objects.filter(user=user).last() serializer = GuardianSerializer(guardian_query) return custom_response(SUCCESS_CODE['3003'], serializer.data, response_status=status.HTTP_200_OK) except User.DoesNotExist: user = User.objects.create(**user_data) if str(user_type) == '1': junior_query = Junior.objects.create(auth=user, is_verified=True, is_active=True, signup_method='3') serializer = JuniorSerializer(junior_query) if str(user_type) == '2': guardian_query = Guardian.objects.create(user=user, is_verified=True, is_active=True, signup_method='3') serializer = GuardianSerializer(guardian_query) return custom_response(SUCCESS_CODE['3003'], serializer.data, response_status=status.HTTP_200_OK) except Exception as e: logging.error(e) return custom_error_response(str(e), response_status=status.HTTP_400_BAD_REQUEST) class UpdateProfileImage(views.APIView): """Update profile image""" permission_classes = [IsAuthenticated] def put(self, request, format=None): if str(request.data['user_type']) == '1': junior_query = Junior.objects.filter(auth=request.user).last() image = request.data['image'] filename = f"images/{image.name}" image_url = upload_image_to_alibaba(image, filename) image_data = image_url serializer = UpdateJuniorProfileImageSerializer(junior_query, data={'image':image_data}, partial=True) if str(request.data['user_type']) == '2': guardian_query = Guardian.objects.filter(user=request.user).last() image = request.data['image'] filename = f"images/{image.name}" image_url = upload_image_to_alibaba(image, filename) image_data = image_url serializer = UpdateGuardianImageSerializer(guardian_query, data={'image':image_data}, partial=True) if serializer.is_valid(): serializer.save() return custom_response(SUCCESS_CODE['3017'], serializer.data, response_status=status.HTTP_200_OK) return custom_error_response(serializer.errors, response_status=status.HTTP_400_BAD_REQUEST) class ChangePasswordAPIView(views.APIView): """change password""" serializer_class = ChangePasswordSerializer permission_classes = [IsAuthenticated] def post(self, request): serializer = ChangePasswordSerializer(context=request.user, data=request.data) if serializer.is_valid(): serializer.save() return custom_response(SUCCESS_CODE['3007'], response_status=status.HTTP_200_OK) return custom_error_response(serializer.errors, response_status=status.HTTP_400_BAD_REQUEST) class ResetPasswordAPIView(views.APIView): """Reset password""" def post(self, request): serializer = ResetPasswordSerializer(data=request.data) if serializer.is_valid(): serializer.save() return custom_response(SUCCESS_CODE['3006'], response_status=status.HTTP_200_OK) return custom_error_response(serializer.errors, response_status=status.HTTP_400_BAD_REQUEST) class ForgotPasswordAPIView(views.APIView): """Forgot password""" def post(self, request): serializer = ForgotPasswordSerializer(data=request.data) if serializer.is_valid(): email = serializer.validated_data['email'] try: User.objects.get(email=email) except User.DoesNotExist: return custom_error_response(ERROR_CODE['2004'], response_status=status.HTTP_404_NOT_FOUND) verification_code = ''.join([str(random.randrange(9)) for _ in range(6)]) # Send the verification code to the user's email from_email = settings.EMAIL_FROM_ADDRESS recipient_list = [email] send_templated_mail( template_name='email_reset_verification.email', from_email=from_email, recipient_list=recipient_list, context={ 'verification_code': verification_code } ) expiry = timezone.now() + timezone.timedelta(days=1) user_data, created = UserEmailOtp.objects.get_or_create(email=email) if created: user_data.expired_at = expiry user_data.save() if user_data: user_data.otp = verification_code user_data.expired_at = expiry user_data.save() return custom_response(SUCCESS_CODE['3015'], response_status=status.HTTP_200_OK) return custom_error_response(serializer.errors, response_status=status.HTTP_400_BAD_REQUEST) class SendPhoneOtp(viewsets.ModelViewSet): """Send otp on phone""" queryset = UserPhoneOtp.objects.all() serializer_class = UserPhoneOtpSerializer def create(self, request, *args, **kwargs): otp = generate_otp() phone_number = self.request.data['phone'] if phone_number.isdigit() and len(phone_number) == 10: phone_otp, created = UserPhoneOtp.objects.get_or_create(country_code=self.request.data['country_code'], phone=self.request.data['phone']) if phone_otp: phone_otp.otp = otp phone_otp.save() return custom_response(None, {'phone_otp':otp}, response_status=status.HTTP_200_OK) return custom_error_response(ERROR_CODE['2020'], response_status=status.HTTP_400_BAD_REQUEST) class UserPhoneVerification(viewsets.ModelViewSet): """Send otp on phone""" queryset = UserPhoneOtp.objects.all() serializer_class = UserPhoneOtpSerializer def list(self, request, *args, **kwargs): try: phone_data = UserPhoneOtp.objects.filter(phone=self.request.GET.get('phone'), otp=self.request.GET.get('otp')).last() if phone_data: phone_data.is_verified = True phone_data.save() return custom_response(SUCCESS_CODE['3012'], response_status=status.HTTP_200_OK) else: return custom_error_response(ERROR_CODE["2008"], response_status=status.HTTP_400_BAD_REQUEST) except Exception: return custom_error_response(ERROR_CODE["2008"], response_status=status.HTTP_400_BAD_REQUEST) class UserLogin(viewsets.ViewSet): """User login""" @action(methods=['post'], detail=False) def login(self, request): username = request.data.get('username') password = request.data.get('password') user = authenticate(request, username=username, password=password) try: if user is not None: login(request, user) guardian_data = Guardian.objects.filter(user__username=username, is_verified=True).last() if guardian_data: serializer = GuardianSerializer(guardian_data).data junior_data = Junior.objects.filter(auth__username=username, is_verified=True).last() if junior_data: serializer = JuniorSerializer(junior_data).data return custom_response(SUCCESS_CODE['3003'], serializer, response_status=status.HTTP_200_OK) else: return custom_error_response(ERROR_CODE["2002"], response_status=status.HTTP_401_UNAUTHORIZED) except Exception as e: logging.error(e) email_verified = UserEmailOtp.objects.filter(email=username).last() refresh = RefreshToken.for_user(user) access_token = str(refresh.access_token) data = {"auth_token":access_token, "is_profile_complete": False, "user_type": email_verified.user_type, } is_verified = False if email_verified: is_verified = email_verified.is_verified if not is_verified: otp = generate_otp() email_verified.otp = otp email_verified.save() data.update({"email_otp":otp}) return custom_response(ERROR_CODE['2024'], {"email_otp": otp, "is_email_verified": is_verified}, response_status=status.HTTP_200_OK) data.update({"is_email_verified": is_verified}) return custom_response(None, data, response_status=status.HTTP_200_OK) @action(methods=['post'], detail=False) def admin_login(self, request): username = request.data.get('username') password = request.data.get('password') user = authenticate(request, username=username, password=password) try: if user is not None: login(request, user) if user.is_superuser: serializer = SuperUserSerializer(user) return custom_response(SUCCESS_CODE['3003'], serializer.data, response_status=status.HTTP_200_OK) else: return custom_error_response(ERROR_CODE["2002"], response_status=status.HTTP_401_UNAUTHORIZED) except Exception as e: logging.error(e) refresh = RefreshToken.for_user(user) access_token = str(refresh.access_token) data = {"auth_token": access_token, "user_role": '3'} return custom_response(None, data, response_status=status.HTTP_200_OK) class UserEmailVerification(viewsets.ModelViewSet): """User Email verification""" serializer_class = EmailVerificationSerializer queryset = UserEmailOtp.objects.all() def list(self, request, *args, **kwargs): try: user_obj = User.objects.filter(username=self.request.GET.get('email')).last() email_data = UserEmailOtp.objects.filter(email=self.request.GET.get('email'), otp=self.request.GET.get('otp')).last() if email_data: input_datetime_str = str(email_data.expired_at) input_format = "%Y-%m-%d %H:%M:%S.%f%z" output_format = "%Y-%m-%d %H:%M:%S.%f" input_datetime = datetime.strptime(input_datetime_str, input_format) output_datetime_str = input_datetime.strftime(output_format) format_str = "%Y-%m-%d %H:%M:%S.%f" datetime_obj = datetime.strptime(output_datetime_str, format_str) if datetime.today() > datetime_obj: return custom_error_response(ERROR_CODE["2029"], response_status=status.HTTP_400_BAD_REQUEST) email_data.is_verified = True email_data.save() if email_data.user_type == '1': junior_data = Junior.objects.filter(auth__email=self.request.GET.get('email')).last() if junior_data: junior_data.is_verified = True junior_data.save() else: guardian_data = Guardian.objects.filter(user__email=self.request.GET.get('email')).last() if guardian_data: guardian_data.is_verified = True guardian_data.save() refresh = RefreshToken.for_user(user_obj) access_token = str(refresh.access_token) return custom_response(SUCCESS_CODE['3011'], {"auth_token":access_token}, response_status=status.HTTP_200_OK) else: return custom_error_response(ERROR_CODE["2008"], response_status=status.HTTP_400_BAD_REQUEST) except Exception as e: logging.error(e) return custom_error_response(ERROR_CODE["2008"], response_status=status.HTTP_400_BAD_REQUEST) class ReSendEmailOtp(viewsets.ModelViewSet): """Send otp on phone""" queryset = UserEmailOtp.objects.all() serializer_class = EmailVerificationSerializer def create(self, request, *args, **kwargs): otp = generate_otp() if User.objects.filter(email=request.data['email']): expiry = timezone.now() + timezone.timedelta(days=1) email_data, created = UserEmailOtp.objects.get_or_create(email=request.data['email']) if created: email_data.expired_at = expiry email_data.save() if email_data: email_data.otp = otp email_data.expired_at = expiry email_data.save() send_otp_email(request.data['email'], otp) return custom_response(SUCCESS_CODE['3016'], response_status=status.HTTP_200_OK) else: return custom_error_response(ERROR_CODE["2023"], response_status=status.HTTP_400_BAD_REQUEST) class ProfileAPIViewSet(viewsets.ModelViewSet): """Profile viewset""" queryset = User.objects.all() serializer_class = JuniorProfileSerializer def list(self, request, *args, **kwargs): """profile view""" if str(self.request.GET.get('user_type')) == '1': junior_data = Junior.objects.filter(auth=self.request.user).last() if junior_data: serializer = JuniorProfileSerializer(junior_data) return custom_response(None, serializer.data, response_status=status.HTTP_200_OK) elif str(self.request.GET.get('user_type')) == '2': guardian_data = Guardian.objects.filter(user=self.request.user).last() if guardian_data: serializer = GuardianProfileSerializer(guardian_data) return custom_response(None, serializer.data, response_status=status.HTTP_200_OK) class UploadImageAPIViewSet(viewsets.ModelViewSet): """Profile viewset""" queryset = DefaultTaskImages.objects.all() serializer_class = DefaultTaskImagesSerializer def create(self, request, *args, **kwargs): """profile view""" image_data = request.data['image_url'] filename = f"default_task_images/{image_data.name}" image = upload_image_to_alibaba(image_data, filename) image_data = image request.data['image_url'] = image_data serializer = DefaultTaskImagesSerializer(data=request.data) if serializer.is_valid(): serializer.save() return custom_response(None, serializer.data, response_status=status.HTTP_200_OK) return custom_error_response(serializer.error, response_status=status.HTTP_400_BAD_REQUEST) class DefaultImageAPIViewSet(viewsets.ModelViewSet): """Profile viewset""" queryset = DefaultTaskImages.objects.all() serializer_class = DefaultTaskImagesDetailsSerializer def list(self, request, *args, **kwargs): """profile view""" queryset = DefaultTaskImages.objects.all() serializer = DefaultTaskImagesSerializer(queryset, many=True) return custom_response(None, serializer.data, response_status=status.HTTP_200_OK) class DeleteUserProfileAPIViewSet(viewsets.GenericViewSet): """ Delete user API view set """ @action(detail=False, methods=['POST'], url_path='user-account',serializer_class=UserDeleteSerializer, permission_classes=[IsAuthenticated]) def account(self, request): user_type = str(request.data['user_type']) password = request.data.get('password') signup_method = str(request.data.get('signup_method')) print("signup_method===>",signup_method,'==>',type(signup_method)) serializer = self.get_serializer(data=request.data, context={'request': request, 'user': request.user, 'user_type': user_type, 'password': password, 'signup_method':signup_method}) if serializer.is_valid(): serializer.save() return custom_response(SUCCESS_CODE['3005'], response_status=status.HTTP_200_OK) return custom_error_response(serializer.errors, response_status=status.HTTP_400_BAD_REQUEST) class UserNotificationAPIViewSet(viewsets.ModelViewSet): """notification viewset""" queryset = UserNotification.objects.all() serializer_class = UserNotificationSerializer def list(self, request, *args, **kwargs): """profile view""" queryset = self.queryset.filter(user=request.user) serializer = UserNotificationSerializer(queryset, many=True) return custom_response(None, serializer.data, response_status=status.HTTP_200_OK) class UpdateUserNotificationAPIViewSet(viewsets.ModelViewSet): """Update notification viewset""" queryset = UserNotification.objects.all() serializer_class = UpdateUserNotificationSerializer def create(self, request, *args, **kwargs): """profile view""" serializer = UpdateUserNotificationSerializer(data=request.data, context=request.user) if serializer.is_valid(): serializer.save() return custom_response(None, serializer.data, response_status=status.HTTP_200_OK) return custom_error_response(serializer.error, response_status=status.HTTP_400_BAD_REQUEST) class SendSupportEmail(views.APIView): """support email api""" def post(self, request): name = request.data.get('name') sender = request.data.get('email') subject = request.data.get('subject') message = request.data.get('message') if name and sender and subject and message: try: send_support_email(name, sender, subject, message) return custom_response(SUCCESS_CODE['3019'], response_status=status.HTTP_200_OK) except Exception as e: return custom_error_response(str(e), response_status=status.HTTP_400_BAD_REQUEST) else: return custom_error_response(ERROR_CODE['2033'], response_status=status.HTTP_400_BAD_REQUEST) class LogoutAPIView(views.APIView): permission_classes = (IsAuthenticated,) def post(self, request): logout(request) request.session.flush() return custom_response(SUCCESS_CODE['3020'], response_status=status.HTTP_200_OK)