import csv import os from datetime import datetime from rest_framework.response import Response from rest_framework.views import APIView from .serializers import * from rest_framework_simplejwt.tokens import RefreshToken from django.contrib.auth import authenticate, login, logout from rest_framework import status from rest_framework.permissions import AllowAny from .models import * from django.conf import settings from django.core.files.storage import default_storage class RegistrationAPIView(APIView): permission_classes = [AllowAny] def post(self, request): serializer = RegisterSerializer(data=request.data) if serializer.is_valid(): user = serializer.save() refresh = RefreshToken.for_user(user) return Response({ 'message': 'Registration successful', 'access_token': str(refresh.access_token), 'refresh_token': str(refresh) }, status=status.HTTP_201_CREATED) else: return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) class LoginView(APIView): authentication_classes = [] permission_classes = [] def post(self, request, *args, **kwargs): username = request.data.get("username") # Changed to username password = request.data.get("password") user = authenticate(request, username=username, password=password) if user: login(request, user) refresh = RefreshToken.for_user(user) return Response({ 'message': 'Login successful', 'access_token': str(refresh.access_token), 'refresh_token': str(refresh), 'user': UserProfileSerializer(user).data }, status=status.HTTP_200_OK) else: return Response( {"error": "Invalid credentials"}, status=status.HTTP_401_UNAUTHORIZED ) class LogoutAPIView(APIView): def post(self, request): logout(request) return Response({'message': 'Logged out successfully'}, status=status.HTTP_200_OK) class UserProfileAPIView(APIView): def get(self, request): serializer = UserProfileSerializer(request.user) return Response(serializer.data, status=status.HTTP_200_OK) def put(self, request, pk): try: user = CustomUser.objects.get(pk=pk) except CustomUser.DoesNotExist: return Response({"error": "User not found."}, status=status.HTTP_404_NOT_FOUND) serializer = RegisterSerializer(user, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response({"message": "User details updated successfully.", "data": serializer.data}, status=status.HTTP_200_OK) return Response({"error": serializer.errors}, status=status.HTTP_400_BAD_REQUEST) class UserListAPIView(APIView): def get(self, request): user_obj = CustomUser.objects.exclude(is_staff=True).order_by('-id') serializer = UserProfileSerializer(user_obj,many=True) return Response(serializer.data, status=status.HTTP_200_OK) class ChangePasswordAPIView(APIView): def post(self, request): user = request.user current_password = request.data.get('current_password') new_password = request.data.get('new_password') if user.check_password(current_password): user.set_password(new_password) user.save() return Response({'message': 'Password changed successfully'}, status=status.HTTP_200_OK) else: return Response({'message': 'Invalid current password'}, status=status.HTTP_400_BAD_REQUEST) # class BulkUserUploadAPIView(APIView): # def post(self, request): # if "file" not in request.FILES: # return Response({"error": "No file uploaded"}, status=status.HTTP_400_BAD_REQUEST) # file = request.FILES["file"] # success_count = 0 # error_entries = [] # timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") # error_filename = f"error_users_{timestamp}.csv" # errors_file_path = os.path.join(settings.MEDIA_ROOT, "uploads", error_filename) # try: # file_content = file.read().decode('utf-8') # reader = csv.DictReader(file_content.splitlines()) # for row in reader: # email = row.get("email") # username = row.get("username") # phone_number = row.get("phone_number") # first_name = row.get("first_name") # last_name = row.get("last_name") # if not username or not email or not phone_number: # error_entries.append([username, email, "Missing required fields"]) # continue # if CustomUser.objects.filter(username=username).exists(): # error_entries.append([username, email, "Username already exists"]) # continue # if CustomUser.objects.filter(email=email).exists(): # error_entries.append([username, email, "Email already exists"]) # continue # try: # user_data = { # "username": username, # "first_name": first_name, # "last_name": last_name, # "phone_number": phone_number, # "email": email, # "password": f"{username}@123", # Auto-generate password # "dob": row.get("dob"), # "gender": row.get("gender"), # } # serializer = RegisterSerializer(data=user_data) # if serializer.is_valid(): # serializer.save() # success_count += 1 # else: # error_entries.append([username, email, str(serializer.errors)]) # except Exception as e: # error_entries.append([username, email, str(e)]) # except Exception as e: # return Response({"error": f"Error processing the file: {str(e)}"}, status=status.HTTP_400_BAD_REQUEST) # # Save errors to CSV # if error_entries: # upload_dir = os.path.join(settings.MEDIA_ROOT, "uploads") # # Ensure the directory exists # if not os.path.exists(upload_dir): # os.makedirs(upload_dir, exist_ok=True) # with open(errors_file_path, mode="w", encoding="utf-8", newline="") as error_file: # writer = csv.writer(error_file) # writer.writerow(["Username", "Email", "Error"]) # CSV headers # writer.writerows(error_entries) # error_file_url = f"{settings.MEDIA_URL}uploads/{error_filename}" # else: # error_file_url = None # return Response( # { # "message": f"{success_count} users created successfully.", # "error_file": error_file_url, # }, # status=status.HTTP_201_CREATED, # )