from rest_framework import status, viewsets from rest_framework.decorators import action from rest_framework.response import Response from rest_framework.views import APIView from .serializers import ( UsersSerializerWrite, UsersSerializerRead, ApproveUsersSerializer, EmailAndPasswordSerializer, EmailSerializer, ChangePasswordSerializer, UserInvitationSerializer, UserInvitationCustomSerializer, UserStaffRegisterSerializer, RegisterSerializer, ) from .models import Users, CustomUsersPagination, UserInvitation from django.db.models import Q from rest_framework.permissions import IsAdminUser, IsAuthenticated from products.permissions import IsAdminOrReadOnly import random, string from django_filters.rest_framework import DjangoFilterBackend from rest_framework.filters import SearchFilter from rest_framework_simplejwt.tokens import RefreshToken from django.contrib.auth.hashers import check_password, make_password from django.core.mail import send_mail from django.template.loader import render_to_string from utils.validators import PasswordValidator from drf_yasg.utils import swagger_auto_schema from django.conf import settings from django.utils import timezone import logging, os from utils.mixins.user_mixins import UserViewSetMixin from utils.email_templates import ( staff_invitation_email, user_registration_notification_email, designer_invitation_email, reset_password_email, ) LOGGER = logging.getLogger(__name__) ENVIRONMENT = settings.ENVIRONMENT IS_HTTPS = True if ENVIRONMENT.lower() in ["dev", "development"]: IS_HTTPS = False # Create your views here. class LoginAPIView(APIView): permission_classes = () authentication_classes = () serializer_class = EmailAndPasswordSerializer def post(self, request): data = request.data serializer = self.serializer_class(data=data) serializer.is_valid(raise_exception=True) validated_data = serializer.validated_data email = validated_data["email"] password = validated_data["password"] try: existing_user = Users.objects.get(is_active=True, email=email) is_password_correct = check_password( password=password, encoded=existing_user.password ) if not is_password_correct: return Response( {"message": "Invalid password. Try again."}, status=status.HTTP_400_BAD_REQUEST, ) refresh = RefreshToken.for_user(existing_user) response = Response( { "message": "Login successful.", "user": UsersSerializerRead(existing_user, many=False).data, }, status=status.HTTP_200_OK, ) response.set_cookie( key="access_token", value=str(refresh.access_token), httponly=True, secure=IS_HTTPS, samesite="Lax", ) response.set_cookie( key="refresh_token", value=str(refresh), httponly=True, secure=IS_HTTPS, samesite="Lax", ) return response except Users.DoesNotExist: return Response( {"message": f"User {email} does not exists, or not yet active."}, status=status.HTTP_404_NOT_FOUND, ) class LogoutAPIView(APIView): permission_classes = [IsAuthenticated] def post(self, request): user = request.user response = Response( {"message": f"User '{user.email}' logged out successfully."}, status=status.HTTP_200_OK, ) # Remove tokens in the cookies and HTTP_AUTHORIZATION response.delete_cookie( "access_token", path="/", domain=None, samesite="Lax", ) response.delete_cookie( "refresh_token", path="/", domain=None, samesite="Lax", ) request.META.pop("HTTP_AUTHORIZATION", None) return response class MeAPIView(APIView): permission_classes = [IsAuthenticated] def get(self, request): user = request.user serialized_data = UsersSerializerRead(user, many=False).data serialized_data["is_authenticated"] = user.is_authenticated return Response(serialized_data, status=status.HTTP_200_OK) class RegisterAPIView(APIView): permission_classes = () serializer_class = RegisterSerializer def post(self, request, *args, **kwargs): data = request.data init_email = data.get("email") existing_user = Users.objects.filter(email=init_email) if existing_user.exists(): return Response( {"message": f"User {init_email} already exists."}, status=status.HTTP_400_BAD_REQUEST, ) serializer = self.serializer_class(data=data) serializer.is_valid(raise_exception=True) validated_data = serializer.validated_data terms_aggrement = validated_data.pop("terms_agreement") if terms_aggrement is False: return Response( { "message": "You must agree to the terms and conditions before registering." }, status=status.HTTP_403_FORBIDDEN, ) try: password = validated_data.get("password") confirm_password = validated_data.pop("confirm_password") pw_validator = PasswordValidator( new_password=password, confirm_password=confirm_password ) pw_validator.validate_password_similarity() pw_validator.validate_password_length() pw_validator.validate_password_complexity() Users.objects.create(is_active=False, **validated_data) is_temp_email_service = bool(os.environ.get("IS_TEMP_EMAIL_SERVICE", False)) if is_temp_email_service: user_registration_notification_email(details=validated_data) else: validated_data["birthdate"] = validated_data["birthdate"].strftime( "%B %-d, %Y, %-I:%M%p" ) email_html_message = render_to_string( "email/staff_invitation.html", validated_data ) send_mail( subject="Caracole Designer Registration", message="", from_email=settings.EMAIL_HOST_USER, recipient_list=settings.INQUIRY_RECEIVER, html_message=email_html_message, auth_user=settings.EMAIL_HOST_USER, auth_password=settings.EMAIL_HOST_PASSWORD, ) return Response( { "message": "Account created. Please wait for the approval to be one of our designers." }, status=status.HTTP_201_CREATED, ) except Exception as e: return Response( { "message": "Error occured during registration.", "detailed_error": str(e), }, status.HTTP_400_BAD_REQUEST, ) class ForgotPasswordAPIView(APIView): permission_classes = () authentication_classes = () serializer_class = EmailSerializer @swagger_auto_schema( request_body=EmailSerializer, responses={ 200: "Reset password link sent to your email.", 404: "User does not exists.", }, ) def post(self, request): data = request.data serializer = self.serializer_class(data=data) serializer.is_valid(raise_exception=True) validated_data = serializer.validated_data email = validated_data["email"] try: existing_user = Users.objects.get(Q(email=email) & (Q(is_active=True))) code = string.ascii_letters + string.digits reset_password_token = "".join(random.choices(code, k=10)) existing_user.reset_password_token = reset_password_token existing_user.save() context = { "link": f"{settings.FRONTEND_URL}/reset-password/{reset_password_token}", } is_temp_email_service = bool(os.environ.get("IS_TEMP_EMAIL_SERVICE", False)) if is_temp_email_service: context["email"] = email reset_password_email(details=context) else: email_html_message = render_to_string( "email/reset_password.html", context ) send_mail( subject="Reset Your Password", message="", from_email=settings.EMAIL_HOST_USER, recipient_list=[existing_user.email], html_message=email_html_message, auth_user=settings.EMAIL_HOST_USER, auth_password=settings.EMAIL_HOST_PASSWORD, ) return Response( {"message": "Reset password link sent to your email."}, status=status.HTTP_200_OK, ) except Users.DoesNotExist: return Response( {"message": f"User {email} does not exists."}, status=status.HTTP_404_NOT_FOUND, ) class ResetPasswordAPIView(APIView): permission_classes = () authentication_classes = () serializer_class = ChangePasswordSerializer @swagger_auto_schema( request_body=ChangePasswordSerializer, responses={ 200: "Password successfully changed.", 400: "Token not found. Please request a new reset password link | Token is required to reset your password | Other Password Validations | Other Errors", }, ) def post(self, request): data = request.data token = data.get("token", None) if not token: return Response( {"message": "Token is required to reset your password."}, status=status.HTTP_400_BAD_REQUEST, ) serializer = self.serializer_class(data=data) serializer.is_valid(raise_exception=True) validated_data = serializer.validated_data try: token = validated_data["token"] new_password = validated_data["new_password"] confirm_password = validated_data["confirm_password"] existing_user_with_token = Users.objects.get(reset_password_token=token) password_validator = PasswordValidator( new_password=new_password, confirm_password=confirm_password ) password_validator.validate_password_similarity() password_validator.validate_new_and_old_password_similarity( user=existing_user_with_token ) password_validator.validate_password_length() password_validator.validate_password_complexity() existing_user_with_token.reset_password_token = None existing_user_with_token.password = make_password(confirm_password) existing_user_with_token.save() return Response( { "message": "Password successfully changed", }, status.HTTP_200_OK, ) except Users.DoesNotExist: return Response( { "message": "Token not found. Please request a new reset password link." }, status=status.HTTP_400_BAD_REQUEST, ) except Exception as e: return Response( {"message": f"Error occurred: {str(e)}"}, status=status.HTTP_400_BAD_REQUEST, ) class ChangePasswordAPIView(APIView): permission_classes = [IsAuthenticated] serializer_class = ChangePasswordSerializer def post(self, request): data = request.data user: Users = request.user old_password = data.get("old_password", None) if old_password is None: return Response( {"message": "Old password is required to change your password."}, status=status.HTTP_400_BAD_REQUEST, ) serializer = self.serializer_class(data=data) serializer.is_valid(raise_exception=True) validated_data = serializer.validated_data try: old_password = validated_data["old_password"] new_password = validated_data["new_password"] confirm_password = validated_data["confirm_password"] password_validator = PasswordValidator( new_password=new_password, confirm_password=confirm_password, old_password=old_password, ) password_validator.validate_old_password(user=user) password_validator.validate_password_similarity() password_validator.validate_password_length() password_validator.validate_password_complexity() user.password = make_password(confirm_password) user.save() return Response( {"message": "Password successfully changed"}, status=status.HTTP_200_OK, ) except Exception as e: return Response( {"message": f"Error occurred: {str(e)}"}, status=status.HTTP_400_BAD_REQUEST, ) class UsersViewset(viewsets.ModelViewSet, UserViewSetMixin): filter_backends = [DjangoFilterBackend, SearchFilter] filterset_fields = ["is_approved", "role", "is_active"] search_fields = ["first_name", "last_name", "email", "company", "mobile_number"] permission_classes = [IsAdminOrReadOnly] pagination_class = CustomUsersPagination http_method_names = ["get", "post", "put", "patch"] def get_serializer_context(self): context = super().get_serializer_context() user = self.request.user if isinstance(user, Users): context["user"] = user else: context["user"] = None return context def get_serializer_class(self): method = self.request.method if method == "POST": return UsersSerializerWrite else: return UsersSerializerRead def get_queryset(self): user = self.request.user queryset = Users.objects.none() # Default queryset if isinstance(user, Users): is_admin = user.is_staff or user.is_superuser if user.role.lower() == "maintainer": queryset = Users.objects.filter(role__iexact="designer").order_by( "last_name" ) elif is_admin: queryset = Users.objects.exclude(id=user.id).order_by("last_name") else: queryset = Users.objects.filter(email=user.email).order_by("last_name") return queryset def create(self, request, *args, **kwargs): data = request.data email = data.get("email") existing_email = Users.objects.filter(email=email) if existing_email.exists(): return Response( {"message": f"A user with email '{email}' already exists."}, status.HTTP_400_BAD_REQUEST, ) return super().create(request, *args, **kwargs) class UserApprovalAPI(APIView): permission_classes = [IsAdminOrReadOnly] serializer_class = ApproveUsersSerializer def post(self, request): """ Approve or Reject a user based on the status input. 'status' key value should be 'approve' or 'reject' only """ data = request.data serializer = self.serializer_class(data=data) serializer.is_valid(raise_exception=True) validated_data = serializer.validated_data try: user_id = validated_data["id"] approval_status = validated_data["status"] user = Users.objects.get(id=user_id) if approval_status.lower() == "reject": user.is_approved = False user.is_active = False response_message = "was rejected." elif approval_status.lower() == "approve": user.is_approved = True user.is_active = True response_message = "was approved." context = { "name": f"{user.first_name} {user.last_name}", "caracole_url": "https://caracole.ph/", } is_temp_email_service = bool( os.environ.get("IS_TEMP_EMAIL_SERVICE", False) ) if is_temp_email_service: context["email"] = user.email designer_invitation_email(details=context) else: email_html_message = render_to_string( "email/invitation.html", context ) send_mail( subject="Welcome to Caracole PH", message="", from_email=settings.EMAIL_HOST_USER, recipient_list=[user.email], html_message=email_html_message, auth_user=settings.EMAIL_HOST_USER, auth_password=settings.EMAIL_HOST_PASSWORD, ) else: return Response( {"message": "Invalid status. 'approve' or 'reject' only."}, status=status.HTTP_400_BAD_REQUEST, ) user.save() return Response( { "id": user.id, "is_approved": user.is_approved, "message": f"{user.email} {response_message}", }, status=status.HTTP_200_OK, ) except Users.DoesNotExist: return Response( {"message": "User not found."}, status=status.HTTP_404_NOT_FOUND ) class UserInvitationViewset(viewsets.ModelViewSet): queryset = UserInvitation.objects.order_by("-created_at") permission_classes = [IsAdminUser] http_method_names = ["post", "get"] def get_serializer_class(self): method = self.request.method path = self.request.path if "register" in path: return UserStaffRegisterSerializer if method == "POST": return UserInvitationCustomSerializer return UserInvitationSerializer @swagger_auto_schema( request_body=UserInvitationCustomSerializer, responses={ 200: "Invitation sent to email", 404: "Account is already active or unexpired invitation already sent", 400: "Invalid Password", }, ) def create(self, request, *args, **kwargs): data = request.data invitor = request.user serializer = self.get_serializer(data=data) serializer.is_valid(raise_exception=True) validated_data = serializer.validated_data user_email = validated_data["email"] first_name = validated_data["first_name"] last_name = validated_data["last_name"] role = validated_data["role"] existing_user = Users.objects.filter(email=user_email) if existing_user.exists(): existing_user = existing_user.first() if existing_user.is_active: return Response( { "message": f"Account with email '{user_email}' is already an active user." }, status=status.HTTP_400_BAD_REQUEST, ) existing_invitation = UserInvitation.objects.filter(invitee__email=user_email) if existing_invitation.exists(): existing_invitee: UserInvitation = existing_invitation.first() if not existing_invitee.is_expired: expires_at = existing_invitee.expires_at local_timezone = timezone.get_default_timezone() expires_at_local = expires_at.astimezone(local_timezone) expires_at = expires_at_local.strftime("%Y-%m-%d %H:%M:%S") return Response( { "message": f"An unexpired invitation for {user_email} has already been sent.", "expires_at": expires_at, }, status=status.HTTP_400_BAD_REQUEST, ) LOGGER.info(f"Deleting expired invitation for {user_email}.") existing_invitee.delete() existing_invitee.invitee.delete() else: existing_user.delete() # Delete to prevent interrupting the invitation process invitation_link = f"{settings.ADMIN_FRONTEND_URL}/user-invite?email={user_email}&first_name={first_name}&last_name={last_name}" context = { "url": invitation_link, "name": f"{first_name} {last_name}", "role": role, } is_temp_email_service = bool(os.environ.get("IS_TEMP_EMAIL_SERVICE", False)) initial_user = Users( first_name=first_name, last_name=last_name, email=user_email, is_staff=True, is_active=False, role=role.title(), ) initial_user._is_invite = True initial_user._invitor = invitor initial_user._invitation_link = invitation_link initial_user.save() if is_temp_email_service: context["email"] = user_email staff_invitation_email(details=context) else: email_html_message = render_to_string( "email/staff_invitation.html", context ) send_mail( subject="Account Invitation", message="", from_email=settings.EMAIL_HOST_USER, recipient_list=[initial_user.email], html_message=email_html_message, auth_user=settings.EMAIL_HOST_USER, auth_password=settings.EMAIL_HOST_PASSWORD, ) LOGGER.info(f"Invitation sent for {user_email}.") return Response( {"message": f"Invitation sent to {user_email}."}, status=status.HTTP_200_OK ) @swagger_auto_schema( request_body=UserStaffRegisterSerializer, responses={ 200: "User Registered Successfully.", 404: "No active invitation found or user does not exists.", 400: "Password Exceptions | Account is already active, not an invited staff, or invitation already expired.", }, ) @action( detail=False, methods=["POST"], permission_classes=[], url_path="register", ) def invitee_register(self, request): data = request.data serializer = self.get_serializer(data=data) serializer.is_valid(raise_exception=True) validated_data = serializer.validated_data user_email = validated_data["email"] password = validated_data["password"] confirm_password = validated_data["confirm_password"] try: existing_user = Users.objects.get(email=user_email) existing_invitation_for_email = UserInvitation.objects.get( invitee=existing_user ) if existing_user.is_active: return Response( { "message": f"Account with email '{user_email}' is already active." }, status=status.HTTP_400_BAD_REQUEST, ) if not existing_user.is_staff: return Response( { "message": f"Account with email '{user_email}' is not an invited staff account." }, status=status.HTTP_400_BAD_REQUEST, ) if existing_invitation_for_email.is_expired: return Response( { "message": "The invitation is already expired. Please request for another invitation." }, status=status.HTTP_400_BAD_REQUEST, ) pw_validator = PasswordValidator( new_password=password, confirm_password=confirm_password ) pw_validator.validate_password_similarity() pw_validator.validate_password_length() pw_validator.validate_password_complexity() existing_user.password = Users.hash_plain_password( plain_password=confirm_password ) existing_user.is_active = True existing_user.save() LOGGER.info(f"User '{user_email}' registered successfully.") return Response( {"message": f"User '{user_email}' registered successfully."}, status=status.HTTP_201_CREATED, ) except UserInvitation.DoesNotExist: return Response( {"message": f"No active invitation found for email '{user_email}'."}, status=status.HTTP_404_NOT_FOUND, ) except Users.DoesNotExist: return Response( {"message": f"User with email '{user_email}' does not exist."}, status=status.HTTP_404_NOT_FOUND, ) except Exception as e: return Response({"message": str(e)}, status=status.HTTP_400_BAD_REQUEST)