"""Account view """ from notifications.utils import remove_fcm_token # django imports from datetime import datetime, timedelta from rest_framework import viewsets, status, views from rest_framework.decorators import action import random import logging from django.utils import timezone import jwt from django.contrib.auth import logout from django.contrib.auth import authenticate, login 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 rest_framework import mixins from django.conf import settings # local imports from guardian.models import Guardian from junior.models import Junior from guardian.utils import upload_image_to_alibaba from account.models import UserDeviceDetails, UserPhoneOtp, UserEmailOtp, DefaultTaskImages, UserNotification from django.contrib.auth.models import User from .serializers import (SuperUserSerializer, GuardianSerializer, JuniorSerializer, EmailVerificationSerializer, ForgotPasswordSerializer, ResetPasswordSerializer, ChangePasswordSerializer, GoogleLoginSerializer, UpdateGuardianImageSerializer, UpdateJuniorProfileImageSerializer, DefaultTaskImagesSerializer, DefaultTaskImagesDetailsSerializer, UserDeleteSerializer, UserNotificationSerializer, UpdateUserNotificationSerializer, UserPhoneOtpSerializer, AdminLoginSerializer) from rest_framework_simplejwt.tokens import RefreshToken from base.messages import ERROR_CODE, SUCCESS_CODE from base.constants import NUMBER, ZOD, JUN, GRD from guardian.tasks import generate_otp from account.utils import (send_otp_email, send_support_email, custom_response, custom_error_response, generate_code, OTP_EXPIRY) from junior.serializers import JuniorProfileSerializer from guardian.serializers import GuardianProfileSerializer class GoogleLoginMixin(object): """google login mixin""" @staticmethod def google_login(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', junior_code=generate_code(JUN, user_obj.id), referral_code=generate_code(ZOD, user_obj.id) ) 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', guardian_code=generate_code(GRD, user_obj.id), referral_code=generate_code(ZOD, user_obj.id) ) 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', junior_code=generate_code(JUN, user.id), referral_code=generate_code(ZOD, user.id)) 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', guardian_code=generate_code(GRD, user.id), referral_code=generate_code(ZOD, user.id)) 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): try: image = request.data['image'] if image and image.size == NUMBER['zero']: return custom_error_response(ERROR_CODE['2035'], response_status=status.HTTP_400_BAD_REQUEST) filename = f"images/{image.name}" image_url = upload_image_to_alibaba(image, filename) image_data = image_url if str(request.data['user_type']) == '1': junior_query = Junior.objects.filter(auth=request.user).last() serializer = UpdateJuniorProfileImageSerializer(junior_query, data={'image':image_data}, partial=True) elif str(request.data['user_type']) == '2': guardian_query = Guardian.objects.filter(user=request.user).last() 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) except Exception as e: logging.error(e) return custom_error_response(ERROR_CODE['2036'],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 = generate_otp() # 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 = OTP_EXPIRY 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') device_id = request.META.get('HTTP_DEVICE_ID') 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 device_details, created = UserDeviceDetails.objects.get_or_create(user=user) if device_details: device_details.device_id = device_id device_details.save() 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) refresh_token = str(refresh) data = {"auth_token":access_token, "refresh_token":refresh_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}) send_otp_email(username, 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(SUCCESS_CODE['3003'], 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) refresh_token = str(refresh) data = {"auth_token": access_token, "refresh_token":refresh_token, "user_type": '3'} return custom_response(None, data, response_status=status.HTTP_200_OK) class AdminLoginViewSet(viewsets.GenericViewSet): """ admin login api """ serializer_class = AdminLoginSerializer @action(methods=['post'], url_name='login', url_path='login', detail=False) def admin_login(self, request, *args, **kwargs): """ :param request: :return: """ serializer = self.serializer_class(data=request.data) serializer.is_valid(raise_exception=True) user = serializer.save() refresh = RefreshToken.for_user(user) access_token = str(refresh.access_token) refresh_token = str(refresh) data = {"auth_token": access_token, "refresh_token": refresh_token, "username": user.username, "email": user.email, "first_name": user.first_name, "last_name": user.last_name, "is_active": user.is_active, "user_type": '3', "is_superuser": user.is_superuser} return custom_response(None, data) 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) refresh_token = str(refresh) return custom_response(SUCCESS_CODE['3011'], {"auth_token":access_token, "refresh_token":refresh_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 permission_classes = [IsAuthenticated] def create(self, request, *args, **kwargs): otp = generate_otp() if User.objects.filter(email=request.data['email']): expiry = OTP_EXPIRY 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 permission_classes = [IsAuthenticated] 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): """upload task image""" queryset = DefaultTaskImages.objects.all() serializer_class = DefaultTaskImagesSerializer def create(self, request, *args, **kwargs): """upload images""" image_data = request.data['image_url'] filename = f"default_task_images/{image_data.name}" if image_data.size == NUMBER['zero']: return custom_error_response(ERROR_CODE['2035'], response_status=status.HTTP_400_BAD_REQUEST) 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 permission_classes = [IsAuthenticated] 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')) 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 permission_classes = [IsAuthenticated] 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 permission_classes = [IsAuthenticated] 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""" permission_classes = (IsAuthenticated,) 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): """Log out API""" permission_classes = (IsAuthenticated,) def post(self, request): remove_fcm_token( request.auth.payload['user_id'], request.META['HTTP_AUTHORIZATION'].split(" ")[1], request.data.get('registration_id', "")) logout(request) request.session.flush() return custom_response(SUCCESS_CODE['3020'], response_status=status.HTTP_200_OK) class AccessTokenAPIView(views.APIView): """generate access token API""" def post(self, request): # Assuming you have a refresh_token string refresh_token = request.data['refresh_token'] # Create a RefreshToken instance from the refresh token string refresh = RefreshToken(refresh_token) # Generate a new access token access_token = str(refresh.access_token) data = {"auth_token": access_token} return custom_response(None, data, response_status=status.HTTP_200_OK)