mirror of
https://github.com/HamzaSha1/zod-backend.git
synced 2025-07-16 02:16:16 +00:00
364 lines
18 KiB
Python
364 lines
18 KiB
Python
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 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
|
|
from django.contrib.auth.models import User
|
|
from .serializers import (SuperUserSerializer, GuardianSerializer, JuniorSerializer, EmailVerificationSerializer,
|
|
ForgotPasswordSerializer, ResetPasswordSerializer, ChangePasswordSerializer,
|
|
GoogleLoginSerializer, UpdateGuardianImageSerializer, UpdateJuniorProfileImageSerializer)
|
|
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
|
|
from account.utils import 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
|
|
from rest_framework.response import Response
|
|
import requests
|
|
from django.conf import settings
|
|
from .utils import get_token
|
|
|
|
|
|
class GoogleLoginMixin:
|
|
def google_login(self, request):
|
|
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)
|
|
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)
|
|
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):
|
|
serializer_class = GoogleLoginSerializer
|
|
|
|
def create(self, request):
|
|
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")
|
|
if not token:
|
|
return custom_error_response(ERROR_CODE['2027'], response_status=status.HTTP_400_BAD_REQUEST)
|
|
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)
|
|
serializer = JuniorSerializer(junior_query)
|
|
if str(user_type) == '2':
|
|
guardian_query = Guardian.objects.create(user=user, is_verified=True, is_active=True)
|
|
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):
|
|
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):
|
|
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):
|
|
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):
|
|
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"""
|
|
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"""
|
|
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):
|
|
@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
|
|
|
|
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"""
|
|
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)
|