from django.contrib.auth import get_user_model from django.contrib.auth.models import Group, Permission from django.db.models import Q from drf_spectacular.utils import extend_schema, extend_schema_view from rest_framework import status from rest_framework.decorators import action from rest_framework.response import Response from rest_framework.views import APIView from rest_framework.viewsets import ModelViewSet, ReadOnlyModelViewSet from apps.usuarios.models import Profile from apps.usuarios.permissions import ADMIN_GROUP_NAME, IsAdmin, IsSuperUser, is_admin_user from apps.usuarios.serializers import ( GroupSerializer, MeResponseSerializer, MeUpdateSerializer, PermissionSerializer, SetGroupPermissionsSerializer, SetOficinasSerializer, SetRolesSerializer, UserAdminSerializer, UserCreateSerializer, UserUpdateSerializer, UserSerializer, ) from utils.pagination import StandardResultsSetPagination User = get_user_model() class MeView(APIView): @extend_schema(tags=["Usuarios"], responses=MeResponseSerializer) def get(self, request): user = request.user return Response(self._serialize(user, request=request)) @extend_schema( tags=["Usuarios"], request=MeUpdateSerializer, responses=MeResponseSerializer, ) def patch(self, request): serializer = MeUpdateSerializer(data=request.data) serializer.is_valid(raise_exception=True) serializer.update(request.user, serializer.validated_data) return Response(self._serialize(request.user, request=request)) def _serialize(self, user, request=None): return MeResponseSerializer(user, context={"request": request}).data @extend_schema_view( list=extend_schema(tags=["Usuarios"]), retrieve=extend_schema(tags=["Usuarios"]), ) class GroupViewSet(ReadOnlyModelViewSet): permission_classes = [IsAdmin] queryset = Group.objects.all().order_by("name") serializer_class = GroupSerializer pagination_class = StandardResultsSetPagination @extend_schema(tags=["Usuarios"], responses=PermissionSerializer(many=True)) @action(detail=True, methods=["get"], url_path="permissions") def permissions(self, request, pk=None): group = self.get_object() perms = group.permissions.select_related("content_type").all().order_by( "content_type__app_label", "codename" ) return Response(PermissionSerializer(perms, many=True).data) @extend_schema(tags=["Usuarios"], request=SetGroupPermissionsSerializer, responses=PermissionSerializer(many=True)) @action( detail=True, methods=["post"], url_path="set-permissions", permission_classes=[IsSuperUser], ) def set_permissions(self, request, pk=None): group = self.get_object() serializer = SetGroupPermissionsSerializer(data=request.data) serializer.is_valid(raise_exception=True) perm_keys = serializer.validated_data["permissions"] pairs: list[tuple[str, str]] = [] invalid = [] for raw in perm_keys: s = str(raw).strip() if not s or "." not in s: invalid.append(raw) continue app_label, codename = s.split(".", 1) app_label = app_label.strip() codename = codename.strip() if not app_label or not codename: invalid.append(raw) continue pairs.append((app_label, codename)) if invalid: return Response( {"detail": "Algunos permisos tienen formato inválido.", "invalid": invalid}, status=status.HTTP_400_BAD_REQUEST, ) if not pairs: group.permissions.clear() return Response([]) q = Q() for app_label, codename in pairs: q |= Q(content_type__app_label=app_label, codename=codename) perms = list(Permission.objects.select_related("content_type").filter(q)) found = {(p.content_type.app_label, p.codename) for p in perms} missing = sorted([f"{a}.{c}" for a, c in pairs if (a, c) not in found]) if missing: return Response( {"detail": "Algunos permisos no existen.", "missing": missing}, status=status.HTTP_400_BAD_REQUEST, ) group.permissions.set(perms) perms_out = group.permissions.select_related("content_type").all().order_by( "content_type__app_label", "codename" ) return Response(PermissionSerializer(perms_out, many=True).data) @extend_schema_view( list=extend_schema(tags=["Usuarios"]), retrieve=extend_schema(tags=["Usuarios"]), ) class PermissionViewSet(ReadOnlyModelViewSet): permission_classes = [IsAdmin] serializer_class = PermissionSerializer pagination_class = StandardResultsSetPagination def get_queryset(self): qs = Permission.objects.select_related("content_type").all().order_by( "content_type__app_label", "codename" ) app_label = self.request.query_params.get("app_label") if app_label: qs = qs.filter(content_type__app_label=app_label) codename = self.request.query_params.get("codename") if codename: qs = qs.filter(codename=codename) q = self.request.query_params.get("q") if q: qs = qs.filter(Q(name__icontains=q) | Q(codename__icontains=q) | Q(content_type__app_label__icontains=q)) return qs @extend_schema_view( list=extend_schema(tags=["Usuarios"]), retrieve=extend_schema(tags=["Usuarios"]), create=extend_schema(tags=["Usuarios"]), update=extend_schema(tags=["Usuarios"]), partial_update=extend_schema(tags=["Usuarios"]), destroy=extend_schema(tags=["Usuarios"]), ) class UserViewSet(ModelViewSet): permission_classes = [IsAdmin] queryset = User.objects.all().order_by("username") filterset_fields = ["is_active", "is_staff"] search_fields = ["username", "email", "first_name", "last_name"] pagination_class = StandardResultsSetPagination def get_serializer_class(self): if self.action == "create": return UserCreateSerializer if self.action in {"update", "partial_update"}: return UserUpdateSerializer return UserAdminSerializer def perform_create(self, serializer): user = serializer.save() actor = getattr(self.request, "user", None) profile, _ = Profile.objects.get_or_create(user=user) if actor and actor.is_authenticated: if profile.created_by_id is None: profile.created_by = actor profile.updated_by = actor profile.save(update_fields=["created_by", "updated_by", "updated_at"]) @extend_schema(tags=["Usuarios"], request=SetRolesSerializer, responses=UserAdminSerializer) @action(detail=True, methods=["post"], url_path="set-roles") def set_roles(self, request, pk=None): user = self.get_object() serializer = SetRolesSerializer(data=request.data) serializer.is_valid(raise_exception=True) group_names = serializer.validated_data["groups"] current_admin = is_admin_user(user) will_be_admin = ADMIN_GROUP_NAME in group_names or user.is_superuser if current_admin and not will_be_admin: admin_group = Group.objects.filter(name=ADMIN_GROUP_NAME).first() if admin_group: admins_count = ( User.objects.filter(is_active=True) .filter(Q(is_superuser=True) | Q(groups=admin_group)) .distinct() .count() ) if admins_count <= 1: return Response( { "detail": "No se puede remover el rol 'Administrador' porque quedarías sin administradores activos." }, status=status.HTTP_400_BAD_REQUEST, ) groups = list(Group.objects.filter(name__in=group_names)) missing = sorted(set(group_names) - set([g.name for g in groups])) if missing: return Response( {"detail": "Algunos grupos no existen.", "missing": missing}, status=status.HTTP_400_BAD_REQUEST, ) user.groups.set(groups) out = self.get_serializer(user).data return Response(out) @extend_schema(tags=["Usuarios"], request=SetOficinasSerializer, responses=UserAdminSerializer) @action(detail=True, methods=["post"], url_path="set-oficinas") def set_oficinas(self, request, pk=None): user = self.get_object() serializer = SetOficinasSerializer(data=request.data) serializer.is_valid(raise_exception=True) office_ids = serializer.validated_data["oficinas"] from apps.catalogos.models import Oficina oficinas = list(Oficina.objects.filter(id__in=office_ids)) found = {o.id for o in oficinas} missing = sorted([oid for oid in set(office_ids) if oid not in found]) if missing: return Response( {"detail": "Algunas oficinas no existen.", "missing": missing}, status=status.HTTP_400_BAD_REQUEST, ) profile, _ = Profile.objects.get_or_create(user=user) profile.oficinas.set(oficinas) out = self.get_serializer(user).data return Response(out)