from rest_framework import status from rest_framework.response import Response from rest_framework.decorators import api_view, permission_classes from rest_framework.permissions import IsAuthenticated, AllowAny from drf_spectacular.utils import extend_schema, OpenApiParameter from rest_framework_simplejwt.tokens import RefreshToken from django.contrib.auth.tokens import PasswordResetTokenGenerator from django.utils.http import urlsafe_base64_encode, urlsafe_base64_decode from django.utils.encoding import force_bytes, force_str from django.core.mail import send_mail from django.conf import settings from django.utils.dateparse import parse_date, parse_datetime from django.utils import timezone from datetime import timedelta from .models import User, LoginSession from .serializers import ( RegisterSerializer, LoginSerializer, ChangePasswordSerializer, ForgotPasswordSerializer, ResetPasswordSerializer, UserSerializer, LoginResponseSerializer, LoginSessionSerializer ) @extend_schema(request=RegisterSerializer, responses=UserSerializer, tags=['Account']) @api_view(['POST']) @permission_classes([AllowAny]) def register(request): """User registration endpoint""" serializer = RegisterSerializer(data=request.data) if serializer.is_valid(): user = serializer.save() return Response({ 'message': 'User registered successfully', 'user': UserSerializer(user).data }, status=status.HTTP_201_CREATED) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) @extend_schema( request=LoginSerializer, responses={200: LoginResponseSerializer}, tags=['Account'] ) @api_view(['POST']) @permission_classes([AllowAny]) def login(request): """User login endpoint""" serializer = LoginSerializer(data=request.data) if serializer.is_valid(): user = serializer.validated_data['user'] branch_id = serializer.validated_data.get('branch_id') # Get IP address ip_address = request.META.get('REMOTE_ADDR', '') user_agent = request.META.get('HTTP_USER_AGENT', '') # Create login session session = LoginSession.objects.create( user=user, branch_id=branch_id, ip_address=ip_address, user_agent=user_agent ) refresh = RefreshToken.for_user(user) return Response({ 'message': 'Login successful', 'access': str(refresh.access_token), 'refresh': str(refresh), 'user': UserSerializer(user).data, 'session': LoginSessionSerializer(session).data }, status=status.HTTP_200_OK) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) @extend_schema( request=ChangePasswordSerializer, responses={200: {'type': 'object', 'properties': {'message': {'type': 'string'}}}}, tags=['Account'] ) @api_view(['POST']) @permission_classes([IsAuthenticated]) def change_password(request): """Change password endpoint""" serializer = ChangePasswordSerializer(data=request.data, context={'request': request}) if serializer.is_valid(): user = request.user user.set_password(serializer.validated_data['new_password']) user.save() return Response({ 'message': 'Password changed successfully' }, status=status.HTTP_200_OK) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) @extend_schema( request=ForgotPasswordSerializer, responses={200: {'type': 'object', 'properties': {'message': {'type': 'string'}}}}, tags=['Account'] ) @api_view(['POST']) @permission_classes([AllowAny]) def forgot_password(request): """Forgot password endpoint - sends reset link via email""" serializer = ForgotPasswordSerializer(data=request.data) if serializer.is_valid(): email = serializer.validated_data['email'] user = User.objects.get(email=email) # Generate token token_generator = PasswordResetTokenGenerator() token = token_generator.make_token(user) uid = urlsafe_base64_encode(force_bytes(user.pk)) # Create reset link reset_link = f"{settings.FRONTEND_URL}/reset-password/{uid}/{token}/" # Send email subject = 'Password Reset Request' message = f'Click the link to reset your password: {reset_link}' send_mail( subject, message, settings.DEFAULT_FROM_EMAIL, [email], fail_silently=False, ) return Response({ 'message': 'Password reset link sent to your email' }, status=status.HTTP_200_OK) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) @extend_schema( request=ResetPasswordSerializer, responses={200: {'type': 'object', 'properties': {'message': {'type': 'string'}}}}, tags=['Account'] ) @api_view(['POST']) @permission_classes([AllowAny]) def reset_password(request): """Reset password endpoint - validates token and resets password""" serializer = ResetPasswordSerializer(data=request.data) if serializer.is_valid(): token = serializer.validated_data['token'] uid = request.data.get('uid') new_password = serializer.validated_data['new_password'] try: user_id = force_str(urlsafe_base64_decode(uid)) user = User.objects.get(pk=user_id) except (TypeError, ValueError, OverflowError, User.DoesNotExist): return Response({ 'error': 'Invalid user ID' }, status=status.HTTP_400_BAD_REQUEST) token_generator = PasswordResetTokenGenerator() if not token_generator.check_token(user, token): return Response({ 'error': 'Invalid or expired token' }, status=status.HTTP_400_BAD_REQUEST) user.set_password(new_password) user.save() return Response({ 'message': 'Password reset successfully' }, status=status.HTTP_200_OK) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) @extend_schema(responses=UserSerializer, tags=['Account']) @api_view(['GET']) @permission_classes([IsAuthenticated]) def profile(request): """Get current user profile""" return Response(UserSerializer(request.user).data) @extend_schema(request=UserSerializer, responses=UserSerializer, tags=['Account']) @api_view(['PUT']) @permission_classes([IsAuthenticated]) def update_profile(request): """Update user profile""" user = request.user serializer = UserSerializer(user, data=request.data, partial=True) if serializer.is_valid(): serializer.save() return Response({ 'message': 'Profile updated successfully', 'user': serializer.data }, status=status.HTTP_200_OK) return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST) def _get_date_range(request): """Helper function to parse date range from query parameters""" start_date = request.query_params.get('start_date') end_date = request.query_params.get('end_date') parsed_start = None parsed_end = None if start_date: parsed_start = parse_datetime(start_date) or parse_date(start_date) if parsed_start and not hasattr(parsed_start, 'hour'): # If date object parsed_start = timezone.make_aware(timezone.datetime.combine(parsed_start, timezone.datetime.min.time())) if end_date: parsed_end = parse_datetime(end_date) or parse_date(end_date) if parsed_end and not hasattr(parsed_end, 'hour'): # If date object parsed_end = timezone.make_aware(timezone.datetime.combine(parsed_end, timezone.datetime.max.time())) return parsed_start, parsed_end @extend_schema( parameters=[ OpenApiParameter( name='start_date', description='Filter sessions from this date (format: YYYY-MM-DD or ISO 8601)', required=False, type=str, ), OpenApiParameter( name='end_date', description='Filter sessions until this date (format: YYYY-MM-DD or ISO 8601)', required=False, type=str, ), OpenApiParameter( name='user_id', description='Filter by user ID (admin only)', required=False, type=int, ), OpenApiParameter( name='branch_id', description='Filter by branch ID', required=False, type=int, ), OpenApiParameter( name='active_only', description='Show only active sessions (no logout_time). Values: true/false', required=False, type=bool, ), OpenApiParameter( name='limit', description='Number of results per page (default: 50)', required=False, type=int, ), OpenApiParameter( name='offset', description='Number of results to skip (default: 0)', required=False, type=int, ), ], responses={200: {'type': 'object'}}, tags=['Account'] ) @api_view(['GET']) @permission_classes([IsAuthenticated]) def login_sessions(request): """ Get login sessions with advanced filtering and pagination. Regular users can only see their own sessions. Admin users can filter by any user using user_id parameter. """ from rest_framework.pagination import LimitOffsetPagination # Base queryset - regular users see only their sessions if request.user.is_staff: sessions = LoginSession.objects.all() else: sessions = LoginSession.objects.filter(user=request.user) # Filter by user_id (admin only) user_id = request.query_params.get('user_id') if user_id: if request.user.is_staff: try: sessions = sessions.filter(user_id=int(user_id)) except ValueError: return Response( {'error': 'Invalid user_id'}, status=status.HTTP_400_BAD_REQUEST ) else: return Response( {'error': 'Only admins can filter by user_id'}, status=status.HTTP_403_FORBIDDEN ) # Filter by branch_id branch_id = request.query_params.get('branch_id') if branch_id: try: sessions = sessions.filter(branch_id=int(branch_id)) except ValueError: return Response( {'error': 'Invalid branch_id'}, status=status.HTTP_400_BAD_REQUEST ) # Filter by date range start_date, end_date = _get_date_range(request) if start_date: sessions = sessions.filter(login_time__gte=start_date) if end_date: sessions = sessions.filter(login_time__lte=end_date) # Filter active sessions only (no logout_time) active_only = request.query_params.get('active_only', '').lower() if active_only in ['true', '1', 'yes']: sessions = sessions.filter(logout_time__isnull=True) # Order by login_time descending sessions = sessions.order_by('-login_time') # Apply pagination paginator = LimitOffsetPagination() paginator.page_size = request.query_params.get('limit', 50) paginated_sessions = paginator.paginate_queryset(sessions, request) serializer = LoginSessionSerializer(paginated_sessions, many=True) return paginator.get_paginated_response(serializer.data) @extend_schema( request=None, responses={200: {'type': 'object', 'properties': {'message': {'type': 'string'}}}}, tags=['Account'] ) @api_view(['POST']) @permission_classes([IsAuthenticated]) def logout(request): """Logout endpoint - records logout time for current session""" session_id = request.data.get('session_id') if not session_id: return Response( {'error': 'session_id required'}, status=status.HTTP_400_BAD_REQUEST ) try: session = LoginSession.objects.get( id=session_id, user=request.user, logout_time__isnull=True ) session.logout_time = timezone.now() session.save() return Response({ 'message': 'Logout successful', 'session': LoginSessionSerializer(session).data }, status=status.HTTP_200_OK) except LoginSession.DoesNotExist: return Response( {'error': 'Session not found or already logged out'}, status=status.HTTP_404_NOT_FOUND )