EFC/backend
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

auditor

Browse Source
This commit is contained in:
Kevin Rosales
2025-10-05 22:48:46 -06:00
parent 38bd8913c7
commit a732fd7efa
4 changed files with 242 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
api/customs/tasks/auditoria.py
View File

@@ -39,6 +39,70 @@ def auditar_procesamiento_remesas(organizacion_id):
organizacion=organizacion_id
)
@shared_task
def auditar_procesamiento_remesa_por_pedimento(pedimento_id):
"""
Audita el procesamiento de remesa para un pedimento específico.
Args:
pedimento_id: UUID del pedimento a auditar
Returns:
dict: Resultado de la auditoría con detalles del procesamiento
"""
try:
pedimento = Pedimento.objects.get(id=pedimento_id)
resultado = {
'pedimento_id': str(pedimento_id),
'pedimento_numero': pedimento.pedimento,
'tiene_remesas': pedimento.remesas,
'procesamiento_creado': False,
'coves_creados': []
}
if not pedimento.remesas:
resultado['mensaje'] = 'El pedimento no tiene remesas para procesar'
return resultado
# Verificar documento tipo remesa
if not pedimento.documents.filter(document_type=3).exists():
# Crear procesamiento si no existe documento de remesa
procesamiento, creado = ProcesamientoPedimento.objects.get_or_create(
pedimento=pedimento,
servicio_id=5, # ID del servicio de remesas
organizacion=pedimento.organizacion_id
)
resultado['procesamiento_creado'] = creado
resultado['mensaje'] = 'Procesamiento de remesa creado - documento no encontrado'
else:
# Procesar XML de remesas
xml_data = extraer_coves(pedimento)
if xml_data:
for remesa in xml_data:
numero_cove = remesa.get('remesaSA')
cove, creado = Cove.objects.get_or_create(
pedimento=pedimento,
numero_cove=numero_cove,
organizacion=pedimento.organizacion_id
)
if creado:
resultado['coves_creados'].append(numero_cove)
resultado['mensaje'] = f"Procesados {len(xml_data)} remesas, creados {len(resultado['coves_creados'])} COVEs nuevos"
else:
resultado['mensaje'] = 'No se encontraron datos de remesas en el XML'
return resultado
except Pedimento.DoesNotExist:
return {
'error': f'Pedimento con ID {pedimento_id} no encontrado',
'pedimento_id': str(pedimento_id)
}
except Exception as e:
return {
'error': f'Error procesando pedimento {pedimento_id}: {str(e)}',
'pedimento_id': str(pedimento_id)
}
@shared_task
def auditar_partidas(organizacion_id):

2
api/customs/tasks/internal_services.py
View File

@@ -167,7 +167,7 @@ def crear_servicios(organizacion_id):
crear_procesamiento_edocument.apply_async(args=[str(pedimento.id)])
@shared_task
def auditar_pedimento(organizacion_id):
def auditar_pedimentos(organizacion_id):
pedimentos = Pedimento.objects.filter(organizacion_id=organizacion_id)
for pedimento in pedimentos:

11
api/customs/urls.py
View File

@@ -30,10 +30,19 @@ router.register(r'importadores', ImportadorViewSet, basename='Importador')
router.register(r'partidas', PartidaViewSet, basename='Partida')
# Import auditor views
from .views_auditor import crear_partidas_organizacion, crear_partidas_pedimento
from .views_auditor import (
crear_partidas_organizacion,
crear_partidas_pedimento,
auditar_pedimentos_endpoint,
auditar_procesamiento_remesas_endpoint,
auditar_procesamiento_remesa_pedimento_endpoint
)
urlpatterns = [
path('', include(router.urls)),
path('auditor/crear-partidas/organizacion/', crear_partidas_organizacion, name='crear-partidas-organizacion'),
path('auditor/crear-partidas/pedimento/', crear_partidas_pedimento, name='crear-partidas-pedimento'),
path('auditor/auditar-pedimentos/', auditar_pedimentos_endpoint, name='auditar-pedimentos'),
path('auditor/auditar-procesamiento-remesas/', auditar_procesamiento_remesas_endpoint, name='auditar-procesamiento-remesas'),
path('auditor/auditar-procesamiento-remesa/pedimento/', auditar_procesamiento_remesa_pedimento_endpoint, name='auditar-procesamiento-remesa-pedimento'),
]

167
api/customs/views_auditor.py
View File

@@ -5,7 +5,13 @@ from rest_framework import status
from drf_yasg.utils import swagger_auto_schema
from drf_yasg import openapi
from core.permissions import IsSuperUser, IsSameOrganizationDeveloper
from .tasks.auditoria import crear_partidas, crear_partidas_por_pedimento
from .tasks.auditoria import (
crear_partidas,
crear_partidas_por_pedimento,
auditar_procesamiento_remesas,
auditar_procesamiento_remesa_por_pedimento
)
from .tasks.internal_services import auditar_pedimentos
from api.customs.models import Pedimento
@@ -110,3 +116,162 @@ def crear_partidas_pedimento(request):
'message': message,
'task_id': task.id
}, status=status.HTTP_200_OK)
@swagger_auto_schema(
method='post',
operation_description="Audita todos los pedimentos de una organización",
request_body=openapi.Schema(
type=openapi.TYPE_OBJECT,
properties={
'organizacion_id': openapi.Schema(type=openapi.TYPE_STRING, description='ID de la organización')
},
required=['organizacion_id']
),
responses={
200: openapi.Response('Tarea de auditoría iniciada correctamente'),
400: openapi.Response('Error en los parámetros'),
403: openapi.Response('No tiene permisos suficientes')
}
)
@api_view(['POST'])
@permission_classes([IsAuthenticated & (IsSuperUser | IsSameOrganizationDeveloper)])
def auditar_pedimentos_endpoint(request):
"""
Inicia una tarea de auditoría para todos los pedimentos de una organización.
Verifica todos los documentos y datos asociados a los pedimentos.
"""
organizacion_id = request.data.get('organizacion_id')
if not organizacion_id:
return Response(
{'error': 'Debe proporcionar organizacion_id'},
status=status.HTTP_400_BAD_REQUEST
)
# Validar permisos
user = request.user
if not user.is_superuser and str(user.organizacion.id) != organizacion_id:
return Response(
{'error': 'No tiene permisos para esta organización'},
status=status.HTTP_403_FORBIDDEN
)
# Ejecutar la tarea de auditoría
task = auditar_pedimentos.delay(organizacion_id)
message = f"Auditoría iniciada para la organización {organizacion_id}"
return Response({
'message': message,
'task_id': task.id
}, status=status.HTTP_200_OK)
@swagger_auto_schema(
method='post',
operation_description="Audita el procesamiento de remesa para un pedimento específico",
request_body=openapi.Schema(
type=openapi.TYPE_OBJECT,
properties={
'pedimento_id': openapi.Schema(type=openapi.TYPE_STRING, description='ID del pedimento a auditar')
},
required=['pedimento_id']
),
responses={
200: openapi.Response('Tarea de auditoría iniciada correctamente'),
400: openapi.Response('Error en los parámetros'),
403: openapi.Response('No tiene permisos suficientes'),
404: openapi.Response('Pedimento no encontrado')
}
)
@api_view(['POST'])
@permission_classes([IsAuthenticated & (IsSuperUser | IsSameOrganizationDeveloper)])
def auditar_procesamiento_remesa_pedimento_endpoint(request):
"""
Inicia una tarea de auditoría de remesa para un pedimento específico.
Verifica el estado del procesamiento de remesa y la creación de COVEs.
"""
pedimento_id = request.data.get('pedimento_id')
if not pedimento_id:
return Response(
{'error': 'Debe proporcionar pedimento_id'},
status=status.HTTP_400_BAD_REQUEST
)
# Validar permisos y existencia del pedimento
try:
pedimento = Pedimento.objects.get(id=pedimento_id)
user = request.user
if not user.is_superuser and str(pedimento.organizacion.id) != str(user.organizacion.id):
return Response(
{'error': 'No tiene permisos para este pedimento'},
status=status.HTTP_403_FORBIDDEN
)
except Pedimento.DoesNotExist:
return Response(
{'error': 'Pedimento no encontrado'},
status=status.HTTP_404_NOT_FOUND
)
# Ejecutar la tarea de auditoría
task = auditar_procesamiento_remesa_por_pedimento.delay(pedimento_id)
message = f"Auditoría de remesa iniciada para el pedimento {pedimento_id}"
return Response({
'message': message,
'task_id': task.id,
'pedimento': {
'id': str(pedimento.id),
'pedimento': pedimento.pedimento,
'tiene_remesas': pedimento.remesas
}
}, status=status.HTTP_200_OK)
@swagger_auto_schema(
method='post',
operation_description="Audita el procesamiento de remesas de una organización",
request_body=openapi.Schema(
type=openapi.TYPE_OBJECT,
properties={
'organizacion_id': openapi.Schema(type=openapi.TYPE_STRING, description='ID de la organización')
},
required=['organizacion_id']
),
responses={
200: openapi.Response('Tarea de auditoría iniciada correctamente'),
400: openapi.Response('Error en los parámetros'),
403: openapi.Response('No tiene permisos suficientes')
}
)
@api_view(['POST'])
@permission_classes([IsAuthenticated & (IsSuperUser | IsSameOrganizationDeveloper)])
def auditar_procesamiento_remesas_endpoint(request):
"""
Inicia una tarea de auditoría para el procesamiento de remesas de una organización.
Verifica el estado y la integridad del procesamiento de remesas.
"""
organizacion_id = request.data.get('organizacion_id')
if not organizacion_id:
return Response(
{'error': 'Debe proporcionar organizacion_id'},
status=status.HTTP_400_BAD_REQUEST
)
# Validar permisos
user = request.user
if not user.is_superuser and str(user.organizacion.id) != organizacion_id:
return Response(
{'error': 'No tiene permisos para esta organización'},
status=status.HTTP_403_FORBIDDEN
)
# Ejecutar la tarea de auditoría
task = auditar_procesamiento_remesas.delay(organizacion_id)
message = f"Auditoría de procesamiento de remesas iniciada para la organización {organizacion_id}"
return Response({
'message': message,
'task_id': task.id
}, status=status.HTTP_200_OK)