""" Management API for the dynamic RBAC. All management endpoints require the caller to hold the relevant ``system_administration.roles_permissions.*`` permission (super admins pass implicitly). Responses follow the project-wide ``{message, payload}`` shape. """ from django.db import transaction from rest_framework.decorators import ( api_view, authentication_classes, permission_classes, ) from rest_framework.authentication import TokenAuthentication from rest_framework.permissions import IsAuthenticated from rest_framework.response import Response from human_resource.models import StaffProfile from .models import Module, Permission, Role, RolePermission, StaffRole from .serializers import ( ModuleSerializer, RoleListSerializer, RoleDetailSerializer, StaffRoleAssignmentSerializer, ) from .permissions import ( get_staff_profile, requires_permission, get_effective_permission_codenames, is_unrestricted, ) from .abilities import build_ability_rules, effective_permissions MANAGE_VIEW = 'system_administration.roles_permissions.view' MANAGE_ADD = 'system_administration.roles_permissions.add' MANAGE_EDIT = 'system_administration.roles_permissions.edit' MANAGE_DELETE = 'system_administration.roles_permissions.delete' def _company_for(staff_profile): branch = getattr(staff_profile, 'company_branch', None) return branch.company_profile if branch is not None else None def _company_roles(staff_profile): company = _company_for(staff_profile) return Role.objects.filter(company_profile=company, recycle_bin=False) # --- catalog --------------------------------------------------------------- @api_view(['GET']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) @requires_permission(MANAGE_VIEW) def get_permission_catalog(request): """Full Module -> Feature -> Permission tree for the matrix UI.""" modules = Module.objects.filter(is_active=True).prefetch_related( 'features__permissions') data = ModuleSerializer(modules, many=True).data return Response({'message': 'ok', 'payload': {'modules': data}}, status=200) # --- roles ----------------------------------------------------------------- @api_view(['GET']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) @requires_permission(MANAGE_VIEW) def list_roles(request): staff_profile = get_staff_profile(request.user) roles = _company_roles(staff_profile).order_by('name') data = RoleListSerializer(roles, many=True).data return Response({'message': 'ok', 'payload': {'roles': data}}, status=200) @api_view(['GET']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) @requires_permission(MANAGE_VIEW) def get_role(request): staff_profile = get_staff_profile(request.user) role_id = request.query_params.get('role_id') or request.data.get('role_id') role = _company_roles(staff_profile).filter(id=role_id).first() if role is None: return Response({'message': 'Role not found.', 'payload': {}}, status=404) return Response( {'message': 'ok', 'payload': {'role': RoleDetailSerializer(role).data}}, status=200, ) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) @requires_permission(MANAGE_ADD) @transaction.atomic def create_role(request): staff_profile = get_staff_profile(request.user) company = _company_for(staff_profile) name = (request.data.get('name') or '').strip() if not name: return Response({'message': 'Role name is required.', 'payload': {}}, status=400) if Role.objects.filter(company_profile=company, name__iexact=name, recycle_bin=False).exists(): return Response({'message': 'A role with that name already exists.', 'payload': {}}, status=409) role = Role.objects.create( company_profile=company, name=name, description=(request.data.get('description') or '').strip(), created_by=staff_profile, last_updated_by=staff_profile, ) _set_role_permissions(role, request.data.get('permission_codenames')) return Response( {'message': 'Role created.', 'payload': {'role': RoleDetailSerializer(role).data}}, status=201, ) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) @requires_permission(MANAGE_EDIT) @transaction.atomic def update_role(request): staff_profile = get_staff_profile(request.user) role = _company_roles(staff_profile).filter(id=request.data.get('role_id')).first() if role is None: return Response({'message': 'Role not found.', 'payload': {}}, status=404) name = request.data.get('name') if name is not None: name = name.strip() if not name: return Response({'message': 'Role name cannot be empty.', 'payload': {}}, status=400) clash = Role.objects.filter( company_profile=role.company_profile, name__iexact=name, recycle_bin=False, ).exclude(id=role.id).exists() if clash: return Response({'message': 'A role with that name already exists.', 'payload': {}}, status=409) role.name = name if 'description' in request.data: role.description = (request.data.get('description') or '').strip() role.last_updated_by = staff_profile role.save() if 'permission_codenames' in request.data: _set_role_permissions(role, request.data.get('permission_codenames')) return Response( {'message': 'Role updated.', 'payload': {'role': RoleDetailSerializer(role).data}}, status=200, ) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) @requires_permission(MANAGE_DELETE) def delete_role(request): staff_profile = get_staff_profile(request.user) role = _company_roles(staff_profile).filter(id=request.data.get('role_id')).first() if role is None: return Response({'message': 'Role not found.', 'payload': {}}, status=404) if role.is_system: return Response( {'message': 'System roles cannot be deleted. You can edit their permissions instead.', 'payload': {}}, status=403, ) role.recycle_bin = True role.last_updated_by = staff_profile role.save() StaffRole.objects.filter(role=role).update(is_active=False) return Response({'message': 'Role deleted.', 'payload': {}}, status=200) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) @requires_permission(MANAGE_ADD) @transaction.atomic def clone_role(request): staff_profile = get_staff_profile(request.user) source = _company_roles(staff_profile).filter(id=request.data.get('role_id')).first() if source is None: return Response({'message': 'Role not found.', 'payload': {}}, status=404) new_name = (request.data.get('name') or f'{source.name} (copy)').strip() if Role.objects.filter(company_profile=source.company_profile, name__iexact=new_name, recycle_bin=False).exists(): return Response({'message': 'A role with that name already exists.', 'payload': {}}, status=409) clone = Role.objects.create( company_profile=source.company_profile, name=new_name, description=source.description, created_by=staff_profile, last_updated_by=staff_profile, ) RolePermission.objects.bulk_create([ RolePermission(role=clone, permission_id=pid) for pid in source.permissions.values_list('id', flat=True) ]) return Response( {'message': 'Role cloned.', 'payload': {'role': RoleDetailSerializer(clone).data}}, status=201, ) def _set_role_permissions(role, codenames): """Replace a role's permission set with the given codenames (idempotent).""" if codenames is None: return perm_ids = list( Permission.objects.filter(codename__in=codenames, is_active=True) .values_list('id', flat=True) ) RolePermission.objects.filter(role=role).delete() RolePermission.objects.bulk_create([ RolePermission(role=role, permission_id=pid) for pid in perm_ids ]) # --- staff assignments ----------------------------------------------------- @api_view(['GET']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) @requires_permission(MANAGE_VIEW) def list_staff_assignments(request): """Staff in the caller's company with their active role assignments.""" staff_profile = get_staff_profile(request.user) company = _company_for(staff_profile) staff_qs = StaffProfile.objects.filter( company_branch__company_profile=company, recycle_bin=False, ).select_related('company_branch', 'company_department').order_by('first_name') assignments = StaffRole.objects.filter( staff_profile__in=staff_qs, role__recycle_bin=False, ).select_related('role') by_staff = {} for sr in assignments: by_staff.setdefault(sr.staff_profile_id, []).append({ 'staff_role_id': sr.id, 'role_id': sr.role_id, 'role_name': sr.role.name, 'is_active': sr.is_active, }) payload = [] for sp in staff_qs: payload.append({ 'staff_id': sp.id, 'staff_number': sp.staff_number, 'name': f'{sp.first_name} {sp.last_name}'.strip(), 'department': sp.company_department.department_name if sp.company_department else None, 'branch': sp.company_branch.branch_name if sp.company_branch else None, 'roles': by_staff.get(sp.id, []), }) return Response({'message': 'ok', 'payload': {'staff': payload}}, status=200) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) @requires_permission(MANAGE_EDIT) def assign_role(request): staff_profile = get_staff_profile(request.user) company = _company_for(staff_profile) target = StaffProfile.objects.filter( id=request.data.get('staff_id'), company_branch__company_profile=company, ).first() role = _company_roles(staff_profile).filter(id=request.data.get('role_id')).first() if target is None or role is None: return Response({'message': 'Staff member or role not found.', 'payload': {}}, status=404) sr, _ = StaffRole.objects.get_or_create( staff_profile=target, role=role, defaults={'assigned_by': staff_profile, 'is_active': True}, ) if not sr.is_active: sr.is_active = True sr.assigned_by = staff_profile sr.save() if 'main_branch_only' in request.data: sr.main_branch_only = bool(request.data.get('main_branch_only')) sr.save() return Response( {'message': 'Role assigned.', 'payload': {'assignment': StaffRoleAssignmentSerializer(sr).data}}, status=200, ) @api_view(['POST']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) @requires_permission(MANAGE_EDIT) def revoke_role(request): staff_profile = get_staff_profile(request.user) company = _company_for(staff_profile) sr = StaffRole.objects.filter( id=request.data.get('staff_role_id'), staff_profile__company_branch__company_profile=company, ).first() if sr is None: # allow revoke by (staff_id, role_id) too sr = StaffRole.objects.filter( staff_profile_id=request.data.get('staff_id'), role_id=request.data.get('role_id'), staff_profile__company_branch__company_profile=company, ).first() if sr is None: return Response({'message': 'Assignment not found.', 'payload': {}}, status=404) sr.delete() return Response({'message': 'Role revoked.', 'payload': {}}, status=200) # --- current user ---------------------------------------------------------- @api_view(['GET']) @authentication_classes([TokenAuthentication]) @permission_classes([IsAuthenticated]) def my_permissions(request): """Effective permissions + CASL rules for the calling user (no extra perm required — every authenticated user may read their own access).""" staff_profile = get_staff_profile(request.user) if staff_profile is None: return Response({'message': 'No staff profile.', 'payload': {}}, status=404) return Response({ 'message': 'ok', 'payload': { 'permissions': effective_permissions(staff_profile), 'ability_rules': build_ability_rules(staff_profile), 'is_unrestricted': is_unrestricted(staff_profile), }, }, status=200)