235 lines
8.6 KiB
Python
235 lines
8.6 KiB
Python
import atexit
|
|
import os
|
|
import tempfile
|
|
|
|
from django.shortcuts import render
|
|
from ..organization.models import Organizacion
|
|
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 api.utils.storage_service import storage_service
|
|
|
|
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 rest_framework import mixins
|
|
|
|
from core.permissions import (
|
|
IsSameOrganizationAndInAllowedGroups,
|
|
get_org_context,
|
|
is_internal_service_request,
|
|
require_permission,
|
|
user_has_permission,
|
|
)
|
|
|
|
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):
|
|
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):
|
|
perms = {
|
|
'list': 'vucem.view',
|
|
'retrieve': 'vucem.view',
|
|
'create': 'vucem.manage',
|
|
'update': 'vucem.manage',
|
|
'partial_update': 'vucem.manage',
|
|
'destroy': 'vucem.manage',
|
|
'download_cer': 'vucem.view',
|
|
'download_key': 'vucem.view',
|
|
}
|
|
codename = perms.get(self.action, 'vucem.view')
|
|
return [IsAuthenticated(), require_permission(codename)()]
|
|
|
|
def get_queryset(self):
|
|
if not self.request.user.is_authenticated:
|
|
return self.queryset.none()
|
|
|
|
if is_internal_service_request(self.request):
|
|
queryset = self.queryset.all()
|
|
importador_rfc = self.request.query_params.get('importador')
|
|
if importador_rfc:
|
|
queryset = queryset.filter(usuarios_importadores__rfc__rfc=importador_rfc).distinct()
|
|
return queryset
|
|
|
|
if not user_has_permission(self.request.user, 'vucem.view'):
|
|
return self.queryset.none()
|
|
|
|
org = get_org_context(self.request.user)
|
|
if not org:
|
|
return self.queryset.none()
|
|
|
|
if self.request.user.is_importador:
|
|
queryset = self.queryset.filter(
|
|
organizacion=org,
|
|
usuario__in=self.request.user.rfc.all(),
|
|
)
|
|
else:
|
|
queryset = self.queryset.filter(organizacion=org)
|
|
|
|
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 is_internal_service_request(self.request):
|
|
serializer.save(updated_by=self.request.user)
|
|
return
|
|
org = get_org_context(self.request.user)
|
|
if not org:
|
|
raise ValueError("El usuario debe tener una organización activa para crear credenciales VUCEM.")
|
|
serializer.save(
|
|
organizacion=org,
|
|
created_by=self.request.user,
|
|
updated_by=self.request.user,
|
|
)
|
|
|
|
def perform_update(self, serializer):
|
|
if is_internal_service_request(self.request):
|
|
instance = self.get_object()
|
|
serializer.save(
|
|
created_by=instance.created_by,
|
|
updated_by=self.request.user,
|
|
)
|
|
return
|
|
org = get_org_context(self.request.user)
|
|
if not org:
|
|
raise ValueError("El usuario debe tener una organización activa para modificar credenciales VUCEM.")
|
|
instance = self.get_object()
|
|
serializer.save(
|
|
organizacion=org,
|
|
created_by=instance.created_by,
|
|
updated_by=self.request.user,
|
|
)
|
|
|
|
@action(detail=True, methods=["get"])
|
|
def download_cer(self, request, pk=None):
|
|
vucem = self.get_object()
|
|
if not vucem.cer:
|
|
return Response({"detail": "No hay archivo cer disponible."}, status=404)
|
|
|
|
ruta = str(vucem.cer)
|
|
|
|
with tempfile.NamedTemporaryFile(delete=False) as tmp:
|
|
tmp_path = tmp.name
|
|
|
|
success = storage_service.download_file(ruta, tmp_path)
|
|
if not success:
|
|
raise Http404("No se pudo descargar el archivo")
|
|
|
|
filename = os.path.basename(ruta)
|
|
response = FileResponse(open(tmp_path, 'rb'), as_attachment=True, filename=filename)
|
|
atexit.register(lambda: os.unlink(tmp_path) if os.path.exists(tmp_path) else None)
|
|
|
|
return response
|
|
|
|
@action(detail=True, methods=["get"])
|
|
def download_key(self, request, pk=None):
|
|
vucem = self.get_object()
|
|
if not vucem.key:
|
|
return Response({"detail": "No hay archivo key disponible."}, status=404)
|
|
|
|
ruta = str(vucem.key)
|
|
|
|
with tempfile.NamedTemporaryFile(delete=False) as tmp:
|
|
tmp_path = tmp.name
|
|
|
|
success = storage_service.download_file(ruta, tmp_path)
|
|
if not success:
|
|
raise Http404("No se pudo descargar el archivo")
|
|
|
|
filename = os.path.basename(ruta)
|
|
response = FileResponse(open(tmp_path, 'rb'), as_attachment=True, filename=filename)
|
|
atexit.register(lambda: os.unlink(tmp_path) if os.path.exists(tmp_path) else None)
|
|
|
|
return response
|
|
|
|
def perform_destroy(self, instance):
|
|
if instance.key:
|
|
storage_service.delete_file(str(instance.key))
|
|
if instance.cer:
|
|
storage_service.delete_file(str(instance.cer))
|
|
instance.delete()
|
|
|
|
|
|
class CredencialesImportadorViewSet(viewsets.ModelViewSet):
|
|
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):
|
|
perms = {
|
|
'list': 'vucem.view',
|
|
'retrieve': 'vucem.view',
|
|
'create': 'vucem.manage',
|
|
'update': 'vucem.manage',
|
|
'partial_update': 'vucem.manage',
|
|
'destroy': 'vucem.manage',
|
|
}
|
|
codename = perms.get(self.action, 'vucem.view')
|
|
return [IsAuthenticated(), require_permission(codename)()]
|
|
|
|
def get_queryset(self):
|
|
if not self.request.user.is_authenticated:
|
|
return self.queryset.none()
|
|
if is_internal_service_request(self.request):
|
|
return self.queryset.all()
|
|
if not user_has_permission(self.request.user, 'vucem.view'):
|
|
return self.queryset.none()
|
|
org = get_org_context(self.request.user)
|
|
if not org:
|
|
return self.queryset.none()
|
|
return self.queryset.filter(organizacion=org)
|
|
|
|
def perform_create(self, serializer):
|
|
if is_internal_service_request(self.request):
|
|
serializer.save()
|
|
return
|
|
org = get_org_context(self.request.user)
|
|
if not org:
|
|
raise ValueError("El usuario debe tener una organización activa.")
|
|
serializer.save(organizacion=org) |