fix/T2025-09-007 corregir documentos duplicados

Reviewed-on: #27

api/cards/views.py

@@ -157,7 +157,7 @@ class ViewPedimentoServicesUtilInformation(LoggingMixin, APIView, FiltroPorOrgan
 
         # Si es importador de la organizacion, devuelve los servicios relacionados con sus pedimentos
         if self.request.user.is_authenticated and self.request.user.groups.filter(name='importador').exists() and self.request.user.is_importador and self.request.user.groups.filter(name='user').exists():
-            return self.request.user.organizacion.procesamiento_pedimentos.filter(pedimento__contribuyente=self.request.user.rfc)
+            return self.request.user.organizacion.procesamiento_pedimentos.filter(pedimento__contribuyente__in=self.request.user.rfc.all())
 
         
         

api/cuser/admin.py

@@ -13,7 +13,7 @@ class CustomUserCreationForm(UserCreationForm):
 class CustomUserChangeForm(UserChangeForm):
     class Meta:
         model = CustomUser
-        fields = ('username', 'email', 'first_name', 'last_name', 'organizacion', 'profile_picture')
+        fields = ('username', 'email', 'first_name', 'last_name', 'organizacion', 'profile_picture', 'is_importador', 'rfc')
 
 
 class CustomUserAdmin(UserAdmin):
@@ -25,6 +25,7 @@ class CustomUserAdmin(UserAdmin):
     list_filter = ('is_staff', 'is_active', 'organizacion')
     search_fields = ('username', 'email', 'first_name', 'last_name')
     ordering = ('username',)
+    filter_horizontal = ('rfc', 'groups', 'user_permissions')
 
     # Fieldsets para editar un usuario
     fieldsets = (

api/cuser/models.py

@@ -12,7 +12,7 @@ class CustomUser(AbstractUser):
     profile_picture = models.ImageField(upload_to='profile_pictures/', null=True, blank=True)
 
     is_importador = models.BooleanField(default=False, help_text="Indicates if the user is an importer")
-    rfc = models.ForeignKey('customs.Importador', on_delete=models.SET_NULL, null=True, blank=True, related_name='users', help_text="RFC associated with the user if they are an importer")
+    rfc = models.ManyToManyField('customs.Importador', blank=True, related_name='users', help_text="RFCs de importadores asociados al usuario")
 
     def __str__(self):
         return self.username

api/cuser/serializers.py

@@ -2,6 +2,7 @@
 from rest_framework import serializers
 from .models import CustomUser
 from django.contrib.auth.models import Group
+from api.customs.models import Importador
 
 class CustomUserSerializer(serializers.ModelSerializer):
     """
@@ -10,8 +11,12 @@ class CustomUserSerializer(serializers.ModelSerializer):
 
     password = serializers.CharField(write_only=True)
     groups = serializers.PrimaryKeyRelatedField(queryset=Group.objects.all(), many=True, required=False)
-    rfc = serializers.CharField(max_length=20, required=False, allow_blank=True)
+    rfc = serializers.PrimaryKeyRelatedField(
-
+        queryset=Importador.objects.all(),
+        many=True,
+        required=False,
+        pk_field=serializers.CharField(),
+    )
 
     class Meta:
         model = CustomUser
@@ -20,10 +25,28 @@ class CustomUserSerializer(serializers.ModelSerializer):
 
     def create(self, validated_data):
         groups = validated_data.pop('groups', [])
+        rfcs = validated_data.pop('rfc', [])
         password = validated_data.pop('password')
         user = CustomUser(**validated_data)
         user.set_password(password)
         user.save()
         if groups:
             user.groups.set(groups)
+        if rfcs:
+            user.rfc.set(rfcs)
         return user
+
+    def update(self, instance, validated_data):
+        groups = validated_data.pop('groups', None)
+        rfcs = validated_data.pop('rfc', None)
+        password = validated_data.pop('password', None)
+        for attr, value in validated_data.items():
+            setattr(instance, attr, value)
+        if password:
+            instance.set_password(password)
+        instance.save()
+        if groups is not None:
+            instance.groups.set(groups)
+        if rfcs is not None:
+            instance.rfc.set(rfcs)
+        return instance

api/customs/serializers.py

@@ -47,55 +47,31 @@ class PartidaSerializer(serializers.ModelSerializer):
     documentos = serializers.SerializerMethodField()
 
     def get_documentos(self, obj):
-        """
-        Busca documentos en la tabla `document` que coincidan EXACTAMENTE con:
-        'documents/vu_PT_{pedimentoApp}_{numero}' al inicio del nombre del archivo.
-        """
-
         if not obj or not getattr(obj, 'pedimento', None):
             return []
-
         if not obj or not getattr(obj, 'numero_partida', None):
             return []
 
         try:
-            pedimentoApp = str(obj.pedimento.pedimento_app).strip()
+            pedimento_app = str(obj.pedimento.pedimento_app).strip()
             numero = str(obj.numero_partida).strip()
+            # Incluir pedimento_app en el patrón para evitar falsos positivos
+            # entre partidas con números cortos (1 matchearía 10, 100, etc.)
+            patron = f"vu_PT_{pedimento_app}_{numero}_"
 
-            # Construir el patrón exacto de búsqueda
+            # 17 = REQUEST partida, 18 = ERROR partida
-            patron_exacto = f'documents/vu_PT_{pedimentoApp}_{numero}.xml'
-
-            # Buscar documentos que empiecen EXACTAMENTE con ese patrón
             qs = Document.objects.filter(
-                archivo=patron_exacto
+                pedimento=obj.pedimento,
-            )
+                archivo__icontains=patron,
+            ).exclude(document_type_id__in=[17, 18])
 
-            # Opción 2: Si puede tener diferentes extensiones
-            # patron_base = f'documents/vu_PT_{pedimentoApp}_{numero}'
-            # qs = Document.objects.filter(
-            #     archivo__startswith=patron_base
-            # ).filter(
-            #     archivo__in=[
-            #         f'{patron_base}.xml',
-            #         f'{patron_base}.pdf',
-            #         f'{patron_base}.zip'
-            #     ]
-            # )
-
-            # Filtro adicional por pedimento si el modelo Document tiene este campo
-            if hasattr(Document, 'pedimento'):
-                qs = qs.filter(pedimento=obj.pedimento)
-
-            # Filtro por organización
             if hasattr(obj, 'organizacion') and obj.organizacion:
                 qs = qs.filter(organizacion=obj.organizacion)
 
             serializer = DocumentSerializer(qs, many=True, context=self.context)
             return serializer.data
 
-            #return []
         except Exception:
-            # En caso de cualquier error (por ejemplo, importaciones circulares), devolver lista vacía
             return []
     class Meta:
         model = Partida
@@ -208,10 +184,11 @@ class EDocumentSerializer(serializers.ModelSerializer):
             numero = str(obj.numero_edocument).strip()
             # id_pedimento = str(obj.pedimento_id).strip()
 
+            # excluir e documents de tipo request y de tipo error
             qs = Document.objects.filter(
                 pedimento=obj.pedimento,
                 archivo__icontains=numero,
-            )
+            ).exclude(document_type_id__in=[21, 25])
 
             # Filtro por organización si aplica
             if hasattr(obj, 'organizacion') and obj.organizacion:
@@ -263,10 +240,15 @@ class CoveSerializer(serializers.ModelSerializer):
         try:
             numero = str(obj.numero_cove).strip()
 
+            # Excluir los tipo de documento 20, 24, 23 y 19
+            # 20 = error solicitud cove 
+            # 24 = error solicitud acuse cove
+            # 23 = request acuse cove
+            # 19 = request cove
             qs = Document.objects.filter(
                 pedimento=obj.pedimento,
                 archivo__icontains=numero,
-            )
+            ).exclude(document_type_id__in=[20, 24, 23, 19])
 
             # Filtro por organización si aplica
             if hasattr(obj, 'organizacion') and obj.organizacion:

api/customs/signals/procesamiento.py

@@ -87,8 +87,11 @@ def trigger_celery_task_on_cove_create(sender, instance, created, **kwargs):
         import logging
         logger = logging.getLogger('api.customs.async_operations')
         logger.info(f"Cove creado: {instance.id}, creando procesamiento...")
-        crear_procesamiento_cove.apply_async(args=[str(instance.pedimento.id)])
+        pedimento_id = str(instance.pedimento.id)
-        crear_procesamiento_acuse_cove.apply_async(args=[str(instance.pedimento.id)])
+        def enqueue_cove_tasks():
+            crear_procesamiento_cove.apply_async(args=[pedimento_id])
+            crear_procesamiento_acuse_cove.apply_async(args=[pedimento_id])
+        transaction.on_commit(enqueue_cove_tasks)
 
 @receiver(post_save, sender=EDocument)
 def trigger_celery_task_on_edocument_create(sender, instance, created, **kwargs):
@@ -96,5 +99,8 @@ def trigger_celery_task_on_edocument_create(sender, instance, created, **kwargs)
         import logging
         logger = logging.getLogger('api.customs.async_operations')
         logger.info(f"EDocument creado: {instance.id}, creando procesamiento...")
-        crear_procesamiento_edocument.apply_async(args=[str(instance.pedimento.id)])
+        pedimento_id = str(instance.pedimento.id)
-        crear_procesamiento_acuse.apply_async(args=[str(instance.pedimento.id)])
+        def enqueue_edocument_tasks():
+            crear_procesamiento_edocument.apply_async(args=[pedimento_id])
+            crear_procesamiento_acuse.apply_async(args=[pedimento_id])
+        transaction.on_commit(enqueue_edocument_tasks)

api/customs/tasks/__init__.py

@@ -1,3 +1,4 @@
 from .microservice import * 
 from .internal_services import *
 from .bulk_upload import *
+from .microservice_v2 import *

api/customs/tasks/auditoria.py

@@ -6,6 +6,8 @@ from api.customs.models import ProcesamientoPedimento, Pedimento, Cove, EDocumen
 from core.utils import xml_controller
 import requests
 from core.utils import xml_remesas_controller
+import logging
+logger = logging.getLogger(__name__)
 
 def obtener_pedimentos(organizacion_id):
     return Pedimento.objects.filter(organizacion_id=organizacion_id)
@@ -35,23 +37,31 @@ def auditor_descargas(pedimento, servicio, related_name, variable, mensaje):
     pedimento_id = pedimento.id
     docs = getattr(pedimento, related_name).all()
     
+    print(f"pedimento: {pedimento}, servicio: {servicio}, related_name: {related_name}, variable: {variable}, mensaje: {mensaje}")
+    logger.info(f"pedimento: {pedimento}, servicio: {servicio}, related_name: {related_name}, variable: {variable}, mensaje: {mensaje}")
+
     # Si no hay documentos, marcar como completado
     if not docs.exists():
         proceso = modificar_estado_procesamiento(pedimento, servicio_id=servicio, nuevo_estado=3)  # Estado "completado"
         print(f"✓ Pedimento {pedimento_id} no tiene {mensaje}s para procesar.")
+        logger.info(f"✓ Pedimento {pedimento_id} no tiene {mensaje}s para procesar.")
     else:
         all_docs = all(getattr(doc, variable) for doc in docs)
         if all_docs:
             proceso = modificar_estado_procesamiento(pedimento, servicio_id=servicio, nuevo_estado=3)  # Estado "completado"
             print(f"✓ Pedimento {pedimento_id} tiene todos sus {mensaje} descargados.")
+            logger.info(f"✓ Pedimento {pedimento_id} tiene todos sus {mensaje} descargados.")
         else:
             proceso = modificar_estado_procesamiento(pedimento, servicio_id=servicio, nuevo_estado=4)  # Estado "en progreso"
             print(f"✗ Pedimento {pedimento_id} NO tiene todos sus {mensaje} descargados.")
+            logger.info(f"✗ Pedimento {pedimento_id} NO tiene todos sus {mensaje} descargados.")
 
     if proceso:
         print(f"✓ Proceso de auditoría para pedimento {pedimento_id} completado.")
+        logger.info(f"✓ Proceso de auditoría para pedimento {pedimento_id} completado.")
     else:
         print(f"✗ No se encontró proceso de auditoría para pedimento {pedimento_id}.")
+        logger.info(f"✗ No se encontró proceso de auditoría para pedimento {pedimento_id}.")
 
 ## Auditar pedimentos
 
@@ -121,44 +131,66 @@ def auditar_procesamiento_remesa_por_pedimento(pedimento_id):
 
 @shared_task
 def crear_partidas(organizacion_id):
+    from api.customs.models import Partida
+
     pedimentos = obtener_pedimentos(organizacion_id)
     total_pedimentos = pedimentos.count()
-    pedimentos_procesados = 0
-    total_partidas_agregadas = 0
 
-    print(f"Iniciando procesamiento de {total_pedimentos} pedimentos para organización {organizacion_id}")
+    completados = []
+    con_pendientes = []
+    sin_datos = []
+    errores = []
 
     for pedimento in pedimentos:
-        pedimentos_procesados += 1
+        try:
-        partidas_agregadas_pedimento = 0
+            if not pedimento.numero_partidas or pedimento.numero_partidas <= 0:
-        
+                sin_datos.append({
-        # Validar que numero_partidas no sea None y sea mayor que 0
+                    'pedimento_id': str(pedimento.id),
-        if pedimento.numero_partidas is not None and pedimento.numero_partidas > 0:
+                    'pedimento': pedimento.pedimento,
-            partidas_existentes = pedimento.partidas.count()
+                    'razon': f'numero_partidas inválido ({pedimento.numero_partidas})',
-            if pedimento.numero_partidas > partidas_existentes:
+                })
-                print(f"Procesando pedimento {pedimento.id} ({pedimentos_procesados}/{total_pedimentos}) - Partidas existentes: {partidas_existentes}, Requeridas: {pedimento.numero_partidas}")
+                continue
 
             for i in range(1, pedimento.numero_partidas + 1):
-                    from api.customs.models import Partida
+                Partida.objects.get_or_create(
-                    partida, created = Partida.objects.get_or_create(
                     pedimento=pedimento,
                     numero_partida=i,
-                        organizacion_id=organizacion_id
+                    defaults={'organizacion_id': organizacion_id}
                 )
-                    if created:
-                        partidas_agregadas_pedimento += 1
-                        total_partidas_agregadas += 1
 
-                print(f"  → Partidas agregadas para pedimento {pedimento.id}: {partidas_agregadas_pedimento}")
+            partidas = list(pedimento.partidas.order_by('numero_partida'))
-            else:
+            no_descargadas = [p.numero_partida for p in partidas if not p.descargado]
-                print(f"Pedimento {pedimento.id} ya tiene todas sus partidas ({partidas_existentes}/{pedimento.numero_partidas})")
-        else:
-            print(f"Pedimento {pedimento.id} omitido - numero_partidas: {pedimento.numero_partidas} (inválido)")
 
-    print(f"\n=== RESUMEN ===")
+            if not no_descargadas:
-    print(f"Pedimentos procesados: {pedimentos_procesados}")
+                completados.append(str(pedimento.id))
-    print(f"Total de partidas agregadas: {total_partidas_agregadas}")
+            else:
-    print(f"Procesamiento completado para organización {organizacion_id}")
+                con_pendientes.append({
+                    'pedimento_id': str(pedimento.id),
+                    'pedimento': pedimento.pedimento,
+                    'total_partidas': len(partidas),
+                    'descargadas': len(partidas) - len(no_descargadas),
+                    'no_descargadas': no_descargadas,
+                })
+
+        except Exception as e:
+            errores.append({
+                'pedimento_id': str(pedimento.id),
+                'pedimento': pedimento.pedimento,
+                'error': str(e),
+            })
+            logger.error(f"Error creando partidas para pedimento {pedimento.id}: {e}")
+
+    return {
+        'organizacion_id': str(organizacion_id),
+        'total_pedimentos': total_pedimentos,
+        'completados': len(completados),
+        'con_pendientes': len(con_pendientes),
+        'sin_datos': len(sin_datos),
+        'con_errores': len(errores),
+        'detalle_pendientes': con_pendientes,
+        'detalle_sin_datos': sin_datos,
+        'detalle_errores': errores,
+    }
 
 @shared_task
 def crear_partidas_por_pedimento(pedimento_id):
@@ -169,6 +201,7 @@ def crear_partidas_por_pedimento(pedimento_id):
         return
 
     print(f"Procesando pedimento individual {pedimento_id}...")
+    logger.info(f"Procesando pedimento individual {pedimento_id}...")
     partidas_agregadas = 0
     
     # Validar que numero_partidas no sea None y sea mayor que 0
@@ -176,6 +209,7 @@ def crear_partidas_por_pedimento(pedimento_id):
         partidas_existentes = pedimento.partidas.count()
         if pedimento.numero_partidas > partidas_existentes:
             print(f"Pedimento {pedimento_id} - Partidas existentes: {partidas_existentes}, Requeridas: {pedimento.numero_partidas}")
+            logger.info(f"Pedimento {pedimento_id} - Partidas existentes: {partidas_existentes}, Requeridas: {pedimento.numero_partidas}")
             
             for i in range(1, pedimento.numero_partidas + 1):
                 from api.customs.models import Partida
@@ -188,62 +222,165 @@ def crear_partidas_por_pedimento(pedimento_id):
                     partidas_agregadas += 1
             
             print(f"✓ Partidas agregadas para pedimento {pedimento_id}: {partidas_agregadas}")
+            logger.info(f"✓ Partidas agregadas para pedimento {pedimento_id}: {partidas_agregadas}")
         else:
             print(f"Pedimento {pedimento_id} ya tiene todas sus partidas ({partidas_existentes}/{pedimento.numero_partidas})")
+            logger.info(f"Pedimento {pedimento_id} ya tiene todas sus partidas ({partidas_existentes}/{pedimento.numero_partidas})")
     else:
         print(f"Error: Pedimento {pedimento_id} tiene numero_partidas inválido: {pedimento.numero_partidas}")    
+        logger.info(f"Error: Pedimento {pedimento_id} tiene numero_partidas inválido: {pedimento.numero_partidas}")    
+     
+def _auditar_organizacion(organizacion_id, servicio, related_name, variable, label):
+    """
+    Itera todos los pedimentos de una organización auditando el campo `variable`
+    en la relación `related_name`. Retorna un resumen estructurado por pedimento.
+    """
+    pedimentos = obtener_pedimentos(organizacion_id)
+    total_pedimentos = pedimentos.count()
+
+    completados = []
+    pendientes = []
+    errores = []
+
+    for pedimento in pedimentos:
+        try:
+            docs = list(getattr(pedimento, related_name).all())
+            total = len(docs)
+            faltantes = [
+                getattr(doc, 'numero_cove', None) or getattr(doc, 'numero_edocument', None)
+                for doc in docs if not getattr(doc, variable)
+            ]
+
+            if total == 0 or len(faltantes) == 0:
+                nuevo_estado = 3
+                completados.append(str(pedimento.id))
+            else:
+                nuevo_estado = 4
+                pendientes.append({
+                    'pedimento_id': str(pedimento.id),
+                    'pedimento': pedimento.pedimento,
+                    f'faltantes_{label}': faltantes,
+                    'total': total,
+                    'descargados': total - len(faltantes),
+                })
+
+            modificar_estado_procesamiento(pedimento, servicio_id=servicio, nuevo_estado=nuevo_estado)
+
+        except Exception as e:
+            errores.append({
+                'pedimento_id': str(pedimento.id),
+                'pedimento': pedimento.pedimento,
+                'error': str(e),
+            })
+            logger.error(f"Error auditando pedimento {pedimento.id} [{label}]: {e}")
+
+    return {
+        'organizacion_id': str(organizacion_id),
+        'auditoria': label,
+        'total_pedimentos': total_pedimentos,
+        'completados': len(completados),
+        'con_pendientes': len(pendientes),
+        'con_errores': len(errores),
+        'detalle_pendientes': pendientes,
+        'detalle_errores': errores,
+    }
+
 
-# Auditar coves      
 @shared_task
 def auditar_coves(organizacion_id):
-    for pedimento in obtener_pedimentos(organizacion_id):
+    return _auditar_organizacion(
-        auditor_descargas(
+        organizacion_id,
-            pedimento, 
         servicio=8,
         related_name='coves',
         variable='cove_descargado',
-            mensaje='COVE'
+        label='cove',
     )
 
 @shared_task
 def auditar_acuse_cove(organizacion_id):
-    for pedimento in obtener_pedimentos(organizacion_id):
+    return _auditar_organizacion(
-        auditor_descargas(
+        organizacion_id,
-            pedimento, 
         servicio=9,
         related_name='coves',
         variable='acuse_cove_descargado',
-            mensaje='acuse de COVE'
+        label='acuse_cove',
     )
 
-# Revisa si el pedimento completo todos sus acuse coves
-
-# Auditar edocuments
 @shared_task
 def auditar_edocuments(organizacion_id):
-    for pedimento in obtener_pedimentos(organizacion_id):
+    return _auditar_organizacion(
-        auditor_descargas(
+        organizacion_id,
-            pedimento, 
         servicio=7,
         related_name='documentos',
         variable='edocument_descargado',
-            mensaje='EDocument'
+        label='edocument',
     )
 
 @shared_task
 def auditar_acuse(organizacion_id):
-    for pedimento in obtener_pedimentos(organizacion_id):
+    return _auditar_organizacion(
-        auditor_descargas(
+        organizacion_id,
-            pedimento, 
         servicio=6,
         related_name='documentos',
         variable='acuse_descargado',
-            mensaje='acuse'
+        label='acuse',
     )
 
+@shared_task
+def auditar_remesas(organizacion_id):
+    """
+    Audita el estado de descarga de remesas para todos los pedimentos de una organización.
+    A diferencia de coves/edocuments, las remesas no tienen campo booleano propio —
+    se verifica la existencia de un documento de tipo 3 (Remesa) en el pedimento.
+    """
+    pedimentos = obtener_pedimentos(organizacion_id)
+    total_pedimentos = pedimentos.count()
+
+    completados = []
+    pendientes = []
+    errores = []
+
+    for pedimento in pedimentos:
+        try:
+            if not pedimento.remesas:
+                # El pedimento no declara remesas — no aplica, marcar como completado
+                modificar_estado_procesamiento(pedimento, servicio_id=5, nuevo_estado=3)
+                completados.append(str(pedimento.id))
+            elif pedimento.documents.filter(document_type=3).exists():
+                # Documento de remesa ya descargado
+                modificar_estado_procesamiento(pedimento, servicio_id=5, nuevo_estado=3)
+                completados.append(str(pedimento.id))
+            else:
+                # Tiene remesas declaradas pero el documento aún no existe
+                modificar_estado_procesamiento(pedimento, servicio_id=5, nuevo_estado=4)
+                pendientes.append({
+                    'pedimento_id': str(pedimento.id),
+                    'pedimento': pedimento.pedimento,
+                })
+        except Exception as e:
+            errores.append({
+                'pedimento_id': str(pedimento.id),
+                'pedimento': pedimento.pedimento,
+                'error': str(e),
+            })
+            logger.error(f"Error auditando remesa de pedimento {pedimento.id}: {e}")
+
+    return {
+        'organizacion_id': str(organizacion_id),
+        'auditoria': 'remesa',
+        'total_pedimentos': total_pedimentos,
+        'completados': len(completados),
+        'con_pendientes': len(pendientes),
+        'con_errores': len(errores),
+        'detalle_pendientes': pendientes,
+        'detalle_errores': errores,
+    }
+
 @shared_task
 def auditar_cove_por_pedimento(pedimento_id):
     try:
+        print(f"auditar_cove_por_pedimento >>>> {pedimento_id}")
+        logger.info(f"auditar_cove_por_pedimento >>>> {pedimento_id}")
         from api.customs.models import Pedimento
         pedimento = Pedimento.objects.get(id=pedimento_id)
         auditor_descargas(

api/customs/tasks/auditoria_xml.py

@@ -1,6 +1,8 @@
 # auditoria_xml.py
 import xml.etree.ElementTree as ET
 from datetime import datetime
+import logging
+logger = logging.getLogger('api.customs.auditoria_xml')
 
 def extraer_info_pedimento_xml(xml_content):
     """
@@ -13,8 +15,10 @@ def extraer_info_pedimento_xml(xml_content):
         # Buscar el namespace (puede variar)
         namespaces = {
             'S': 'http://schemas.xmlsoap.org/soap/envelope/',
+            's': 'http://schemas.xmlsoap.org/soap/envelope/',
             'ns2': 'http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/consultarpedimentocompleto',
-            'ns3': 'http://www.ventanillaunica.gob.mx/common/ws/oxml/respuesta'
+            'ns3': 'http://www.ventanillaunica.gob.mx/common/ws/oxml/respuesta',
+            
         }
         
         resultado = {}
@@ -181,10 +185,37 @@ def extraer_info_pedimento_xml(xml_content):
             if edocs_encontrados:
                 resultado['edocuments_en_xml'] = edocs_encontrados
         
-        # Verificar si hay error en la respuesta
+        # Verificar si hay error en la respuesta — 3 variantes según el servicio VUCEM:
+        # 1) Remesas/pedimentos: <ns3:tieneError> en namespace oxml/respuesta
+        # 2) eDocuments:        <TieneError> en namespace tempuri.org, mensaje en <Errores>
+        # 3) Acuses:            <error> sin namespace dentro de responseConsultaAcuses
         tiene_error = root.find('.//ns3:tieneError', namespaces)
         if tiene_error is not None:
             resultado['tiene_error'] = tiene_error.text.lower() == 'true'
+            if resultado['tiene_error']:
+                mensaje = root.find('.//ns3:error/ns3:mensaje', namespaces)
+                if mensaje is not None and mensaje.text:
+                    resultado['error_mensaje'] = mensaje.text.strip()
+        else:
+            # Variante eDocuments (tempuri.org)
+            tiene_error_edoc = root.find('.//{http://tempuri.org/}TieneError')
+            if tiene_error_edoc is not None:
+                resultado['tiene_error'] = tiene_error_edoc.text.lower() == 'true'
+                if resultado['tiene_error']:
+                    errores_elem = root.find('.//{http://tempuri.org/}Errores')
+                    if errores_elem is not None and errores_elem.text:
+                        resultado['error_mensaje'] = errores_elem.text.strip()
+            else:
+                # Variante acuses: <error> sin namespace
+                error_acuses = root.find('.//error')
+                if error_acuses is not None and error_acuses.text is not None:
+                    resultado['tiene_error'] = error_acuses.text.lower() == 'true'
+                    if resultado['tiene_error']:
+                        descripciones = root.findall('.//mensajeErrores/descripcion')
+                        if descripciones:
+                            resultado['error_mensaje'] = ' | '.join(
+                                d.text.strip() for d in descripciones if d.text
+                            )
         
         return resultado
         

api/customs/tasks/bulk_upload.py

@@ -27,16 +27,27 @@ def normalize_filename(filename):
     return filename
 
 
+def extract_django_suffix(filename):
+    """
+    Extrae el sufijo UUID de 8 chars que storage_service añade a los archivos.
+    """
+    name_without_ext = os.path.splitext(filename)[0]
+    match = re.search(r'_([a-zA-Z0-9]{8})$', name_without_ext)
+    if match:
+        return match.group(1)
+    return None
+
+
 def get_clean_base_filename(filename):
     """
-    Obtiene el nombre base limpio sin el sufijo de Django.
+    Obtiene el nombre base limpio sin el sufijo UUID de storage_service.
     """
     normalized = normalize_filename(filename)
     name_without_ext, ext = os.path.splitext(normalized)
 
     django_suffix = extract_django_suffix(name_without_ext)
     if django_suffix:
-        base_name = name_without_ext[:-8]
+        base_name = name_without_ext[:-9]  # elimina _XXXXXXXX (underscore + 8 chars UUID)
     else:
         base_name = name_without_ext
 
@@ -45,17 +56,6 @@ def get_clean_base_filename(filename):
     return base_name.lower().strip('_')
 
 
-def extract_django_suffix(filename):
-    """
-    Extrae el sufijo único que Django añade a los archivos.
-    """
-    name_without_ext = os.path.splitext(filename)[0]
-    match = re.search(r'_([a-zA-Z0-9]{7})$', name_without_ext)
-    if match:
-        return match.group(1)
-    return None
-
-
 def is_same_document(existing_doc, new_filename):
     """
     Compara si un documento existente y un nuevo archivo son el mismo documento.

api/customs/tasks/internal_services.py

@@ -1,6 +1,14 @@
 from celery import shared_task, group
 from api.customs.models import ProcesamientoPedimento, Pedimento, Cove, EDocument
 from core.utils import xml_controller
+from api.customs.tasks.microservice import (
+    procesar_cove_individual,
+    procesar_acuse_individual,
+    procesar_acuse_cove_individual,
+    procesar_edoc_individual,
+    procesar_partida_individual,
+    procesar_remesa_individual,
+)
 
 @shared_task
 def crear_procesamiento_remesa(pedimento_id):
@@ -11,7 +19,7 @@ def crear_procesamiento_remesa(pedimento_id):
     if pedimento.remesas:
         existe = ProcesamientoPedimento.objects.filter(
             pedimento=pedimento,
-            servicio_id=5,  # ID del servicio de remesas
+            servicio_id=5,
             organizacion=pedimento.organizacion,
             estado_id__in=[1, 2, 3, 4]
         ).exists()
@@ -19,10 +27,11 @@ def crear_procesamiento_remesa(pedimento_id):
             logger.info(f"[TAREA] ProcesamientoPedimento remesa creado para pedimento {pedimento_id}")
             ProcesamientoPedimento.objects.create(
                 pedimento=pedimento,
-                estado_id=1,  # Estado "pendiente"
+                estado_id=1,
                 servicio_id=5,
                 organizacion=pedimento.organizacion
             )
+            procesar_remesa_individual.apply_async(args=[str(pedimento.id), str(pedimento.organizacion.id)])
 
 @shared_task
 def crear_procesamiento_partida(pedimento_id):
@@ -32,7 +41,7 @@ def crear_procesamiento_partida(pedimento_id):
     logger.info(f"[TAREA] crear_procesamiento_partida para pedimento {pedimento_id}")
     existe = ProcesamientoPedimento.objects.filter(
         pedimento=pedimento,
-        servicio_id=4,  # ID del servicio de partidas
+        servicio_id=4,
         organizacion=pedimento.organizacion,
         estado_id__in=[1, 2, 3, 4]
     ).exists()
@@ -40,10 +49,11 @@ def crear_procesamiento_partida(pedimento_id):
         logger.info(f"[TAREA] ProcesamientoPedimento partida creado para pedimento {pedimento_id}")
         ProcesamientoPedimento.objects.create(
             pedimento=pedimento,
-            estado_id=1,  # Estado "pendiente"
+            estado_id=1,
             servicio_id=4,
             organizacion=pedimento.organizacion
         )
+        procesar_partida_individual.apply_async(args=[str(pedimento.id), str(pedimento.organizacion.id)])
 
 @shared_task
 def crear_procesamiento_cove(pedimento_id):
@@ -54,7 +64,7 @@ def crear_procesamiento_cove(pedimento_id):
     if pedimento.coves.exists():
         existe = ProcesamientoPedimento.objects.filter(
             pedimento=pedimento,
-            servicio_id=8,  # ID del servicio de Coves
+            servicio_id=8,
             organizacion=pedimento.organizacion,
             estado_id__in=[1, 2, 3, 4]
         ).exists()
@@ -62,10 +72,11 @@ def crear_procesamiento_cove(pedimento_id):
             logger.info(f"[TAREA] ProcesamientoPedimento cove creado para pedimento {pedimento_id}")
             ProcesamientoPedimento.objects.create(
                 pedimento=pedimento,
-                estado_id=1,  # Estado "pendiente"
+                estado_id=1,
                 servicio_id=8,
                 organizacion=pedimento.organizacion
             )
+            procesar_cove_individual.apply_async(args=[str(pedimento.id), str(pedimento.organizacion.id)])
 
 @shared_task
 def crear_procesamiento_acuse(pedimento_id):
@@ -73,10 +84,10 @@ def crear_procesamiento_acuse(pedimento_id):
     logger = logging.getLogger('api.customs.async_operations')
     pedimento = Pedimento.objects.get(id=pedimento_id)
     logger.info(f"[TAREA] crear_procesamiento_acuse para pedimento {pedimento_id}")
-    if pedimento.coves.exists():
+    if pedimento.documentos.exists():
         existe = ProcesamientoPedimento.objects.filter(
             pedimento=pedimento,
-            servicio_id=6,  # ID del servicio de Acuse Cove
+            servicio_id=6,
             organizacion=pedimento.organizacion,
             estado_id__in=[1, 2, 3, 4]
         ).exists()
@@ -84,10 +95,11 @@ def crear_procesamiento_acuse(pedimento_id):
             logger.info(f"[TAREA] ProcesamientoPedimento acuse creado para pedimento {pedimento_id}")
             ProcesamientoPedimento.objects.create(
                 pedimento=pedimento,
-                estado_id=1,  # Estado "pendiente"
+                estado_id=1,
                 servicio_id=6,
                 organizacion=pedimento.organizacion
             )
+            procesar_acuse_individual.apply_async(args=[str(pedimento.id), str(pedimento.organizacion.id)])
 
 @shared_task
 def crear_procesamiento_acuse_cove(pedimento_id):
@@ -98,7 +110,7 @@ def crear_procesamiento_acuse_cove(pedimento_id):
     if pedimento.coves.exists():
         existe = ProcesamientoPedimento.objects.filter(
             pedimento=pedimento,
-            servicio_id=9,  # ID del servicio de Acuse Cove
+            servicio_id=9,
             organizacion=pedimento.organizacion,
             estado_id__in=[1, 2, 3, 4]
         ).exists()
@@ -106,10 +118,11 @@ def crear_procesamiento_acuse_cove(pedimento_id):
             logger.info(f"[TAREA] ProcesamientoPedimento acuse_cove creado para pedimento {pedimento_id}")
             ProcesamientoPedimento.objects.create(
                 pedimento=pedimento,
-                estado_id=1,  # Estado "pendiente"
+                estado_id=1,
                 servicio_id=9,
                 organizacion=pedimento.organizacion
             )
+            procesar_acuse_cove_individual.apply_async(args=[str(pedimento.id), str(pedimento.organizacion.id)])
 
 @shared_task
 def crear_procesamiento_edocument(pedimento_id):
@@ -120,7 +133,7 @@ def crear_procesamiento_edocument(pedimento_id):
     if pedimento.documentos.exists():
         existe = ProcesamientoPedimento.objects.filter(
             pedimento=pedimento,
-            servicio_id=7,  # ID del servicio de EDocument
+            servicio_id=7,
             organizacion=pedimento.organizacion,
             estado_id__in=[1, 2, 3, 4]
         ).exists()
@@ -128,10 +141,11 @@ def crear_procesamiento_edocument(pedimento_id):
             logger.info(f"[TAREA] ProcesamientoPedimento edocument creado para pedimento {pedimento_id}")
             ProcesamientoPedimento.objects.create(
                 pedimento=pedimento,
-                estado_id=1,  # Estado "pendiente"
+                estado_id=1,
                 servicio_id=7,
                 organizacion=pedimento.organizacion
             )
+            procesar_edoc_individual.apply_async(args=[str(pedimento.id), str(pedimento.organizacion.id)])
 
 @shared_task
 def crear_procesamiento_pedimento_completo(organizacion_id):

api/customs/tasks/microservice_v2.py

@@ -1,3 +1,4 @@
+from api.organization.models import Organizacion
 from celery import group
 from celery import shared_task, group
 from api.customs.models import *
@@ -8,6 +9,11 @@ import requests
 from config.settings import SERVICE_API_URL_V2
 from datetime import datetime
 import json
+import logging
+import uuid
+# este solo fue para pruebas personales, lo dejo por si en un futuro lo requiero
+TEST_ORG_ID = uuid.UUID('defc7848-4f39-4d67-9dba-5bb445248d23')
+logger = logging.getLogger('api.customs.microservice_v2')
 
 def credenciales_to_dict(credenciales):
     if not credenciales:
@@ -132,7 +138,7 @@ def procesar_edocs_pedimento(pedimento_id):
         }
 
         response = requests.post(
-            f"{SERVICE_API_URL_V2}/services/download/edoc/", 
+            f"{SERVICE_API_URL_V2}/services/download/all/edocs/",
             data=json.dumps(payload),
             headers={"Content-Type": "application/json"}
         )
@@ -277,10 +283,23 @@ def procesar_remesas(organizacion_id):
     pedimentos = Pedimento.objects.filter(organizacion_id=organizacion_id)
 
     for pedimento in pedimentos:
-        if not pedimento.documents.filter(document_type=3).exists():  # Tipo 3: Remesa
+        logger.info(f"pedimento >>>> {pedimento}")
-            # Convertir el pedimento a JSON usando el serializer
+        try:
+            # if pedimento.documents.filter(document_type=3).exists():  # Remesa ya descargada
+            #     logger.info(f"Pedimento {pedimento.pedimento} ya tiene remesa descargada, omitiendo.")
+            #     continue
+
             pedimento_dict = pedimento_to_dict(pedimento)
-            credenciales = Vucem.objects.filter(id=CredencialesImportador.objects.filter(rfc=pedimento.contribuyente).first().vucem.id).first()
+
+            credencial_importador = CredencialesImportador.objects.filter(rfc=pedimento.contribuyente).first()
+            if not credencial_importador:
+                logger.warning(f"Sin credenciales para RFC {pedimento.contribuyente} (pedimento {pedimento.pedimento}), omitiendo.")
+                continue
+
+            credenciales = Vucem.objects.filter(id=credencial_importador.vucem.id).first()
+            if not credenciales:
+                logger.warning(f"Credencial Vucem no encontrada para pedimento {pedimento.pedimento}, omitiendo.")
+                continue
 
             credenciales_dict = credenciales_to_dict(credenciales)
 
@@ -289,15 +308,15 @@ def procesar_remesas(organizacion_id):
                 "credencial": credenciales_dict
             }
 
-            
             response = requests.post(
-                f"{SERVICE_API_URL_V2}/services/remesas", 
+                f"{SERVICE_API_URL_V2}/services/remesas/",
                 data=json.dumps(payload),
                 headers={"Content-Type": "application/json"}
             )
-            # Aquí puedes continuar con el resto de tu lógica
+            logger.info(f"Servicio enviado para pedimento {pedimento.pedimento} — status {response.status_code}")
 
-            print(f"Servicio enviado para pedimento {pedimento.pedimento}")
+        except Exception as e:
+            logger.error(f"Error procesando remesa para pedimento {pedimento.pedimento}: {e}", exc_info=True)
             
 @shared_task
 def procesar_coves(organizacion_id):
@@ -522,6 +541,34 @@ def ejecutar_todos_por_organizacion(organizacion_id):
     procesar_pedimentos_completos.delay(organizacion_id)
     procesar_remesas.delay(organizacion_id)
 
+def ejecutar_basicos_organizacion(organizacion_id):
+    # solo coves y e documents, si es necesario ya en un futuro se agregan los de partidas, pedimento completo y esas madres
+    procesar_coves.delay(organizacion_id)
+    procesar_acuse_coves.delay(organizacion_id)
+    procesar_edocs.delay(organizacion_id)
+    procesar_acuses.delay(organizacion_id)
+    # procesar_partidas.delay(organizacion_id)
+    # procesar_pedimentos_completos.delay(organizacion_id)
+    # procesar_remesas.delay(organizacion_id)
 
+@shared_task
+def process_organization_batch(org_id):
+    """
+    Procesa todos los tipos de documentos pendientes para una organización.
+    """
+    ejecutar_basicos_organizacion(org_id)
 
+@shared_task
+def process_all_organizations():
+    """
+    Envía una tarea por organización activa a la cola org_processing.
+    """
+    active_orgs = Organizacion.objects.filter(is_active=True, is_verified=True)
+    
+    for org in active_orgs:
+        process_organization_batch.apply_async(
+            args=[org.id],
+            queue='org_processing'
+        )
+    return f"Dispatched {active_orgs.count()} organizations"
 

api/customs/tests.py

@@ -3,7 +3,12 @@ from django.urls import reverse
 from rest_framework.test import APITestCase, APIClient
 from rest_framework import status
 from django.contrib.auth import get_user_model
+from django.core.files.uploadedfile import SimpleUploadedFile
+from unittest.mock import patch
+from io import BytesIO
+import zipfile
 from api.organization.models import Organizacion
+from api.licence.models import Licencia
 from .models import Pedimento, TipoOperacion, ProcesamientoPedimento, EDocument
 
 User = get_user_model()
@@ -75,3 +80,147 @@ class CustomsViewsTests(APITestCase):
         self.client.force_authenticate(user=self.admin)
         response = self.client.get(url)
         self.assertEqual(response.status_code, status.HTTP_200_OK)
+
+
+# ---------------------------------------------------------------------------
+# Tests de integración para bulk-create (ViewSetPedimento.bulk_create)
+# Verifica que al re-cargar un pedimento existente sus documentos se actualicen
+# ---------------------------------------------------------------------------
+
+class BulkCreateDocumentReplaceTests(APITestCase):
+    """Verifica que bulk-create actualiza los documentos de pedimentos existentes
+    en vez de ignorarlos, y que no quedan archivos residuales en el storage."""
+
+    PEDIMENTO_APP = "24-01-3420-1234567"
+
+    def setUp(self):
+        self.licencia = Licencia.objects.create(nombre="Lic100GB", almacenamiento=100)
+        self.org = Organizacion.objects.create(
+            nombre="OrgBulkCreate",
+            licencia=self.licencia,
+            is_active=True,
+            is_verified=True,
+        )
+        self.user = User.objects.create_user(
+            username="bulkcreateuser", password="pass", organizacion=self.org
+        )
+        self.pedimento = Pedimento.objects.create(
+            organizacion=self.org,
+            pedimento="1234567",
+            pedimento_app=self.PEDIMENTO_APP,
+        )
+        from api.record.models import DocumentType, Fuente
+        self.doc_type = DocumentType.objects.get_or_create(nombre="Pedimento")[0]
+        # bulk_create usa fuente_id=4 hardcodeado; debe existir en la DB de test
+        Fuente.objects.get_or_create(id=4, defaults={"nombre": "Bulk Create"})
+        self.url = reverse("Pedimento-bulk-create")
+        self.client.force_authenticate(user=self.user)
+
+    def _make_zip(self, files_dict):
+        """Crea un ZIP en memoria. files_dict = {nombre_archivo: contenido_bytes}"""
+        buf = BytesIO()
+        with zipfile.ZipFile(buf, "w") as zf:
+            for name, content in files_dict.items():
+                zf.writestr(name, content)
+        buf.seek(0)
+        return SimpleUploadedFile(
+            f"{self.PEDIMENTO_APP}.zip", buf.read(), content_type="application/zip"
+        )
+
+    def _post_zip(self, files_dict):
+        return self.client.post(
+            self.url,
+            {"contribuyente": "XAXX010101000", "archivos": [self._make_zip(files_dict)]},
+            format="multipart",
+        )
+
+    @patch("api.customs.views.storage_service")
+    def test_existing_pedimento_not_duplicated(self, mock_st):
+        """Re-subir un pedimento existente NO debe crear un segundo Pedimento."""
+        mock_st.save_document_from_path.return_value = "org_1/documents/ped/informe_a1b2c3d4.pdf"
+
+        self._post_zip({"informe.pdf": b"contenido"})
+
+        self.assertEqual(
+            Pedimento.objects.filter(
+                organizacion=self.org, pedimento_app=self.PEDIMENTO_APP
+            ).count(),
+            1,
+        )
+
+    @patch("api.customs.views.storage_service")
+    def test_existing_pedimento_document_replaced_not_duplicated(self, mock_st):
+        """Documento existente con el mismo nombre base se reemplaza, no se duplica."""
+        from api.record.models import Document
+
+        old_path = f"org_1/documents/{self.PEDIMENTO_APP}/informe_a1b2c3d4.pdf"
+        old_doc = Document.objects.create(
+            organizacion=self.org,
+            pedimento=self.pedimento,
+            document_type=self.doc_type,
+            archivo=old_path,
+            size=500,
+            extension="pdf",
+        )
+        new_path = f"org_1/documents/{self.PEDIMENTO_APP}/informe_b5c6d7e8.pdf"
+        mock_st.save_document_from_path.return_value = new_path
+        mock_st.delete_file.return_value = True
+
+        self._post_zip({"informe.pdf": b"contenido actualizado"})
+
+        docs = Document.objects.filter(pedimento=self.pedimento)
+        # Sin duplicados
+        self.assertEqual(docs.count(), 1)
+        # Mismo registro
+        self.assertEqual(docs.first().id, old_doc.id)
+        # Archivo actualizado
+        old_doc.refresh_from_db()
+        self.assertEqual(old_doc.archivo.name, new_path)
+
+    @patch("api.customs.views.storage_service")
+    def test_existing_pedimento_stale_file_deleted_from_storage(self, mock_st):
+        """Al reemplazar un documento, el archivo viejo debe eliminarse del storage."""
+        from api.record.models import Document
+
+        old_path = f"org_1/documents/{self.PEDIMENTO_APP}/informe_a1b2c3d4.pdf"
+        Document.objects.create(
+            organizacion=self.org,
+            pedimento=self.pedimento,
+            document_type=self.doc_type,
+            archivo=old_path,
+            size=500,
+            extension="pdf",
+        )
+        mock_st.save_document_from_path.return_value = f"org_1/documents/{self.PEDIMENTO_APP}/informe_b5c6d7e8.pdf"
+        mock_st.delete_file.return_value = True
+
+        self._post_zip({"informe.pdf": b"contenido"})
+
+        # delete_file debe haberse llamado con la ruta del archivo viejo
+        mock_st.delete_file.assert_called()
+        called_arg = str(mock_st.delete_file.call_args[0][0])
+        self.assertIn("informe_a1b2c3d4", called_arg)
+
+    @patch("api.customs.views.storage_service")
+    def test_existing_pedimento_new_file_added(self, mock_st):
+        """Archivo nuevo en el ZIP se añade al pedimento existente."""
+        from api.record.models import Document
+
+        mock_st.save_document_from_path.return_value = "org_1/documents/ped/nuevo_b5c6d7e8.pdf"
+
+        self._post_zip({"nuevo_documento.pdf": b"contenido nuevo"})
+
+        self.assertGreaterEqual(
+            Document.objects.filter(pedimento=self.pedimento).count(), 1
+        )
+
+    @patch("api.customs.views.storage_service")
+    def test_already_existing_count_in_response(self, mock_st):
+        """La respuesta debe indicar que el pedimento ya existía (already_existing_count >= 1)."""
+        mock_st.save_document_from_path.return_value = "org_1/documents/ped/f_a1b2c3d4.pdf"
+
+        response = self._post_zip({"archivo.pdf": b"contenido"})
+
+        self.assertIn(response.status_code, [status.HTTP_200_OK, status.HTTP_207_MULTI_STATUS, status.HTTP_201_CREATED])
+        data = response.json()
+        self.assertGreaterEqual(data.get("already_existing_count", 0), 1)

api/customs/urls.py

@@ -39,6 +39,7 @@ from .views_auditor import (
     auditar_acuse_cove_endpoint,
     auditar_edocuments_endpoint,
     auditar_acuse_endpoint,
+    auditar_remesas_endpoint,
     auditar_cove_pedimento_endpoint,
     auditar_acuse_cove_pedimento_endpoint,
     auditar_edocument_pedimento_endpoint,
@@ -72,6 +73,7 @@ urlpatterns = [
     path('auditor/auditar-acuse-cove/', auditar_acuse_cove_endpoint, name='auditar-acuse-cove'),
     path('auditor/auditar-edocuments/', auditar_edocuments_endpoint, name='auditar-edocuments'),
     path('auditor/auditar-acuse/', auditar_acuse_endpoint, name='auditar-acuse'),
+    path('auditor/auditar-remesas/', auditar_remesas_endpoint, name='auditar-remesas'),
     path('auditor/auditar-cove/pedimento/', auditar_cove_pedimento_endpoint, name='auditar-cove-pedimento'),
     path('auditor/auditar-acuse-cove/pedimento/', auditar_acuse_cove_pedimento_endpoint, name='auditar-acuse-cove-pedimento'),
     path('auditor/auditar-edocument/pedimento/', auditar_edocument_pedimento_endpoint, name='auditar-edocument-pedimento'),

api/customs/views.py

@@ -1,3 +1,4 @@
+from api.utils.storage_service import storage_service
 from config.settings import SERVICE_API_URL
 from django.shortcuts import render
 from rest_framework import viewsets
@@ -61,7 +62,6 @@ except ImportError:
 
 # Importar tarea de procesamiento de pedimento (Celery)
 from api.customs.tasks.microservice import procesar_pedimento_completo_individual
-from api.utils.storage_service import storage_service
 
 def get_available_extractors():
     """
@@ -394,6 +394,131 @@ class ViewSetPedimento(LoggingMixin, viewsets.ModelViewSet, OrganizacionFiltrada
         except Exception as e:
             return Response({"error": f"Error inesperado al llamar al servicio API: {str(e)}"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
 
+    @action(detail=True, methods=['post'], url_path='procesar-partidas')
+    def procesar_partidas(self, request, pk=None):
+        """
+        Acción para disparar el procesamiento de un partidas de un pedimento existente.
+        Dispara la tarea `procesar_partidas_individual` de forma asíncrona
+        y devuelve el `task_id`.
+        """
+        pedimento = self.get_object()
+        try:
+            from api.customs.tasks import microservice_v2
+
+            # Usar el nombre del servicio de Docker Compose en lugar de localhost
+            task = microservice_v2.procesar_partidas_pedimento.delay(pedimento.id)
+            # Verificar si la respuesta fue exitosa
+            if task.id:
+                return Response({"status": "Iniciando Procesamiento de Partidas", "task_id": task.id}, status=status.HTTP_202_ACCEPTED)
+            else:
+                return Response({"status": "El Servicio respondió con error", "task_id": 0}, status=status.HTTP_202_ACCEPTED)
+        except Exception as e:
+            return Response({"error": f"Error inesperado al llamar al servicio API: {str(e)}"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+
+    @action(detail=True, methods=['post'], url_path='procesar-coves')
+    def procesar_coves(self, request, pk=None):
+        """
+        Acción para disparar el procesamiento de un cove de un pedimento existente.
+        Dispara la tarea `procesar_coves_individual` de forma asíncrona
+        y devuelve el `task_id`.
+        """
+        pedimento = self.get_object()
+        try:
+            from api.customs.tasks import microservice_v2
+
+            # Usar el nombre del servicio de Docker Compose en lugar de localhost
+            task = microservice_v2.procesar_coves_pedimento.delay(pedimento.id)
+            # Verificar si la respuesta fue exitosa
+            if task.id:
+                return Response({"status": "Iniciando Procesamiento de COVES", "task_id": task.id}, status=status.HTTP_202_ACCEPTED)
+            else:
+                return Response({"status": "Servicio API respondió con error", "task_id": 0}, status=status.HTTP_202_ACCEPTED)
+        except Exception as e:
+            return Response({"error": f"Error inesperado al llamar al servicio API: {str(e)}"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+    @action(detail=True, methods=['post'], url_path='procesar-acuse-coves')
+    def procesar_acuse_coves(self, request, pk=None):
+        """
+        Acción para disparar el procesamiento de un acuse cove de un pedimento existente.
+        Dispara la tarea `procesar_acuse_coves_individual` de forma asíncrona
+        y devuelve el `task_id`.
+        """
+        pedimento = self.get_object()
+        try:
+        # Usar el nombre del servicio de Docker Compose en lugar de localhost
+            from api.customs.tasks import microservice_v2
+
+            # Usar el nombre del servicio de Docker Compose en lugar de localhost
+            task = microservice_v2.procesar_acuse_coves_pedimento.delay(pedimento.id)
+            # Verificar si la respuesta fue exitosa
+            if task.id:
+                return Response({"status": "Iniciando Procesamiento de Acuse COVES", "task_id": task.id}, status=status.HTTP_202_ACCEPTED)
+            else:
+                return Response({"status": "Servicio API respondió con error", "task_id": 0}, status=status.HTTP_202_ACCEPTED)
+        except Exception as e:
+            return Response({"error": f"Error inesperado al llamar al servicio API: {str(e)}"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+    @action(detail=True, methods=['post'], url_path='procesar-edocuments')
+    def procesar_edocs(self, request, pk=None):
+        """
+        Acción para disparar el procesamiento de un edocuments de un pedimento existente.
+        Dispara la tarea `procesar_edocuments_individual` de forma asíncrona
+        y devuelve el `task_id`.
+        """
+        pedimento = self.get_object()
+        try:
+        # Usar el nombre del servicio de Docker Compose en lugar de localhost
+            from api.customs.tasks import microservice_v2
+
+            # Usar el nombre del servicio de Docker Compose en lugar de localhost
+            task = microservice_v2.procesar_edocs_pedimento.delay(pedimento.id)
+            # Verificar si la respuesta fue exitosa
+            if task.id:
+                return Response({"status": "Iniciando Procesamiento de EDOCS", "task_id": task.id}, status=status.HTTP_202_ACCEPTED)
+            else:
+                return Response({"status": "Servicio API respondió con error", "task_id": 0}, status=status.HTTP_202_ACCEPTED)
+        except Exception as e:
+            return Response({"error": f"Error inesperado al llamar al servicio API: {str(e)}"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+    @action(detail=True, methods=['post'], url_path='procesar-acuses')
+    def procesar_acuses(self, request, pk=None):
+        """
+        Acción para disparar el procesamiento de un acuses de un pedimento existente.
+        Dispara la tarea `procesar_acuses_individual` de forma asíncrona
+        y devuelve el `task_id`.
+        """
+        pedimento = self.get_object()
+        try:
+            from api.customs.tasks import microservice_v2
+            # Usar el nombre del servicio de Docker Compose en lugar de localhost
+            task = microservice_v2.procesar_acuses_pedimento.delay(pedimento.id)
+            # Verificar si la respuesta fue exitosa
+            if task.id:
+                return Response({"status": "Iniciando Procesamiento de Acuses", "task_id": task.id}, status=status.HTTP_202_ACCEPTED)
+            else:
+                return Response({"status": "Servicio API respondió con error", "task_id": 0}, status=status.HTTP_202_ACCEPTED)
+        except Exception as e:
+            return Response({"error": f"Error inesperado al llamar al servicio API: {str(e)}"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+
+    @action(detail=True, methods=['post'], url_path='procesar-remesas')
+    def procesar_remesas(self, request, pk=None):
+        """
+        Acción para disparar el procesamiento de remesas de un pedimento existente.
+        Dispara la tarea `procesar_remesas_pedimento` de forma asíncrona
+        y devuelve el `task_id`.
+        """
+        pedimento = self.get_object()
+        try:
+            from api.customs.tasks import microservice_v2
+            task = microservice_v2.procesar_remesas_pedimento.delay(pedimento.id)
+            if task.id:
+                return Response({"status": "Iniciando Procesamiento de Remesas", "task_id": task.id}, status=status.HTTP_202_ACCEPTED)
+            else:
+                return Response({"status": "Servicio API respondió con error", "task_id": 0}, status=status.HTTP_202_ACCEPTED)
+        except Exception as e:
+            return Response({"error": f"Error inesperado al llamar al servicio API: {str(e)}"}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+
     @action(detail=False, methods=['post'], url_path='bulk-delete')
     def bulk_delete(self, request):
         import traceback
@@ -657,11 +782,7 @@ class ViewSetPedimento(LoggingMixin, viewsets.ModelViewSet, OrganizacionFiltrada
                             "contribuyente": existing_pedimento.contribuyente.rfc if existing_pedimento.contribuyente else None,
                             "archivo_original": archivo.name
                         })
-                        # NO procesamos este archivo, pasamos al siguiente
+                        # Continuar al procesamiento de documentos del pedimento existente
-                        continue
-                    
-                    # Si el pedimento no existe, continuar con el procesamiento normal
-                    print("📝 Pedimento no existe, continuando con procesamiento...")
                     
                     # Crear subdirectorio para cada archivo usando el nombre del archivo sin extensión
                     sub_dir = os.path.join(temp_dir, archivo_name_sin_extension)
@@ -713,7 +834,10 @@ class ViewSetPedimento(LoggingMixin, viewsets.ModelViewSet, OrganizacionFiltrada
                                 f.write(chunk)
                         print(f"Archivo individual {archivo.name} guardado en sub_dir:", archivo_path)
                     
-                    # Ahora crear el pedimento (ya verificamos que no existe)
+                    if existing_pedimento:
+                        pedimento = existing_pedimento
+                    else:
+                        # Crear el pedimento nuevo
                         try:
                             print("🔄 Iniciando creación de pedimento...")
 
@@ -2248,6 +2372,7 @@ class ViewSetEDocument(LoggingMixin, viewsets.ModelViewSet, OrganizacionFiltrada
             serializer.save()
             return
         
+        print(f"self.request.user.groups >>>> {self.request.user.groups}")
         if (self.request.user.groups.filter(name='developer').exists() or self.request.user.groups.filter(name='admin').exists() or self.request.user.groups.filter(name='user').exists()) and self.request.user.groups.filter(name='Agente Aduanal').exists():
             # Para usuarios normales, usar siempre su organización
             if not hasattr(self.request.user, 'organizacion') or not self.request.user.organizacion:
@@ -2355,6 +2480,15 @@ class ImportadorViewSet(viewsets.ModelViewSet, OrganizacionFiltradaMixin):
     model = Importador
 
     def get_queryset(self):
+        user = self.request.user
+        grupos = user.groups.values_list('name', flat=True)
+
+        if user.is_superuser:
+            return Importador.objects.all()
+
+        if 'Importador' in grupos:
+            return user.rfc.all()
+
         return self.get_queryset_filtrado_por_organizacion()
 
     def perform_create(self, serializer):
@@ -2889,7 +3023,7 @@ def extract_django_suffix(filename):
     """
     name_without_ext = os.path.splitext(filename)[0]
     
-    match = re.search(r'_([a-zA-Z0-9]{7})$', name_without_ext)
+    match = re.search(r'_([a-zA-Z0-9]{8})$', name_without_ext)
     if match:
         return match.group(1)
     return None
@@ -2903,7 +3037,7 @@ def get_clean_base_filename(filename):
 
     django_suffix = extract_django_suffix(name_without_ext)
     if django_suffix:
-        base_name = name_without_ext[:-8]  
+        base_name = name_without_ext[:-9]  # elimina _XXXXXXXX (underscore + 8 chars UUID)
     else:
         base_name = name_without_ext
     

api/customs/views_auditor.py

@@ -8,27 +8,24 @@ from drf_yasg import openapi
 from core.permissions import IsSuperUser, IsSameOrganizationDeveloper
 from .tasks.auditoria import (
     crear_partidas,
-    crear_partidas_por_pedimento,
-    auditar_procesamiento_remesa_por_pedimento,
     auditar_coves,
     auditar_acuse_cove,
     auditar_edocuments,
     auditar_acuse,
-    auditar_cove_por_pedimento,
+    auditar_remesas,
-    auditar_acuse_cove_por_pedimento,
-    auditar_edocument_por_pedimento,
-    auditar_acuse_por_pedimento
 )
 from .tasks.internal_services import auditar_pedimentos
 from .tasks.microservice_v2 import procesar_pedimentos_completos
 from api.customs.models import Pedimento
 from api.organization.models import Organizacion
 from api.record.models import Document
-from .tasks.auditoria import auditar_pedimento_por_id
 from .tasks.auditoria_xml import extraer_info_pedimento_xml
 import tempfile
 import os
 from api.utils.storage_service import storage_service
+import logging
+import uuid
+logger = logging.getLogger('api.customs.views_auditor')
 
 def get_document_content(documento):
     """
@@ -72,7 +69,7 @@ def get_document_path(documento):
 
 @swagger_auto_schema(
     method='post',
-    operation_description="Crea partidas para todos los pedimentos de una organización",
+    operation_description="Crea partidas faltantes para todos los pedimentos de una organización e informa cuáles están descargadas",
     request_body=openapi.Schema(
         type=openapi.TYPE_OBJECT,
         properties={
@@ -81,7 +78,7 @@ def get_document_path(documento):
         required=['organizacion_id']
     ),
     responses={
-        200: openapi.Response('Tarea iniciada correctamente'),
+        202: openapi.Response('Tarea iniciada — usar task_id para consultar resultado'),
         400: openapi.Response('Error en los parámetros'),
         403: openapi.Response('No tiene permisos suficientes')
     }
@@ -89,37 +86,27 @@ def get_document_path(documento):
 @api_view(['POST'])
 @permission_classes([IsAuthenticated & (IsSuperUser | IsSameOrganizationDeveloper)])
 def crear_partidas_organizacion(request):
-    """
-    Crea partidas para todos los pedimentos de una organización específica.
-    """
     organizacion_id = request.data.get('organizacion_id')
 
     if not organizacion_id:
-        return Response(
+        return Response({'error': 'Debe proporcionar organizacion_id'}, status=status.HTTP_400_BAD_REQUEST)
-            {'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(
+        return Response({'error': 'No tiene permisos para esta organización'}, status=status.HTTP_403_FORBIDDEN)
-            {'error': 'No tiene permisos para esta organización'},
-            status=status.HTTP_403_FORBIDDEN
-        )
 
-    # Ejecutar la tarea
     task = crear_partidas.delay(organizacion_id)
-    message = f"Creación de partidas iniciada para la organización {organizacion_id}"
 
     return Response({
-        'message': message,
+        'organizacion_id': organizacion_id,
-        'task_id': task.id
+        'auditoria': 'partidas',
-    }, status=status.HTTP_200_OK)
+        'task_id': task.id,
+        'mensaje': f'Creación de partidas iniciada. Consulta el resultado en GET /api/tasks/status/{task.id}/',
+    }, status=status.HTTP_202_ACCEPTED)
 
 @swagger_auto_schema(
     method='post',
-    operation_description="Crea partidas para un pedimento específico",
+    operation_description="Crea partidas faltantes para un pedimento e informa cuáles están descargadas",
     request_body=openapi.Schema(
         type=openapi.TYPE_OBJECT,
         properties={
@@ -128,48 +115,74 @@ def crear_partidas_organizacion(request):
         required=['pedimento_id']
     ),
     responses={
-        200: openapi.Response('Tarea iniciada correctamente'),
+        200: openapi.Response('Resultado de creación y estado de descarga de partidas'),
         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 ])
+@permission_classes([IsAuthenticated])
 def crear_partidas_pedimento(request):
-    """
-    Crea partidas para un pedimento específico.
-    """
     pedimento_id = request.data.get('pedimento_id')
 
     if not pedimento_id:
-        return Response(
+        return Response({'error': 'Debe proporcionar pedimento_id'}, status=status.HTTP_400_BAD_REQUEST)
-            {'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)
+        pedimento = Pedimento.objects.prefetch_related('partidas').select_related('organizacion').get(id=pedimento_id)
+    except Pedimento.DoesNotExist:
+        return Response({'error': 'Pedimento no encontrado'}, status=status.HTTP_404_NOT_FOUND)
+
     user = request.user
     if not user.is_superuser and str(pedimento.organizacion.id) != str(user.organizacion.id):
-            return Response(
+        return Response({'error': 'No tiene permisos para este pedimento'}, status=status.HTTP_403_FORBIDDEN)
-                {'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
+    if not pedimento.numero_partidas or pedimento.numero_partidas <= 0:
-    task = crear_partidas_por_pedimento.delay(pedimento_id)
+        return Response({
-    message = f"Creación de partidas iniciada para el pedimento {pedimento_id}"
+            'pedimento_id': str(pedimento_id),
+            'pedimento': pedimento.pedimento,
+            'estado': 'sin_datos',
+            'mensaje': f'El pedimento no tiene número de partidas definido (numero_partidas={pedimento.numero_partidas})',
+        }, status=status.HTTP_200_OK)
+
+    # Crear partidas faltantes (get_or_create por número)
+    from api.customs.models import Partida
+    partidas_creadas = 0
+    for i in range(1, pedimento.numero_partidas + 1):
+        _, created = Partida.objects.get_or_create(
+            pedimento=pedimento,
+            numero_partida=i,
+            defaults={'organizacion_id': pedimento.organizacion_id}
+        )
+        if created:
+            partidas_creadas += 1
+
+    # Evaluar estado de descarga sobre el conjunto completo
+    partidas = list(pedimento.partidas.order_by('numero_partida'))
+    total = len(partidas)
+    descargadas = [p.numero_partida for p in partidas if p.descargado]
+    no_descargadas = [p.numero_partida for p in partidas if not p.descargado]
+
+    if not no_descargadas:
+        estado = 'completado'
+        mensaje = f'Todas las partidas están descargadas ({total}/{total})'
+    else:
+        estado = 'en_proceso'
+        mensaje = f'{len(no_descargadas)} de {total} partidas pendientes de descarga'
 
     return Response({
-        'message': message,
+        'pedimento_id': str(pedimento_id),
-        'task_id': task.id
+        'pedimento': pedimento.pedimento,
+        'estado': estado,
+        'mensaje': mensaje,
+        'resumen': {
+            'total_partidas': total,
+            'partidas_creadas_ahora': partidas_creadas,
+            'descargadas': len(descargadas),
+            'no_descargadas': len(no_descargadas),
+        },
+        'no_descargadas': no_descargadas,
     }, status=status.HTTP_200_OK)
 
 @swagger_auto_schema(
@@ -223,7 +236,7 @@ def auditar_pedimentos_endpoint(request):
 
 @swagger_auto_schema(
     method='post',
-    operation_description="Audita el procesamiento de remesa para un pedimento específico",
+    operation_description="Audita el estado de procesamiento de remesa de un pedimento específico",
     request_body=openapi.Schema(
         type=openapi.TYPE_OBJECT,
         properties={
@@ -232,7 +245,7 @@ def auditar_pedimentos_endpoint(request):
         required=['pedimento_id']
     ),
     responses={
-        200: openapi.Response('Tarea de auditoría iniciada correctamente'),
+        200: openapi.Response('Estado de procesamiento de remesa del pedimento'),
         400: openapi.Response('Error en los parámetros'),
         403: openapi.Response('No tiene permisos suficientes'),
         404: openapi.Response('Pedimento no encontrado')
@@ -241,247 +254,179 @@ def auditar_pedimentos_endpoint(request):
 @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(
+        return Response({'error': 'Debe proporcionar pedimento_id'}, status=status.HTTP_400_BAD_REQUEST)
-            {'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)
+        pedimento = Pedimento.objects.select_related('organizacion').prefetch_related('coves').get(id=pedimento_id)
+    except Pedimento.DoesNotExist:
+        return Response({'error': 'Pedimento no encontrado'}, status=status.HTTP_404_NOT_FOUND)
+
     user = request.user
     if not user.is_superuser and str(pedimento.organizacion.id) != str(user.organizacion.id):
-            return Response(
+        return Response({'error': 'No tiene permisos para este pedimento'}, status=status.HTTP_403_FORBIDDEN)
-                {'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
+    if not pedimento.remesas:
-    task = auditar_procesamiento_remesa_por_pedimento.delay(pedimento_id)
+        return Response({
-    message = f"Auditoría de remesa iniciada para el pedimento {pedimento_id}"
+            'pedimento_id': str(pedimento_id),
+            'pedimento': pedimento.pedimento,
+            'tiene_remesas': False,
+            'estado': 'completado',
+            'mensaje': 'El pedimento no tiene remesas para procesar',
+        }, status=status.HTTP_200_OK)
+
+    tiene_documento_remesa = pedimento.documents.filter(document_type=3).exists()
+    coves = list(pedimento.coves.all())
+    total_coves = len(coves)
+
+    if not tiene_documento_remesa:
+        estado = 'en_proceso'
+        mensaje = 'Documento XML de remesa aún no descargado'
+    elif total_coves == 0:
+        estado = 'en_proceso'
+        mensaje = 'Documento de remesa disponible pero no se han creado COVEs'
+    else:
+        estado = 'completado'
+        mensaje = f'Remesa procesada — {total_coves} COVE(s) registrados'
 
     return Response({
-        'message': message,
+        'pedimento_id': str(pedimento_id),
-        'task_id': task.id,
-        'pedimento': {
-            'id': str(pedimento.id),
         'pedimento': pedimento.pedimento,
-            'tiene_remesas': pedimento.remesas
+        'tiene_remesas': True,
-        }
+        'estado': estado,
+        'mensaje': mensaje,
+        'resumen': {
+            'tiene_documento_remesa': tiene_documento_remesa,
+            'total_coves_registrados': total_coves,
+        },
+        'coves': [c.numero_cove for c in coves],
     }, status=status.HTTP_200_OK)
 
 
+def _lanzar_auditoria_organizacion(request, task_fn, label):
+    """Helper compartido para los 4 endpoints de auditoría masiva por organización."""
+    organizacion_id = request.data.get('organizacion_id')
+    if not organizacion_id:
+        return Response({'error': 'Debe proporcionar organizacion_id'}, status=status.HTTP_400_BAD_REQUEST)
+
+    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)
+
+    task = task_fn.delay(organizacion_id)
+    return Response({
+        'organizacion_id': organizacion_id,
+        'auditoria': label,
+        'task_id': task.id,
+        'mensaje': f'Auditoría de {label} iniciada. Consulta el resultado en GET /api/tasks/status/{task.id}/',
+    }, status=status.HTTP_202_ACCEPTED)
+
+
 @swagger_auto_schema(
     method='post',
-    operation_description="Audita los COVEs de una organización",
+    operation_description="Audita el estado de descarga de COVEs de todos los pedimentos de una organización",
     request_body=openapi.Schema(
         type=openapi.TYPE_OBJECT,
-        properties={
+        properties={'organizacion_id': openapi.Schema(type=openapi.TYPE_STRING)},
-            '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'),
+        202: openapi.Response('Tarea iniciada — usar task_id para consultar resultado'),
         400: openapi.Response('Error en los parámetros'),
-        403: openapi.Response('No tiene permisos suficientes')
+        403: openapi.Response('No tiene permisos suficientes'),
     }
 )
 @api_view(['POST'])
 @permission_classes([IsAuthenticated & (IsSuperUser | IsSameOrganizationDeveloper)])
 def auditar_coves_endpoint(request):
-    """
+    return _lanzar_auditoria_organizacion(request, auditar_coves, 'COVEs')
-    Inicia una tarea de auditoría para los COVEs de una organización.
-    Verifica la existencia y validez de los COVEs generados.
-    """
-    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_coves.delay(organizacion_id)
-    message = f"Auditoría de COVEs 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 los acuses de COVE de una organización",
+    operation_description="Audita el estado de descarga de acuses de COVE de todos los pedimentos de una organización",
     request_body=openapi.Schema(
         type=openapi.TYPE_OBJECT,
-        properties={
+        properties={'organizacion_id': openapi.Schema(type=openapi.TYPE_STRING)},
-            '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'),
+        202: openapi.Response('Tarea iniciada — usar task_id para consultar resultado'),
         400: openapi.Response('Error en los parámetros'),
-        403: openapi.Response('No tiene permisos suficientes')
+        403: openapi.Response('No tiene permisos suficientes'),
     }
 )
 @api_view(['POST'])
 @permission_classes([IsAuthenticated & (IsSuperUser | IsSameOrganizationDeveloper)])
 def auditar_acuse_cove_endpoint(request):
-    """
+    return _lanzar_auditoria_organizacion(request, auditar_acuse_cove, 'acuses de COVE')
-    Inicia una tarea de auditoría para los acuses de COVE de una organización.
-    Verifica la recepción y validez de los acuses de COVE.
-    """
-    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_acuse_cove.delay(organizacion_id)
-    message = f"Auditoría de acuses de COVE 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 los EDocuments de una organización",
+    operation_description="Audita el estado de descarga de EDocuments de todos los pedimentos de una organización",
     request_body=openapi.Schema(
         type=openapi.TYPE_OBJECT,
-        properties={
+        properties={'organizacion_id': openapi.Schema(type=openapi.TYPE_STRING)},
-            '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'),
+        202: openapi.Response('Tarea iniciada — usar task_id para consultar resultado'),
         400: openapi.Response('Error en los parámetros'),
-        403: openapi.Response('No tiene permisos suficientes')
+        403: openapi.Response('No tiene permisos suficientes'),
     }
 )
 @api_view(['POST'])
 @permission_classes([IsAuthenticated & (IsSuperUser | IsSameOrganizationDeveloper)])
 def auditar_edocuments_endpoint(request):
-    """
+    return _lanzar_auditoria_organizacion(request, auditar_edocuments, 'EDocuments')
-    Inicia una tarea de auditoría para los EDocuments de una organización.
-    Verifica la existencia y validez de los EDocuments generados.
-    """
-    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_edocuments.delay(organizacion_id)
-    message = f"Auditoría de EDocuments 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 los acuses de una organización",
+    operation_description="Audita el estado de descarga de acuses de EDocument de todos los pedimentos de una organización",
     request_body=openapi.Schema(
         type=openapi.TYPE_OBJECT,
-        properties={
+        properties={'organizacion_id': openapi.Schema(type=openapi.TYPE_STRING)},
-            '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'),
+        202: openapi.Response('Tarea iniciada — usar task_id para consultar resultado'),
         400: openapi.Response('Error en los parámetros'),
-        403: openapi.Response('No tiene permisos suficientes')
+        403: openapi.Response('No tiene permisos suficientes'),
     }
 )
 @api_view(['POST'])
 @permission_classes([IsAuthenticated & (IsSuperUser | IsSameOrganizationDeveloper)])
 def auditar_acuse_endpoint(request):
-    """
+    return _lanzar_auditoria_organizacion(request, auditar_acuse, 'acuses de EDocument')
-    Inicia una tarea de auditoría para los acuses de una organización.
-    Verifica la recepción y validez de los acuses.
-    """
-    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_acuse.delay(organizacion_id)
-    message = f"Auditoría de acuses 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 COVE de un pedimento específico",
+    operation_description="Audita el estado de descarga de remesas de todos los pedimentos de una organización",
+    request_body=openapi.Schema(
+        type=openapi.TYPE_OBJECT,
+        properties={'organizacion_id': openapi.Schema(type=openapi.TYPE_STRING)},
+        required=['organizacion_id']
+    ),
+    responses={
+        202: openapi.Response('Tarea iniciada — usar task_id para consultar resultado'),
+        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_remesas_endpoint(request):
+    return _lanzar_auditoria_organizacion(request, auditar_remesas, 'remesas')
+
+
+@swagger_auto_schema(
+    method='post',
+    operation_description="Audita el estado de descarga de COVEs de un pedimento específico",
     request_body=openapi.Schema(
         type=openapi.TYPE_OBJECT,
         properties={
@@ -490,7 +435,7 @@ def auditar_acuse_endpoint(request):
         required=['pedimento_id']
     ),
     responses={
-        200: openapi.Response('Tarea de auditoría iniciada correctamente'),
+        200: openapi.Response('Estado de descarga de COVEs del pedimento'),
         400: openapi.Response('Error en los parámetros'),
         403: openapi.Response('No tiene permisos suficientes'),
         404: openapi.Response('Pedimento no encontrado')
@@ -502,19 +447,48 @@ def auditar_cove_pedimento_endpoint(request):
     pedimento_id = request.data.get('pedimento_id')
     if not pedimento_id:
         return Response({'error': 'Debe proporcionar pedimento_id'}, status=status.HTTP_400_BAD_REQUEST)
+
     try:
-        pedimento = Pedimento.objects.get(id=pedimento_id)
+        pedimento = Pedimento.objects.select_related('organizacion').prefetch_related('coves').get(id=pedimento_id)
+    except Pedimento.DoesNotExist:
+        return Response({'error': 'Pedimento no encontrado'}, status=status.HTTP_404_NOT_FOUND)
+
     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)
+    coves = list(pedimento.coves.all())
-    task = auditar_cove_por_pedimento.delay(pedimento_id)
+    total = len(coves)
-    return Response({'message': f'Auditoría de COVE iniciada para el pedimento {pedimento_id}', 'task_id': task.id}, status=status.HTTP_200_OK)
+    descargados = sum(1 for c in coves if c.cove_descargado)
+    pendientes = [c.numero_cove for c in coves if not c.cove_descargado]
+
+    if total == 0:
+        nuevo_estado = 3
+        mensaje = 'El pedimento no tiene COVEs registrados'
+    elif descargados == total:
+        nuevo_estado = 3
+        mensaje = 'Todos los COVEs están descargados'
+    else:
+        nuevo_estado = 4
+        mensaje = f'{total - descargados} de {total} COVEs pendientes de descarga'
+
+    from api.customs.tasks.auditoria import modificar_estado_procesamiento
+    modificar_estado_procesamiento(pedimento, servicio_id=8, nuevo_estado=nuevo_estado)
+
+    return Response({
+        'pedimento_id': str(pedimento_id),
+        'estado': 'completado' if nuevo_estado == 3 else 'en_proceso',
+        'mensaje': mensaje,
+        'resumen': {
+            'total_coves': total,
+            'coves_descargados': descargados,
+        },
+        'pendientes': pendientes,
+    }, status=status.HTTP_200_OK)
 
 @swagger_auto_schema(
     method='post',
-    operation_description="Audita el acuse de COVE de un pedimento específico",
+    operation_description="Audita el estado de descarga de acuses de COVE de un pedimento específico",
     request_body=openapi.Schema(
         type=openapi.TYPE_OBJECT,
         properties={
@@ -523,7 +497,7 @@ def auditar_cove_pedimento_endpoint(request):
         required=['pedimento_id']
     ),
     responses={
-        200: openapi.Response('Tarea de auditoría iniciada correctamente'),
+        200: openapi.Response('Estado de descarga de acuses de COVE del pedimento'),
         400: openapi.Response('Error en los parámetros'),
         403: openapi.Response('No tiene permisos suficientes'),
         404: openapi.Response('Pedimento no encontrado')
@@ -535,19 +509,48 @@ def auditar_acuse_cove_pedimento_endpoint(request):
     pedimento_id = request.data.get('pedimento_id')
     if not pedimento_id:
         return Response({'error': 'Debe proporcionar pedimento_id'}, status=status.HTTP_400_BAD_REQUEST)
+
     try:
-        pedimento = Pedimento.objects.get(id=pedimento_id)
+        pedimento = Pedimento.objects.select_related('organizacion').prefetch_related('coves').get(id=pedimento_id)
+    except Pedimento.DoesNotExist:
+        return Response({'error': 'Pedimento no encontrado'}, status=status.HTTP_404_NOT_FOUND)
+
     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)
+    coves = list(pedimento.coves.all())
-    task = auditar_acuse_cove_por_pedimento.delay(pedimento_id)
+    total = len(coves)
-    return Response({'message': f'Auditoría de acuse de COVE iniciada para el pedimento {pedimento_id}', 'task_id': task.id}, status=status.HTTP_200_OK)
+    descargados = sum(1 for c in coves if c.acuse_cove_descargado)
+    pendientes = [c.numero_cove for c in coves if not c.acuse_cove_descargado]
+
+    if total == 0:
+        nuevo_estado = 3
+        mensaje = 'El pedimento no tiene COVEs registrados, no hay acuses que auditar'
+    elif descargados == total:
+        nuevo_estado = 3
+        mensaje = 'Todos los acuses de COVE están descargados'
+    else:
+        nuevo_estado = 4
+        mensaje = f'{total - descargados} de {total} acuses de COVE pendientes de descarga'
+
+    from api.customs.tasks.auditoria import modificar_estado_procesamiento
+    modificar_estado_procesamiento(pedimento, servicio_id=9, nuevo_estado=nuevo_estado)
+
+    return Response({
+        'pedimento_id': str(pedimento_id),
+        'estado': 'completado' if nuevo_estado == 3 else 'en_proceso',
+        'mensaje': mensaje,
+        'resumen': {
+            'total_coves': total,
+            'acuses_descargados': descargados,
+        },
+        'pendientes': pendientes,
+    }, status=status.HTTP_200_OK)
 
 @swagger_auto_schema(
     method='post',
-    operation_description="Audita el EDocument de un pedimento específico",
+    operation_description="Audita el estado de descarga de EDocuments de un pedimento específico",
     request_body=openapi.Schema(
         type=openapi.TYPE_OBJECT,
         properties={
@@ -556,7 +559,7 @@ def auditar_acuse_cove_pedimento_endpoint(request):
         required=['pedimento_id']
     ),
     responses={
-        200: openapi.Response('Tarea de auditoría iniciada correctamente'),
+        200: openapi.Response('Estado de descarga de EDocuments del pedimento'),
         400: openapi.Response('Error en los parámetros'),
         403: openapi.Response('No tiene permisos suficientes'),
         404: openapi.Response('Pedimento no encontrado')
@@ -568,19 +571,48 @@ def auditar_edocument_pedimento_endpoint(request):
     pedimento_id = request.data.get('pedimento_id')
     if not pedimento_id:
         return Response({'error': 'Debe proporcionar pedimento_id'}, status=status.HTTP_400_BAD_REQUEST)
+
     try:
-        pedimento = Pedimento.objects.get(id=pedimento_id)
+        pedimento = Pedimento.objects.select_related('organizacion').prefetch_related('documentos').get(id=pedimento_id)
+    except Pedimento.DoesNotExist:
+        return Response({'error': 'Pedimento no encontrado'}, status=status.HTTP_404_NOT_FOUND)
+
     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)
+    edocuments = list(pedimento.documentos.all())
-    task = auditar_edocument_por_pedimento.delay(pedimento_id)
+    total = len(edocuments)
-    return Response({'message': f'Auditoría de EDocument iniciada para el pedimento {pedimento_id}', 'task_id': task.id}, status=status.HTTP_200_OK)
+    descargados = sum(1 for d in edocuments if d.edocument_descargado)
+    pendientes = [d.numero_edocument for d in edocuments if not d.edocument_descargado]
+
+    if total == 0:
+        nuevo_estado = 3
+        mensaje = 'El pedimento no tiene EDocuments registrados'
+    elif descargados == total:
+        nuevo_estado = 3
+        mensaje = 'Todos los EDocuments están descargados'
+    else:
+        nuevo_estado = 4
+        mensaje = f'{total - descargados} de {total} EDocuments pendientes de descarga'
+
+    from api.customs.tasks.auditoria import modificar_estado_procesamiento
+    modificar_estado_procesamiento(pedimento, servicio_id=7, nuevo_estado=nuevo_estado)
+
+    return Response({
+        'pedimento_id': str(pedimento_id),
+        'estado': 'completado' if nuevo_estado == 3 else 'en_proceso',
+        'mensaje': mensaje,
+        'resumen': {
+            'total_edocuments': total,
+            'edocuments_descargados': descargados,
+        },
+        'pendientes': pendientes,
+    }, status=status.HTTP_200_OK)
 
 @swagger_auto_schema(
     method='post',
-    operation_description="Audita el acuse de un pedimento específico",
+    operation_description="Audita el estado de descarga de acuses de EDocument de un pedimento específico",
     request_body=openapi.Schema(
         type=openapi.TYPE_OBJECT,
         properties={
@@ -589,28 +621,56 @@ def auditar_edocument_pedimento_endpoint(request):
         required=['pedimento_id']
     ),
     responses={
-        200: openapi.Response('Tarea de auditoría iniciada correctamente'),
+        200: openapi.Response('Estado de descarga de acuses de EDocument del pedimento'),
         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])
 def auditar_acuse_pedimento_endpoint(request):
     pedimento_id = request.data.get('pedimento_id')
     if not pedimento_id:
         return Response({'error': 'Debe proporcionar pedimento_id'}, status=status.HTTP_400_BAD_REQUEST)
+
     try:
-        pedimento = Pedimento.objects.get(id=pedimento_id)
+        pedimento = Pedimento.objects.select_related('organizacion').prefetch_related('documentos').get(id=pedimento_id)
+    except Pedimento.DoesNotExist:
+        return Response({'error': 'Pedimento no encontrado'}, status=status.HTTP_404_NOT_FOUND)
+
     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)
+    edocuments = list(pedimento.documentos.all())
-    task = auditar_acuse_por_pedimento.delay(pedimento_id)
+    total = len(edocuments)
-    return Response({'message': f'Auditoría de acuse iniciada para el pedimento {pedimento_id}', 'task_id': task.id}, status=status.HTTP_200_OK)
+    descargados = sum(1 for d in edocuments if d.acuse_descargado)
+    pendientes = [d.numero_edocument for d in edocuments if not d.acuse_descargado]
+
+    if total == 0:
+        nuevo_estado = 3
+        mensaje = 'El pedimento no tiene EDocuments registrados, no hay acuses que auditar'
+    elif descargados == total:
+        nuevo_estado = 3
+        mensaje = 'Todos los acuses de EDocument están descargados'
+    else:
+        nuevo_estado = 4
+        mensaje = f'{total - descargados} de {total} acuses de EDocument pendientes de descarga'
+
+    from api.customs.tasks.auditoria import modificar_estado_procesamiento
+    modificar_estado_procesamiento(pedimento, servicio_id=6, nuevo_estado=nuevo_estado)
+
+    return Response({
+        'pedimento_id': str(pedimento_id),
+        'estado': 'completado' if nuevo_estado == 3 else 'en_proceso',
+        'mensaje': mensaje,
+        'resumen': {
+            'total_edocuments': total,
+            'acuses_descargados': descargados,
+        },
+        'pendientes': pendientes,
+    }, status=status.HTTP_200_OK)
 
 ### Procesamiento de pedimentos ###
 @swagger_auto_schema(
@@ -1663,6 +1723,10 @@ def auditar_pedimento_endpoint(request):
         informacion_extraida = []
         
         for documento in documentos_xml:
+
+            print(f"documento >>>> {documento}")
+            logger.info(f"documento >>>> {documento}")
+
             try:
                 xml_info = {
                     'documento_id': str(documento.id),
@@ -1695,9 +1759,6 @@ def auditar_pedimento_endpoint(request):
                     'error': f'Error procesando archivo: {str(e)}'
                 })
         
-        # Ejecutar la tarea de auditoría completa
-        task = auditar_pedimento_por_id.delay(pedimento_id)
-        
         response_data = {
             'pedimento_id': str(pedimento_id),
             'pedimento': pedimento.pedimento,
@@ -1706,7 +1767,6 @@ def auditar_pedimento_endpoint(request):
             'xmls_analizados': xmls_analizados,
             'informacion_extraida': informacion_extraida,
             'auditoria_completa': True,
-            'task_id': task.id,
             'mensaje': f'Auditoría completada para el pedimento {pedimento.pedimento}'
         }
         

api/datastage/models.py

@@ -85,6 +85,8 @@ class Registro501(models.Model):
     consulta = models.CharField(max_length=50, null=True, blank=True)
     datastage = models.ForeignKey(DataStage, on_delete=models.CASCADE, related_name='registro501s', null=True, blank=True)
 
+    created_at = models.DateTimeField(auto_now_add=True)
+
     class Meta:
         db_table = 'registro501'
 
@@ -104,6 +106,8 @@ class Registro502(models.Model):
     datastage = models.ForeignKey(DataStage, on_delete=models.CASCADE, related_name='registro502s', null=True, blank=True)
     patente = models.CharField(max_length=50, null=True, blank=True)
 
+    created_at = models.DateTimeField(auto_now_add=True)
+
     class Meta:
         db_table = 'registro502'
 
@@ -120,6 +124,8 @@ class Registro503(models.Model):
     consulta = models.CharField(max_length=50, null=True, blank=True)
     datastage = models.ForeignKey(DataStage, on_delete=models.CASCADE, related_name='registro503s', null=True, blank=True)
 
+    created_at = models.DateTimeField(auto_now_add=True)
+
     class Meta:
         db_table = 'registro503'
 
@@ -136,6 +142,8 @@ class Registro504(models.Model):
     consulta = models.CharField(max_length=50, null=True, blank=True)
     datastage = models.ForeignKey(DataStage, on_delete=models.CASCADE, related_name='registro504s', null=True, blank=True)
 
+    created_at = models.DateTimeField(auto_now_add=True)
+
     class Meta:
         db_table = 'registro504'
 
@@ -165,6 +173,8 @@ class Registro505(models.Model):
     datastage = models.ForeignKey(DataStage, on_delete=models.CASCADE, related_name='registro505s', null=True, blank=True)
     patente = models.CharField(max_length=50, null=True, blank=True)
 
+    created_at = models.DateTimeField(auto_now_add=True)
+
     class Meta:
         db_table = 'registro505'
 
@@ -181,6 +191,8 @@ class Registro506(models.Model):
     consulta = models.CharField(max_length=50, null=True, blank=True)
     datastage = models.ForeignKey(DataStage, on_delete=models.CASCADE, related_name='registro506s', null=True, blank=True)
 
+    created_at = models.DateTimeField(auto_now_add=True)
+
     class Meta:
         db_table = 'registro506'
 
@@ -199,6 +211,8 @@ class Registro507(models.Model):
     consulta = models.CharField(max_length=50, null=True, blank=True)
     datastage = models.ForeignKey(DataStage, on_delete=models.CASCADE, related_name='registro507s', null=True, blank=True)
 
+    created_at = models.DateTimeField(auto_now_add=True)
+
     class Meta:
         db_table = 'registro507'
 
@@ -223,6 +237,8 @@ class Registro508(models.Model):
     consulta = models.CharField(max_length=50, null=True, blank=True)
     datastage = models.ForeignKey(DataStage, on_delete=models.CASCADE, related_name='registro508s', null=True, blank=True)
 
+    created_at = models.DateTimeField(auto_now_add=True)
+
     class Meta:
         db_table = 'registro508'
 
@@ -241,6 +257,8 @@ class Registro509(models.Model):
     consulta = models.CharField(max_length=50, null=True, blank=True)
     datastage = models.ForeignKey(DataStage, on_delete=models.CASCADE, related_name='registro509s', null=True, blank=True)
 
+    created_at = models.DateTimeField(auto_now_add=True)
+
     class Meta:
         db_table = 'registro509'
 
@@ -261,6 +279,8 @@ class Registro510(models.Model):
     forma_pago = models.CharField(max_length=3, null=True, blank=True)
     importe_pago = models.CharField(max_length=12, null=True, blank=True)
 
+    created_at = models.DateTimeField(auto_now_add=True)
+
     class Meta:
         db_table = 'registro510'
 
@@ -278,6 +298,8 @@ class Registro511(models.Model):
     consulta = models.CharField(max_length=50, null=True, blank=True)
     datastage = models.ForeignKey(DataStage, on_delete=models.CASCADE, related_name='registro511s', null=True, blank=True)
 
+    created_at = models.DateTimeField(auto_now_add=True)
+
     class Meta:
         db_table = 'registro511'
 
@@ -301,6 +323,8 @@ class Registro512(models.Model):
     consulta = models.CharField(max_length=50, null=True, blank=True)
     datastage = models.ForeignKey(DataStage, on_delete=models.CASCADE, related_name='registro512s', null=True, blank=True)
 
+    created_at = models.DateTimeField(auto_now_add=True)
+
     class Meta:
         db_table = 'registro512'
 
@@ -363,6 +387,8 @@ class Registro551(models.Model):
     datastage = models.ForeignKey(DataStage, on_delete=models.CASCADE, related_name='registro551s', null=True, blank=True)
     entidad_fed_destino = models.CharField(max_length=50, null=True, blank=True)
 
+    created_at = models.DateTimeField(auto_now_add=True)
+
     class Meta:
         db_table = 'registro551'
 
@@ -381,6 +407,8 @@ class Registro552(models.Model):
     consulta = models.CharField(max_length=50, null=True, blank=True)
     datastage = models.ForeignKey(DataStage, on_delete=models.CASCADE, related_name='registro552s', null=True, blank=True)
 
+    created_at = models.DateTimeField(auto_now_add=True)
+
     class Meta:
         db_table = 'registro552'
 
@@ -402,6 +430,8 @@ class Registro553(models.Model):
     datastage = models.ForeignKey(DataStage, on_delete=models.CASCADE, related_name='registro553s', null=True, blank=True)
     consulta = models.CharField(max_length=50, null=True, blank=True)
 
+    created_at = models.DateTimeField(auto_now_add=True)
+
     class Meta:
         db_table = 'registro553'
 
@@ -421,6 +451,8 @@ class Registro554(models.Model):
     consulta = models.CharField(max_length=50, null=True, blank=True)
     datastage = models.ForeignKey(DataStage, on_delete=models.CASCADE, related_name='registro554s', null=True, blank=True)
 
+    created_at = models.DateTimeField(auto_now_add=True)
+
     class Meta:
         db_table = 'registro554'
 
@@ -446,6 +478,8 @@ class Registro555(models.Model):
     created_by = models.IntegerField(null=True, blank=True)
     consulta = models.CharField(max_length=50, null=True, blank=True)
 
+    created_at = models.DateTimeField(auto_now_add=True)
+
     class Meta:
         db_table = 'registro555'
 
@@ -465,6 +499,8 @@ class Registro556(models.Model):
     fraccion = models.CharField(max_length=8, null=True, blank=True)
     secuencia_fraccion = models.CharField(max_length=50, null=True, blank=True)
 
+    created_at = models.DateTimeField(auto_now_add=True)
+
     class Meta:
         db_table = 'registro556'
 
@@ -484,6 +520,8 @@ class Registro557(models.Model):
     datastage = models.ForeignKey(DataStage, on_delete=models.CASCADE, related_name='registro557s', null=True, blank=True)
     consulta = models.CharField(max_length=50, null=True, blank=True)
 
+    created_at = models.DateTimeField(auto_now_add=True)
+
     class Meta:
         db_table = 'registro557'
 
@@ -502,6 +540,8 @@ class Registro558(models.Model):
     datastage = models.ForeignKey(DataStage, on_delete=models.CASCADE, related_name='registro558s', null=True, blank=True)
     consulta = models.CharField(max_length=50, null=True, blank=True)
 
+    created_at = models.DateTimeField(auto_now_add=True)
+
     class Meta:
         db_table = 'registro558'
 
@@ -522,6 +562,8 @@ class RegistroSel(models.Model):
     consulta = models.CharField(max_length=50, null=True, blank=True)
     datastage = models.ForeignKey(DataStage, on_delete=models.CASCADE, related_name='registro_sel', null=True, blank=True)
 
+    created_at = models.DateTimeField(auto_now_add=True)
+
     class Meta:
         db_table = 'registro_sel'
 
@@ -546,6 +588,8 @@ class Registro701(models.Model):
     consulta = models.CharField(max_length=50, null=True, blank=True)
     datastage = models.ForeignKey(DataStage, on_delete=models.CASCADE, related_name='registro701s', null=True, blank=True)
 
+    created_at = models.DateTimeField(auto_now_add=True)
+
     class Meta:
         db_table = 'registro701'
 
@@ -564,6 +608,8 @@ class Registro702(models.Model):
     consulta = models.CharField(max_length=50, null=True, blank=True)
     datastage = models.ForeignKey(DataStage, on_delete=models.CASCADE, related_name='registro702s', null=True, blank=True)
 
+    created_at = models.DateTimeField(auto_now_add=True)
+
     class Meta:
         db_table = 'registro702'
 

api/datastage/tasks.py

@@ -9,6 +9,8 @@ import zipfile
 import re
 from api.utils.storage_service import storage_service
 
+logger = logging.getLogger(__name__)
+
 @shared_task
 def procesar_datastage_task(datastage_id, user_organizacion_id=None):
     import traceback
@@ -167,7 +169,10 @@ def procesar_archivo_asc_task(datastage_id, user_organizacion_id, asc_name):
                         continue
                     
                     if first:
-                        field_names = [f for f in line_decoded.split('|')]
+                        field_names = line_decoded.split('|')
+                        # Eliminar columnas vacías del final (líneas terminan con |)
+                        while field_names and field_names[-1] == '':
+                            field_names.pop()
                         field_names_snake = [to_snake_case(f) for f in field_names]
                         first = False
                         continue
@@ -176,6 +181,10 @@ def procesar_archivo_asc_task(datastage_id, user_organizacion_id, asc_name):
                     while values and values[-1] == '':
                         values.pop()
                     if len(values) != len(field_names_snake):
+                        logger.debug(
+                            "%s línea %d: esperados %d campos, recibidos %d — se omite",
+                            asc_name, line_count, len(field_names_snake), len(values)
+                        )
                         continue
                     
                     data = dict(zip(field_names_snake, values))
@@ -185,27 +194,35 @@ def procesar_archivo_asc_task(datastage_id, user_organizacion_id, asc_name):
                     if hasattr(Model, 'datastage_id'):
                         data['datastage_id'] = datastage.id
                     
-                    # Limpiar fechas vacías
+                    # Parsear y normalizar todos los campos de fecha/datetime
                     for field in Model._meta.get_fields():
-                        if hasattr(field, 'get_internal_type') and field.get_internal_type() in ["DateField", "DateTimeField"]:
+                        if not hasattr(field, 'get_internal_type'):
-                            if data.get(field.name) == "":
+                            continue
+                        field_type = field.get_internal_type()
+                        val = data.get(field.name)
+                        if val == '' or val is None:
                             data[field.name] = None
-                    
+                            continue
-                    # Convertir fecha_pago_real
+                        if field_type == 'DateTimeField' and isinstance(val, str):
-                    if 'fecha_pago_real' in data and data['fecha_pago_real']:
-                        fecha_val = data['fecha_pago_real']
-                        if isinstance(fecha_val, str):
-                            try:
-                                dt = datetime.datetime.strptime(fecha_val, '%Y-%m-%d %H:%M:%S')
-                            except ValueError:
-                                try:
-                                    dt = datetime.datetime.strptime(fecha_val, '%Y-%m-%d')
-                                except Exception:
                             dt = None
+                            for fmt in ('%Y-%m-%d %H:%M:%S', '%Y-%m-%d'):
+                                try:
+                                    dt = datetime.datetime.strptime(val, fmt)
+                                    break
+                                except ValueError:
+                                    continue
                             if dt and timezone.is_naive(dt):
                                 dt = timezone.make_aware(dt)
-                            if dt:
+                            data[field.name] = dt
-                                data['fecha_pago_real'] = dt
+                    
+                    # Filtrar data para solo incluir campos válidos del modelo
+                    valid_fields = set()
+                    for f in Model._meta.get_fields():
+                        if hasattr(f, 'name'):
+                            valid_fields.add(f.name)
+                        if hasattr(f, 'attname'):
+                            valid_fields.add(f.attname)
+                    data = {k: v for k, v in data.items() if k in valid_fields}
 
                     try:
                         obj = Model(**data)
@@ -284,8 +301,9 @@ def procesar_archivo_asc_task(datastage_id, user_organizacion_id, asc_name):
                             try:
                                 Pedimento.objects.create(**pedimento_data)
                             except Exception as ped_exc:
-                                pass
+                                logger.warning("No se pudo crear Pedimento %s: %s", pedimento_app, ped_exc)
                     except Exception as e:
+                        logger.error("%s línea %d: error creando objeto %s: %s", asc_name, line_count, model_name, e)
                         continue
                 
         # Bulk create

api/notificaciones/signals/notificaciones.py

@@ -19,7 +19,7 @@ def trigger_notificacion(sender, instance, created, **kwargs):
         for usuario in usuarios_org:
             # Notificar solo a importadores cuyo RFC coincide
             if (usuario.is_importador or usuario.groups.filter(name='Importador').exists()):
-                if usuario.rfc == instance.pedimento.contribuyente:
+                if instance.pedimento.contribuyente in usuario.rfc.all():
                     Notificacion.objects.create(
                         tipo=tipo_info,
                         dirigido=usuario,

api/record/tests.py

@@ -1,12 +1,16 @@
 
 from django.urls import reverse
+from django.test import TestCase
 from rest_framework.test import APITestCase, APIClient
 from rest_framework import status
 from django.core.files.uploadedfile import SimpleUploadedFile
+from unittest.mock import patch, MagicMock
 from api.organization.models import Organizacion, UsoAlmacenamiento
 from api.cuser.models import CustomUser
 from api.customs.models import Pedimento
-from .models import Document
+from api.licence.models import Licencia
+from api.customs.views import is_same_document, get_clean_base_filename
+from .models import Document, DocumentType
 import io
 
 class DocumentViewSetTests(APITestCase):
@@ -95,3 +99,177 @@ class DocumentViewSetTests(APITestCase):
         url = reverse('descargar-documento', args=[doc.id])
         response = self.client.get(url)
         self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+
+
+# ---------------------------------------------------------------------------
+# Tests unitarios para las funciones helper de comparación de documentos
+# ---------------------------------------------------------------------------
+
+class DocumentNameHelperTests(TestCase):
+    """Verifica que get_clean_base_filename e is_same_document manejan
+    correctamente el sufijo UUID de 8 chars que añade storage_service."""
+
+    def test_strips_uuid_suffix(self):
+        self.assertEqual(get_clean_base_filename('informe_a1b2c3d4.pdf'), 'informe')
+
+    def test_no_suffix_unchanged(self):
+        self.assertEqual(get_clean_base_filename('informe.pdf'), 'informe')
+
+    def test_is_same_document_matches_stored_uuid_name(self):
+        """El archivo guardado tiene sufijo, el nuevo no — deben coincidir."""
+        doc = MagicMock()
+        doc.archivo.name = 'org_1/documents/24-01-3420-1234567/informe_a1b2c3d4.pdf'
+        doc.extension = 'pdf'
+        self.assertTrue(is_same_document(doc, 'informe.pdf'))
+
+    def test_is_same_document_different_name_no_match(self):
+        doc = MagicMock()
+        doc.archivo.name = 'org_1/documents/ped/informe_a1b2c3d4.pdf'
+        doc.extension = 'pdf'
+        self.assertFalse(is_same_document(doc, 'otro.pdf'))
+
+    def test_is_same_document_different_extension_no_match(self):
+        doc = MagicMock()
+        doc.archivo.name = 'org_1/documents/ped/informe_a1b2c3d4.pdf'
+        doc.extension = 'pdf'
+        self.assertFalse(is_same_document(doc, 'informe.xml'))
+
+    def test_both_clean_names_equal(self):
+        """Dos archivos con UUID distintos pero mismo nombre base deben coincidir."""
+        doc = MagicMock()
+        doc.archivo.name = 'org_1/documents/ped/pedimento_a1b2c3d4.xml'
+        doc.extension = 'xml'
+        self.assertTrue(is_same_document(doc, 'pedimento_b5c6d7e8.xml'))
+
+
+# ---------------------------------------------------------------------------
+# Tests de integración para bulk-upload (DocumentViewSet.bulk_upload)
+# ---------------------------------------------------------------------------
+
+class BulkUploadReplaceTests(APITestCase):
+    """Verifica que bulk-upload reemplaza documentos existentes en vez de duplicar
+    y que no quedan archivos residuales en el storage."""
+
+    def setUp(self):
+        self.licencia = Licencia.objects.create(nombre="Lic100GB", almacenamiento=100)
+        self.org = Organizacion.objects.create(
+            nombre="OrgBulkUpload",
+            licencia=self.licencia,
+            is_active=True,
+            is_verified=True,
+        )
+        self.user = CustomUser.objects.create_user(
+            username="bulkuploaduser", password="pass", organizacion=self.org
+        )
+        self.pedimento = Pedimento.objects.create(
+            organizacion=self.org,
+            pedimento="1234567",
+            pedimento_app="24-01-3420-1234567",
+        )
+        self.doc_type = DocumentType.objects.get_or_create(nombre="Documento General")[0]
+        self.url = reverse("Document-bulk-upload")
+        self.client.force_authenticate(user=self.user)
+
+    def _post_file(self, filename, content=b"contenido de prueba"):
+        archivo = SimpleUploadedFile(filename, content, content_type="application/pdf")
+        return self.client.post(
+            self.url,
+            {"pedimento_id": str(self.pedimento.id), "files": [archivo]},
+            format="multipart",
+        )
+
+    @patch("api.record.views.storage_service")
+    def test_new_file_creates_document(self, mock_st):
+        """Subir un archivo nuevo crea exactamente un Document."""
+        mock_st.save_document.return_value = "org_1/documents/ped/informe_a1b2c3d4.pdf"
+
+        response = self._post_file("informe.pdf")
+
+        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+        self.assertEqual(Document.objects.filter(pedimento=self.pedimento).count(), 1)
+        mock_st.delete_file.assert_not_called()
+
+    @patch("api.record.views.storage_service")
+    def test_duplicate_replaces_not_creates(self, mock_st):
+        """Re-subir el mismo archivo debe actualizar el Document existente,
+        no crear uno nuevo."""
+        old_path = "org_1/documents/24-01-3420-1234567/informe_a1b2c3d4.pdf"
+        old_doc = Document.objects.create(
+            organizacion=self.org,
+            pedimento=self.pedimento,
+            document_type=self.doc_type,
+            archivo=old_path,
+            size=500,
+            extension="pdf",
+        )
+        new_path = "org_1/documents/24-01-3420-1234567/informe_b5c6d7e8.pdf"
+        mock_st.save_document.return_value = new_path
+        mock_st.delete_file.return_value = True
+
+        response = self._post_file("informe.pdf", b"contenido actualizado")
+
+        self.assertIn(response.status_code, [status.HTTP_201_CREATED, status.HTTP_207_MULTI_STATUS])
+        docs = Document.objects.filter(pedimento=self.pedimento)
+        # Un único Document — sin duplicados
+        self.assertEqual(docs.count(), 1)
+        # Es el mismo registro (mismo UUID)
+        self.assertEqual(docs.first().id, old_doc.id)
+        # El campo archivo fue actualizado
+        old_doc.refresh_from_db()
+        self.assertEqual(old_doc.archivo.name, new_path)
+
+    @patch("api.record.views.storage_service")
+    def test_replace_deletes_old_storage_file(self, mock_st):
+        """Al reemplazar, delete_file debe llamarse con la ruta del archivo viejo."""
+        old_path = "org_1/documents/24-01-3420-1234567/informe_a1b2c3d4.pdf"
+        Document.objects.create(
+            organizacion=self.org,
+            pedimento=self.pedimento,
+            document_type=self.doc_type,
+            archivo=old_path,
+            size=500,
+            extension="pdf",
+        )
+        mock_st.save_document.return_value = "org_1/documents/24-01-3420-1234567/informe_b5c6d7e8.pdf"
+        mock_st.delete_file.return_value = True
+
+        self._post_file("informe.pdf")
+
+        mock_st.delete_file.assert_called_once_with(old_path)
+
+    @patch("api.record.views.storage_service")
+    def test_different_filename_creates_new_document(self, mock_st):
+        """Archivo con nombre diferente debe crear un Document adicional."""
+        Document.objects.create(
+            organizacion=self.org,
+            pedimento=self.pedimento,
+            document_type=self.doc_type,
+            archivo="org_1/documents/ped/informe_a1b2c3d4.pdf",
+            size=500,
+            extension="pdf",
+        )
+        mock_st.save_document.return_value = "org_1/documents/ped/otro_b5c6d7e8.pdf"
+
+        self._post_file("otro.pdf")
+
+        self.assertEqual(Document.objects.filter(pedimento=self.pedimento).count(), 2)
+        mock_st.delete_file.assert_not_called()
+
+    @patch("api.record.views.storage_service")
+    def test_multiple_files_no_cross_replacement(self, mock_st):
+        """Subir dos archivos distintos en la misma petición crea dos Documents."""
+        mock_st.save_document.side_effect = [
+            "org_1/documents/ped/a_a1b2c3d4.pdf",
+            "org_1/documents/ped/b_a1b2c3d4.pdf",
+        ]
+        archivos = [
+            SimpleUploadedFile("a.pdf", b"contenido a", content_type="application/pdf"),
+            SimpleUploadedFile("b.pdf", b"contenido b", content_type="application/pdf"),
+        ]
+        self.client.post(
+            self.url,
+            {"pedimento_id": str(self.pedimento.id), "files": archivos},
+            format="multipart",
+        )
+        self.assertEqual(Document.objects.filter(pedimento=self.pedimento).count(), 2)
+        mock_st.delete_file.assert_not_called()

api/record/views.py

@@ -273,6 +273,9 @@ class DocumentViewSet(viewsets.ModelViewSet, DocumentosFiltradosMixin):
             if ruta:
                 documento.archivo = ruta
                 documento.save()
+                # si no agrego esto, el proceso no retorna todos los campos necesarios como id, si lo agrega a minIO pero no 
+                # actualiza su status.
+                serializer.instance = documento
             else:
                 documento.delete()
                 raise ValidationError({"archivo": "Error al guardar el archivo"})
@@ -1321,6 +1324,12 @@ class DocumentViewSet(viewsets.ModelViewSet, DocumentosFiltradosMixin):
                         "codigo": "bulk_storage_limit_exceeded"
                     }, status=status.HTTP_400_BAD_REQUEST)
 
+                # Cargar documentos existentes del pedimento para detectar y reemplazar duplicados
+                existing_docs = list(Document.objects.filter(
+                    pedimento_id=pedimento_id,
+                    organizacion=organizacion
+                ))
+
                 # Procesar cada archivo
                 espacio_usado_temp = espacio_inicial
 
@@ -1335,7 +1344,39 @@ class DocumentViewSet(viewsets.ModelViewSet, DocumentosFiltradosMixin):
                         # Obtener extensión del archivo
                         extension = file.name.split('.')[-1].lower() if '.' in file.name else ''
 
-                        # Crear el documento
+                        # Detectar si ya existe un documento con el mismo nombre base + extensión.
+                        # storage_service agrega un sufijo UUID de 8 chars al guardar, hay que ignorarlo.
+                        new_name_base = re.sub(r'_[a-zA-Z0-9]{8}$', '', os.path.splitext(file.name)[0]).lower().strip('_')
+                        existing_doc = None
+                        for doc in existing_docs:
+                            if doc.archivo:
+                                doc_basename = os.path.basename(doc.archivo.name)
+                                doc_base = re.sub(r'_[a-zA-Z0-9]{8}$', '', os.path.splitext(doc_basename)[0]).lower().strip('_')
+                                doc_ext = (doc.extension or '').lower()
+                                if new_name_base == doc_base and extension == doc_ext:
+                                    existing_doc = doc
+                                    break
+
+                        if existing_doc:
+                            # Reemplazar archivo del documento existente
+                            if existing_doc.archivo:
+                                storage_service.delete_file(existing_doc.archivo.name)
+                            ruta = storage_service.save_document(
+                                file=file,
+                                organizacion_id=organizacion.id,
+                                pedimento_app=pedimento.pedimento_app,
+                                metadata={'source': 'bulk_upload_replace'}
+                            )
+                            if ruta:
+                                existing_doc.archivo = ruta
+                                existing_doc.size = file.size
+                                existing_doc.extension = extension
+                                existing_doc.save()
+                            else:
+                                raise Exception(f"Error al guardar archivo: {file.name}")
+                            document = existing_doc
+                        else:
+                            # Crear nuevo documento
                             document = Document.objects.create(
                                 organizacion=organizacion,
                                 pedimento_id=pedimento_id,
@@ -1343,14 +1384,12 @@ class DocumentViewSet(viewsets.ModelViewSet, DocumentosFiltradosMixin):
                                 size=file.size,
                                 extension=extension
                             )
-
                             ruta = storage_service.save_document(
                                 file=file,
                                 organizacion_id=organizacion.id,
                                 pedimento_app=pedimento.pedimento_app,
                                 metadata={'source': 'bulk_upload'}
                             )
-
                             if ruta:
                                 document.archivo = ruta
                                 document.save()

api/reports/views.py

@@ -135,6 +135,33 @@ class ExportDataStageView(APIView):
         else:
             return str(value)
 
+    def get(self, request, *args, **kwargs):
+        """Retorna RFCs distintos de Registro501 para la organización indicada. El parámetro organizacion es obligatorio."""
+        try:
+            Registro501 = apps.get_model('datastage', 'Registro501')
+
+            if not request.user.is_superuser:
+                qs = Registro501.objects.filter(organizacion=request.user.organizacion)
+            else:
+                org_id = request.query_params.get('organizacion')
+                if not org_id:
+                    return Response({'error': 'El parámetro organizacion es obligatorio'}, status=status.HTTP_400_BAD_REQUEST)
+                try:
+                    qs = Registro501.objects.filter(organizacion_id=uuid.UUID(org_id))
+                except (ValueError, AttributeError):
+                    return Response({'error': 'UUID de organización inválido'}, status=status.HTTP_400_BAD_REQUEST)
+
+            rfcs = (
+                qs.exclude(rfc__isnull=True)
+                  .exclude(rfc='')
+                  .values_list('rfc', flat=True)
+                  .distinct()
+                  .order_by('rfc')
+            )
+            return Response({'rfcs': list(rfcs)})
+        except LookupError:
+            return Response({'rfcs': []})
+
     @swagger_auto_schema(request_body=ExportModelSerializer, responses={200: 'Archivo generado (Excel o CSV)'})
     def post(self, request, *args, **kwargs):
         """
@@ -148,6 +175,27 @@ class ExportDataStageView(APIView):
         else:
             return self.handle_simple_export(request)
     
+    def _resolve_org_filter(self, global_filters, user):
+        """
+        Devuelve los global_filters asegurando que siempre haya una organización.
+        - Superuser sin org → error (no mezclar tenants).
+        - No-superuser sin org → se inyecta la org del usuario.
+        Retorna (filters_dict, error_response_or_None).
+        """
+        org_value = (global_filters or {}).get('organizacion', '')
+        if not org_value:
+            if user.is_superuser:
+                return None, Response(
+                    {'error': 'El parámetro organizacion es obligatorio'},
+                    status=status.HTTP_400_BAD_REQUEST
+                )
+            # No-superuser: inyectar su propia org
+            if hasattr(user, 'organizacion') and user.organizacion:
+                filters = dict(global_filters or {})
+                filters['organizacion'] = str(user.organizacion.id)
+                return filters, None
+        return dict(global_filters or {}), None
+
     def handle_simple_export(self, request):
         """Maneja exportación simple de DataStage (un solo modelo)"""
         model_name = request.data.get('model')
@@ -159,6 +207,10 @@ class ExportDataStageView(APIView):
         if not model_name or not fields:
             return Response({'error': 'model and fields are required'}, status=status.HTTP_400_BAD_REQUEST)
 
+        global_filters, err = self._resolve_org_filter(global_filters, request.user)
+        if err:
+            return err
+
         try:
             model = apps.get_model(module, model_name)
             filters = self.apply_global_filters_to_model(global_filters, model, request.user)
@@ -190,18 +242,16 @@ class ExportDataStageView(APIView):
         if not models_data:
             return Response({'error': 'models are required for multiple export'}, status=status.HTTP_400_BAD_REQUEST)
 
+        global_filters, err = self._resolve_org_filter(global_filters, request.user)
+        if err:
+            return err
+
         related_keys = self.get_related_keys_from_filters(global_filters, models_data, request.user)
         
         if export_type == 'excel':
-            # Siempre usar el método particionado inteligente para Excel
             return self.export_datastage_multiple_partitioned_excel_agrupados(request, models_data, global_filters, related_keys)
         else:
-            # Para CSV, podemos mantener la lógica actual o mejorarla
+            return self.export_datastage_multiple_to_csv_combined(request, models_data, global_filters, related_keys)
-            total_estimated_records = self.estimate_total_records(models_data, global_filters, related_keys, request.user)
-            if total_estimated_records > self.MAX_RECORDS_PER_FILE:
-                return self.export_datastage_multiple_partitioned_csv(request, models_data, global_filters, related_keys)
-            else:
-                return self.export_datastage_multiple_to_csv(request, models_data, global_filters, related_keys)
 
     def estimate_total_records(self, models_data, global_filters, related_keys, user):
         """Estima el total de registros para todos los modelos"""
@@ -282,17 +332,11 @@ class ExportDataStageView(APIView):
     def export_datastage_multiple_partitioned_excel_agrupados(self, request, models_data, global_filters, related_keys):
         """Exporta múltiples modelos de DataStage agrupados en la misma hoja de Excel, con particionado por límite de registros"""
         try:
-            zip_buffer = io.BytesIO()
-
-            # 🔥 PRECARGAR ORGANIZACIONES para mapeo rápido
             from api.organization.models import Organizacion
-            organizaciones = Organizacion.objects.all()
+            org_mapping = {str(org.id): org.nombre for org in Organizacion.objects.all()}
-            org_mapping = {str(org.id): org.nombre for org in organizaciones}
 
-            with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zip_file:
+            # 1. Recopilar todos los datos FUERA del contexto ZIP
-
+            all_models_data = {}
-                # 1. Recopilar todos los datos de cada modelo
-                all_models_data = {}  # Ahora será una lista por clave
             model_field_mappings = {}
 
             for model_data in models_data:
@@ -302,8 +346,6 @@ class ExportDataStageView(APIView):
                 if not model_name or not fields:
                     continue
 
-                    # Normalizar nombres de campo entrantes: si se pasó "Organizacion"
-                    # (cualquier capitalización), usar el campo real de la BD `organizacion_id`.
                 normalized_fields = []
                 for f in fields:
                     try:
@@ -320,13 +362,11 @@ class ExportDataStageView(APIView):
 
                 fields = normalized_fields
 
-                    # Asegurar que tenemos los campos de relación
                 required_fields = ['seccion_aduanera', 'patente', 'pedimento']
                 for field in required_fields:
                     if field not in fields:
                         fields.append(field)
 
-                    # 🔥 Añadir organizacion_id a los campos si no está y existe en el modelo
                 if 'organizacion_id' not in fields and 'organizacion_id' in [f.name for f in apps.get_model('datastage', model_name)._meta.get_fields()]:
                     fields.append('organizacion_id')
 
@@ -339,233 +379,182 @@ class ExportDataStageView(APIView):
                     else:
                         queryset = model.objects.none()
 
-                        total_records = queryset.count()
+                    if queryset.count() == 0:
-
-                        if total_records == 0:
                         continue
 
-                        # Determinar campos de relación disponibles en este modelo
+                    relation_fields = [fn for fn in ['seccion_aduanera', 'patente', 'pedimento'] if fn in fields]
-                        relation_fields = []
-                        for field_name in ['seccion_aduanera', 'patente', 'pedimento']:
-                            if field_name in fields:
-                                relation_fields.append(field_name)
-
                     if not relation_fields:
-                            # Si no hay campos de relación, usar un identificador único
                         relation_fields = ['datastage_id'] if 'datastage_id' in fields else [fields[0]]
 
-                        # Guardar mapeo de campos para este modelo
                     if model_name not in model_field_mappings:
                         model_field_mappings[model_name] = fields
 
-                        # Procesar cada registro
                     for record in queryset:
-                            # Crear clave de relación
+                        key_parts = [str(record[rf]) for rf in relation_fields if rf in record and record[rf] is not None]
-                            key_parts = []
-                            for rel_field in relation_fields:
-                                if rel_field in record and record[rel_field] is not None:
-                                    key_parts.append(str(record[rel_field]))
-
                         if not key_parts:
-                                # Si no hay campos de relación, usar un hash del registro
                             import hashlib
-                                record_str = str(sorted(record.items()))
+                            key = hashlib.md5(str(sorted(record.items())).encode()).hexdigest()[:10]
-                                key = hashlib.md5(record_str.encode()).hexdigest()[:10]
                         else:
                             key = "_".join(key_parts)
 
-                            # 🔥 PROCESAR CAMPO organizacion_id para convertirlo a nombre
                         processed_record = {}
                         for field_name, value in record.items():
-                                # Convertir organizacion_id a nombre
                             if field_name == 'organizacion_id' and value:
                                 org_id_str = str(value)
-                                    # Usar el nombre de la organización si está en el mapeo
                                 if org_id_str in org_mapping:
                                     processed_value = org_mapping[org_id_str]
                                 else:
-                                        # Si no se encuentra, intentar obtener de la base de datos
                                     try:
                                         org = Organizacion.objects.filter(id=value).first()
-                                            processed_value = org.nombre if org else str(value)
+                                        processed_value = org.nombre if org else org_id_str
-                                            # Actualizar mapeo para futuras referencias
                                         org_mapping[org_id_str] = processed_value
-                                        except:
+                                    except Exception:
-                                            processed_value = str(value)
+                                        processed_value = org_id_str
                             else:
                                 processed_value = value
 
-                                # Agregar prefijo del modelo a los campos para evitar colisiones
                             if field_name in relation_fields:
                                 prefixed_field_name = field_name
                             else:
                                 prefixed_field_name = f"{model_name}_{field_name}"
 
-                                # 🔥 RENOMBRAR organizacion_id a organizacion_nombre
                             if field_name == 'organizacion_id':
                                 prefixed_field_name = prefixed_field_name.replace('organizacion_id', 'organizacion_nombre')
 
                             processed_record[prefixed_field_name] = self.safe_excel_value(processed_value)
 
-                            # 🔥 CORRECIÓN: Ahora almacenamos una LISTA de registros por clave
                         if key not in all_models_data:
-                                all_models_data[key] = {
+                            all_models_data[key] = {'relation_fields': {}, 'model_records': {}}
-                                    'relation_fields': {},  # Campos de relación compartidos
-                                    'model_records': {}     # Diccionario de listas por modelo
-                                }
 
-                            # Guardar campos de relación (solo una vez, ya que son los mismos)
                         for rel_field in relation_fields:
                             if rel_field in record:
                                 all_models_data[key]['relation_fields'][rel_field] = record[rel_field]
 
-                            # 🔥 GUARDAR COMO LISTA: Crear lista si no existe
                         if model_name not in all_models_data[key]['model_records']:
                             all_models_data[key]['model_records'][model_name] = []
 
-                            # Agregar este registro a la lista del modelo
                         all_models_data[key]['model_records'][model_name].append(processed_record)
 
                 except LookupError:
                     continue
 
-                # Si no hay datos, retornar error
+            # 2. Sin datos → Excel vacío (no JSON 404 que rompe la descarga en el frontend)
             if not all_models_data:
-                    return Response({'error': 'No se encontraron datos para exportar'}, status=status.HTTP_404_NOT_FOUND)
+                wb = openpyxl.Workbook()
+                ws = wb.active
+                ws.title = "Sin datos"
+                ws.append(["No se encontraron datos para los filtros especificados"])
+                output = io.BytesIO()
+                wb.save(output)
+                output.seek(0)
+                resp = HttpResponse(
+                    output.read(),
+                    content_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
+                )
+                resp['Content-Disposition'] = 'attachment; filename="datastage_sin_datos.xlsx"'
+                return resp
 
-                # 2. Crear estructura de filas combinadas
+            # 3. Construir filas combinadas — repetir el último registro en lugar de dejar vacíos
-                # Ahora necesitamos expandir las filas cuando hay múltiples registros con la misma clave
             combined_rows = []
-
             for key, data in all_models_data.items():
-                    relation_fields = data['relation_fields']
+                relation_fields_data = data['relation_fields']
                 model_records = data['model_records']
 
-                    # 🔥 NUEVO: Calcular cuántas filas necesitamos para esta clave
+                max_records_per_key = max((len(recs) for recs in model_records.values()), default=1)
-                    # Encontrar el modelo con más registros para esta clave
-                    max_records_per_key = 1
-                    for model_name, records in model_records.items():
-                        if len(records) > max_records_per_key:
-                            max_records_per_key = len(records)
 
-                    # 🔗 CREAR UNA FILA POR CADA COMBINACIÓN
                 for i in range(max_records_per_key):
                     row_data = {}
 
-                        # Campos de relación (mismos para todas las filas con esta clave)
+                    for rel_field, rel_value in relation_fields_data.items():
-                        for rel_field, rel_value in relation_fields.items():
                         row_data[rel_field] = self.safe_excel_value(rel_value)
 
-                        # Datos de cada modelo
                     for model_name, records in model_records.items():
-                            # Si hay un registro en esta posición i
+                        # Usar posición i o el último registro disponible
-                            if i < len(records):
+                        record = records[i] if i < len(records) else records[-1]
-                                record = records[i]
                         for field_name, value in record.items():
                             row_data[field_name] = value
-                            else:
-                                # Si no hay más registros para este modelo, poner campos vacíos
-                                for field_name in model_field_mappings.get(model_name, []):
-                                    if field_name in ['seccion_aduanera', 'patente', 'pedimento', 'organizacion_id']:
-                                        # Los campos de relación ya están llenados o transformados
-                                        continue
-                                    prefixed_field_name = f"{model_name}_{field_name}"
-                                    # 🔥 RENOMBRAR organizacion_id a organizacion_nombre
-                                    if field_name == 'organizacion_id':
-                                        prefixed_field_name = prefixed_field_name.replace('organizacion_id', 'organizacion_nombre')
-                                    row_data[prefixed_field_name] = ''
 
                     combined_rows.append(row_data)
 
-                # 3. Determinar todos los campos únicos para los encabezados
+            # 4. Encabezados ordenados
             all_fields_set = set()
-
-                # Campos de relación primero
-                common_relation_fields = ['seccion_aduanera', 'patente', 'pedimento']
-
-                # Agregar todos los campos de todas las filas
             for row in combined_rows:
                 all_fields_set.update(row.keys())
 
-                # Ordenar campos: relación primero, luego alfabéticamente
             all_fields = []
-                for rel_field in common_relation_fields:
+            for rel_field in ['seccion_aduanera', 'patente', 'pedimento']:
                 if rel_field in all_fields_set:
                     all_fields.append(rel_field)
-                        all_fields_set.remove(rel_field)
+                    all_fields_set.discard(rel_field)
 
-                # 🔥 Mover organizacion_nombre cerca de los campos de relación
+            org_fields = sorted(f for f in all_fields_set if 'organizacion' in f.lower())
-                org_fields = [f for f in all_fields_set if 'organizacion' in f.lower()]
+            for org_field in org_fields:
-                for org_field in sorted(org_fields):
                 all_fields.append(org_field)
-                    all_fields_set.remove(org_field)
+                all_fields_set.discard(org_field)
 
-                # Agregar el resto de campos ordenados alfabéticamente
             all_fields.extend(sorted(all_fields_set))
 
-                total_records = len(combined_rows)
+            # 5. Filas de título y fecha de generación
+            now_str = datetime.datetime.now().strftime('%d/%m/%Y %H:%M:%S')
+            title_row = ["Reporte Datastage"]
+            date_row = [f"Generado: {now_str}"]
 
-                # 4. Manejar particionado
+            def _write_sheet(ws, sheet_name, page_rows):
-                from django.core.paginator import Paginator
+                ws.title = sheet_name[:31]
-                paginator = Paginator(combined_rows, self.MAX_RECORDS_PER_FILE)
+                ws.append(title_row)
-
+                ws.append(date_row)
-                for page_num in paginator.page_range:
+                ws.append([])
-                    page = paginator.page(page_num)
+                ws.append(all_fields)
-
+                for row_data in page_rows:
-                    # Crear nuevo workbook para cada partición
+                    ws.append([row_data.get(field, '') for field in all_fields])
-                    current_wb = openpyxl.Workbook()
+                for column in ws.columns:
-                    current_ws = current_wb.active
-
-                    # Nombre de hoja limitado a 31 caracteres
-                    sheet_name = f"Datastage_p{page_num}"
-                    if len(sheet_name) > 31:
-                        sheet_name = sheet_name[:31]
-                    current_ws.title = sheet_name
-
-                    # Escribir encabezados
-                    current_ws.append(all_fields)
-
-                    # Escribir datos de esta página
-                    for row_data in page.object_list:
-                        row_values = [row_data.get(field, '') for field in all_fields]
-                        current_ws.append(row_values)
-
-                    # Autoajustar anchos de columna
-                    for column in current_ws.columns:
                     max_length = 0
-                        column_letter = column[0].column_letter
+                    col_letter = column[0].column_letter
-
                     for cell in column:
                         try:
                             if len(str(cell.value)) > max_length:
                                 max_length = len(str(cell.value))
-                            except:
+                        except Exception:
                             pass
+                    ws.column_dimensions[col_letter].width = min(max_length + 2, 50)
 
-                        adjusted_width = min(max_length + 2, 50)
+            # 6. Excel directo si cabe en un archivo; ZIP solo si se necesita particionar
-                        current_ws.column_dimensions[column_letter].width = adjusted_width
+            from django.core.paginator import Paginator
+            paginator = Paginator(combined_rows, self.MAX_RECORDS_PER_FILE)
 
-                    # Guardar archivo en ZIP
+            if paginator.num_pages == 1:
+                wb = openpyxl.Workbook()
+                _write_sheet(wb.active, "Datastage", paginator.page(1).object_list)
+                output = io.BytesIO()
+                wb.save(output)
+                output.seek(0)
+                resp = HttpResponse(
+                    output.read(),
+                    content_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
+                )
+                resp['Content-Disposition'] = 'attachment; filename="datastage_reporte.xlsx"'
+                return resp
+
+            zip_buffer = io.BytesIO()
+            with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zip_file:
+                for page_num in paginator.page_range:
+                    page = paginator.page(page_num)
+                    current_wb = openpyxl.Workbook()
+                    _write_sheet(current_wb.active, f"Datastage_p{page_num}", page.object_list)
                     part_buffer = io.BytesIO()
                     current_wb.save(part_buffer)
                     part_buffer.seek(0)
                     zip_file.writestr(f"datastage_part{page_num}.xlsx", part_buffer.getvalue())
 
-                    # Información de depuración
-                    print(f"Creada partición {page_num} con {len(page.object_list)} registros combinados")
-                    print(f"Total de claves únicas: {len(all_models_data)}")
-                    print(f"Total de filas expandidas: {total_records}")
-
             zip_buffer.seek(0)
-
+            resp = HttpResponse(zip_buffer.read(), content_type='application/zip')
-            response = HttpResponse(zip_buffer.read(), content_type='application/zip')
+            resp['Content-Disposition'] = 'attachment; filename="datastage_combinado.zip"'
-            response['Content-Disposition'] = 'attachment; filename="datastage_combinado.zip"'
+            return resp
-            return response
 
         except Exception as e:
             import traceback
-            error_details = traceback.format_exc()
+            import logging
-            print(f"Error en exportación: {error_details}")
+            logging.getLogger(__name__).error("Error en exportación combinada: %s", traceback.format_exc())
             return Response({'error': f'Error en exportación combinada: {str(e)}'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
 
 
@@ -782,10 +771,6 @@ class ExportDataStageView(APIView):
                     part_buffer.seek(0)
                     zip_file.writestr(f"datastage_part{page_num}.xlsx", part_buffer.getvalue())
 
-                    # Información de depuración
-                    print(f"Creada partición {page_num} con {len(page.object_list)} registros combinados")
-                    print(f"Total de claves únicas: {len(all_models_data)}")
-                    print(f"Total de filas expandidas: {total_records}")
 
             zip_buffer.seek(0)
 
@@ -795,12 +780,11 @@ class ExportDataStageView(APIView):
 
         except Exception as e:
             import traceback
-            error_details = traceback.format_exc()
+            import logging
-            print(f"Error en exportación: {error_details}")
+            logging.getLogger(__name__).error("Error en exportación combinada: %s", traceback.format_exc())
             return Response({'error': f'Error en exportación combinada: {str(e)}'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
 
 
-
     def export_datastage_multiple_partitioned_excel_test_2(self, request, models_data, global_filters, related_keys):
         """Exporta múltiples modelos de DataStage agrupados en la misma hoja de Excel, con particionado por límite de registros"""
         try:
@@ -1009,8 +993,8 @@ class ExportDataStageView(APIView):
 
         except Exception as e:
             import traceback
-            error_details = traceback.format_exc()
+            import logging
-            print(f"Error en exportación: {error_details}")
+            logging.getLogger(__name__).error("Error en exportación combinada: %s", traceback.format_exc())
             return Response({'error': f'Error en exportación combinada: {str(e)}'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
 
 
@@ -1126,8 +1110,6 @@ class ExportDataStageView(APIView):
                     part_buffer.seek(0)
                     zip_file.writestr(f"datastage_combinado_part{page_num}.xlsx", part_buffer.getvalue())
 
-                    # Información de depuración (opcional)
-                    print(f"Creada partición {page_num} con {len(page.object_list)} registros")
 
             zip_buffer.seek(0)
 
@@ -1137,8 +1119,8 @@ class ExportDataStageView(APIView):
 
         except Exception as e:
             import traceback
-            error_details = traceback.format_exc()
+            import logging
-            print(f"Error en exportación: {error_details}")
+            logging.getLogger(__name__).error("Error en exportación combinada: %s", traceback.format_exc())
             return Response({'error': f'Error en exportación combinada: {str(e)}'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
 
     def export_datastage_multiple_partitioned_excel(self, request, models_data, global_filters, related_keys):
@@ -1265,6 +1247,144 @@ class ExportDataStageView(APIView):
         except Exception as e:
             return Response({'error': f'Error en exportación particionada: {str(e)}'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
         
+    def export_datastage_multiple_to_csv_combined(self, request, models_data, global_filters, related_keys):
+        """Exporta múltiples modelos combinados en un único CSV plano (misma lógica de agrupación que el Excel)."""
+        import hashlib
+        import logging
+        import traceback
+        logger = logging.getLogger(__name__)
+        try:
+            from api.organization.models import Organizacion
+            org_mapping = {str(org.id): org.nombre for org in Organizacion.objects.all()}
+
+            all_models_data = {}
+            model_field_mappings = {}
+
+            for model_data in models_data:
+                model_name = model_data.get('model')
+                fields = model_data.get('fields', [])
+                if not model_name or not fields:
+                    continue
+
+                normalized_fields = []
+                for f in fields:
+                    key = f.strip() if isinstance(f, str) else f
+                    if isinstance(key, str) and key.lower() == 'organizacion':
+                        if 'organizacion_id' not in normalized_fields:
+                            normalized_fields.append('organizacion_id')
+                    else:
+                        if key not in normalized_fields:
+                            normalized_fields.append(key)
+                fields = normalized_fields
+
+                for req_field in ['seccion_aduanera', 'patente', 'pedimento']:
+                    if req_field not in fields:
+                        fields.append(req_field)
+
+                try:
+                    model = apps.get_model('datastage', model_name)
+                    model_field_names = [f.name for f in model._meta.get_fields() if hasattr(f, 'name')]
+                    if 'organizacion_id' not in fields and 'organizacion_id' in model_field_names:
+                        fields.append('organizacion_id')
+
+                    filters = self.apply_related_filters(global_filters, model, related_keys, request.user)
+                    queryset = model.objects.filter(**filters).values(*fields) if filters else model.objects.none()
+                    if queryset.count() == 0:
+                        continue
+
+                    relation_fields = [fn for fn in ['seccion_aduanera', 'patente', 'pedimento'] if fn in fields]
+                    if not relation_fields:
+                        relation_fields = ['datastage_id'] if 'datastage_id' in fields else [fields[0]]
+
+                    if model_name not in model_field_mappings:
+                        model_field_mappings[model_name] = fields
+
+                    for record in queryset:
+                        key_parts = [str(record[rf]) for rf in relation_fields if rf in record and record[rf] is not None]
+                        key = "_".join(key_parts) if key_parts else hashlib.md5(str(sorted(record.items())).encode()).hexdigest()[:10]
+
+                        processed_record = {}
+                        for field_name, value in record.items():
+                            if field_name == 'organizacion_id' and value:
+                                org_id_str = str(value)
+                                processed_value = org_mapping.get(org_id_str, org_id_str)
+                            else:
+                                processed_value = value
+
+                            if field_name in relation_fields:
+                                prefixed = field_name
+                            else:
+                                prefixed = f"{model_name}_{field_name}"
+                            if field_name == 'organizacion_id':
+                                prefixed = prefixed.replace('organizacion_id', 'organizacion_nombre')
+                            processed_record[prefixed] = self.safe_excel_value(processed_value)
+
+                        if key not in all_models_data:
+                            all_models_data[key] = {'relation_fields': {}, 'model_records': {}}
+                        for rel_field in relation_fields:
+                            if rel_field in record:
+                                all_models_data[key]['relation_fields'][rel_field] = record[rel_field]
+                        if model_name not in all_models_data[key]['model_records']:
+                            all_models_data[key]['model_records'][model_name] = []
+                        all_models_data[key]['model_records'][model_name].append(processed_record)
+
+                except LookupError:
+                    continue
+
+            # Sin datos → CSV con mensaje, no error HTTP
+            if not all_models_data:
+                buf = io.StringIO()
+                csv.writer(buf).writerow(['No se encontraron datos para los filtros especificados'])
+                resp = HttpResponse(buf.getvalue(), content_type='text/csv; charset=utf-8')
+                resp['Content-Disposition'] = 'attachment; filename="datastage_sin_datos.csv"'
+                return resp
+
+            # Construir filas planas
+            combined_rows = []
+            for key, data in all_models_data.items():
+                relation_fields_data = data['relation_fields']
+                model_records = data['model_records']
+                max_records = max((len(recs) for recs in model_records.values()), default=1)
+                for i in range(max_records):
+                    row_data = {}
+                    for rel_field, rel_value in relation_fields_data.items():
+                        row_data[rel_field] = self.safe_excel_value(rel_value)
+                    for mn, records in model_records.items():
+                        record = records[i] if i < len(records) else records[-1]
+                        for field_name, value in record.items():
+                            row_data[field_name] = value
+                    combined_rows.append(row_data)
+
+            # Encabezados: campos de relación primero, luego org, luego el resto
+            all_fields_set = set()
+            for row in combined_rows:
+                all_fields_set.update(row.keys())
+
+            all_fields = []
+            for rel_field in ['seccion_aduanera', 'patente', 'pedimento']:
+                if rel_field in all_fields_set:
+                    all_fields.append(rel_field)
+                    all_fields_set.discard(rel_field)
+            org_fields = sorted(f for f in all_fields_set if 'organizacion' in f.lower())
+            for org_field in org_fields:
+                all_fields.append(org_field)
+                all_fields_set.discard(org_field)
+            all_fields.extend(sorted(all_fields_set))
+
+            buf = io.StringIO()
+            writer = csv.writer(buf)
+            writer.writerow(all_fields)
+            for row_data in combined_rows:
+                writer.writerow([row_data.get(field, '') for field in all_fields])
+
+            resp = HttpResponse(buf.getvalue(), content_type='text/csv; charset=utf-8')
+            resp['Content-Disposition'] = 'attachment; filename="datastage_reporte.csv"'
+            return resp
+
+        except Exception as e:
+            logger.error("Error en exportación CSV combinada: %s", traceback.format_exc())
+            return Response({'error': f'Error en exportación CSV combinada: {str(e)}'}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
+
     def export_datastage_multiple_to_csv(self, request, models_data, global_filters, related_keys):
         """Exporta múltiples modelos de DataStage a múltiples archivos CSV en ZIP"""
         zip_buffer = io.BytesIO()
@@ -1472,8 +1592,13 @@ class ExportDataStageView(APIView):
         
     def get_related_keys_from_filters(self, global_filters, models_data, user):
         """
-        Obtiene patentes, pedimentos y datastages que cumplen EXACTAMENTE con TODOS los filtros globales
+        Construye el conjunto de (patente, pedimento, datastage_id) que servirá como
-        VERSIÓN SIMPLIFICADA - Usa la MISMA lógica que apply_global_filters_to_model
+        llave de cruce entre modelos.
+
+        Regla clave: si el filtro RFC está activo, solo los modelos que tienen el campo
+        'rfc' pueden contribuir a related_keys. Los modelos sin 'rfc' (ej. 505, 506)
+        no se usan como semilla — solo se filtrarán más tarde usando las claves ya
+        construidas, evitando que contaminen el resultado con pedimentos de otros RFC.
         """
         related_keys = {
             'patentes': set(),
@@ -1481,40 +1606,34 @@ class ExportDataStageView(APIView):
             'datastage_ids': set()
         }
 
-        # Si no hay filtros, retornar vacío
+        # Sin filtros significativos → sin cruce
         if not any(v for v in global_filters.values() if v not in [None, '']):
             return {}
 
+        rfc_filter_active = bool(global_filters.get('rfc'))
+        date_filter_active = bool(global_filters.get('fecha_pago_desde') or global_filters.get('fecha_pago_hasta'))
         all_records_with_filters = []
 
         for model_data in models_data:
             model_name = model_data.get('model')
-            
             try:
                 model = apps.get_model('datastage', model_name)
+                model_field_names = {f.name for f in model._meta.get_fields() if hasattr(f, 'name')}
+
+                # Un modelo puede ser semilla de related_keys SOLO si tiene campos
+                # para aplicar TODOS los filtros activos. Un modelo sin 'rfc' no puede
+                # ser semilla cuando hay filtro de RFC (contaminaría con pedimentos de
+                # otros RFCs). Igual para fecha_pago_real cuando hay filtro de fechas.
+                if rfc_filter_active and 'rfc' not in model_field_names:
+                    continue
+                if date_filter_active and 'fecha_pago_real' not in model_field_names:
+                    continue
 
-                # ¡USAR LA MISMA FUNCIÓN QUE EN MODO SINGULAR!
                 filters = self.apply_global_filters_to_model(global_filters, model, user)
+                if not filters:
+                    continue
 
-                if filters:
+                records = model.objects.filter(**filters).values('patente', 'pedimento', 'datastage_id')
-                    # EJECUTAR CONSULTA - IDÉNTICO A MODO SINGULAR
-                    queryset = model.objects.filter(**filters)
-                    total = queryset.count()
-                    
-                    # VERIFICACIÓN ESPECIAL PARA RFC
-                    if 'rfc' in filters:
-                        rfc_value = filters['rfc']
-                        # Doble verificación: contar registros con ese RFC exacto
-                        rfc_exact_count = queryset.filter(rfc=rfc_value).count()
-                        
-                        if rfc_exact_count != total:
-                            try:
-                                other_rfcs = queryset.exclude(rfc=rfc_value).values_list('rfc', flat=True).distinct()[:5]
-                            except:
-                                pass
-                    
-                    # Obtener registros
-                    records = queryset.values('patente', 'pedimento', 'datastage_id')
                 all_records_with_filters.extend(list(records))
 
             except LookupError:
@@ -1585,9 +1704,17 @@ class ExportDataStageView(APIView):
         filters = {}
         model_fields = [f.name for f in model._meta.get_fields()]
         
-        # 1. Organización
+        # 1. Organización — convertir a UUID igual que apply_global_filters_to_model
         if 'organizacion' in model_fields and global_filters.get('organizacion'):
-            filters['organizacion'] = global_filters['organizacion']
+            org_value = global_filters['organizacion']
+            try:
+                field = model._meta.get_field('organizacion')
+                if hasattr(field, 'related_model'):
+                    filters['organizacion_id'] = uuid.UUID(org_value)
+                else:
+                    filters['organizacion'] = org_value
+            except Exception:
+                filters['organizacion_id'] = org_value
         
         # 2. RFC (¡ESTO ES LO QUE FALTA!)
         if 'rfc' in model_fields and global_filters.get('rfc'):

api/tasks/views.py

@@ -57,42 +57,57 @@ from celery.result import AsyncResult
 
 
 class TaskStatusView(APIView):
-    """
-    Vista para consultar el estado de tareas de Celery.
-    """
     permission_classes = [IsAuthenticated]
 
     def get(self, request, task_id):
         """
-        Consulta el estado de una tarea de Celery.
+        Consulta el estado de una tarea Celery.
 
-        Returns:
+        Estados posibles:
-            - PENDING: La tarea está esperando ser procesada
+          PENDING  — en cola, aún no inició
-            - STARTED: La tarea ha sido iniciada
+          STARTED  — worker la tomó y está ejecutando
-            - SUCCESS: La tarea se completó exitosamente
+          SUCCESS  — terminó correctamente, `result` contiene el resumen
-            - FAILURE: La tarea falló
+          FAILURE  — lanzó una excepción no capturada, `error` describe el problema
-            - RETRY: La tarea está reintentando
+          RETRY    — el worker la está reintentando
         """
         try:
             task_result = AsyncResult(task_id)
+            state = task_result.state
 
             response_data = {
                 'task_id': task_id,
-                'status': task_result.state,
+                'status': state,
                 'ready': task_result.ready(),
                 'successful': task_result.successful() if task_result.ready() else None,
             }
 
-            if task_result.ready() and task_result.successful():
+            if state == 'SUCCESS':
-                try:
+                result = task_result.result
-                    response_data['result'] = task_result.result
+                response_data['result'] = result
-                except Exception:
-                    pass
 
-            if task_result.state == 'FAILURE':
+                # Resumen legible cuando es auditoría masiva de organización
+                if isinstance(result, dict) and 'total_pedimentos' in result:
+                    total = result.get('total_pedimentos', 0)
+                    completados = result.get('completados', 0)
+                    con_pendientes = result.get('con_pendientes', 0)
+                    con_errores = result.get('con_errores', 0)
+
+                    if con_pendientes == 0 and con_errores == 0:
+                        mensaje = f'Auditoría completa — {completados}/{total} pedimentos sin pendientes'
+                    else:
+                        partes = []
+                        if con_pendientes:
+                            partes.append(f'{con_pendientes} con documentos pendientes')
+                        if con_errores:
+                            partes.append(f'{con_errores} con error')
+                        mensaje = f'{completados}/{total} pedimentos completos — {", ".join(partes)}'
+
+                    response_data['mensaje'] = mensaje
+
+            elif state == 'FAILURE':
                 response_data['error'] = str(task_result.info)
 
-            if task_result.state == 'STARTED':
+            elif state == 'STARTED':
                 response_data['info'] = str(task_result.info) if task_result.info else None
 
             return Response(response_data, status=status.HTTP_200_OK)

api/vucem/views.py

@@ -84,7 +84,7 @@ class VucemView(viewsets.ModelViewSet):
         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)
+            queryset = queryset.filter(organizacion=self.request.user.organizacion, usuario__in=self.request.user.rfc.all())
         else:
             queryset = queryset.filter(organizacion=self.request.user.organizacion)
 

config/celery.py

@@ -1,8 +1,11 @@
 import os
 from celery import Celery
+from datetime import timedelta
 
 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')
 
 app = Celery('config')
 app.config_from_object('django.conf:settings', namespace='CELERY')
+# corroborar que las tareas esten programadas, se cambio el horario a hora denver
+# print("Beat schedule cargado:", app.conf.beat_schedule)
 app.autodiscover_tasks()

config/settings.py

@@ -30,8 +30,14 @@ from celery.schedules import crontab
 from config.stg.storage import *
 
 CELERY_BEAT_SCHEDULE = {
-
+    'process_all_organizations': {
-
+        'task': 'api.customs.tasks.microservice_v2.process_all_organizations',
+        'schedule': crontab(hour=7, minute=1),  # analizar si se requiere otra en un futuro
+    },
+    # 'process_all_organizations': {
+    #     'task': 'api.customs.tasks.microservice_v2.process_all_organizations',
+    #     'schedule': crontab(hour=11, minute=39),  # analizar si se requiere otra en un futuro
+    # },
 }
 
 # Cargar variables de entorno desde un archivo .env
@@ -305,7 +311,8 @@ DEFAULT_FROM_EMAIL = EMAIL_HOST_USER
 # Configuración Celery
 CELERY_BROKER_URL = os.getenv('CELERY_BROKER_URL', 'redis://redis:6379/0')
 CELERY_RESULT_BACKEND = os.getenv('CELERY_RESULT_BACKEND', 'redis://redis:6379/0')
-CELERY_TIMEZONE = 'America/Mexico_City'
+# CELERY_TIMEZONE = 'America/Mexico_City'
+CELERY_TIMEZONE = 'America/Denver'
 
 # Configuración para procesamiento asíncrono nativo de Django
 ASGI_APPLICATION = 'config.asgi.application'

mixins/filtrado_organizacion.py

@@ -62,8 +62,8 @@ class OrganizacionFiltradaMixin:
                 return model.objects.filter(**filtros_base)
         
         # if hasattr(model, self.campo_contribuyente):
-        if self.request.user.is_authenticated and 'Importador' in grupos :
+        if self.request.user.is_authenticated and 'Importador' in grupos:
-            filtros_base[f"{self.campo_contribuyente}__rfc"] = self.request.user.rfc.rfc
+            filtros_base[f"{self.campo_contribuyente}__in"] = self.request.user.rfc.all()
             return model.objects.filter(**filtros_base)
 
         # Si no entra en los roles válidos
@@ -98,7 +98,7 @@ class DocumentosFiltradosMixin:
         
         if hasattr(model, self.campo_contribuyente):
             if self.request.user.is_authenticated and 'Importador' in grupos and getattr(self.request.user, 'is_importador', False):
-                filtros_base[f"{self.campo_contribuyente}__contribuyente"] = self.request.user.rfc
+                filtros_base[f"{self.campo_contribuyente}__contribuyente__in"] = self.request.user.rfc.all()
                 return model.objects.filter(**filtros_base)
 
         # Si no entra en los roles válidos
@@ -133,8 +133,8 @@ class ProcesosPorOrganizacionMixin:
                 return model.objects.filter(**filtros_base)
         
         if hasattr(model, self.campo_pedimento):
-            if self.request.user.is_authenticated and'Importador' in grupos and getattr(self.request.user, 'is_importador', False):
+            if self.request.user.is_authenticated and 'Importador' in grupos and getattr(self.request.user, 'is_importador', False):
-                filtros_base[f"{self.campo_pedimento}__contribuyente"] = self.request.user.rfc
+                filtros_base[f"{self.campo_pedimento}__contribuyente__in"] = self.request.user.rfc.all()
                 return model.objects.filter(**filtros_base)
 
         # Si no entra en los roles válidos