from django.shortcuts import render from rest_framework import viewsets, status from rest_framework.decorators import action from rest_framework.response import Response from django.db import transaction from django.utils import timezone from django.contrib.auth.hashers import check_password, make_password from .models import User from .serializers import UserSerializer, UserReadSerializer from ErrorLogs.utils import log_error # Create your views here. class UserViewSet(viewsets.ModelViewSet): queryset = User.objects.all() serializer_class = UserSerializer lookup_field = 'id' def get_serializer_class(self): if self.action in ['list', 'retrieve']: return UserReadSerializer return UserSerializer def get_queryset(self): """ Override queryset to filter by supervisor group when for_time_stamps parameter is present """ queryset = super().get_queryset() # Check if this is a request for time-stamps view for_time_stamps = self.request.query_params.get('for_time_stamps', '').lower() == 'true' # If for_time_stamps and user is authenticated, filter by supervisor group if for_time_stamps and self.request.user and hasattr(self.request.user, 'id'): try: from supervisorgroup.models import SupervisorGroup from userSettings.models import UserSettings # Get supervisor groups for the current logged-in user supervisor_groups = SupervisorGroup.objects.filter(user_id=self.request.user.id) if supervisor_groups.exists(): # Get all supervisor codes (groups) for this supervisor supervisor_codes = list(supervisor_groups.values_list('group', flat=True).distinct()) # Filter users by supervisor_code in their settings # Get user IDs that have matching supervisor_code user_ids = UserSettings.objects.filter( supervisor_code__in=supervisor_codes ).values_list('user_id', flat=True) # Filter queryset by these user IDs queryset = queryset.filter(id__in=user_ids) else: # If supervisor has no groups, return empty queryset queryset = queryset.none() except Exception as filter_error: # If filtering fails, log error but continue with default behavior log_error('UserViewSet.get_queryset.filter_by_supervisor', filter_error) # Continue with default queryset return queryset def list(self, request, *args, **kwargs): try: return super().list(request, *args, **kwargs) except Exception as e: log_error('UserViewSet.list', e) return Response( {"error": "An error occurred while fetching users"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) def create(self, request, *args, **kwargs): try: with transaction.atomic(): # Check if email should be sent send_email = request.data.get('send_email', False) if isinstance(send_email, str): send_email = send_email.lower() == 'true' # Generate temporary password if not provided temp_password = None if 'password' not in request.data or not request.data.get('password') or request.data.get('password') == 'default_password': from configurations.email_utils import generate_temp_password, send_welcome_email from configurations.models import AppSettings app_settings = AppSettings.get_settings() temp_password = generate_temp_password(app_settings.temp_password_length) request.data['password'] = temp_password # Set must_change_password flag based on app settings, but allow override from request if 'must_change_password' in request.data: # Explicit override provided in request request.data['must_change_password'] = request.data.get('must_change_password') in [True, 'true', 'True', 1, '1'] else: # Use app settings default request.data['must_change_password'] = app_settings.force_password_change_on_temp_password # Create user response = super().create(request, *args, **kwargs) # Send welcome email if requested and SMTP is enabled if send_email and temp_password and response.status_code == 201: try: user_data = response.data user_email = user_data.get('email') or request.data.get('email') user_name = f"{request.data.get('firstname', '')} {request.data.get('lastname', '')}".strip() if user_email and user_name and temp_password: # Get login URL from request or use default try: login_url = request.build_absolute_uri('/login') if hasattr(request, 'build_absolute_uri') else None except: login_url = None email_sent = send_welcome_email( user_email=user_email, user_name=user_name, temp_password=temp_password, login_url=login_url ) if email_sent: # Add email status to response response.data['email_sent'] = True response.data['temp_password'] = temp_password # Include temp password in response for display else: response.data['email_sent'] = False response.data['email_error'] = 'Failed to send email. Please check SMTP configuration.' except Exception as email_error: log_error('UserViewSet.create.send_email', email_error) # Don't fail user creation if email fails if hasattr(response, 'data'): response.data['email_sent'] = False response.data['email_error'] = str(email_error) return response except Exception as e: log_error('UserViewSet.create', e) return Response( {"error": "An error occurred while creating user"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) def retrieve(self, request, *args, **kwargs): try: return super().retrieve(request, *args, **kwargs) except Exception as e: log_error('UserViewSet.retrieve', e) return Response( {"error": "An error occurred while fetching user details"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) def update(self, request, *args, **kwargs): try: with transaction.atomic(): instance = self.get_object() # Handle send_email and must_change_password flags send_email = request.data.get('send_email', False) if isinstance(send_email, str): send_email = send_email.lower() == 'true' must_change_password = request.data.get('must_change_password') if isinstance(must_change_password, str): must_change_password = must_change_password.lower() == 'true' temp_password = None email_sent = False email_error = None # Generate and send temporary password if requested if send_email: from configurations.email_utils import generate_temp_password, send_welcome_email from configurations.models import AppSettings app_settings = AppSettings.get_settings() temp_password = generate_temp_password(app_settings.temp_password_length) # Set the temporary password (make_password is already imported at top) instance.password = make_password(temp_password) # Set must_change_password if explicitly provided, otherwise use app settings if must_change_password is not None: instance.must_change_password = must_change_password elif send_email: # If sending email but not explicitly set, use app settings instance.must_change_password = app_settings.force_password_change_on_temp_password instance.save() # Send welcome email try: user_email = instance.email user_name = f"{instance.firstname or ''} {instance.lastname or ''}".strip() or instance.email # Get login URL from request or use default try: login_url = request.build_absolute_uri('/login') if hasattr(request, 'build_absolute_uri') else None except: login_url = None email_sent = send_welcome_email( user_email=user_email, user_name=user_name, temp_password=temp_password, login_url=login_url ) except Exception as email_err: email_error = str(email_err) log_error('UserViewSet.update.send_email', email_err) elif must_change_password is not None: # Only update must_change_password flag if explicitly provided instance.must_change_password = must_change_password instance.save() # Perform the standard update response = super().update(request, *args, **kwargs) # Add email status to response if email was attempted if send_email: if hasattr(response, 'data') and isinstance(response.data, dict): response.data['email_sent'] = email_sent if email_error: response.data['email_error'] = email_error return response except Exception as e: log_error('UserViewSet.update', e) return Response( {"error": "An error occurred while updating user"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) def destroy(self, request, *args, **kwargs): try: with transaction.atomic(): instance = self.get_object() # Delete both user and associated settings instance.settings.delete() # This will delete the associated UserSettings instance.delete() return Response(status=status.HTTP_204_NO_CONTENT) except Exception as e: log_error('UserViewSet.destroy', e) return Response( {"error": "An error occurred while deleting user"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) @action(detail=True, methods=['patch']) def update_language(self, request, id=None): """Update user's language preference""" try: user = self.get_object() lang = request.data.get('lang') if lang not in ['en', 'fi', 'sv']: return Response( {"error": "Invalid language. Must be 'en', 'fi', or 'sv'"}, status=status.HTTP_400_BAD_REQUEST ) user.lang = lang user.save() return Response({ "message": "Language preference updated successfully", "lang": user.lang }) except Exception as e: log_error('UserViewSet.update_language', e) return Response( {"error": "An error occurred while updating language preference"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) @action(detail=True, methods=['patch']) def update_compliance_preference(self, request, id=None): """Update user's compliance monitoring preference""" try: user = self.get_object() compliance_enabled = request.data.get('compliance_enabled') if compliance_enabled is None: return Response( {"error": "compliance_enabled is required"}, status=status.HTTP_400_BAD_REQUEST ) user.compliance_enabled = compliance_enabled user.save() return Response({ "message": "Compliance preference updated successfully", "compliance_enabled": user.compliance_enabled }) except Exception as e: log_error('UserViewSet.update_compliance_preference', e) return Response( {"error": "An error occurred while updating compliance preference"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) @action(detail=True, methods=['put', 'patch', 'delete']) def working_time_policy(self, request, id=None): """Assign or remove working time policy from user""" try: user = self.get_object() if request.method == 'DELETE': # Remove policy assignment user.working_time_policy = None user.save() return Response({ "message": "Working time policy removed successfully", "working_time_policy": None }) # PUT/PATCH: Assign policy policy_id = request.data.get('policy_id') if not policy_id: return Response( {"error": "policy_id is required"}, status=status.HTTP_400_BAD_REQUEST ) from company.models import WorkingTimePolicy try: # Get company_id from user if available company_id = getattr(user, 'company_id', None) if not company_id and hasattr(request.user, 'company_id'): company_id = request.user.company_id if company_id: policy = WorkingTimePolicy.objects.get(id=policy_id, company_id=company_id) else: policy = WorkingTimePolicy.objects.get(id=policy_id) user.working_time_policy = policy user.save() from company.serializers import WorkingTimePolicySerializer policy_serializer = WorkingTimePolicySerializer(policy) return Response({ "message": "Working time policy assigned successfully", "working_time_policy": policy_serializer.data }) except WorkingTimePolicy.DoesNotExist: return Response( {"error": "Working time policy not found"}, status=status.HTTP_404_NOT_FOUND ) except Exception as e: log_error('UserViewSet.working_time_policy', e) return Response( {"error": "An error occurred while updating working time policy"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) @action(detail=True, methods=['post']) def change_password(self, request, id=None): """Change user's password""" try: user = None user_id = id or request.data.get('user_id') # Try to get user from URL parameter first if id: try: user = self.get_object() except Exception as e: log_error('UserViewSet.change_password.get_object', e) # Continue to try alternative methods # If get_object failed, try to get user from request data if not user and user_id: try: # Handle UUID format (with or without dashes) import uuid as uuid_lib try: # Try to parse as UUID if isinstance(user_id, str): # Remove dashes if present and try to parse user_id_clean = user_id.replace('-', '') if len(user_id_clean) == 32: # Format as UUID user_id_formatted = f"{user_id_clean[:8]}-{user_id_clean[8:12]}-{user_id_clean[12:16]}-{user_id_clean[16:20]}-{user_id_clean[20:]}" user_id = uuid_lib.UUID(user_id_formatted) else: user_id = uuid_lib.UUID(user_id) user = User.objects.get(id=user_id) except (ValueError, User.DoesNotExist): # Try direct lookup user = User.objects.get(id=user_id) except User.DoesNotExist: return Response( {"error": f"User not found with ID: {user_id}. Please check your user ID and try again."}, status=status.HTTP_404_NOT_FOUND ) except Exception as e: log_error('UserViewSet.change_password.get_user', e) return Response( {"error": f"Error finding user: {str(e)}"}, status=status.HTTP_400_BAD_REQUEST ) if not user: return Response( {"error": "User ID is required. Please provide a valid user ID."}, status=status.HTTP_400_BAD_REQUEST ) old_password = request.data.get('old_password') new_password = request.data.get('new_password') confirm_password = request.data.get('confirm_password') # If user must change password, we don't require old password # (because temp password was invalidated after first login) require_old_password = not user.must_change_password # Validate required fields if require_old_password and not old_password: return Response( {"error": "Old password is required"}, status=status.HTTP_400_BAD_REQUEST ) if not new_password: return Response( {"error": "New password is required"}, status=status.HTTP_400_BAD_REQUEST ) if not confirm_password: return Response( {"error": "Password confirmation is required"}, status=status.HTTP_400_BAD_REQUEST ) # Validate old password (only if required) if require_old_password: if not check_password(old_password, user.password): return Response( {"error": "Old password is incorrect"}, status=status.HTTP_400_BAD_REQUEST ) # Validate new password matches confirmation if new_password != confirm_password: return Response( {"error": "New password and confirmation do not match"}, status=status.HTTP_400_BAD_REQUEST ) # Validate new password is different from old password (only if old password is required) # When must_change_password is True, old password is already invalidated, so skip this check if require_old_password and check_password(new_password, user.password): return Response( {"error": "New password must be different from old password"}, status=status.HTTP_400_BAD_REQUEST ) # Validate password length (minimum 8 characters) if len(new_password) < 8: return Response( {"error": "New password must be at least 8 characters long"}, status=status.HTTP_400_BAD_REQUEST ) # Update password user.password = make_password(new_password) user.must_change_password = False # Clear the flag after successful password change user.password_changed_at = timezone.now() user.save() return Response({ "message": "Password changed successfully" }, status=status.HTTP_200_OK) except Exception as e: log_error('UserViewSet.change_password', e) return Response( {"error": "An error occurred while changing password"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR )