feature/implementacion de gestor de informacion y archivos minIO

api/datastage/models.py

@@ -3,7 +3,8 @@ from django.db import models
 # Create your models here.
 class DataStage(models.Model):
     organizacion = models.ForeignKey('organization.Organizacion', on_delete=models.CASCADE, related_name='datastages', null=True, blank=True)
-    archivo = models.FileField(upload_to='datastages/', blank=False, null=False)
+    # archivo = models.FileField(upload_to='datastages/', blank=False, null=False)
+    archivo = models.CharField(max_length=500, blank=True, null=True)
     contribuyente = models.CharField(max_length=100, blank=False, null=False)
     procesado = models.BooleanField(default=False)
 

api/datastage/serializer.py

@@ -1,12 +1,86 @@
+from api.utils.storage_service import storage_service
 from rest_framework import serializers
 
 from .models import DataStage
 from api.organization.models import Organizacion
 
 class DataStageSerializer(serializers.ModelSerializer):
+    archivo = serializers.FileField(write_only=True, required=False, allow_null=True)
+    download_url = serializers.SerializerMethodField(read_only=True)
+
     organizacion = serializers.PrimaryKeyRelatedField(required=False, allow_null=True, queryset=Organizacion.objects.all())
 
     class Meta:
         model = DataStage
         fields = '__all__'
-        read_only_fields = ('id', 'created_at', 'updated_at')
+        read_only_fields = ('id', 'created_at', 'updated_at')
+        # extra_kwargs = {'archivo': {'read_only': True},}
+
+    def get_download_url(self, obj):
+        """Retorna URL de descarga según dónde esté el archivo"""
+        if not obj.archivo:
+            return None
+        
+        if storage_service.is_minio_path(obj.archivo):
+            return storage_service.get_file_url(obj.archivo)
+        else:
+            request = self.context.get('request')
+            if request:
+                return request.build_absolute_uri(
+                    f"/api/v1/datastage/datastages/{obj.id}/download-datastage/"
+                )
+            return f"/api/v1/datastage/datastages/{obj.id}/download-datastage/"
+
+    def create(self, validated_data):
+        """Override para manejar la subida del archivo a MinIO"""
+        archivo_file = validated_data.pop('archivo', None)
+        organizacion = validated_data.get('organizacion')
+        datastage = super().create(validated_data)
+        print(f"ENDPOINT DE CREATE >>>>")
+        # guardarlo en MinIO
+        if archivo_file:
+            ruta = storage_service.save_datastage(
+                file=archivo_file,
+                organizacion_id=organizacion.id if organizacion else datastage.organizacion.id,
+                metadata={
+                    'datastage_id': str(datastage.id),
+                    'nombre': datastage.nombre if hasattr(datastage, 'nombre') else ''
+                }
+            )
+            
+            if ruta:
+                datastage.archivo = ruta
+                datastage.save()
+            else:
+                # eliminar el registro creado
+                datastage.delete()
+                raise serializers.ValidationError({"archivo": "Error al guardar el archivo en el almacenamiento"})
+        
+        return datastage
+    
+    def update(self, instance, validated_data):
+        """Override para manejar actualización de archivo"""
+        archivo_file = validated_data.pop('archivo', None)
+        organizacion = validated_data.get('organizacion', instance.organizacion)
+        instance = super().update(instance, validated_data)
+        
+        # Si hay nuevo archivo, reemplazarlo
+        if archivo_file:
+            if instance.archivo:
+                storage_service.delete_file(instance.archivo)
+            
+            ruta = storage_service.save_datastage(
+                file=archivo_file,
+                organizacion_id=organizacion.id,
+                metadata={
+                    'datastage_id': str(instance.id),
+                    'updated': 'true'
+                }
+            )
+
+            if ruta:
+                instance.archivo = ruta
+                instance.save()
+            else:
+                raise serializers.ValidationError({"archivo": "Error al guardar el nuevo archivo"})
+        return instance

api/datastage/tasks.py

@@ -1,3 +1,4 @@
+import tempfile
 from celery import group
 from celery import shared_task
 import logging
@@ -6,81 +7,130 @@ from django.utils import timezone
 import os
 import zipfile
 import re
+from api.utils.storage_service import storage_service
 
 @shared_task
 def procesar_datastage_task(datastage_id, user_organizacion_id=None):
     import traceback
+    tmp_path = None
     try:
-        logger = logging.getLogger(__name__)
         from api.datastage.models import DataStage
         from api.organization.models import Organizacion
-        from api.customs.models import Pedimento, TipoOperacion, Regimen
 
-        datastage = DataStage.objects.get(id=datastage_id)
+        # Obtener datastage
+        try:
+            datastage = DataStage.objects.get(id=datastage_id)
+        except DataStage.DoesNotExist:
+            return {'error': f'DataStage {datastage_id} no encontrado'}
+
+        # Validar archivo
         if not datastage.archivo:
+            print("DataStage no tiene archivo asociado")
             return {'detail': 'No hay archivo asociado a este DataStage.'}
-        file_path = datastage.archivo.path
-        if not os.path.exists(file_path):
-            return {'detail': 'El archivo no existe en el servidor.'}
-        if not file_path.endswith('.zip'):
+        
+        ruta_archivo = str(datastage.archivo)
+        
+        if not ruta_archivo.lower().endswith('.zip'):
             return {'detail': 'El archivo no es un .zip.'}
-
-        documentos_encontrados = []
-        registros_cargados = {}
-        registros_por_archivo = {}
-        errores_por_archivo = {}
-        errores_pedimento = []
+        
+        # Descargar archivo
+        with tempfile.NamedTemporaryFile(delete=False, suffix='.zip') as tmp:
+            tmp_path = tmp.name
+        
+        success = storage_service.download_file(ruta_archivo, tmp_path)
+        
+        if not success:
+            print(f"No se pudo descargar: {ruta_archivo}")
+            return {'detail': f'No se pudo descargar el archivo: {ruta_archivo}'}
+        
+        file_path = tmp_path
+        
+        # Obtener organización
         user_organizacion = None
-
         if user_organizacion_id:
-            user_organizacion = Organizacion.objects.get(id=user_organizacion_id)
+            try:
+                user_organizacion = Organizacion.objects.get(id=user_organizacion_id)
+            except Organizacion.DoesNotExist:
+                print(f"Organización no encontrada: {user_organizacion_id}")
 
-        def to_snake_case(name):
-            s1 = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', name)
-            s2 = re.sub('([a-z0-9])([A-Z])', r'\1_\2', s1)
-            return s2.replace('__', '_').lower()
-
-        # Lanzar una subtarea por cada archivo ASC
+        # Leer ZIP y lanzar subtareas
         subtasks = []
         with zipfile.ZipFile(file_path, 'r') as zip_ref:
-            for asc_name in zip_ref.namelist():
+            namelist = zip_ref.namelist()
+            
+            for asc_name in namelist:
                 if asc_name.endswith('.asc'):
-                    subtasks.append(procesar_archivo_asc_task.s(datastage_id, user_organizacion_id, asc_name))
+                    subtasks.append(
+                        procesar_archivo_asc_task.s(datastage_id, user_organizacion_id, asc_name)
+                    )
+        
         if subtasks:
             job = group(subtasks).apply_async()
+            print(f"Grupo de tareas lanzado: {job.id}")
             return {
                 'group_id': job.id,
                 'subtask_ids': [t.id for t in job.results],
-                'detail': 'Procesamiento lanzado. Monitorea el estado de cada subtask_id.'
+                'detail': f'Procesamiento lanzado. {len(subtasks)} archivos .ASC en cola.'
             }
+        
+        print("No se encontraron archivos .ASC")
         return {'detail': 'No se encontraron archivos .asc'}
+        
     except Exception as e:
         import traceback
         return {'error': str(e), 'traceback': traceback.format_exc()}
+        
+    finally:
+        # Limpiar temporal
+        if tmp_path and os.path.exists(tmp_path):
+            try:
+                os.unlink(tmp_path)
+            except Exception as e:
+                print(f"No se pudo eliminar temporal: {e}")
 
 @shared_task
 def procesar_archivo_asc_task(datastage_id, user_organizacion_id, asc_name):
-    import traceback
+    """
+    Procesa un archivo .ASC individual dentro del ZIP
+    """
+    tmp_path = None
     try:
-        logger = logging.getLogger(__name__)
         from api.datastage.models import DataStage
         from api.organization.models import Organizacion
         from api.customs.models import Pedimento, TipoOperacion, Regimen
-        from django.apps import apps
-        import zipfile
-        import re
+        import datetime
+        
+        # Obtener datastage
         datastage = DataStage.objects.get(id=datastage_id)
         user_organizacion = None
         if user_organizacion_id:
             user_organizacion = Organizacion.objects.get(id=user_organizacion_id)
-        file_path = datastage.archivo.path
+        
+        ruta_archivo = str(datastage.archivo)
+        
+        # Descargar archivo
+        with tempfile.NamedTemporaryFile(delete=False, suffix='.zip') as tmp:
+            tmp_path = tmp.name
+        
+        success = storage_service.download_file(ruta_archivo, tmp_path)
+        if not success:
+            return {'errores': [f'No se pudo descargar el archivo: {ruta_archivo}']}
+        
+        file_path = tmp_path
+        
         def to_snake_case(name):
             s1 = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', name)
             s2 = re.sub('([a-z0-9])([A-Z])', r'\1_\2', s1)
             return s2.replace('__', '_').lower()
+        
+        objects_to_create = []
+        
         with zipfile.ZipFile(file_path, 'r') as zip_ref:
             if asc_name not in zip_ref.namelist():
+                print(f"❌ {asc_name} no encontrado en el ZIP")
                 return {'errores': [f'{asc_name} no encontrado en el zip']}
+            
+            # Determinar modelo
             match = re.match(r'.*_(\d+)\.asc$', asc_name)
             if match:
                 registro_key = match.group(1)
@@ -96,53 +146,53 @@ def procesar_archivo_asc_task(datastage_id, user_organizacion_id, asc_name):
                 Model = apps.get_model('datastage', model_name)
             except LookupError:
                 return {'errores': [f"No existe el modelo para {model_name}"]}
+            
+            # Procesar archivo
             with zip_ref.open(asc_name) as asc_file:
                 first = True
-                field_names = []
                 field_names_snake = []
-                objects_to_create = []
-                errores_pedimento = []
+                line_count = 0
+                
                 for line in asc_file:
-                    line_decoded = None
+                    line_count += 1
                     try:
                         line_decoded = line.decode('utf-8').strip()
                     except UnicodeDecodeError:
                         try:
                             line_decoded = line.decode('latin-1').strip()
-                        except Exception as e:
+                        except Exception:
                             continue
-                    except Exception as e:
-                        continue
+                    
                     if not line_decoded:
                         continue
+                    
                     if first:
                         field_names = [f for f in line_decoded.split('|')]
                         field_names_snake = [to_snake_case(f) for f in field_names]
                         first = False
                         continue
+                    
                     values = line_decoded.split('|')
                     while values and values[-1] == '':
                         values.pop()
-                    if len(values) == len(field_names_snake) + 1 and values[-1] == '':
-                        values = values[:-1]
-                    if len(values) < len(field_names_snake):
-                        values += [None] * (len(field_names_snake) - len(values))
                     if len(values) != len(field_names_snake):
                         continue
+                    
                     data = dict(zip(field_names_snake, values))
+                    
                     if hasattr(Model, 'organizacion_id'):
                         data['organizacion_id'] = user_organizacion.id if user_organizacion else None
                     if hasattr(Model, 'datastage_id'):
                         data['datastage_id'] = datastage.id
-                    # Limpiar campos de fecha vacíos ('') a None
+                    
+                    # Limpiar fechas vacías
                     for field in Model._meta.get_fields():
                         if hasattr(field, 'get_internal_type') and field.get_internal_type() in ["DateField", "DateTimeField"]:
                             if data.get(field.name) == "":
                                 data[field.name] = None
-                    # Convertir fecha_pago_real a timezone-aware si existe
+                    
+                    # Convertir fecha_pago_real
                     if 'fecha_pago_real' in data and data['fecha_pago_real']:
-                        from django.utils import timezone
-                        import datetime
                         fecha_val = data['fecha_pago_real']
                         if isinstance(fecha_val, str):
                             try:
@@ -156,11 +206,11 @@ def procesar_archivo_asc_task(datastage_id, user_organizacion_id, asc_name):
                                 dt = timezone.make_aware(dt)
                             if dt:
                                 data['fecha_pago_real'] = dt
-                        elif isinstance(fecha_val, datetime.datetime) and timezone.is_naive(fecha_val):
-                            data['fecha_pago_real'] = timezone.make_aware(fecha_val)
+                    
                     try:
                         obj = Model(**data)
                         objects_to_create.append(obj)
+                        
                         # Si es Registro501, crear Pedimento
                         if model_name == 'Registro501':
                             organizacion_instance = None
@@ -169,7 +219,7 @@ def procesar_archivo_asc_task(datastage_id, user_organizacion_id, asc_name):
                                 try:
                                     organizacion_instance = Organizacion.objects.get(id=org_id)
                                 except Exception as org_exc:
-                                    logger.warning(f"No se encontró la organización con id {org_id}: {org_exc}")
+                                    print(f"No se encontró la organización con id {org_id}: {org_exc}")
                             if not organizacion_instance:
                                 organizacion_instance = user_organizacion
                             fecha_pago_raw = data.get('fecha_pago_real')
@@ -198,7 +248,7 @@ def procesar_archivo_asc_task(datastage_id, user_organizacion_id, asc_name):
                                     pedimento_app = f"{year[-2:]}-{str(aduana).zfill(2)[:2]}-{str(patente).zfill(4)[-4:]}-{str(pedimento_num).zfill(7)[-7:]}"
                                     # logger.info(f"pedimento_app >>>> {pedimento_app}")
                             except Exception as ped_app_exc:
-                                logger.warning(f"No se pudo generar pedimento_app: {ped_app_exc}")
+                                print(f"No se pudo generar pedimento_app: {ped_app_exc}")
                             tipo_operacion_val = data.get('tipo_operacion')
                             tipo_operacion = TipoOperacion.objects.filter(id=int(tipo_operacion_val)).first() if tipo_operacion_val else None
                             regimen = Regimen.objects.filter(claveped=data.get('clave_documento', '').strip(), tipo=tipo_operacion.id if tipo_operacion else None).first() if tipo_operacion else None
@@ -237,11 +287,14 @@ def procesar_archivo_asc_task(datastage_id, user_organizacion_id, asc_name):
                                 pass
                     except Exception as e:
                         continue
-                if objects_to_create:
-                    try:
-                        Model.objects.bulk_create(objects_to_create, batch_size=1000)
-                    except Exception as e:
-                        return {'archivo': asc_name, 'error': str(e), 'traceback': traceback.format_exc()}
+                
+        # Bulk create
+        if objects_to_create:
+            try:
+                Model.objects.bulk_create(objects_to_create, batch_size=1000)
+            except Exception as e:
+                return {'archivo': asc_name, 'error': str(e)}
+        
         return {
             'archivo': asc_name,
             'insertados': len(objects_to_create)
@@ -249,33 +302,11 @@ def procesar_archivo_asc_task(datastage_id, user_organizacion_id, asc_name):
     except Exception as e:
         import traceback
         return {'archivo': asc_name, 'error': str(e), 'traceback': traceback.format_exc()}
-
-    detalles = {}
-    for key in ['502', '503', '504']:
-        model_name = f'Registro{key}'
-        asc_file = None
-        encabezado = None
-        errores = []
-        for asc_name in registros_por_archivo:
-            if asc_name.endswith(f'_{key}.asc'):
-                asc_file = asc_name
-                break
-        if asc_file:
+        
+    finally:
+        # Limpiar temporal
+        if tmp_path and os.path.exists(tmp_path):
             try:
-                with zipfile.ZipFile(file_path, 'r') as zip_ref:
-                    with zip_ref.open(asc_file) as f:
-                        for line in f:
-                            try:
-                                encabezado = line.decode('utf-8').strip()
-                            except UnicodeDecodeError:
-                                encabezado = line.decode('latin-1').strip()
-                            break
+                os.unlink(tmp_path)
             except Exception as e:
-                encabezado = f'Error leyendo encabezado: {e}'
-            errores = errores_por_archivo.get(asc_file, [])
-        detalles[model_name] = {
-            'archivo': asc_file,
-            'encabezado': encabezado,
-            'errores': errores
-        }
-    return {'registros_cargados': registros_cargados, 'errores_pedimento': errores_pedimento}
+                print(f"No se pudo eliminar temporal: {e}")

api/datastage/views.py

@@ -1,3 +1,8 @@
+import atexit
+import tempfile
+
+from api.utils.storage_service import storage_service
+from config import settings
 from rest_framework.pagination import PageNumberPagination
 from api.customs.models import Pedimento, TipoOperacion, Regimen
 from django.shortcuts import render
@@ -112,18 +117,65 @@ class DataStageViewSet(LoggingMixin, viewsets.ModelViewSet, OrganizacionFiltrada
     def download_datastage(self, request, pk=None):
         """
         Endpoint para descargar el archivo asociado a un DataStage.
+        Soporta tanto archivos en MinIO como archivos locales antiguos.
         """
         try:
             datastage = self.get_object()
             if not datastage.archivo:
                 raise Http404("No hay archivo asociado a este DataStage.")
-            file_path = datastage.archivo.path
-            if not os.path.exists(file_path):
-                raise Http404("El archivo no existe en el servidor.")
-            response = FileResponse(open(file_path, 'rb'), as_attachment=True, filename=os.path.basename(file_path))
-            return response
+            
+            # Detectar si es ruta de MinIO o local
+            is_minio_path = datastage.archivo.startswith('org_')
+            
+            if is_minio_path:
+                import tempfile
+                
+                with tempfile.NamedTemporaryFile(delete=False) as tmp:
+                    tmp_path = tmp.name
+                
+                success = storage_service.download_file(datastage.archivo, tmp_path)
+                
+                if not success:
+                    raise Http404("No se pudo descargar el archivo de MinIO")
+                
+                filename = os.path.basename(datastage.archivo)
+                
+                response = FileResponse(
+                    open(tmp_path, 'rb'),
+                    as_attachment=True,
+                    filename=filename
+                )
+                
+                import atexit
+                atexit.register(lambda: os.unlink(tmp_path) if os.path.exists(tmp_path) else None)
+                
+                return response
+                
+            else:
+                file_path = os.path.join(settings.MEDIA_ROOT, str(datastage.archivo))
+                
+                if not os.path.exists(file_path):
+                    raise Http404(f"El archivo no existe: {file_path}")
+                
+                filename = os.path.basename(file_path)
+                
+                response = FileResponse(
+                    open(file_path, 'rb'),
+                    as_attachment=True,
+                    filename=filename
+                )
+                return response
+                
         except Exception as e:
             return Response({'detail': str(e)}, status=404)
+    
+    def perform_destroy(self, instance):
+        """
+        Al eliminar un DataStage, también eliminar su archivo asociado.
+        """
+        if instance.archivo:
+            storage_service.delete_file(instance.archivo)
+        instance.delete()
 
     @action(detail=True, methods=['post'], url_path='procesar')
     def procesar(self, request, pk=None):