feat: Implement asynchronous invoice report generation with email delivery and frontend status polling.

backend/api/v1/modules/a76/reports/movements/invoices/csv_utils.py

@@ -0,0 +1,87 @@
+"""
+CSV generation utilities for invoice movement reports.
+"""
+import csv
+import io
+from typing import List, Union
+from datetime import datetime
+
+from .schemas import MovementItem, MovementItemDetailed
+
+
+def generate_csv_from_movements(
+    movements: List[Union[MovementItem, MovementItemDetailed]],
+    report_type: str = "normal"
+) -> str:
+    """
+    Generate CSV content from movement items.
+    
+    Args:
+        movements: List of movement items (normal or detailed)
+        report_type: "normal" or "detailed"
+        
+    Returns:
+        CSV content as string
+    """
+    output = io.StringIO()
+    
+    if report_type.lower() == "normal":
+        # Normal report columns
+        fieldnames = [
+            'Pedimento', 'ClavePed', 'Factura', 'FechaFactura', 
+            'ValorComercialMN', 'TipoMovTemDef', 'Estatus', 'BaseDeDatos',
+            'TipoCambio', 'Fecha_Pago', 'ValorMPTemp', 'ValorAgre',
+            'TipoExpo', 'EsCambioRegimen', 'Regimen'
+        ]
+        
+        writer = csv.DictWriter(output, fieldnames=fieldnames, extrasaction='ignore')
+        writer.writeheader()
+        
+        for movement in movements:
+            row = movement.model_dump()
+            # Format datetime fields
+            if row.get('FechaFactura'):
+                row['FechaFactura'] = _format_datetime(row['FechaFactura'])
+            if row.get('Fecha_Pago'):
+                row['Fecha_Pago'] = _format_datetime(row['Fecha_Pago'])
+            writer.writerow(row)
+    
+    else:
+        # Detailed report columns
+        fieldnames = [
+            'Linea', 'Pedimento', 'Factura', 'FechaFactura', 
+            'Proveedor', 'VendidoA', 'CantidadIE', 'DescripcionE',
+            'DescripcionI', 'NumParte', 'UniMed', 'ValorComercialMN',
+            'TipoMovTemDef', 'ClavePed', 'Estatus', 'TipoCambio',
+            'PesoNeto', 'PesoBruto', 'OrdenCompraVenta', 'Regimen',
+            'AgenteAduanal', 'Patente', 'BaseDeDatos'
+        ]
+        
+        writer = csv.DictWriter(output, fieldnames=fieldnames, extrasaction='ignore')
+        writer.writeheader()
+        
+        for movement in movements:
+            row = movement.model_dump()
+            # Format datetime fields
+            if row.get('FechaFactura'):
+                row['FechaFactura'] = _format_datetime(row['FechaFactura'])
+            if row.get('Fecha_Pago'):
+                row['Fecha_Pago'] = _format_datetime(row['Fecha_Pago'])
+            if row.get('Fecha_Inicio'):
+                row['Fecha_Inicio'] = _format_datetime(row['Fecha_Inicio'])
+            if row.get('Fecha_Fin'):
+                row['Fecha_Fin'] = _format_datetime(row['Fecha_Fin'])
+            writer.writerow(row)
+    
+    csv_content = output.getvalue()
+    output.close()
+    return csv_content
+
+
+def _format_datetime(dt) -> str:
+    """Format datetime for CSV export."""
+    if isinstance(dt, datetime):
+        return dt.strftime('%Y-%m-%d %H:%M:%S')
+    elif isinstance(dt, str):
+        return dt
+    return ''

backend/api/v1/modules/a76/reports/movements/invoices/movement_service.py

@@ -154,16 +154,25 @@ class MovementService:
         # Convert AllMovementsFilter to individual filter types
         # We'll use the same filter parameters for all queries
         
-        # Determine which services to call based on operation_type filter
-        should_fetch_imports = filters.operation_type in [None, 'imp']
-        should_fetch_exports = filters.operation_type in [None, 'exp']
+        # Determine which services to call based on granular flags
+        # Default behavior: If granular flags are all defaults (True) but operation_type is set, 
+        # we might need to respect operation_type.
+        # But for simplicity, we assume granular flags from frontend are the source of truth.
+        # If frontend didn't set them (legacy call?), they default to True.
         
-        logger.info(f"Operation type filter: {filters.operation_type}")
-        logger.info(f"Should fetch imports: {should_fetch_imports}")
-        logger.info(f"Should fetch exports: {should_fetch_exports}")
+        # Override based on operation_type if provided (legacy compatibility or coarse filter)
+        if filters.operation_type == 'imp':
+            filters.export_def = False
+            filters.export_rep = False
+        elif filters.operation_type == 'exp':
+            filters.import_temp = False
+            filters.import_def = False
+            filters.import_rep = False
+            
+        logger.info(f"Fetching movements with flags: Temp={filters.import_temp}, Def={filters.import_def}, Rep={filters.import_rep}, ExpDef={filters.export_def}, ExpRep={filters.export_rep}")
         
         # 1. Temporary Imports
-        if should_fetch_imports:
+        if filters.import_temp:
                 temp_filter = ImportTemporaryFilter(
                     range_type=filters.range_type,
                     start_date=filters.start_date,
@@ -172,11 +181,11 @@ class MovementService:
                     provider=filters.provider,
                     buyer=filters.buyer,
                     pedimento_code=filters.pedimento_code,
-                    report_type=filters.report_type,  # Use filter's report_type
+                    report_type=filters.report_type,
                     currency_type=filters.currency_type,
                     exchange_rate_type=filters.exchange_rate_type,
                     is_shelter=filters.is_shelter,
-                    database_name='default'  # Required field
+                    database_name='default'
                 )
                 if filters.report_type == ReportType.DETAILED:
                     temp_movements = self.temporary_service.get_movements_detailed(db, temp_filter)
@@ -186,7 +195,7 @@ class MovementService:
                 logger.info(f"Added {len(temp_movements)} temporary import movements")
         
         # 2. Definitive Imports
-        if should_fetch_imports:
+        if filters.import_def:
             def_filter = ImportDefinitiveFilter(
                 range_type=filters.range_type,
                 start_date=filters.start_date,
@@ -211,7 +220,7 @@ class MovementService:
             logger.info(f"Added {len(def_movements)} definitive import movements")
         
         # 3. Repair Imports
-        if should_fetch_imports:
+        if filters.import_rep:
             repair_filter = ImportRepairFilter(
                 range_type=filters.range_type,
                 start_date=filters.start_date,
@@ -236,7 +245,7 @@ class MovementService:
             logger.info(f"Added {len(repair_movements)} repair import movements")
         
         # 4. Exports (Definitive)
-        if should_fetch_exports:
+        if filters.export_def:
             export_filter = ExportFilter(
                 range_type=filters.range_type,
                 start_date=filters.start_date,
@@ -262,7 +271,7 @@ class MovementService:
             logger.info(f"Added {len(export_movements)} export movements")
         
         # 5. Export Repairs
-        if should_fetch_exports:
+        if filters.export_rep:
             export_repair_filter = ExportRepairFilter(
                 range_type=filters.range_type,
                 start_date=filters.start_date,

backend/api/v1/modules/a76/reports/movements/invoices/routes.py

@@ -604,9 +604,11 @@ def get_export_repair_movements_detailed(
     
     Use this when "TODAS" checkbox is selected to get a comprehensive view of all movements
     regardless of their specific type.
+    
+    If send_email is True, the report will be sent to the authenticated user's email address.
     """
 )
-def get_all_movements(
+async def get_all_movements(
     filters: AllMovementsFilter,
     db: Session = Depends(get_core_db),
     current_user: dict = Depends(get_current_user)
@@ -628,13 +630,52 @@ def get_all_movements(
     try:
         logger.info(
             f"User {current_user.get('preferred_username', 'unknown')} "
-            f"requesting all invoice movements"
+            f"requesting all invoice movements (send_email={filters.send_email})"
         )
         movements = movement_service.get_all_movements(
             db=db, 
             filters=filters
         )
         logger.info(f"Successfully retrieved {len(movements)} total movements")
+        
+        # Send email if requested
+        if filters.send_email:
+            user_email = current_user.get('email')
+            if not user_email:
+                logger.warning(f"User {current_user.get('sub')} has no email address - skipping email")
+            else:
+                try:
+                    from core.email import EmailService
+                    from .csv_utils import generate_csv_from_movements
+                    from datetime import datetime
+                    
+                    # Generate CSV
+                    csv_content = generate_csv_from_movements(
+                        movements=movements,
+                        report_type=filters.report_type.value.lower()
+                    )
+                    
+                    # Generate filename
+                    filename = f"reporte_facturas_{filters.start_date}_{filters.end_date}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
+                    
+                    # Send email
+                    email_sent = await EmailService.send_report_email(
+                        recipient_email=user_email,
+                        subject=f"Reporte de Facturas - {filters.start_date} al {filters.end_date}",
+                        body_text=f"Se ha generado el reporte de facturas solicitado con {len(movements)} registros.",
+                        csv_content=csv_content,
+                        filename=filename
+                    )
+                    
+                    if email_sent:
+                        logger.info(f"Report emailed successfully to {user_email}")
+                    else:
+                        logger.warning(f"Failed to send email to {user_email} - SMTP may not be configured correctly")
+                        
+                except Exception as email_error:
+                    logger.warning(f"Email sending failed: {str(email_error)} - continuing with report generation")
+        
+        return movements
         return movements
     except ValueError as e:
         logger.warning(f"Validation error fetching all movements: {str(e)}")
@@ -642,9 +683,66 @@ def get_all_movements(
             status_code=status.HTTP_400_BAD_REQUEST,
             detail=str(e)
         )
+    except HTTPException:
+        raise
     except Exception as e:
         logger.error(f"Error fetching all movements: {str(e)}", exc_info=True)
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail=f"Error processing all movements: {str(e)}"
         )
+
+
+@router.post(
+    "/generate",
+    summary="Generate Invoice Report (Async)",
+    description="Trigger background generation of invoice report."
+)
+def generate_invoice_report_async(
+    filters: AllMovementsFilter,
+    current_user: dict = Depends(get_current_user)
+):
+    """
+    Trigger background generation of invoice report.
+    Returns task_id to poll status.
+    """
+    from .tasks import generate_invoice_movements_async
+    
+    logger.info(f"User {current_user.get('preferred_username', 'unknown')} triggering async report generation")
+    
+    # Serialize filters to dict for Celery
+    filter_data = filters.model_dump()
+    user_email = current_user.get('email')
+    
+    # Trigger task
+    task = generate_invoice_movements_async.delay(filter_data, user_email)
+    
+    return {"task_id": task.id}
+
+
+@router.get(
+    "/task/{task_id}",
+    summary="Get Async Task Status",
+    description="Check status of background report generation task."
+)
+def get_task_status(task_id: str):
+    """
+    Get status of background task.
+    """
+    from celery.result import AsyncResult
+    from core.celery_app import celery_app
+    
+    task_result = AsyncResult(task_id, app=celery_app)
+    
+    response = {
+        "task_id": task_id,
+        "status": task_result.status,
+    }
+    
+    if task_result.state == 'PROCESSING':
+         response["meta"] = task_result.info
+    
+    if task_result.ready():
+        response["result"] = task_result.result
+        
+    return response

backend/api/v1/modules/a76/reports/movements/invoices/schemas.py

@@ -103,6 +103,27 @@ class AllMovementsFilter(BaseModel):
         default=None,
         description="Filter by operation type: 'imp' for imports only, 'exp' for exports only, None for all"
     )
+    send_email: bool = Field(
+        default=False,
+        description="Send report via email to current user"
+    )
+    
+    # Granular movement selection
+    import_temp: bool = Field(default=True, description="Include temporary imports (IMTEM)")
+    import_def: bool = Field(default=True, description="Include definitive imports (IMPDF/COMEX)")
+    import_rep: bool = Field(default=True, description="Include repair imports (IMPRE)")
+    export_def: bool = Field(default=True, description="Include definitive exports")
+    export_rep: bool = Field(default=True, description="Include repair exports")
+    
+    # Specific filters
+    export_types: Optional[list[str]] = Field(
+        default=None, 
+        description="Specific export legacy codes to include (AFIJO, NODES, etc)"
+    )
+    discharge_filter: DischargeFilter = Field(
+        default=DischargeFilter.ALL,
+        description="Global discharge filter for repair movements"
+    )
 
 
 class ImportTemporaryFilter(BaseModel):
@@ -155,6 +176,10 @@ class ImportTemporaryFilter(BaseModel):
         ...,
         description="Legacy database name to query from"
     )
+    send_email: bool = Field(
+        default=False,
+        description="Send report via email to current user"
+    )
 
 
 class ImportDefinitiveFilter(BaseModel):
@@ -215,6 +240,10 @@ class ImportDefinitiveFilter(BaseModel):
         ...,
         description="Legacy database name to query from"
     )
+    send_email: bool = Field(
+        default=False,
+        description="Send report via email to current user"
+    )
 
 class MovementItem(BaseModel):
     """Movement item representing a temporary import invoice"""
@@ -317,6 +346,10 @@ class ImportRepairFilter(BaseModel):
         ...,
         description="Legacy database name to query from"
     )
+    send_email: bool = Field(
+        default=False,
+        description="Send report via email to current user"
+    )
 
 
 class MovementItemDetailed(BaseModel):
@@ -464,6 +497,10 @@ class ExportFilter(BaseModel):
         ...,
         description="Legacy database name"
     )
+    send_email: bool = Field(
+        default=False,
+        description="Send report via email to current user"
+    )
 
     model_config = {
         "json_schema_extra": {
@@ -545,6 +582,10 @@ class ExportRepairFilter(BaseModel):
         ...,
         description="Legacy database name"
     )
+    send_email: bool = Field(
+        default=False,
+        description="Send report via email to current user"
+    )
 
     model_config = {
         "json_schema_extra": {

backend/api/v1/modules/a76/reports/movements/invoices/services/query_builders.py

@@ -353,7 +353,14 @@ class RepairImportQueries:
     @staticmethod
     def build_aggregated_query(db_name: str, where_str: str, discharge_clause: str = "") -> str:
         """Build optimized query for NORMAL mode (grouped by invoice with totals)."""
-        discharge_filter = ""  # Temporarily disabled until schema migration
+        # Use fa_item_lines.download field (true = discharged, false = not discharged)
+        if "SiDes" in discharge_clause:
+            discharge_filter = "AND fil.download = true"
+        elif "NoDes" in discharge_clause:
+            discharge_filter = "AND fil.download = false"
+        else:
+            discharge_filter = ""
+
         return f"""
         SELECT 
             ih.invoice_number AS C2,
@@ -403,9 +410,13 @@ class RepairImportQueries:
     @staticmethod
     def build_main_query(db_name: str, where_str: str, discharge_clause: str = "") -> str:
         """Build main SQL query for repair import data."""
-        # Note: is_discharged field not yet migrated to PostgreSQL schema
-        # discharge_filter = "AND il.is_discharged = true" if "descargado" in discharge_clause.lower() else ""
-        discharge_filter = ""  # Temporarily disabled until schema migration
+        # Use fa_item_lines.download field (true = discharged, false = not discharged)
+        if "SiDes" in discharge_clause:
+            discharge_filter = "AND fil.download = true"
+        elif "NoDes" in discharge_clause:
+            discharge_filter = "AND fil.download = false"
+        else:
+            discharge_filter = ""
         return f"""
         SELECT 
             il.line_number,
@@ -479,9 +490,13 @@ class RepairImportQueries:
     @staticmethod
     def build_totals_query(db_name: str, discharge_clause: str = "") -> str:
         """Build query to get totals for a repair import invoice."""
-        # Note: is_discharged field not yet migrated to PostgreSQL schema
-        # discharge_filter = "AND il.is_discharged = true" if "descargado" in discharge_clause.lower() else ""
-        discharge_filter = ""  # Temporarily disabled until schema migration
+        # Use fa_item_lines.download field (true = discharged, false = not discharged)
+        if "SiDes" in discharge_clause:
+            discharge_filter = "AND fil.download = true"
+        elif "NoDes" in discharge_clause:
+            discharge_filter = "AND fil.download = false"
+        else:
+            discharge_filter = ""
         return f"""
         SELECT 
             COALESCE(SUM(CASE WHEN COALESCE(fil.is_subitem, false) THEN 0 ELSE lf.value_usd END), 0), 
@@ -651,7 +666,13 @@ class ExportQueries:
         
         Only sums partidas where is_subitem is false (main partidas, not sub-items).
         """
-        discharge_filter = "AND il.is_discharged = true" if "descargado" in discharge_clause.lower() else ""
+        # Use fa_item_lines.download field (true = discharged, false = not discharged)
+        if "SiDes" in discharge_clause:
+            discharge_filter = "AND fil.download = true"
+        elif "NoDes" in discharge_clause:
+            discharge_filter = "AND fil.download = false"
+        else:
+            discharge_filter = ""
         return f"""
         SELECT 
             COALESCE(SUM(CASE WHEN COALESCE(fil.is_subitem, false) THEN 0 ELSE lf.value_usd END), 0), 
@@ -689,8 +710,14 @@ class ExportRepairQueries:
     @staticmethod
     def build_aggregated_query(db_name: str, where_str: str, discharge_clause: str = "") -> str:
         """Build optimized query for NORMAL mode (grouped by invoice with totals)."""
-        # Note: discharge_clause temporarily disabled until is_discharged field migrated
-        discharge_filter = ""  # Will be: " AND il.is_discharged = true/false" when ready
+        # Use fa_item_lines.download field (true = discharged, false = not discharged)
+        if "SiDes" in discharge_clause:
+            discharge_filter = "AND fil.download = true"
+        elif "NoDes" in discharge_clause:
+            discharge_filter = "AND fil.download = false"
+        else:
+            discharge_filter = ""
+
         return f"""
         SELECT 
             ih.invoice_number AS C1,
@@ -817,7 +844,13 @@ class ExportRepairQueries:
     @staticmethod
     def build_totals_query(db_name: str, discharge_clause: str = "") -> str:
         """Build query to get totals for an export repair invoice."""
-        discharge_filter = "AND il.is_discharged = true" if "descargado" in discharge_clause.lower() else ""
+        # Use fa_item_lines.download field (true = discharged, false = not discharged)
+        if "SiDes" in discharge_clause:
+            discharge_filter = "AND fil.download = true"
+        elif "NoDes" in discharge_clause:
+            discharge_filter = "AND fil.download = false"
+        else:
+            discharge_filter = ""
         return f"""
         SELECT 
             COALESCE(SUM(CASE WHEN COALESCE(fil.is_subitem, false) THEN 0 ELSE lf.value_usd END), 0), 

backend/api/v1/modules/a76/reports/movements/invoices/services/temporary.py

@@ -233,10 +233,10 @@ class TemporaryImportService:
                 # ValorComercialMN should always be in MXN
                 # If currency_type is ME, valor_comercial is in USD, so multiply by tipo_cambio
                 if filters.currency_type.value == "ME" and tipo_cambio:
-                    valor_comercial_mn = valor_comercial * tipo_cambio
+                    valor_comercial_mn = float(valor_comercial) * float(tipo_cambio)
                 else:
                     # If currency_type is MN, valor_comercial is already in MXN
-                    valor_comercial_mn = valor_comercial
+                    valor_comercial_mn = float(valor_comercial)
                 
                 # Set peso values based on subpartida flag
                 if row[39] == 'P':  # C40 - EsSubPartida

backend/api/v1/modules/a76/reports/movements/invoices/tasks.py

@@ -0,0 +1,108 @@
+
+import base64
+import logging
+import traceback
+from typing import Dict, Any
+
+from core.celery_app import celery_app
+from core.database import CoreSessionLocal
+from core.email import EmailService
+from datetime import datetime
+
+from .movement_service import movement_service
+from .schemas import AllMovementsFilter
+from .csv_utils import generate_csv_from_movements
+
+logger = logging.getLogger(__name__)
+
+@celery_app.task(bind=True, name="generate_invoice_movements_async")
+def generate_invoice_movements_async(self, filter_data: Dict[str, Any], user_email: str = None):
+    """
+    Async task to generate invoice movements report.
+    FETCHES data -> GENERATES CSV -> SENDS EMAIL (optional) -> RETURNS CSV (base64)
+    """
+    db = CoreSessionLocal()
+    try:
+        # 1. Update Progress
+        self.update_state(state='PROCESSING', meta={'current': 10, 'total': 100, 'status': 'Inicializando reporte...'})
+        
+        # 2. Reconstruct Filter
+        filters = AllMovementsFilter(**filter_data)
+        
+        # 3. Fetch Data
+        self.update_state(state='PROCESSING', meta={'current': 30, 'total': 100, 'status': 'Obteniendo movimientos de base de datos...'})
+        logger.info(f"Async Task: Fetching movements for {filters}")
+        
+        movements = movement_service.get_all_movements(db=db, filters=filters)
+        
+        self.update_state(state='PROCESSING', meta={'current': 70, 'total': 100, 'status': f'Procesando {len(movements)} registros...'})
+        
+        # 4. Generate CSV
+        csv_content = generate_csv_from_movements(
+            movements=movements,
+            report_type=filters.report_type.value.lower()
+        )
+        
+        # 5. Send Email if requested
+        email_sent = False
+        if filters.send_email and user_email:
+            self.update_state(state='PROCESSING', meta={'current': 90, 'total': 100, 'status': 'Enviando correo electrónico...'})
+            try:
+                # Generate filename
+                filename = f"reporte_facturas_{filters.start_date}_{filters.end_date}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
+                
+                # Send email (using the new async wrapper or run_until_complete if needed, 
+                # but since we are in a sync celery task we might need to be careful with async/await.
+                # Actually EmailService.send_report_email is async. 
+                # We need to run it synchronously here or make the task async.
+                # Celery tasks are sync by default. We can use asgiref.sync.async_to_sync
+                
+                import asyncio
+                from asgiref.sync import async_to_sync
+                
+                # Helper to run async method
+                result = async_to_sync(EmailService.send_report_email)(
+                    recipient_email=user_email,
+                    subject=f"Reporte de Facturas - {filters.start_date} al {filters.end_date}",
+                    body_text=f"Se ha generado el reporte de facturas solicitado con {len(movements)} registros.",
+                    csv_content=csv_content,
+                    filename=filename
+                )
+                
+                if result:
+                    email_sent = True
+                    logger.info(f"Async Task: Email sent to {user_email}")
+                else:
+                    logger.warning(f"Async Task: Failed to send email to {user_email}")
+                    
+            except Exception as e:
+                logger.error(f"Async Task: Email error: {str(e)}")
+        
+        # 6. Encode and Return
+        self.update_state(state='PROCESSING', meta={'current': 95, 'total': 100, 'status': 'Finalizando...'})
+        
+        # Convert string csv to bytes then base64
+        pdf_b64 = base64.b64encode(csv_content.encode('utf-8')).decode('utf-8')
+        
+        return {
+            'status': 'success',
+            'file_name': f"reporte_facturas_{datetime.now().strftime('%Y%m%d')}.csv",
+            'content': pdf_b64,
+            'media_type': 'text/csv',
+            'email_sent': email_sent,
+            'total_records': len(movements)
+        }
+
+    except Exception as e:
+        logger.error(f"Error in generate_invoice_movements_async: {str(e)}", exc_info=True)
+        self.update_state(
+            state='FAILURE',
+            meta={
+                'exc_type': type(e).__name__,
+                'exc_message': str(e),
+                'custom': 'Error generating report'
+            }
+        )
+        raise e
+    finally:
+        db.close()

backend/api/v1/modules/a76/router.py

@@ -56,7 +56,6 @@ from .reports.exportacion.aviso_consolidado.routes import router as aviso_consol
 from .reports.movements.invoices.routes import router as movement_invoices_router
 
 
-
 # Router principal
 router = APIRouter()