from rest_framework.response import Response from rest_framework import status from rest_framework.views import APIView from authentication.serializers import ( UserRegistrationSerializer, UserLoginSerializer, UserChangePasswordSerializer, SendPasswordResetEmailSerializer, UserPasswordResetSerializer, UserTypeSerializer, UserDetailSerializer, send_verification_email ) from authentication.renderers import UserRenderer from rest_framework_simplejwt.tokens import RefreshToken from rest_framework_simplejwt.views import TokenRefreshView from rest_framework.permissions import IsAuthenticated from rest_framework import generics from django.contrib.auth import get_user_model from .models import User, usertype, CodeConfirmation, EmailVerification from rest_framework import generics from django.core.exceptions import ValidationError as DjangoValidationError from utils.exception import ValidationError from utils.helper import success_response, failed_response, generate_random_text from rest_framework_simplejwt.authentication import JWTAuthentication User = get_user_model() import re from utils.data import logger_settings from django.core.validators import validate_email from utils.validation import ( validate_required_params, ) from utils.logger import record from utils.data import validation_msg from rest_framework.permissions import AllowAny from django.conf import settings from django.template.loader import render_to_string from utils.validation import is_email from django.shortcuts import render from django.utils.html import strip_tags from django.core.mail import EmailMultiAlternatives import os from django.utils.http import urlsafe_base64_encode from django.contrib.auth.tokens import default_token_generator from django.utils.encoding import force_bytes from django.shortcuts import get_object_or_404 from datetime import timedelta from django.utils import timezone log_msg = logger_settings["msg"] def get_tokens_for_user(user): refresh = RefreshToken.for_user(user) return { "refresh": str(refresh), "access": str(refresh.access_token), } def login_page(request): return render(request, 'login.html') def register_page(request): return render(request, 'register.html') class CustomTokenRefreshView(TokenRefreshView): def post(self, request, *args, **kwargs): refresh_token = request.data.get('refresh') try: refresh = RefreshToken(refresh_token) new_access_token = refresh.access_token return Response({"access": str(new_access_token)}, status=status.HTTP_200_OK) except Exception as e: return Response({"detail": str(e)}, status=status.HTTP_400_BAD_REQUEST) class UserRegistrationView(APIView): def post(self, request, *args, **kwargs): process_code = generate_random_text(6) try: serializer = UserRegistrationSerializer(data=request.data) if serializer.is_valid(): user = serializer.save() # Generate verification code and send email verification_code = generate_random_text(6) expires_at = timezone.now() + timedelta(minutes=30) EmailVerification.objects.create( user=user, code=verification_code, expires_at=expires_at ) # Send verification email send_verification_email(user.email, verification_code) return success_response( msg="Registration successful. Please check your email for verification code.", code=status.HTTP_201_CREATED ) return failed_response( msg=serializer.errors, code=status.HTTP_400_BAD_REQUEST ) except Exception as e: return failed_response( msg=str(e), code=status.HTTP_500_INTERNAL_SERVER_ERROR ) class VerifyEmailView(APIView): def post(self, request): email = request.data.get('email') verification_code = request.data.get('code') try: user = User.objects.get(email=email) verification = EmailVerification.objects.filter( user=user, code=verification_code, is_used=False, expires_at__gt=timezone.now() ).first() if not verification: return failed_response( msg="Invalid or expired verification code", code=status.HTTP_400_BAD_REQUEST ) # Mark user as active and verification as used user.is_active = True user.email_verified_at = timezone.now() user.save() verification.is_used = True verification.save() return success_response( msg="Email verified successfully. You can now login.", code=status.HTTP_200_OK ) except User.DoesNotExist: return failed_response( msg="User not found", code=status.HTTP_404_NOT_FOUND ) class ResendVerificationCodeView(APIView): def post(self, request): email = request.data.get('email') try: user = User.objects.get(email=email) # Check if user is already verified if user.is_active: return failed_response( msg="Email is already verified", code=status.HTTP_400_BAD_REQUEST ) # Generate new verification code verification_code = generate_random_text(6) expires_at = timezone.now() + timedelta(minutes=30) # Invalidate old verifications EmailVerification.objects.filter(user=user).update(is_used=True) # Create new verification EmailVerification.objects.create( user=user, code=verification_code, expires_at=expires_at ) # Send new verification email send_verification_email(user.email, verification_code) return success_response( msg="New verification code sent. Please check your email.", code=status.HTTP_200_OK ) except User.DoesNotExist: return failed_response( msg="User not found", code=status.HTTP_404_NOT_FOUND ) # class UserRegistrationView(generics.CreateAPIView): # serializer_class = UserRegistrationSerializer # def post(self, request, *args, **kwargs): # process_code = generate_random_text(6) # record( # "info", f'{process_code}: {log_msg["started"]} - User Registration Started' # ) # data = request.data # record("info", f'{process_code}: {log_msg["validated"]}') # try: # validate_required_params(request, ["email", "username", "password"]) # data = request.data # email = data.get("email") # if not email: # record( # "warn", # f'{process_code}: {validation_msg["required"]} - Email is missing', # ) # return failed_response( # msg=validation_msg["required"], code=status.HTTP_400_BAD_REQUEST # ) # try: # validate_email(email) # except DjangoValidationError: # record( # "error", # f'{process_code}: {validation_msg["invalid_email"]} - Invalid email address', # ) # return failed_response( # msg=validation_msg["invalid_email"], # code=status.HTTP_400_BAD_REQUEST, # ) # if User.objects.filter(email=email).exists(): # record( # "error", # f'{process_code}: {validation_msg["duplicate_email"]} - A user with this email already exists', # ) # return failed_response( # msg=validation_msg["duplicate_email"], # code=status.HTTP_400_BAD_REQUEST, # ) # username = data.get("username") # if not username: # record( # "warn", # f'{process_code}: {validation_msg["required"]} - Username is missing', # ) # return failed_response( # msg=validation_msg["required"], code=status.HTTP_400_BAD_REQUEST # ) # if User.objects.filter(username=username).exists(): # record( # "error", # f'{process_code}: {validation_msg["duplicate_username"]} - This username is already taken', # ) # return failed_response( # msg=validation_msg["duplicate_username"], # code=status.HTTP_400_BAD_REQUEST, # ) # if len(username) < 3: # record( # "warn", # f'{process_code}: {validation_msg["invalid_username"]} - Username must be at least three characters long', # ) # return failed_response( # msg=validation_msg["invalid_username"], # code=status.HTTP_400_BAD_REQUEST, # ) # password = data.get("password") # if not password: # record( # "warn", # f'{process_code}: {validation_msg["required"]} - Password is missing', # ) # return failed_response( # msg=validation_msg["required"], code=status.HTTP_400_BAD_REQUEST # ) # if len(password) < 8: # record( # "warn", # f'{process_code}: {validation_msg["min_length"]} - Password must be at least eight characters long', # ) # return failed_response( # msg=validation_msg["min_length"], code=status.HTTP_400_BAD_REQUEST # ) # if not re.search(r"[A-Z]", password): # record( # "warn", # f'{process_code}: {validation_msg["missing_upper_case"]} - Password must contain at least one uppercase letter', # ) # return failed_response( # msg=validation_msg["missing_upper_case"], # code=status.HTTP_400_BAD_REQUEST, # ) # if not re.search(r"[a-z]", password): # record( # "warn", # f'{process_code}: {validation_msg["missing_lower_case"]} - Password must contain at least one lowercase letter', # ) # return failed_response( # msg=validation_msg["missing_lower_case"], # code=status.HTTP_400_BAD_REQUEST, # ) # if not re.search(r"\d", password): # record( # "warn", # f'{process_code}: {validation_msg["missing_number"]} - Password must contain at least one digit', # ) # return failed_response( # msg=validation_msg["missing_number"], # code=status.HTTP_400_BAD_REQUEST, # ) # serializer = self.get_serializer(data=data) # if serializer.is_valid(): # user = serializer.save() # confirmation_code = generate_random_text(6) # CodeConfirmation.objects.create(user=user, code=confirmation_code) # record( # "info", # f'{process_code}: {log_msg["completed"]} - User registered successfully', # ) # return success_response( # msg="Registration Successful. Please verify using the code." # ) # record( # "error", # f"{process_code}: {serializer.errors} - Validation errors during registration", # ) # return failed_response( # msg=serializer.errors, code=status.HTTP_400_BAD_REQUEST # ) # except ValidationError as e: # record("error", f"{process_code}: {str(e)} - ValidationError occurred") # return failed_response(msg=str(e), code=status.HTTP_400_BAD_REQUEST) # except Exception as e: # record("error", f"{process_code}: {str(e)} - Unexpected error occurred") # return failed_response( # msg= f"An unexpected error occurred: {str(e)}", # code=status.HTTP_500_INTERNAL_SERVER_ERROR, # ) class UserLoginView(APIView): def post(self, request, format=None): process_code = generate_random_text(6) record("info", f'{process_code}: {log_msg["started"]} - User Login Started') try: serializer = UserLoginSerializer(data=request.data) if serializer.is_valid(raise_exception=True): record("info", f'{process_code}: {log_msg["validated"]}') username = serializer.data.get("username") email = serializer.data.get("email") password = serializer.data.get("password") user = None if email: user = User.objects.filter(email=email).first() if not user: record( "warn", f'{process_code}: {validation_msg["invalid_email"]}' ) return failed_response( msg=validation_msg["invalid_email"], code=status.HTTP_400_BAD_REQUEST, ) elif username: user = User.objects.filter(username=username).first() if not user: record( "warn", f'{process_code}: {validation_msg["invalid_username"]}', ) return failed_response( msg=validation_msg["invalid_username"], code=status.HTTP_400_BAD_REQUEST, ) if user and not user.check_password(password): record("warn", f'{process_code}: {validation_msg["password"]}') return failed_response( msg=validation_msg["password"], code=status.HTTP_400_BAD_REQUEST ) if not user.is_active: return failed_response( msg="Please verify your email before logging in.", code=status.HTTP_403_FORBIDDEN ) token = get_tokens_for_user(user) record( "info", f'{process_code}: {log_msg["main_process"]} - User login successful', ) return success_response(data={"token": token}, msg="Login Success") record("warn", f"{process_code}: {serializer.errors}") return failed_response( msg=serializer.errors, code=status.HTTP_400_BAD_REQUEST ) except DjangoValidationError as e: record("error", f"{process_code}: {str(e)} - ValidationError occurred") return failed_response(msg=str(e), code=status.HTTP_400_BAD_REQUEST) except Exception as e: record("error", f"{process_code}: {str(e)} - Unexpected error occurred") return failed_response( msg="An unexpected error occurred", code=status.HTTP_500_INTERNAL_SERVER_ERROR, ) class UserChangePasswordView(APIView): authentication_classes = [JWTAuthentication] permission_classes = [IsAuthenticated] def post(self, request, format=None): process_code = generate_random_text(6) record( "info", f'{process_code}: {log_msg["started"]} - UserChangePasswordView POST request started', ) try: serializer = UserChangePasswordSerializer( data=request.data, context={"user": request.user} ) if serializer.is_valid(raise_exception=True): record( "info", f'{process_code}: {log_msg["completed"]} - Password changed successfully', ) return success_response( msg="Password Changed Successfully" ) record( "warn", f"{process_code}: {serializer.errors} - Validation errors during password change", ) return failed_response( msg="Validation errors occurred during password change", code=status.HTTP_400_BAD_REQUEST, ) except Exception as e: record( "error", f"{process_code}: {str(e)} - Unexpected error occurred during password change", ) return failed_response( msg=f"An unexpected error occurred: {str(e)}", code=status.HTTP_500_INTERNAL_SERVER_ERROR, ) class SendPasswordResetEmailView(APIView): renderer_classes = [UserRenderer] permission_classes = [AllowAny] def post(self, request, format=None): process_code = generate_random_text(6) record( "info", f'{process_code}: {log_msg["started"]} - SendPasswordResetEmailView POST request started', ) try: serializer = SendPasswordResetEmailSerializer(data=request.data) if serializer.is_valid(raise_exception=True): email = serializer.validated_data['email'] user = User.objects.get(email=email) token = default_token_generator.make_token(user) uid = urlsafe_base64_encode(force_bytes(user.pk)) reset_link = f"http://127.0.0.1:8000/user-auth/reset-password/{uid}/{token}" self.send_password_reset_email(email, reset_link) record( "info", f'{process_code}: {log_msg["completed"]} - Password reset email sent successfully to {email}', ) return success_response( msg="Password Reset Link Sent. Please check your email.", ) record( "warn", f"{process_code}: {serializer.errors} - Validation errors during password reset email request", ) return failed_response( msg="Validation errors occurred during password reset email request", code=status.HTTP_400_BAD_REQUEST, ) except Exception as e: record( "error", f"{process_code}: {str(e)} - Please try again with a valid email address", ) return failed_response( msg=f"An unexpected error occurred: {str(e)}", code=status.HTTP_500_INTERNAL_SERVER_ERROR, ) def send_password_reset_email(self, email, reset_link): subject = 'Your Password Reset Link' html_message = render_to_string('password_reset.html', {'reset_link': reset_link}) plain_message = strip_tags(html_message) from_email = os.environ.get('EMAIL_USER') email_message = EmailMultiAlternatives(subject, plain_message, from_email, [email]) email_message.attach_alternative(html_message, "text/html") email_message.send(fail_silently=False) class UserPasswordResetView(APIView): renderer_classes = [UserRenderer] def post(self, request, uid, token, format=None): process_code = generate_random_text(6) record( "info", f'{process_code}: {log_msg["started"]} - UserPasswordResetView POST request started', ) try: if not token or not uid: return failed_response( msg="Missing token or uid in the request", code=status.HTTP_400_BAD_REQUEST ) serializer = UserPasswordResetSerializer( data=request.data, context={"uid": uid, "token": token} ) if serializer.is_valid(raise_exception=True): record( "info", f'{process_code}: {log_msg["completed"]} - Password reset successfully', ) return success_response( msg="Password Reset Successfully", ) record( "warn", f"{process_code}: {serializer.errors} - Validation errors during password reset request", ) return failed_response( msg="Validation errors occurred during password reset request", code=status.HTTP_400_BAD_REQUEST, ) except Exception as e: record( "error", f"{process_code}: {str(e)} - Unexpected error occurred during password reset request", ) return failed_response( msg=f"An unexpected error occurred: {str(e)}", code=status.HTTP_500_INTERNAL_SERVER_ERROR, ) class LogoutView(APIView): def post(self, request, *args, **kwargs): process_code = generate_random_text(6) record( "info", f'{process_code}: {log_msg["started"]} - LogoutView POST request started', ) refresh_token = request.data.get("refresh_token") if not refresh_token: record("warn", f"{process_code}: Refresh token is missing") return success_response( msg = "Refresh token is required.", code=status.HTTP_400_BAD_REQUEST, ) try: token = RefreshToken(refresh_token) token.blacklist() record( "info", f'{process_code}: {log_msg["completed"]} - Logout successful' ) return success_response(msg ="Logout successful.", code=status.HTTP_200_OK) except Exception as e: record( "error", f"{process_code}: {str(e)} - Unexpected error occurred during logout", ) return failed_response(msg=f"An internal server error occurred: {str(e)}", code=status.HTTP_400_BAD_REQUEST) class UserTypeListView(generics.ListAPIView): serializer_class = UserTypeSerializer def get_queryset(self): queryset = usertype.objects.all() subtype = self.request.query_params.get('subtype',None) if subtype is not None: queryset = queryset.filter(subtype= subtype) return queryset def list(self,request,*args,**kwargs): process_code = generate_random_text(6) queryset = self.get_queryset() serializer = self.get_serializer(queryset, many = True) if queryset.exists(): record( "info", f'{process_code}: {log_msg["completed"]} - User Types fetched successfully' ) return success_response(msg=serializer.data, code = status.HTTP_200_OK) else: record("warn", f"{process_code}: No user types found for the given subtype") return failed_response(msg="No user types found for given subtype",code=status.HTTP_400_BAD_REQUEST) class UserTypeBySubtypeView(generics.ListAPIView): serializer_class = UserTypeSerializer def get(self, request, subtype): process_code = generate_random_text(6) user_types = usertype.objects.filter(subtype=subtype) serializer = self.get_serializer(user_types, many=True) record( "info", f'{process_code}: {log_msg["completed"]} - f"User types fetched for subtype: {subtype}"' ) return success_response(msg = serializer.data, code=status.HTTP_200_OK) class CheckUniqueUsernameOrEmailAPIView(APIView): def post(self, request, *args, **kwargs): process_code = generate_random_text(6) try: data = request.data if 'email' in data: email = data.get('email') if self.is_email(email): if User.objects.filter(email=email).exists(): record("warn", f"{process_code}: Email is already registered") return failed_response(msg="Email is already registered.", code=status.HTTP_400_BAD_REQUEST) else: record("warn", f"{process_code}: Invalid email format") return failed_response(msg="Invalid email format.", code=status.HTTP_400_BAD_REQUEST) if 'username' in data: username = data.get('username') if User.objects.filter(username=username).exists(): record("warn", f"{process_code}: Username is already taken") return failed_response(msg="Username is already taken.", code=status.HTTP_400_BAD_REQUEST) return success_response(msg="Both email and username are available.") except Exception as e: record("error", f"{process_code}: Unexpected error occurred: {str(e)}") return failed_response(msg=f"An unexpected error occurred: {str(e)}", code=status.HTTP_500_INTERNAL_SERVER_ERROR) def is_email(self, value): return is_email(value) class UserCheckView(APIView): def get(self, request, user_id, format=None): process_code = generate_random_text(6) try: user = User.objects.get(id=user_id) record("info", f"{process_code}: User with ID {user_id} found") return Response({"exists": True, "user_id": user.id}, status=status.HTTP_200_OK) except User.DoesNotExist: record("warn", f"{process_code}: User with ID {user_id} not found") return Response({"exists": False, "message": "User not found"}, status=status.HTTP_404_NOT_FOUND) class UserDetailView(APIView): authentication_classes = [JWTAuthentication] def post(self, request, *args, **kwargs): process_code = generate_random_text(6) email = request.data.get('email') if not email: return Response() user = get_object_or_404(User, email=email) serializer = UserDetailSerializer(user) return Response(serializer.data, status=status.HTTP_200_OK)