from django.contrib.auth import get_user_model from django_filters.rest_framework import DjangoFilterBackend from drf_spectacular.utils import extend_schema, inline_serializer from rest_framework import serializers, status from rest_framework.filters import OrderingFilter, SearchFilter from rest_framework.generics import ListAPIView, UpdateAPIView from rest_framework.parsers import FormParser, MultiPartParser from rest_framework.permissions import AllowAny, IsAuthenticated from rest_framework.response import Response from rest_framework.views import APIView from rest_framework_simplejwt.authentication import JWTAuthentication from rest_framework.mixins import UpdateModelMixin, RetrieveModelMixin, DestroyModelMixin from rest_framework.viewsets import GenericViewSet from core.paginations.limit_offset import CustomLimitOffsetPagination from apps.users.api.serializers import ( ChangePasswordSerializer, LoginOtpSerializer, LoginSerializer, RegisterSerializer, ResetPasswordSerializer, SendOTPSerializer, UserListSerializer, UserProfilePictureSerializer, LogoutSerializer, TokenPairSerializer, RegisterWithPasswordSerializer, UserProfileSerializer, UserSearchSerializer, ) from apps.users.services.auth import ( register_user_with_password, register_user_with_otp, generate_and_send_otp, login_with_password, login_with_otp, reset_password_with_otp, change_password, logout_user ) User = get_user_model() class RegisterWithPasswordView(APIView): """ Sign-up with mobile and password """ permission_classes = (AllowAny,) @extend_schema(request=RegisterWithPasswordSerializer, responses=TokenPairSerializer) def post(self, request): mobile = request.data.get("mobile") password = request.data.get("password") if not mobile or not password: return Response( {"detail": "mobile and password required"}, status=status.HTTP_400_BAD_REQUEST, ) tokens = register_user_with_password(mobile, password) return Response(tokens, status=status.HTTP_201_CREATED) class RegisterWithOTPView(APIView): """ A view for registring with OTP and issuing JWT tokens. """ permission_classes = (AllowAny,) @extend_schema(request=RegisterSerializer, responses=TokenPairSerializer) def post(self, request): serializer = RegisterSerializer(data=request.data) serializer.is_valid(raise_exception=True) data = serializer.validated_data.copy() data.pop("re_password", None) tokens = register_user_with_otp(**data) return Response(tokens, status=status.HTTP_201_CREATED) class SendOTPView(APIView): """ send OTP code for on of the following actions: + registrations + login + password reset """ permission_classes = (AllowAny,) @extend_schema(request=SendOTPSerializer, responses=None) def post(self, request): serializer = SendOTPSerializer(data=request.data) serializer.is_valid(raise_exception=True) generate_and_send_otp( mobile=serializer.validated_data["mobile"], mode=serializer.validated_data["mode"] ) return Response({"detail": "OTP sent successfully"}, status=status.HTTP_200_OK) class LoginView(APIView): permission_classes = (AllowAny,) @extend_schema(request=LoginSerializer, responses=TokenPairSerializer) def post(self, request): serializer = LoginSerializer(data=request.data) serializer.is_valid(raise_exception=True) tokens = login_with_password( mobile=serializer.validated_data["mobile"], password=serializer.validated_data["password"], request=request ) return Response(tokens, status=status.HTTP_200_OK) class LoginOTPView(APIView): permission_classes = (AllowAny,) @extend_schema(request=LoginOtpSerializer, responses=TokenPairSerializer) def post(self, request): serializer = LoginOtpSerializer(data=request.data) serializer.is_valid(raise_exception=True) tokens = login_with_otp( mobile=serializer.validated_data["mobile"], code=serializer.validated_data["code"], request=request ) return Response(tokens, status=status.HTTP_200_OK) class ResetPasswordView(APIView): permission_classes = (AllowAny,) serializer_class = ResetPasswordSerializer @extend_schema(request=ResetPasswordSerializer) def post(self, request): serializer = ResetPasswordSerializer(data=request.data) serializer.is_valid(raise_exception=True) reset_password_with_otp( mobile=serializer.validated_data["mobile"], code=serializer.validated_data["code"], password=serializer.validated_data["password"] ) return Response({"detail": "رمز عبور با موفقیت تغییر کرد."}, status=status.HTTP_200_OK) class ChangePasswordView(APIView): permission_classes = [IsAuthenticated] serializer_class = ChangePasswordSerializer @extend_schema(request=ChangePasswordSerializer) def patch(self, request, *args, **kwargs): serializer = ChangePasswordSerializer(data=request.data) serializer.is_valid(raise_exception=True) change_password( user=request.user, old_password=serializer.validated_data["old_password"], new_password=serializer.validated_data["new_password"] ) return Response({"detail": "رمز عبور با موفقیت تغییر کرد."}, status=status.HTTP_200_OK) class LogoutView(APIView): permission_classes = (IsAuthenticated,) serializer_class = LogoutSerializer @extend_schema( request=inline_serializer( name="LogoutRequest", fields={"refresh": serializers.CharField()} ) ) def post(self, request): refresh_token = request.data.get("refresh") logout_user(refresh_token) return Response(status=status.HTTP_205_RESET_CONTENT) class SetPasswordView(UpdateAPIView): permission_classes = [IsAuthenticated] authentication_classes = [JWTAuthentication] serializer_class = ChangePasswordSerializer @extend_schema(request=ChangePasswordSerializer, responses=None) def patch(self, request, *args, **kwargs): return super().patch(request, *args, **kwargs) def get_object(self): return self.request.user class ProfilePictureView(APIView): """ Update the authenticated user's profile picture. """ permission_classes = [IsAuthenticated] authentication_classes = [JWTAuthentication] parser_classes = [MultiPartParser, FormParser] @extend_schema( request=UserProfilePictureSerializer, responses=UserProfilePictureSerializer, operation_id="users_profile_picture_self_create", ) def post(self, request): serializer = UserProfilePictureSerializer( instance=request.user, data=request.data, context={"request": request}, partial=True, ) serializer.is_valid(raise_exception=True) serializer.save() return Response(serializer.data, status=status.HTTP_200_OK) class UserListView(ListAPIView): permission_classes = [IsAuthenticated] authentication_classes = [JWTAuthentication] serializer_class = UserListSerializer queryset = User.objects.all() pagination_class = CustomLimitOffsetPagination filter_backends = (DjangoFilterBackend, OrderingFilter, SearchFilter) search_fields = ("first_name", "last_name", "mobile") ordering = ("-created_at",) ordering_fields = ("first_name", "created_at") @extend_schema(responses=UserListSerializer(many=True)) def get(self, request, *args, **kwargs): return super().get(request, *args, **kwargs) class UserProfileViewSet(RetrieveModelMixin, UpdateModelMixin, DestroyModelMixin, GenericViewSet): serializer_class = UserProfileSerializer permission_classes = [IsAuthenticated] def get_object(self): return self.request.user class UserSearchAPIView(APIView): permission_classes = [IsAuthenticated] def get(self, request): mobile = request.query_params.get('mobile') if not mobile: return Response( {"detail": "Mobile parameter is required."}, status=status.HTTP_400_BAD_REQUEST ) user = User.objects.filter(mobile=mobile).first() if not user: return Response( {"detail": "User not found."}, status=status.HTTP_404_NOT_FOUND ) serializer = UserSearchSerializer(user, context={"request": request}) return Response(serializer.data, status=status.HTTP_200_OK)