import secrets from datetime import timedelta from django.conf import settings from django.contrib.auth import get_user_model from django.core.cache import cache from django.core.mail import send_mail from django.utils import timezone from rest_framework import status from rest_framework.permissions import AllowAny from rest_framework.response import Response from rest_framework.views import APIView from rest_framework_simplejwt.tokens import RefreshToken from .models import OTPCode from .models import GymSettings from .permissions import IsGymOwner from rest_framework.permissions import IsAuthenticated from .serializers import ( ChangePasswordSerializer, ForgotPasswordSerializer, GymSettingsSerializer, LoginSerializer, RegisterSerializer, ResetPasswordSerializer, UpdateProfileSerializer, UserSerializer, VerifyOTPSerializer, ) User = get_user_model() OTP_EXPIRY_MINUTES = 5 MAX_OTP_ATTEMPTS = 5 # lock after 5 wrong attempts OTP_LOCKOUT_MINUTES = 15 # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def generate_otp(): """Cryptographically secure 6-digit OTP.""" return str(secrets.randbelow(900000) + 100000) def send_otp_email(email, code): send_mail( subject="Your FitMtaani verification code", message=f"Your verification code is: {code}\n\nThis code expires in {OTP_EXPIRY_MINUTES} minutes.", from_email=settings.DEFAULT_FROM_EMAIL, recipient_list=[email], fail_silently=False, ) def _otp_attempt_key(email): return f"otp_attempts:{email}" def _login_attempt_key(email): return f"login_attempts:{email}" GENERIC_AUTH_ERROR = "Invalid email or password." # --------------------------------------------------------------------------- # Login # --------------------------------------------------------------------------- class LoginView(APIView): """Validate credentials and send OTP to the user's email.""" permission_classes = [AllowAny] def post(self, request): serializer = LoginSerializer(data=request.data) serializer.is_valid(raise_exception=True) email = serializer.validated_data["email"] password = serializer.validated_data["password"] # Check login attempt rate limit attempt_key = _login_attempt_key(email) attempts = cache.get(attempt_key, 0) if attempts >= MAX_OTP_ATTEMPTS: return Response( {"detail": f"Too many attempts. Try again in {OTP_LOCKOUT_MINUTES} minutes."}, status=status.HTTP_429_TOO_MANY_REQUESTS, ) try: user = User.objects.get(email=email) except User.DoesNotExist: # Generic message — don't reveal whether email exists cache.set(attempt_key, attempts + 1, OTP_LOCKOUT_MINUTES * 60) return Response( {"detail": GENERIC_AUTH_ERROR}, status=status.HTTP_401_UNAUTHORIZED, ) if not user.check_password(password): cache.set(attempt_key, attempts + 1, OTP_LOCKOUT_MINUTES * 60) return Response( {"detail": GENERIC_AUTH_ERROR}, status=status.HTTP_401_UNAUTHORIZED, ) # Reset login attempts on success cache.delete(attempt_key) # Invalidate any existing unused OTPs OTPCode.objects.filter(user=user, is_used=False).update(is_used=True) # Generate and save new OTP code = generate_otp() OTPCode.objects.create( user=user, code=code, expires_at=timezone.now() + timedelta(minutes=OTP_EXPIRY_MINUTES), ) # Send OTP via email send_otp_email(user.email, code) return Response( {"detail": "Verification code sent to your email.", "email": email}, status=status.HTTP_200_OK, ) # --------------------------------------------------------------------------- # OTP Verification # --------------------------------------------------------------------------- class VerifyOTPView(APIView): """Verify OTP and return JWT tokens.""" permission_classes = [AllowAny] def post(self, request): serializer = VerifyOTPSerializer(data=request.data) serializer.is_valid(raise_exception=True) email = serializer.validated_data["email"] code = serializer.validated_data["code"] # Check OTP attempt rate limit attempt_key = _otp_attempt_key(email) attempts = cache.get(attempt_key, 0) if attempts >= MAX_OTP_ATTEMPTS: return Response( {"detail": f"Too many failed attempts. Try again in {OTP_LOCKOUT_MINUTES} minutes."}, status=status.HTTP_429_TOO_MANY_REQUESTS, ) try: user = User.objects.get(email=email) except User.DoesNotExist: return Response( {"detail": "Invalid or expired verification code."}, status=status.HTTP_400_BAD_REQUEST, ) # Find the most recent valid OTP otp = ( OTPCode.objects.filter(user=user, code=code, is_used=False) .order_by("-created_at") .first() ) if not otp or not otp.is_valid: # Increment failed attempt counter cache.set(attempt_key, attempts + 1, OTP_LOCKOUT_MINUTES * 60) return Response( {"detail": "Invalid or expired verification code."}, status=status.HTTP_400_BAD_REQUEST, ) # Mark OTP as used otp.is_used = True otp.save(update_fields=["is_used"]) # Reset attempt counter cache.delete(attempt_key) # Generate JWT tokens refresh = RefreshToken.for_user(user) return Response( { "access": str(refresh.access_token), "refresh": str(refresh), "user": UserSerializer(user).data, }, status=status.HTTP_200_OK, ) # --------------------------------------------------------------------------- # Registration — only for gym owners # --------------------------------------------------------------------------- class RegisterView(APIView): """Register a new gym owner account.""" permission_classes = [AllowAny] def post(self, request): serializer = RegisterSerializer(data=request.data) serializer.is_valid(raise_exception=True) data = serializer.validated_data name_parts = data["name"].strip().split(" ", 1) first_name = name_parts[0] last_name = name_parts[1] if len(name_parts) > 1 else "" # All registrations create gym owners (role=owner). # Staff and members are created by owners through the dashboard. user = User.objects.create_user( username=data["email"], email=data["email"], password=data["password"], first_name=first_name, last_name=last_name, phone=data.get("phone", ""), role=User.Role.OWNER, ) # Save gym name if provided — per-gym settings gym_name = data.get("gym_name", "").strip() if gym_name: from .models import GymSettings GymSettings.objects.create(gym=user, gym_name=gym_name) else: from .models import GymSettings GymSettings.objects.create(gym=user) # Auto-create subscription if package_id provided (for free packages) package_id = data.get("package_id", "").strip() if package_id: try: from apps.billing.models import Subscription, SubscriptionPackage package = SubscriptionPackage.objects.filter(id=package_id).first() Subscription.objects.create( user=user, package=package, package_name=package.name if package else "Free", amount=package.price if package else 0, billing_cycle="monthly", payment_reference=f"free_{user.id}", paid_at=timezone.now(), expires_at=None, # Free subscriptions never expire ) except Exception: # If package_id is not a valid UUID, still create a free sub try: from apps.billing.models import Subscription Subscription.objects.create( user=user, package=None, package_name="Free", amount=0, billing_cycle="monthly", payment_reference=f"free_{user.id}", paid_at=timezone.now(), expires_at=None, ) except Exception: pass # Generate JWT tokens refresh = RefreshToken.for_user(user) return Response( { "access": str(refresh.access_token), "refresh": str(refresh), "user": UserSerializer(user).data, }, status=status.HTTP_201_CREATED, ) # --------------------------------------------------------------------------- # Check email (safe — no existence leak) # --------------------------------------------------------------------------- class CheckEmailView(APIView): """Check email format. Does NOT reveal whether an account exists.""" permission_classes = [AllowAny] def post(self, request): # Always return the same response to prevent user enumeration return Response({"valid": True}) # --------------------------------------------------------------------------- # Admin list (superuser only) # --------------------------------------------------------------------------- class AdminListView(APIView): """List, update, or delete gym owner accounts. Superuser only.""" def _check_superuser(self, request): if not request.user.is_superuser: return Response( {"detail": "Not authorized."}, status=status.HTTP_403_FORBIDDEN, ) return None def get(self, request): denied = self._check_superuser(request) if denied: return denied admins = User.objects.filter(role__in=[User.Role.ADMIN, User.Role.OWNER]).order_by("-date_joined") return Response(UserSerializer(admins, many=True).data) def patch(self, request): """Update a gym owner. Expects { id, ...fields }.""" denied = self._check_superuser(request) if denied: return denied user_id = request.data.get("id") if not user_id: return Response({"detail": "id is required."}, status=status.HTTP_400_BAD_REQUEST) try: user = User.objects.get(pk=user_id, role__in=[User.Role.ADMIN, User.Role.OWNER]) except User.DoesNotExist: return Response({"detail": "User not found."}, status=status.HTTP_404_NOT_FOUND) serializer = UpdateProfileSerializer(user, data=request.data, partial=True) serializer.is_valid(raise_exception=True) serializer.save() return Response(UserSerializer(user).data) def delete(self, request): """Delete a gym owner. Expects { id }.""" denied = self._check_superuser(request) if denied: return denied user_id = request.data.get("id") if not user_id: return Response({"detail": "id is required."}, status=status.HTTP_400_BAD_REQUEST) try: user = User.objects.get(pk=user_id, role__in=[User.Role.ADMIN, User.Role.OWNER]) except User.DoesNotExist: return Response({"detail": "User not found."}, status=status.HTTP_404_NOT_FOUND) if user == request.user: return Response({"detail": "Cannot delete yourself."}, status=status.HTTP_400_BAD_REQUEST) user.delete() return Response(status=status.HTTP_204_NO_CONTENT) class AdminDetailView(APIView): """Get gym owner details including their members. Superuser only.""" def get(self, request, pk): if not request.user.is_superuser: return Response({"detail": "Not authorized."}, status=status.HTTP_403_FORBIDDEN) try: user = User.objects.get(pk=pk, role__in=[User.Role.ADMIN, User.Role.OWNER]) except User.DoesNotExist: return Response({"detail": "User not found."}, status=status.HTTP_404_NOT_FOUND) from apps.members.models import MemberProfile from apps.members.serializers import MemberProfileSerializer members = MemberProfile.objects.select_related("user").filter(user__gym=user) return Response({ "owner": UserSerializer(user).data, "members": MemberProfileSerializer(members, many=True).data, "member_count": members.count(), "staff_count": user.gym_users.exclude(role__in=[User.Role.MEMBER]).count(), }) class DashboardStatsView(APIView): """Return dashboard stats for the current gym owner.""" def get(self, request): from django.db.models import Q, Count from apps.members.models import MemberProfile from apps.staff.models import StaffProfile user = request.user gym_owner = None if user.is_superuser: # Superuser: all data total_members = MemberProfile.objects.count() active_members = MemberProfile.objects.filter(status="active").count() total_staff = StaffProfile.objects.filter(is_active=True).count() else: from .permissions import get_gym_owner gym_owner = get_gym_owner(user) owner = gym_owner if (gym_owner and gym_owner != user) else user member_qs = MemberProfile.objects.filter(user__gym=owner) total_members = member_qs.count() active_members = member_qs.filter(status="active").count() total_staff = StaffProfile.objects.filter( user__gym=owner, is_active=True, ).count() return Response({ "total_members": total_members, "active_members": active_members, "inactive_members": total_members - active_members, "total_staff": total_staff, }) # --------------------------------------------------------------------------- # Change Password (authenticated) # --------------------------------------------------------------------------- class ChangePasswordView(APIView): """Allow authenticated users to change their password.""" permission_classes = [IsAuthenticated] def post(self, request): serializer = ChangePasswordSerializer(data=request.data) serializer.is_valid(raise_exception=True) user = request.user if not user.check_password(serializer.validated_data["current_password"]): return Response( {"detail": "Current password is incorrect."}, status=status.HTTP_400_BAD_REQUEST, ) user.set_password(serializer.validated_data["new_password"]) user.save() return Response({"detail": "Password changed successfully."}) # --------------------------------------------------------------------------- # Forgot Password (public — sends OTP) # --------------------------------------------------------------------------- class ForgotPasswordView(APIView): """Send a password-reset OTP to the user's email.""" permission_classes = [AllowAny] def post(self, request): serializer = ForgotPasswordSerializer(data=request.data) serializer.is_valid(raise_exception=True) email = serializer.validated_data["email"] # Rate limit attempt_key = f"forgot_pw:{email}" attempts = cache.get(attempt_key, 0) if attempts >= MAX_OTP_ATTEMPTS: return Response( {"detail": f"Too many attempts. Try again in {OTP_LOCKOUT_MINUTES} minutes."}, status=status.HTTP_429_TOO_MANY_REQUESTS, ) # Always return success to avoid email enumeration try: user = User.objects.get(email=email) except User.DoesNotExist: return Response({"detail": "If that email exists, a reset code has been sent."}) # Invalidate existing OTPs OTPCode.objects.filter(user=user, is_used=False).update(is_used=True) code = generate_otp() OTPCode.objects.create( user=user, code=code, expires_at=timezone.now() + timedelta(minutes=OTP_EXPIRY_MINUTES), ) send_mail( subject="Password Reset Code", message=f"Your password reset code is: {code}\n\nThis code expires in {OTP_EXPIRY_MINUTES} minutes.\n\nIf you did not request this, please ignore this email.", from_email=settings.DEFAULT_FROM_EMAIL, recipient_list=[email], fail_silently=True, ) cache.set(attempt_key, attempts + 1, OTP_LOCKOUT_MINUTES * 60) return Response({"detail": "If that email exists, a reset code has been sent."}) # --------------------------------------------------------------------------- # Reset Password (public — verifies OTP + sets new password) # --------------------------------------------------------------------------- class ResetPasswordView(APIView): """Verify OTP and set a new password.""" permission_classes = [AllowAny] def post(self, request): serializer = ResetPasswordSerializer(data=request.data) serializer.is_valid(raise_exception=True) email = serializer.validated_data["email"] code = serializer.validated_data["code"] new_password = serializer.validated_data["new_password"] # Rate limit attempt_key = _otp_attempt_key(f"reset:{email}") attempts = cache.get(attempt_key, 0) if attempts >= MAX_OTP_ATTEMPTS: return Response( {"detail": f"Too many failed attempts. Try again in {OTP_LOCKOUT_MINUTES} minutes."}, status=status.HTTP_429_TOO_MANY_REQUESTS, ) try: user = User.objects.get(email=email) except User.DoesNotExist: return Response( {"detail": "Invalid or expired reset code."}, status=status.HTTP_400_BAD_REQUEST, ) otp = ( OTPCode.objects.filter(user=user, code=code, is_used=False) .order_by("-created_at") .first() ) if not otp or not otp.is_valid: cache.set(attempt_key, attempts + 1, OTP_LOCKOUT_MINUTES * 60) return Response( {"detail": "Invalid or expired reset code."}, status=status.HTTP_400_BAD_REQUEST, ) otp.is_used = True otp.save(update_fields=["is_used"]) cache.delete(attempt_key) user.set_password(new_password) user.save() return Response({"detail": "Password has been reset successfully. You can now log in."}) # --------------------------------------------------------------------------- # Me # --------------------------------------------------------------------------- class MeView(APIView): """Return or update the current authenticated user.""" def get(self, request): return Response(UserSerializer(request.user).data) def patch(self, request): serializer = UpdateProfileSerializer(request.user, data=request.data, partial=True) serializer.is_valid(raise_exception=True) serializer.save() return Response(UserSerializer(request.user).data) class GymSettingsView(APIView): """Get or update gym settings (owner only).""" permission_classes = [IsGymOwner] def _get_owner(self, request): from .permissions import get_gym_owner owner = get_gym_owner(request.user) return owner if (owner and owner != request.user) else request.user def get(self, request): owner = self._get_owner(request) settings_obj, _ = GymSettings.objects.get_or_create(gym=owner) return Response(GymSettingsSerializer(settings_obj).data) def patch(self, request): owner = self._get_owner(request) settings_obj, _ = GymSettings.objects.get_or_create(gym=owner) serializer = GymSettingsSerializer(settings_obj, data=request.data, partial=True) serializer.is_valid(raise_exception=True) serializer.save() return Response(GymSettingsSerializer(settings_obj).data)