"""GroupViewSet — CRUD и операции с группами.""" from __future__ import annotations from typing import List, Optional from django.contrib.auth.models import Group, Permission from django.core.exceptions import FieldError from django.db.models import Q from rest_framework import filters, status, viewsets from rest_framework.decorators import action from rest_framework.permissions import IsAuthenticated from rest_framework.response import Response from ...permissions import AdminOrActionOrModelPerms from ..serializers import GroupSerializer from ._helpers import Employee class GroupViewSet(viewsets.ModelViewSet): """CRUD и LDAP-операции с группами. Базовые маршруты: GET/POST /api/v1/groups/ GET/PATCH/DELETE /api/v1/groups/{id}/ Экшены: permissions, set-permissions, add-permissions, remove-permissions, rename, set-description, members, add-members, remove-members, replace-members """ queryset = Group.objects.all() serializer_class = GroupSerializer pagination_class = None filter_backends = [filters.SearchFilter, filters.OrderingFilter] search_fields = ["name"] ordering_fields = ["name", "id"] ordering = ["name"] permission_classes = [IsAuthenticated, AdminOrActionOrModelPerms] required_perms_by_action = { "set_permissions": "employees.assign_group_permissions", "add_permissions": "employees.assign_group_permissions", "remove_permissions": "employees.assign_group_permissions", "rename": "employees.assign_group_permissions", "set_description": "employees.assign_group_permissions", "add_members": "employees.assign_group_permissions", "remove_members": "employees.assign_group_permissions", "replace_members": "employees.assign_group_permissions", } # ---------- queryset ---------- def get_queryset(self): qs = super().get_queryset() member_raw = self.request.query_params.get( "member" ) or self.request.query_params.get("member_id") if member_raw is None: return qs try: member_id = int(str(member_raw).strip()) except (TypeError, ValueError): return qs.none() try: return qs.filter( Q(user__id=member_id) | Q(user_set__id=member_id) ).distinct() except FieldError: try: return qs.filter(user__id=member_id).distinct() except FieldError: try: return qs.filter(user_set__id=member_id).distinct() except FieldError: return qs.none() # ---------- helpers ---------- def _validate_permissions_payload( self, request ) -> tuple[Optional[List[Permission]], Optional[Response]]: """Валидирует payload с ID permissions.""" ids = request.data.get("permissions") if not isinstance(ids, list): return None, Response( {"detail": "Поле 'permissions' должно быть списком id"}, status=status.HTTP_400_BAD_REQUEST, ) try: pid_list = [int(x) for x in ids] except (TypeError, ValueError): return None, Response( {"detail": "Список 'permissions' должен содержать целые числа"}, status=status.HTTP_400_BAD_REQUEST, ) qs = Permission.objects.filter(id__in=pid_list) if qs.count() != len(set(pid_list)): return None, Response( {"detail": "Некоторые permissions не найдены"}, status=status.HTTP_400_BAD_REQUEST, ) return list(qs), None # ---------- override CRUD ---------- def create(self, request, *args, **kwargs) -> Response: """Создаёт Group в БД. Синхронизация в LDAP через сигналы.""" ser = self.get_serializer(data=request.data) ser.is_valid(raise_exception=True) grp = Group(name=ser.validated_data["name"]) # Устанавливаем LDAP-специфичные атрибуты для сигнала grp._ldap_parent_dn = request.data.get("ldap_parent_dn") grp._ldap_description = request.data.get("ldap_description") grp._ldap_scope = request.data.get("ldap_scope", "global") grp._ldap_security_enabled = bool(request.data.get("ldap_security", True)) perms = ser.validated_data.get("permissions") grp.save() if perms: grp.permissions.set(perms) out = self.get_serializer(grp) return Response( out.data, status=status.HTTP_201_CREATED, headers=self.get_success_headers(out.data), ) def partial_update(self, request, *args, **kwargs) -> Response: """Частичное обновление Group. Синхронизация в LDAP через сигналы.""" grp = self.get_object() new_name = request.data.get("name") new_desc = request.data.get("ldap_description", "__NO_CHANGE__") # Устанавливаем LDAP-специфичные атрибуты для сигнала if new_name and new_name != grp.name: grp._ldap_old_name = grp.name if new_desc != "__NO_CHANGE__": grp._ldap_description = new_desc return super().partial_update(request, *args, **kwargs) def destroy(self, request, *args, **kwargs) -> Response: """Удаляет Group из БД. Синхронизация в LDAP через сигналы.""" return super().destroy(request, *args, **kwargs) def list(self, request, *args, **kwargs) -> Response: """Список групп.""" return super().list(request, *args, **kwargs) # ---------- Actions: Django permissions ---------- @action(detail=True, methods=["get"]) def permissions(self, request, pk=None) -> Response: """Permissions, привязанные к группе.""" grp = self.get_object() perms = grp.permissions.select_related("content_type").distinct() data = [ { "id": p.id, "codename": f"{p.content_type.app_label}.{p.codename}", "name": p.name, "app": p.content_type.app_label, "model": p.content_type.model, } for p in perms ] return Response( {"count": len(data), "results": data}, status=status.HTTP_200_OK ) @action(detail=True, methods=["post"], url_path="set-permissions") def set_permissions(self, request, pk=None) -> Response: """Полностью заменяет набор permissions у группы.""" grp = self.get_object() qs, error = self._validate_permissions_payload(request) if error: return error grp.permissions.set(qs) return Response( { "ok": True, "permission_ids": list(grp.permissions.values_list("id", flat=True)), }, status=status.HTTP_200_OK, ) @action(detail=True, methods=["post"], url_path="add-permissions") def add_permissions(self, request, pk=None) -> Response: """Добавляет permissions к группе.""" grp = self.get_object() qs, error = self._validate_permissions_payload(request) if error: return error grp.permissions.add(*qs) return Response( { "ok": True, "permission_ids": list(grp.permissions.values_list("id", flat=True)), }, status=status.HTTP_200_OK, ) @action(detail=True, methods=["post"], url_path="remove-permissions") def remove_permissions(self, request, pk=None) -> Response: """Удаляет указанные permissions у группы.""" grp = self.get_object() qs, error = self._validate_permissions_payload(request) if error: return error grp.permissions.remove(*qs) return Response( { "ok": True, "permission_ids": list(grp.permissions.values_list("id", flat=True)), }, status=status.HTTP_200_OK, ) # ---------- Actions: LDAP ---------- @action(detail=True, methods=["post"]) def rename(self, request, pk=None) -> Response: """Переименовывает группу. Синхронизация в LDAP через сигналы.""" grp = self.get_object() new_name = (request.data.get("new_name") or "").strip() if not new_name: return Response( {"detail": "new_name обязателен"}, status=status.HTTP_400_BAD_REQUEST, ) grp._ldap_old_name = grp.name grp.name = new_name grp.save(update_fields=["name"]) return Response({"ok": True, "name": grp.name}, status=status.HTTP_200_OK) @action(detail=True, methods=["post"], url_path="set-description") def set_description(self, request, pk=None) -> Response: """Устанавливает описание группы. Синхронизация в LDAP через сигналы.""" grp = self.get_object() grp._ldap_description = request.data.get("description") grp.save() return Response({"ok": True}, status=status.HTTP_200_OK) @action(detail=True, methods=["get"]) def members(self, request, pk=None) -> Response: """Состав группы.""" grp = self.get_object() users = grp.user_set.all() employees = [ { "id": u.id, "email": u.email, "first_name": u.first_name, "last_name": u.last_name, } for u in users ] return Response({"employees": employees}, status=status.HTTP_200_OK) @action(detail=True, methods=["post"], url_path="add-members") def add_members(self, request, pk=None) -> Response: """Добавляет участников в группу. Синхронизация в LDAP через сигналы.""" grp = self.get_object() member_ids = request.data.get("member_ids") or [] if not isinstance(member_ids, list): return Response( {"detail": "member_ids must be a list"}, status=status.HTTP_400_BAD_REQUEST, ) requested_ids = set(member_ids) users = Employee.objects.visible_in_ui().filter( id__in=requested_ids, is_active=True, ) if requested_ids - set(users.values_list("id", flat=True)): return Response( {"member_ids": ["Добавлять можно только активных сотрудников."]}, status=status.HTTP_400_BAD_REQUEST, ) grp.user_set.add(*users) ok_user_ids = [u.id for u in users] return Response( { "ok": True, "db_added": len(users), "ok_user_ids": ok_user_ids, }, status=status.HTTP_200_OK, ) @action(detail=True, methods=["post"], url_path="remove-members") def remove_members(self, request, pk=None) -> Response: """Удаляет участников из группы. Синхронизация в LDAP через сигналы.""" grp = self.get_object() member_ids = request.data.get("member_ids") or [] if not isinstance(member_ids, list): return Response( {"detail": "member_ids must be a list"}, status=status.HTTP_400_BAD_REQUEST, ) users = Employee.objects.filter(id__in=member_ids) grp.user_set.remove(*users) ok_user_ids = [u.id for u in users] return Response( { "ok": True, "db_removed": len(users), "ok_user_ids": ok_user_ids, }, status=status.HTTP_200_OK, ) @action(detail=True, methods=["post"], url_path="replace-members") def replace_members(self, request, pk=None) -> Response: """Полностью заменяет состав группы. Синхронизация в LDAP выполняется через сигналы. """ grp = self.get_object() member_ids = request.data.get("member_ids") or [] if not isinstance(member_ids, list): return Response( {"detail": "member_ids must be a list"}, status=status.HTTP_400_BAD_REQUEST, ) requested_ids = set(member_ids) existing_ids = set(grp.user_set.values_list("id", flat=True)) users = ( Employee.objects.visible_in_ui(include_ids=existing_ids) .filter(id__in=requested_ids) .filter(Q(is_active=True) | Q(id__in=existing_ids)) ) if requested_ids - set(users.values_list("id", flat=True)): return Response( { "member_ids": [ "Добавлять можно только активных сотрудников; " "уже состоящие в группе неактивные сотрудники могут быть сохранены." ] }, status=status.HTTP_400_BAD_REQUEST, ) grp.user_set.set(users) ok_user_ids = [u.id for u in users] return Response( { "ok": True, "db_total": len(users), "ok_user_ids": ok_user_ids, }, status=status.HTTP_200_OK, )