182 lines
7.3 KiB
Python
182 lines
7.3 KiB
Python
from django.shortcuts import render
|
|
from rest_framework import viewsets
|
|
from rest_framework.pagination import PageNumberPagination
|
|
from django_filters.rest_framework import DjangoFilterBackend
|
|
from rest_framework.filters import SearchFilter, OrderingFilter
|
|
from rest_framework.permissions import IsAuthenticated
|
|
from rest_framework.decorators import action
|
|
from rest_framework.response import Response
|
|
from django.http import FileResponse, Http404
|
|
|
|
from .serializers import VucemSerializer, CredencialesImportadorSerializer, CredencialesImportadorSimpleSerializer
|
|
from rest_framework import serializers
|
|
|
|
# Serializer para update donde key y cer no son requeridos
|
|
class VucemUpdateSerializer(VucemSerializer):
|
|
key = serializers.FileField(required=False, allow_null=True)
|
|
cer = serializers.FileField(required=False, allow_null=True)
|
|
|
|
class Meta(VucemSerializer.Meta):
|
|
fields = VucemSerializer.Meta.fields
|
|
from .models import Vucem, CredencialesImportador
|
|
from core.permissions import IsSameOrganizationDeveloper
|
|
from rest_framework import mixins
|
|
|
|
from core.permissions import (
|
|
IsSameOrganization,
|
|
IsSameOrganizationDeveloper,
|
|
IsSameOrganizationAndAdmin,
|
|
IsSuperUser,
|
|
IsSameOrganizationAndInAllowedGroups
|
|
)
|
|
|
|
class CustomVucemPagination(PageNumberPagination):
|
|
"""
|
|
Paginación personalizada para VUCEM
|
|
"""
|
|
page_size = None # Sin paginación por defecto
|
|
page_size_query_param = 'page_size'
|
|
max_page_size = 1000
|
|
page_query_param = 'page'
|
|
|
|
def paginate_queryset(self, queryset, request, view=None):
|
|
page_size = request.query_params.get(self.page_size_query_param)
|
|
if page_size is None:
|
|
return None
|
|
return super().paginate_queryset(queryset, request, view)
|
|
# Create your views here.
|
|
|
|
class VucemView(viewsets.ModelViewSet):
|
|
permission_classes = [IsAuthenticated , (IsSuperUser | IsSameOrganization | IsSameOrganizationAndAdmin | IsSameOrganizationDeveloper )]
|
|
|
|
queryset = Vucem.objects.all()
|
|
pagination_class = CustomVucemPagination
|
|
filterset_fields = ['organizacion', 'patente', 'usuario', 'is_importador', 'acusecove', 'acuseedocument', 'is_active']
|
|
search_fields = ['usuario', 'patente']
|
|
ordering_fields = ['created_at', 'updated_at', 'usuario', 'patente']
|
|
ordering = ['-created_at']
|
|
|
|
def get_serializer_class(self):
|
|
if self.action in ['update', 'partial_update']:
|
|
return VucemUpdateSerializer
|
|
return VucemSerializer
|
|
|
|
def get_permissions(self):
|
|
if self.action in ['create', 'update', 'partial_update', 'destroy']:
|
|
return [IsAuthenticated(), IsSameOrganizationAndInAllowedGroups()]
|
|
return super().get_permissions()
|
|
|
|
def get_queryset(self):
|
|
# Verificar que el usuario esté autenticado y tenga organización
|
|
if not self.request.user.is_authenticated:
|
|
return self.queryset.none()
|
|
|
|
queryset = self.queryset
|
|
|
|
if self.request.user.is_superuser:
|
|
queryset = queryset.all()
|
|
elif not hasattr(self.request.user, 'organizacion') or not self.request.user.organizacion:
|
|
return queryset.none()
|
|
elif self.request.user.groups.filter(name='Importador').exists():
|
|
queryset = queryset.filter(organizacion=self.request.user.organizacion, usuario=self.request.user.rfc)
|
|
else:
|
|
queryset = queryset.filter(organizacion=self.request.user.organizacion)
|
|
|
|
# Filtro por importador (RFC)
|
|
importador_rfc = self.request.query_params.get('importador')
|
|
if importador_rfc:
|
|
queryset = queryset.filter(usuarios_importadores__rfc__rfc=importador_rfc).distinct()
|
|
|
|
return queryset
|
|
|
|
def perform_create(self, serializer):
|
|
if not self.request.user.is_authenticated or not hasattr(self.request.user, 'organizacion'):
|
|
raise ValueError("El usuario debe estar autenticado y tener una organización asignada.")
|
|
if self.request.user.is_superuser:
|
|
serializer.save(created_by=self.request.user, updated_by=self.request.user)
|
|
return
|
|
else:
|
|
serializer.save(
|
|
organizacion=self.request.user.organizacion,
|
|
created_by=self.request.user,
|
|
updated_by=self.request.user
|
|
)
|
|
return
|
|
|
|
def perform_update(self, serializer):
|
|
if not self.request.user.is_authenticated or not hasattr(self.request.user, 'organizacion'):
|
|
raise ValueError("El usuario debe estar autenticado y tener una organización asignada.")
|
|
instance = self.get_object()
|
|
if self.request.user.is_superuser:
|
|
serializer.save(
|
|
created_by=instance.created_by,
|
|
updated_by=self.request.user
|
|
)
|
|
return
|
|
else:
|
|
serializer.save(
|
|
organizacion=self.request.user.organizacion,
|
|
created_by=instance.created_by,
|
|
updated_by=self.request.user
|
|
)
|
|
return
|
|
|
|
@action(detail=True, methods=["get"], permission_classes=[IsAuthenticated])
|
|
def download_cer(self, request, pk=None):
|
|
"""
|
|
Descarga directa del archivo cer.
|
|
"""
|
|
vucem = self.get_object()
|
|
if not vucem.cer:
|
|
return Response({"detail": "No hay archivo cer disponible."}, status=404)
|
|
response = FileResponse(vucem.cer.open('rb'), as_attachment=True, filename=vucem.cer.name.split('/')[-1])
|
|
return response
|
|
|
|
@action(detail=True, methods=["get"], permission_classes=[IsAuthenticated])
|
|
def download_key(self, request, pk=None):
|
|
"""
|
|
Descarga directa del archivo key.
|
|
"""
|
|
vucem = self.get_object()
|
|
if not vucem.key:
|
|
return Response({"detail": "No hay archivo key disponible."}, status=404)
|
|
response = FileResponse(vucem.key.open('rb'), as_attachment=True, filename=vucem.key.name.split('/')[-1])
|
|
return response
|
|
|
|
|
|
class CredencialesImportadorViewSet(viewsets.ModelViewSet):
|
|
permission_classes = [IsAuthenticated]
|
|
queryset = CredencialesImportador.objects.all()
|
|
serializer_class = CredencialesImportadorSimpleSerializer
|
|
filterset_fields = ['organizacion', 'vucem', 'rfc']
|
|
search_fields = ['rfc']
|
|
ordering_fields = ['created_at', 'updated_at', 'rfc']
|
|
ordering = ['-created_at']
|
|
|
|
my_tags = ['Credenciales por Importador']
|
|
|
|
def get_permissions(self):
|
|
if self.action in ['create', 'update', 'partial_update', 'destroy']:
|
|
return [IsAuthenticated()]
|
|
return super().get_permissions()
|
|
|
|
def get_queryset(self):
|
|
|
|
if self.request.user.is_superuser:
|
|
# Si es superusuario, devolver todos los registros
|
|
return self.queryset.all()
|
|
|
|
# Verificar que el usuario esté autenticado y tenga organización
|
|
if not self.request.user.is_authenticated or not hasattr(self.request.user, 'organizacion'):
|
|
return self.queryset.none()
|
|
|
|
queryset = self.queryset.filter(organizacion=self.request.user.organizacion)
|
|
|
|
|
|
return queryset
|
|
|
|
def perform_create(self, serializer):
|
|
if not self.request.user.is_authenticated or not hasattr(self.request.user, 'organizacion'):
|
|
raise ValueError("El usuario debe estar autenticado y tener una organización asignada.")
|
|
serializer.save(organizacion=self.request.user.organizacion)
|
|
return |