from rest_framework.generics import ListAPIView from rest_framework.permissions import IsAdminUser from rest_framework.viewsets import ModelViewSet from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status, serializers from rest_framework.decorators import action from django.contrib.auth import get_user_model from django.db import IntegrityError from .models import Role, Permission, UserCompanyRole from .serializers import RoleSerializer, PermissionSerializer, UserSerializer from .permissions import CompanyPermission from apps.companies.models import Company from .permissions import SuperAdminOnly from rest_framework import filters from utils.paginations import OurLimitOffsetPagination import logging User = get_user_model() logging.basicConfig(level=logging.INFO, format='%(levelname)s %(name)s %(message)s') logger = logging.getLogger(__name__) class PermissionViewSet(ModelViewSet): """Manage permissions (super admin only)""" queryset = Permission.objects.all() serializer_class = PermissionSerializer permission_classes = [SuperAdminOnly] ordering = ['name'] class CompanyRolesView(ListAPIView): """List roles for a specific company""" serializer_class = RoleSerializer permission_classes = [CompanyPermission] def get_permissions(self): """Return different permissions based on action""" company_id = self.kwargs.get('company_id') company = Company.objects.filter(id=company_id).first() return [CompanyPermission('view_roles', company) or CompanyPermission('create_users', company)] def list(self, request, *args, **kwargs): """Override list to return proper error responses""" user = request.user company_id = kwargs.get('company_id') # Check if company exists try: company = Company.objects.get(id=company_id, is_active=True) except Company.DoesNotExist: return Response( {'detail': 'Company not found.'}, status=status.HTTP_404_NOT_FOUND ) # Check permissions if not user.is_superuser: if not user.active_company: return Response( {'detail': 'No active company.'}, status=status.HTTP_400_BAD_REQUEST ) # if user.active_company.id != company_id: # return Response( # {'detail': 'Access denied.'}, # status=status.HTTP_403_FORBIDDEN # ) # Get roles roles = Role.objects.filter(company=company, is_active=True) serializer = self.get_serializer(roles, many=True) return Response(serializer.data) class RoleViewSet(ModelViewSet): """Manage roles and their permissions""" serializer_class = RoleSerializer permission_classes = [CompanyPermission] filter_backends = [filters.SearchFilter] search_fields = [ 'name', 'description', 'company__name', ] ordering = ['-created_at'] def get_permissions(self): """Return different permissions based on action""" if self.action == 'create': return [CompanyPermission('create_roles')] elif self.action in ['update', 'partial_update']: return [CompanyPermission('edit_roles')] elif self.action == 'destroy': return [CompanyPermission('delete_roles')] elif self.action == 'assign_to_user': return [CompanyPermission('assign_roles')] else: return [CompanyPermission('view_roles')] def get_queryset(self): """Filter roles based on user permissions""" user = self.request.user if user.is_superuser: company_id = self.request.query_params.get('company_id') if company_id: return Role.objects.filter(company_id=company_id, is_active=True) else: return Role.objects.filter(is_active=True) if user.active_company: return Role.objects.filter(company=user.active_company, is_active=True) return Role.objects.none() def perform_create(self, serializer): try: company_id = self.request.data.get('company') if company_id: try: company = Company.objects.get(id=company_id, is_active=True) if not self.request.user.is_superuser and not self.request.user.belongs_to_company(company): raise serializers.ValidationError({ 'company': f'You can only create roles for companies you belong to.' }) serializer.save(company=company) except Company.DoesNotExist: raise serializers.ValidationError({ 'company': f'Company with ID {company_id} not found or inactive.' }) else: serializer.save(company=self.request.user.active_company) except IntegrityError as e: if 'unique' in str(e).lower() and 'name' in str(e).lower(): raise serializers.ValidationError({ 'name': f"Role with name {serializer.validated_data.get('name')} already exists in the specified company." }) raise @action(detail=True, methods=['post']) def assign_to_user(self, request, pk=None): """Assign this role to a user""" try: # 1. Get role by ID from URL role_id = pk try: role = Role.objects.get(id=role_id, is_active=True) if (role.permissions.count() == 0): return Response( { 'detail': f'Role {role.name} has no permissions. Please assign some permissions or choose another one.'}, status=status.HTTP_400_BAD_REQUEST ) except Role.DoesNotExist: return Response( {'detail': f'Role with ID {role_id} not found or inactive.'}, status=status.HTTP_404_NOT_FOUND ) # 2. Get user by ID from request user_id = request.data.get('user_id') if not user_id: return Response( {'detail': 'user_id is required.'}, status=status.HTTP_400_BAD_REQUEST ) try: user = User.objects.get(id=user_id) except User.DoesNotExist: return Response( {'detail': f'User with ID {user_id} not found.'}, status=status.HTTP_404_NOT_FOUND ) # 3. Determine target company based on user type if request.user.is_superuser: # Superuser can specify company_id or use role's company company_id = request.data.get('company_id') if company_id: try: target_company = Company.objects.get(id=company_id, is_active=True) except Company.DoesNotExist: return Response( {'detail': f'Company with ID {company_id} not found or inactive.'}, status=status.HTTP_404_NOT_FOUND ) else: # Use role's company as fallback target_company = role.company else: # Regular user can only use their active company if not request.user.active_company: return Response( {'detail': 'You must have an active company to assign roles.'}, status=status.HTTP_400_BAD_REQUEST ) target_company = request.user.active_company # 4. Validate role belongs to target company if role.company != target_company: return Response( {'detail': f'Role {role.name} does not belong to company {target_company.name}.'}, status=status.HTTP_400_BAD_REQUEST ) # 5. Validate user belongs to target company if not user.belongs_to_company(target_company): return Response( { 'detail': f'{user.profile.first_name} {user.profile.last_name} does not belong to company {target_company.name}.'}, status=status.HTTP_400_BAD_REQUEST ) # 6. Check if user already has a role in this company existing_user_role = UserCompanyRole.objects.filter( user=user, company=target_company ).first() # If have no roles in user.active_company, set target_company to user.active_company logger.info("=============Permissions==============") has_role = UserCompanyRole.objects.filter(user=user, company=user.active_company, role__isnull=False).exists() logger.info(f"has_role: {has_role} in company: {user.active_company}") if not has_role: user.active_company = target_company user.save() if existing_user_role: if existing_user_role.role == role: return Response( { 'detail': f'{user.profile.first_name} {user.profile.last_name} already has role {role.name} in company {target_company.name}.'}, status=status.HTTP_400_BAD_REQUEST ) else: # Replace existing role existing_user_role.role = role existing_user_role.save() return Response({ 'detail': f'Role for {user.profile.first_name} {user.profile.last_name} updated to {role.name} in company {target_company.name}.', 'user_id': user.id, 'role_id': role.id, 'company_id': target_company.id, 'replaced': True }, status=status.HTTP_200_OK) # 7. Create new role assignment user_role = UserCompanyRole.objects.create( user=user, company=target_company, role=role ) return Response({ 'detail': f'Role {role.name} assigned to {user.profile.first_name} {user.profile.last_name} in company {target_company.name}.', 'user_id': user.id, 'role_id': role.id, 'company_id': target_company.id, 'created': True }, status=status.HTTP_201_CREATED) except Exception as e: return Response( {'detail': f'An error occurred: {str(e)}'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR ) class RoleUserListView(ListAPIView): """List users with a specific role""" permission_classes = [CompanyPermission] serializer_class = UserSerializer pagination_class = OurLimitOffsetPagination filter_backends = [filters.SearchFilter, filters.OrderingFilter] search_fields = ['email', 'profile__first_name', 'profile__last_name'] ordering_fields = ['email', 'profile__first_name', 'profile__last_name', 'created_at'] ordering = ['-created_at'] def get_permissions(self): """Return permissions with specific permission check""" return [CompanyPermission('view_roles')] def get_queryset(self): """Get users for the specified role""" role_id = self.kwargs.get('role_id') user = self.request.user try: role = Role.objects.get(id=role_id) except Role.DoesNotExist: return User.objects.none() if not user.is_superuser: if not user.active_company or role.company != user.active_company: return User.objects.none() user_role_ids = UserCompanyRole.objects.filter( role=role, company=role.company, is_active=True ).values_list('user_id', flat=True) return User.objects.filter( id__in=user_role_ids ).select_related('profile') class PermissionListView(APIView): """List all available permissions""" permission_classes = [CompanyPermission] def get_permissions(self): """Return permissions with specific permission check""" return [CompanyPermission('view_roles')] def get(self, request): # Only show permissions that are relevant to the user's active company # This prevents users from seeing permissions they shouldn't know about permissions = Permission.objects.filter(is_active=True) # If user is superuser, show all permissions if request.user.is_superuser: return Response(PermissionSerializer(permissions, many=True).data) # For company users, only show permissions that are used in their active company if request.user.active_company: company_permissions = Permission.objects.filter( is_active=True, roles__company=request.user.active_company ).distinct() return Response(PermissionSerializer(company_permissions, many=True).data) return Response([])