fix/de los tickets T2026-05-027, T2025-09-004 y T2025-09-056

api/reports/tasks/__init__.py

@@ -0,0 +1,3 @@
+# Importa los módulos de tasks para que autodiscover_tasks() los registre en el worker
+from .report_document import generate_report_document, generate_report_control_pedimento
+from .report_datastage import generate_report_datastage

api/reports/tasks/report_datastage.py

@@ -0,0 +1,105 @@
+import logging
+import traceback
+
+from celery import shared_task
+from celery.exceptions import SoftTimeLimitExceeded
+from django.core.files.uploadedfile import SimpleUploadedFile
+from django.utils import timezone
+
+from api.reports.models import ReportDocument
+from api.reports.services.datastage_export import build_datastage_export
+from api.utils.storage_service import storage_service
+from core.redis_events import publish_task_event
+
+logger = logging.getLogger('api.reports.tasks')
+
+
+@shared_task(bind=True, queue='reports', soft_time_limit=1800, time_limit=1860)
+def generate_report_datastage(self, report_id):
+    task_id = self.request.id
+    report = None
+
+    def _fail(msg, exc=None):
+        """Marca el reporte como error, notifica al frontend y loguea. Sin re-raise."""
+        tb = traceback.format_exc() if exc else ''
+        full_msg = f"{msg}\n\n{tb}".strip() if tb else msg
+        logger.error('[reporte_datastage] report=%s FALLO: %s', report_id, full_msg)
+        if report:
+            report.status = 'error'
+            report.error_message = full_msg
+            report.finished_at = timezone.now()
+            report.save(update_fields=['status', 'error_message', 'finished_at'])
+        publish_task_event(task_id, 'failed', msg, progress=0)
+
+    # ── 1. Obtener reporte ────────────────────────────────────────────────────
+    try:
+        report = ReportDocument.objects.get(id=report_id)
+    except ReportDocument.DoesNotExist:
+        logger.error('[reporte_datastage] ReportDocument %s no existe', report_id)
+        publish_task_event(task_id, 'failed', f'Reporte {report_id} no encontrado', progress=0)
+        return
+
+    logger.info('[reporte_datastage] Iniciando report=%s user=%s', report_id, report.user_id)
+    report.status = 'processing'
+    report.save(update_fields=['status'])
+    publish_task_event(task_id, 'processing', 'Iniciando generación de reporte...', progress=5)
+
+    try:
+        # La organización ya viene resuelta en el payload (la vista la fija antes de encolar)
+        payload = report.filters or {}
+        org_id = payload.get('organizacion_id')
+
+        def _progress(pct, msg):
+            publish_task_event(task_id, 'processing', msg, progress=pct)
+
+        # ── 2. Generar archivo (xlsx / csv / zip según modo, formato y volumen) ──
+        content, filename, content_type, total_rows = build_datastage_export(payload, _progress)
+
+        # ── 3. Subir a almacenamiento ─────────────────────────────────────────
+        logger.info('[reporte_datastage] report=%s archivo=%s size=%.1fKB filas=%d',
+                    report_id, filename, len(content) / 1024, total_rows)
+        publish_task_event(task_id, 'processing', 'Subiendo a almacenamiento...', progress=93)
+
+        final_name = f"datastage_{report.id}_{timezone.now().strftime('%Y%m%d%H%M%S')}_{filename}"
+        ruta = storage_service.save_report(
+            file=SimpleUploadedFile(
+                name=final_name,
+                content=content,
+                content_type=content_type,
+            ),
+            organizacion_id=org_id,
+            metadata={
+                'report_id': str(report.id),
+                'report_type': 'datastage',
+                'user_id': str(report.user.id) if report.user else None,
+            },
+        )
+
+        if ruta:
+            logger.info('[reporte_datastage] report=%s guardado en storage=%s', report_id, ruta)
+            report.file = ruta
+            report.status = 'ready'
+        else:
+            _fail('Error al guardar el archivo en almacenamiento (storage retornó None)')
+            return
+
+        report.finished_at = timezone.now()
+        report.save(update_fields=['status', 'file', 'finished_at', 'error_message'])
+
+        resultado = {
+            'report_id': str(report.id),
+            'total_registros': total_rows,
+            'archivo': final_name,
+        }
+        publish_task_event(task_id, 'completed', 'Reporte generado exitosamente.', progress=100, resultado=resultado)
+        logger.info('[reporte_datastage] report=%s COMPLETADO filas=%d', report_id, total_rows)
+        return resultado
+
+    except SoftTimeLimitExceeded:
+        _fail('El reporte tardó más de 30 minutos y fue cancelado. Intenta con filtros más acotados.')
+
+    except ValueError as exc:
+        _fail(str(exc))
+
+    except Exception as exc:
+        _fail(str(exc), exc=exc)