# users/views.py from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status from .models import CustomUser, OTP , Device from .serializers import OTPSendSerializer, OTPVerifySerializer, SetPasswordSerializer, LoginSerializer, UserDetailSerializer from django.core.mail import send_mail from django.conf import settings from django.utils import timezone from django.contrib.auth import authenticate from django.db import IntegrityError import random import string from rest_framework.permissions import IsAdminUser from app.models import Country, Category, QuestionAnswer class OTPSendView(APIView): def post(self, request): serializer = OTPSendSerializer(data=request.data) if serializer.is_valid(): email = serializer.validated_data['email'] # Check if user exists or create unverified user user, created = CustomUser.objects.get_or_create( email=email, defaults={'is_verified': False} ) if user.is_verified: return Response({"error": "User already verified"}, status=status.HTTP_400_BAD_REQUEST) # Generate 4-digit OTP otp = ''.join(random.choices(string.digits, k=4)) # Check if an OTP already exists for this user otp_obj = OTP.objects.filter(user=user).first() if otp_obj: # Update the existing OTP otp_obj.otp = otp otp_obj.created_at = timezone.now() otp_obj.expires_at = otp_obj.created_at + timezone.timedelta(minutes=10) otp_obj.save() else: # Create new OTP record OTP.objects.create(user=user, otp=otp) # Send OTP via email subject = 'Your OTP for Smart Prep MCQ App Verification' message = f'Your OTP is {otp}. It is valid for 10 minutes.' from_email = settings.EMAIL_HOST_USER send_mail(subject, message, from_email, [email]) return Response({"message": "OTP sent to email"}, status=status.HTTP_200_OK) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) class OTPVerifyView(APIView): def post(self, request): serializer = OTPVerifySerializer(data=request.data) if serializer.is_valid(): email = serializer.validated_data['email'] otp = serializer.validated_data['otp'] try: user = CustomUser.objects.get(email=email) except CustomUser.DoesNotExist: return Response({"error": "User not found"}, status=status.HTTP_404_NOT_FOUND) if user.is_verified: return Response({"error": "User already verified"}, status=status.HTTP_200_OK) try: otp_obj = OTP.objects.get(user=user, otp=otp) if otp_obj.expires_at < timezone.now(): otp_obj.delete() return Response({"error": "OTP expired"}, status=status.HTTP_400_BAD_REQUEST) # Mark OTP as used (delete after verification) user.is_verified = True user.save() otp_obj.delete() return Response({"message": "OTP verified successfully"}, status=status.HTTP_200_OK) except OTP.DoesNotExist: return Response({"error": "Invalid OTP"}, status=status.HTTP_400_BAD_REQUEST) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) class SetPasswordView(APIView): def post(self, request): serializer = SetPasswordSerializer(data=request.data) if serializer.is_valid(): email = serializer.validated_data['email'] password = serializer.validated_data['password'] first_name = serializer.validated_data['first_name'] last_name = serializer.validated_data['last_name'] phone_number = serializer.validated_data.get('phone_number', '') country_name = serializer.validated_data.get('country_name', '') try: user = CustomUser.objects.get(email=email) except CustomUser.DoesNotExist: return Response({"error": "User not found"}, status=status.HTTP_404_NOT_FOUND) if not user.is_verified: return Response({"error": "User not verified"}, status=status.HTTP_400_BAD_REQUEST) # Update user information user.first_name = first_name user.last_name = last_name user.phone_number = phone_number if phone_number else None user.country_name = country_name if country_name else None # Set password and save user.set_password(password) user.save() return Response({"message": "Password and user information set successfully"}, status=status.HTTP_200_OK) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) class LoginView(APIView): def post(self, request): serializer = LoginSerializer(data=request.data) if serializer.is_valid(): email = serializer.validated_data['email'] password = serializer.validated_data['password'] device_id = request.data.get('device_id') # Get device_id from frontend try: user = CustomUser.objects.get(email=email) if user.check_password(password) and user.is_verified: # Store device information user_agent = request.META.get('HTTP_USER_AGENT', 'Unknown') ip_address = request.META.get('REMOTE_ADDR', '0.0.0.0') # Update or create device with device_id if provided device_data = { 'user_agent': user_agent, 'ip_address': ip_address } if device_id: device_data['device_id'] = device_id try: Device.objects.update_or_create( user=user, defaults=device_data ) except IntegrityError: # Handle duplicate device_id for same user return Response({"error": "Device already registered for this user"}, status=status.HTTP_400_BAD_REQUEST) return Response({"message": "Login successful" , "is_admin" : user.is_superuser, "email" : email}, status=status.HTTP_200_OK) return Response({"error": "Invalid credentials or unverified user"}, status=status.HTTP_401_UNAUTHORIZED) except CustomUser.DoesNotExist: return Response({"error": "User not found"}, status=status.HTTP_404_NOT_FOUND) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) class AutoLoginView(APIView): def post(self, request): device_id = request.data.get('device_id') # Get device_id from frontend if device_id: # Find device with matching device_id device = Device.objects.filter(device_id=device_id).first() if device: user = device.user if user.is_verified: return Response({"message": "Auto-login successful", "email": user.email , "is_admin" : user.is_superuser}, status=status.HTTP_200_OK) else: # Fallback to IP-based matching for backward compatibility user_agent = request.META.get('HTTP_USER_AGENT', 'Unknown') ip_address = request.META.get('REMOTE_ADDR', '0.0.0.0') # Find devices with matching IP only matching_devices = Device.objects.filter(ip_address=ip_address) if matching_devices.exists(): # If more than one device, pick the most recent by last_login most_recent_device = matching_devices.order_by('-last_login').first() user = most_recent_device.user if user.is_verified: return Response({"message": "Auto-login successful", "email": user.email , "is_admin" : user.is_superuser}, status=status.HTTP_200_OK) return Response({"error": "No matching device or unverified user"}, status=status.HTTP_401_UNAUTHORIZED) class CreateSuperUserView(APIView): def post(self, request): consent_user = request.data.get('consent_user') if consent_user != 'kalash@2025': return Response({'error': 'You do not have permission to create superuser.'}, status=status.HTTP_403_FORBIDDEN) email = request.data.get('email') password = request.data.get('password') if not email or not password: return Response({'error': 'Email and password are required.'}, status=status.HTTP_400_BAD_REQUEST) if CustomUser.objects.filter(email=email).exists(): return Response({'error': 'User with this email already exists.'}, status=status.HTTP_400_BAD_REQUEST) CustomUser.objects.create_superuser(email=email, password=password) return Response({'message': 'Superuser created successfully.'}, status=status.HTTP_201_CREATED) class LogoutView(APIView): def post(self, request): email = request.data.get('email') device_id = request.data.get('device_id') # Get device_id from frontend try: user = CustomUser.objects.get(email=email) except CustomUser.DoesNotExist: return Response({'error': 'User not found.'}, status=status.HTTP_404_NOT_FOUND) if device_id: # Remove the device entry for this user and device_id Device.objects.filter(user=user, device_id=device_id).delete() else: # Fallback to old method for backward compatibility user_agent = request.META.get('HTTP_USER_AGENT', 'Unknown') ip_address = request.META.get('REMOTE_ADDR', '0.0.0.0') Device.objects.filter(user=user, user_agent=user_agent, ip_address=ip_address).delete() return Response({'message': 'Logged out successfully.'}, status=status.HTTP_200_OK) class UserAnalyticsView(APIView): def get(self, request): # Fetch all users and filter in Python to avoid Djongo boolean issues all_users = list(CustomUser.objects.all()) non_superusers = [u for u in all_users if not getattr(u, 'is_superuser', False)] total_users = len(non_superusers) current_active_users = len([u for u in non_superusers if u.is_active]) # Get active and verified users (currently using the app) active_verified_users = [u for u in non_superusers if u.is_active and u.is_verified] total_active_verified_users = len(active_verified_users) # Get active users list (currently using the app) active_users_list = [] for u in active_verified_users: device = Device.objects.filter(user=u).first() active_user_data = { 'id': u.id, 'email': u.email, 'is_active': u.is_active, 'is_verified': u.is_verified, 'last_login_device': { 'user_agent': device.user_agent if device else None, 'ip_address': device.ip_address if device else None, 'last_login': device.last_login.isoformat() if device and device.last_login else None } if device else None } active_users_list.append(active_user_data) total_categories = Category.objects.count() total_countries = Country.objects.count() total_questions = QuestionAnswer.objects.count() return Response({ 'total_users': total_users, 'current_active_users': current_active_users, 'total_active_verified_users': total_active_verified_users, 'active_users_list': active_users_list, 'total_categories': total_categories, 'total_countries': total_countries, 'total_questions': total_questions }, status=status.HTTP_200_OK) class UserDetailView(APIView): def get(self, request): """Get user details by email""" try: email = request.query_params.get('email') if not email: return Response({'error': 'email parameter is required'}, status=status.HTTP_400_BAD_REQUEST) try: user = CustomUser.objects.get(email=email) except CustomUser.DoesNotExist: return Response({'error': 'User not found'}, status=status.HTTP_404_NOT_FOUND) serializer = UserDetailSerializer(user) return Response({ 'message': 'User details retrieved successfully', 'user': serializer.data }, status=status.HTTP_200_OK) except Exception as e: return Response({ 'error': f'Failed to retrieve user details: {str(e)}' }, status=status.HTTP_400_BAD_REQUEST) def put(self, request): """Update user details""" try: print(request.data) email = request.data.get('email') if not email: return Response({'error': 'email is required'}, status=status.HTTP_400_BAD_REQUEST) try: user = CustomUser.objects.get(email=email) except CustomUser.DoesNotExist: return Response({'error': 'User not found'}, status=status.HTTP_404_NOT_FOUND) # Check if user is verified if not user.is_verified: return Response({'error': 'User not verified'}, status=status.HTTP_400_BAD_REQUEST) serializer = UserDetailSerializer(user, data=request.data, partial=True) if serializer.is_valid(): # Manually update the user object since we're using Serializer, not ModelSerializer validated_data = serializer.validated_data user.first_name = validated_data.get('first_name', user.first_name) user.last_name = validated_data.get('last_name', user.last_name) user.phone_number = validated_data.get('phone_number', user.phone_number) user.country_name = validated_data.get('country_name', user.country_name) user.save() # Return updated user data updated_serializer = UserDetailSerializer(user) return Response({ 'message': 'User details updated successfully', 'user': updated_serializer.data }, status=status.HTTP_200_OK) else: return Response({ 'error': 'Invalid data provided', 'details': serializer.errors }, status=status.HTTP_400_BAD_REQUEST) except Exception as e: return Response({ 'error': f'Failed to update user details: {str(e)}' }, status=status.HTTP_400_BAD_REQUEST) def post(self, request): """Add or update gender for a user by email""" try: email = request.data.get('email') gender = request.data.get('gender') if not email or not gender: return Response({'error': 'email and gender are required'}, status=status.HTTP_400_BAD_REQUEST) try: user = CustomUser.objects.get(email=email) except CustomUser.DoesNotExist: return Response({'error': 'User not found'}, status=status.HTTP_404_NOT_FOUND) if not user.is_verified: return Response({'error': 'User not verified'}, status=status.HTTP_400_BAD_REQUEST) user.gender = gender user.save() serializer = UserDetailSerializer(user) return Response({'message': 'Gender updated successfully', 'user': serializer.data}, status=status.HTTP_200_OK) except Exception as e: return Response({'error': f'Failed to update gender: {str(e)}'}, status=status.HTTP_400_BAD_REQUEST) def patch(self, request): """Update music_sound field for a user by email""" try: email = request.data.get('email') music_sound = request.data.get('music_sound') if not email: return Response({'error': 'email is required'}, status=status.HTTP_400_BAD_REQUEST) if music_sound is None: return Response({'error': 'music_sound is required'}, status=status.HTTP_400_BAD_REQUEST) # Validate music_sound is boolean if not isinstance(music_sound, bool): return Response({'error': 'music_sound must be true or false'}, status=status.HTTP_400_BAD_REQUEST) try: user = CustomUser.objects.get(email=email) except CustomUser.DoesNotExist: return Response({'error': 'User not found'}, status=status.HTTP_404_NOT_FOUND) if not user.is_verified: return Response({'error': 'User not verified'}, status=status.HTTP_400_BAD_REQUEST) # Update music_sound field directly without using serializer user.music_sound = music_sound user.save() # Return updated user data using serializer for consistent response format serializer = UserDetailSerializer(user) return Response({ 'message': 'Music sound setting updated successfully', 'user': serializer.data }, status=status.HTTP_200_OK) except Exception as e: return Response({ 'error': f'Failed to update music sound setting: {str(e)}' }, status=status.HTTP_400_BAD_REQUEST) class SkipLoginView(APIView): def post(self, request): try: username = request.data.get('username') device_id = request.data.get('device_id') # Get device_id from frontend if not username: return Response({"error": "username is required"}, status=status.HTTP_400_BAD_REQUEST) # Check if user exists with this username existing_user = CustomUser.objects.filter(last_name=username).first() if existing_user: # User already exists, use existing user user = existing_user email = user.email else: # Create email from username email = f"{username.lower()}@guest.com" # Create new user user = CustomUser.objects.create( email=email, first_name="", # Empty first name last_name=username, is_verified=True # Auto verify the user ) # Ensure user is saved and has proper ID user.save() # Store device information user_agent = request.META.get('HTTP_USER_AGENT', 'Unknown') ip_address = request.META.get('REMOTE_ADDR', '0.0.0.0') # Create device with device_id if provided device_data = { 'user_id': user.id, # Use user.id instead of user object to avoid ObjectId issues 'user_agent': user_agent, 'ip_address': ip_address } if device_id: device_data['device_id'] = device_id try: Device.objects.create(**device_data) except IntegrityError: # Handle duplicate device_id for same user (though unlikely in skip login) return Response({"error": "Device already registered"}, status=status.HTTP_400_BAD_REQUEST) return Response({"message": "logged in successfully", "email": email , "is_admin" : user.is_superuser}, status=status.HTTP_200_OK) except Exception as e: return Response({ "error": f"Failed to create user: {str(e)}" }, status=status.HTTP_400_BAD_REQUEST)