Refactor invoice processing logic to remove legacy returned quantity fields and enhance data handling

- Updated invoice processing functions to eliminate legacy fields related to returned quantities, now relying on new balance movement and discharge records.
- Adjusted various components and services to reflect changes in quantity handling, including updates to the frontend for displaying used quantities instead of returned ones.
- Improved the logic for invoice total updates and validation processes to ensure consistency with the new data structure.

These changes aim to streamline invoice management and improve data integrity across the application.

backend/api/v1/modules/a76/invoices/exports/process/sub_process/finalize_invoice.py

@@ -277,7 +277,7 @@ def finalize_invoice_with_discharge(
         if to_discharge:
             # Write BalanceMovement (CONSUMPTION) + DischargeHeader + DischargeDetail
             register_discharge_ledger(db, invoice, to_discharge)
-            # Update quantity_returned / value_returned on the import lines
+            # Update returned values on the import lines
             register_import_discharge(db, invoice, to_discharge)
             register_discharge_series(db, invoice, to_discharge)
 

backend/api/v1/modules/a76/invoices/exports/process/sub_process/register_discharge_ledger.py

@@ -167,6 +167,13 @@ def register_discharge_ledger(
             # lot_consumed_total == consume for single-lot entries (most cases)
             value_me = _proportional_value(consume, consume, lot.value_me)
             value_mn = _proportional_value(consume, consume, lot.value_mn)
+            imp_qty = import_line_obj.quantity if import_line_obj else None
+            imp_total_qty = imp_qty.quantity if imp_qty else None
+            net_weight = _proportional_qty(
+                consume,
+                imp_qty.net_weight if imp_qty else None,
+                imp_total_qty,
+            )
 
             movement = BalanceMovement(
                 tenant_id=export_invoice.tenant_id,
@@ -178,6 +185,7 @@ def register_discharge_ledger(
                 quantity=consume,
                 value_me=value_me,
                 value_mn=value_mn,
+                net_weight=net_weight,
                 source_invoice_id=export_invoice.id,
                 source_item_line_id=export_line_id,
                 order_peps=0,   # placeholder — set after flush (rule 4)
@@ -194,10 +202,7 @@ def register_discharge_ledger(
             # ── 3. DischargeDetail ─────────────────────────────────────────
             # Denormalized fields expected by reports:
             imp_cust = import_line_obj.customs if import_line_obj else None
-            imp_qty = import_line_obj.quantity if import_line_obj else None
-            imp_total_qty = imp_qty.quantity if imp_qty else None
-
-            net_weight = _proportional_qty(consume, imp_qty.net_weight if imp_qty else None, imp_total_qty)
+            # Reuse already computed net_weight for consistency with movement.
             gross_weight = _proportional_qty(consume, imp_qty.gross_weight if imp_qty else None, imp_total_qty)
 
             detail = DischargeDetail(

backend/api/v1/modules/a76/invoices/exports/process/sub_process/register_import_discharge.py

@@ -6,8 +6,8 @@ quantities and values consumed by this export invoice.
 For each entry in ``to_discharge`` (QSaldoActual in the legacy) the routine:
   · Looks up the source import invoice header (TEM → QFacImp, DEF → QFacImpDef).
   · Looks up the corresponding import line item.
-  · Increments quantity_returned, value_returned_mxn, value_returned_usd on the
-    import line's quantity/financial sub-records.
+  · Increments value_returned_mxn, value_returned_usd on the import line's
+    financial sub-record.
   · For TEM invoices, also calculates vat_used_mxn / vat_used_usd when the
     import invoice date is on or after 2014-12-31 (Clarion date 78165).
 
@@ -148,9 +148,7 @@ def register_import_discharge(
         returned_mn  = qty_used * value_mn  / original_qty
         returned_usd = qty_used * value_usd / original_qty
 
-        # ── Accumulate returned qty and value ─────────────────────────────────
-        qty_rec.quantity_returned = (qty_rec.quantity_returned or Decimal(0)) + qty_used
-
+        # ── Accumulate returned value ──────────────────────────────────────────
         fin.value_returned_mxn = (fin.value_returned_mxn or Decimal(0)) + returned_mn
         fin.value_returned_usd = (fin.value_returned_usd or Decimal(0)) + returned_usd
 

backend/api/v1/modules/a76/invoices/exports/revert/main_process.py

@@ -0,0 +1,373 @@
+import datetime
+from decimal import Decimal
+from typing import List, Optional
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from api.v1.modules.a24.discharges.models import (
+    DischargeDetail,
+    DischargeHeader,
+    DischargeStatus,
+)
+from api.v1.modules.a24.balance_movements.models import BalanceMovement, MovementType
+from api.v1.modules.a76.invoices.models import (
+    InvoiceHeader,
+    InvoiceStatus,
+    OperationType,
+)
+from api.v1.modules.a76.items.models import LineItem
+from api.v1.modules.a76.items.series.models import Serie
+from core.exceptions import ErrorCollector
+
+_VAT_CUTOFF = datetime.date(2014, 12, 31)  # Clarion day 78165
+
+
+def _validate_regime_change_definitive_invoice_exists(
+    db: Session,
+    invoice: InvoiceHeader,
+    errors: ErrorCollector,
+) -> None:
+    """
+    Clarion mapping:
+      If EqiFex:EsCambioRegimen='S' then count QFacImpDef where
+      FacturaImpoDef = FacturaExpo and ProvImpoDefCR='C'.
+
+    Python approximation:
+      Search an import invoice with same invoice_number and invoice_type='IMD'.
+    """
+    if not (invoice.compliance_mx and invoice.compliance_mx.is_regime_change):
+        return
+
+    if not invoice.invoice_number:
+        return
+
+    exists = (
+        db.query(InvoiceHeader.id)
+        .filter(
+            InvoiceHeader.tenant_id == invoice.tenant_id,
+            InvoiceHeader.company_id == invoice.company_id,
+            InvoiceHeader.operation_type == OperationType.IMP,            
+            InvoiceHeader.invoice_type == "DEF",  # Importación Definitiva generada
+            InvoiceHeader.invoice_number == invoice.invoice_number,
+        )
+        .first()
+        is not None
+    )
+
+    if exists:
+        errors.add_error(
+            field="invoice_number",
+            message=(
+                "Error: Existe una Factura de Importación Definitiva a partir "
+                "de esta Factura."
+            ),
+            solution=[
+                f"Desactualizar y borrar la Factura: {invoice.invoice_number} "
+                "de Importación Definitiva."
+            ],
+            code="DEFINITIVE_IMPORT_ALREADY_EXISTS",
+            value=invoice.invoice_number,
+        )
+
+
+def _todo_check_access_lock(invoice: InvoiceHeader) -> None:
+    # TODO: DO VALIDACION_USO_FACTURA_OTRO_USUARIO
+    # Clarion block against GAccesosModulos (security lock by terminal/user).
+    _ = invoice
+
+
+
+def _return_discharged_quantities(
+    db: Session,
+    export_invoice: InvoiceHeader,
+    lines: List[LineItem],
+) -> None:
+    """
+    REGRESA_CANT_RETORNADAS
+    Returns discharged quantities/values to their source import invoice lines.
+
+    Uses the new specialized discharge tables:
+      a24.discharge_header + a24.discharge_detail
+
+    We only revert details that belong to this export invoice and are applied.
+    This guarantees parity with the actual discharge ledger (instead of relying
+    on editable UI references in fa_data).
+    """
+    _ = lines  # source of truth is discharge tables
+
+    details: List[DischargeDetail] = (
+        db.query(DischargeDetail)
+        .join(DischargeHeader, DischargeHeader.id == DischargeDetail.discharge_header_id)
+        .filter(
+            DischargeHeader.source_invoice_id == export_invoice.id,
+            DischargeHeader.status == DischargeStatus.APPLIED,
+            DischargeDetail.tenant_id == export_invoice.tenant_id,
+            DischargeDetail.company_id == export_invoice.company_id,
+        )
+        .all()
+    )
+
+    for detail in details:
+        qty_exported = Decimal(str(detail.quantity_discharged or 0))
+        if qty_exported <= 0:
+            continue
+
+        import_line = db.get(LineItem, detail.import_item_line_id)
+        if (
+            import_line is None
+            or import_line.quantity is None
+            or import_line.financial is None
+        ):
+            continue
+
+        import_invoice = db.get(InvoiceHeader, import_line.invoice_id)
+        if import_invoice is None:
+            continue
+
+        qty_rec = import_line.quantity
+        fin = import_line.financial
+        original_qty = Decimal(str(qty_rec.quantity or 0))
+        if original_qty == 0:
+            continue
+
+        value_mxn = Decimal(str(fin.value_mxn or 0))
+        value_usd = Decimal(str(fin.value_usd or 0))
+        returned_mxn = qty_exported * value_mxn / original_qty
+        returned_usd = qty_exported * value_usd / original_qty
+
+        # Reverse monetary returned values
+        # Clarion shows '-' for TEM and '+' for DEF; in the current ledger migration,
+        # register_import_discharge adds both TEM/DEF, so revert subtracts both.
+        fin.value_returned_mxn = (fin.value_returned_mxn or Decimal(0)) - returned_mxn
+        fin.value_returned_usd = (fin.value_returned_usd or Decimal(0)) - returned_usd
+
+        # TEM VAT recalculation by header date cutoff
+        if (detail.procedence or "").upper() == "TEM":
+            inv_date = import_invoice.invoice_date
+            if isinstance(inv_date, datetime.datetime):
+                inv_date = inv_date.date()
+            if inv_date and inv_date >= _VAT_CUTOFF:
+                iva_factor = Decimal(0)
+                if import_invoice.financials and import_invoice.financials.iva_factor:
+                    iva_factor = Decimal(str(import_invoice.financials.iva_factor))
+                fin.vat_used_mxn = (fin.value_returned_mxn or Decimal(0)) * iva_factor / 100
+                fin.vat_used_usd = (fin.value_returned_usd or Decimal(0)) * iva_factor / 100
+            else:
+                fin.vat_used_mxn = Decimal(0)
+                fin.vat_used_usd = Decimal(0)
+
+
+def _unmark_returned_series(
+    db: Session,
+    export_invoice: InvoiceHeader,
+    lines: List[LineItem],
+) -> None:
+    """
+    DESMARCA_SERIES_RETORNADAS
+    Resets import serie discharge marks that were set by this export invoice.
+
+    New data-source logic:
+    - Uses a24.discharge_header/detail to identify which import lines were
+      consumed by this export invoice.
+    - Uses a76.item_line_series on export lines (discharge=True) to resolve the
+      import serie row (serie_row) or by serial number fallback.
+    """
+    line_ids = [ln.id for ln in lines if ln.id is not None]
+    if not line_ids:
+        return
+
+    # All discharge details belonging to this export invoice, grouped by export line.
+    details: List[DischargeDetail] = (
+        db.query(DischargeDetail)
+        .join(DischargeHeader, DischargeHeader.id == DischargeDetail.discharge_header_id)
+        .filter(
+            DischargeHeader.source_invoice_id == export_invoice.id,
+            DischargeHeader.status == DischargeStatus.APPLIED,
+            DischargeDetail.export_item_line_id.in_(line_ids),
+            DischargeDetail.tenant_id == export_invoice.tenant_id,
+            DischargeDetail.company_id == export_invoice.company_id,
+        )
+        .all()
+    )
+
+    import_lines_by_export: dict[int, set[int]] = {}
+    for d in details:
+        if d.export_item_line_id is None:
+            continue
+        import_lines_by_export.setdefault(d.export_item_line_id, set()).add(d.import_item_line_id)
+
+    if not import_lines_by_export:
+        return
+
+    # Export series marked for discharge (equivalent to SerExpo.Marca = 1)
+    export_series: List[Serie] = (
+        db.execute(
+            select(Serie).where(
+                Serie.line_item_id.in_(line_ids),
+                Serie.discharge == True,  # noqa: E712
+            )
+        )
+        .scalars()
+        .all()
+    )
+
+    for ex_serie in export_series:
+        candidate_import_lines = import_lines_by_export.get(ex_serie.line_item_id, set())
+        if not candidate_import_lines:
+            continue
+
+        import_serie: Optional[Serie] = None
+        # 1) Prefer explicit mapped row from export serie
+        if ex_serie.serie_row is not None:
+            import_serie = db.execute(
+                select(Serie).where(
+                    Serie.line_item_id.in_(candidate_import_lines),
+                    Serie.row == ex_serie.serie_row,
+                )
+            ).scalar_one_or_none()
+
+        # 2) Fallback by serial number if row is absent
+        if import_serie is None and ex_serie.serial_numbers:
+            import_serie = db.execute(
+                select(Serie).where(
+                    Serie.line_item_id.in_(candidate_import_lines),
+                    Serie.serial_numbers == ex_serie.serial_numbers,
+                )
+            ).scalar_one_or_none()
+
+        if import_serie is None:
+            continue
+
+        # Clarion equivalent: SerImp:SerieExpo = 0 / SerDef:SerieExpo = 0
+        import_serie.discharge = False
+
+
+def _cancel_discharge_records(
+    db: Session,
+    export_invoice: InvoiceHeader,
+    cancelled_by: Optional[str] = None,
+) -> None:
+    """
+    Cancels discharge records created by this export invoice in the new
+    specialized tables:
+      - a24.discharge_header: status -> CANCELLED
+      - a24.balance_movement: insert RETURN per discharge_detail row
+
+    This is the ledger-safe equivalent of undoing "Descarga=1" effects.
+    """
+    headers: List[DischargeHeader] = (
+        db.query(DischargeHeader)
+        .filter(
+            DischargeHeader.source_invoice_id == export_invoice.id,
+            DischargeHeader.tenant_id == export_invoice.tenant_id,
+            DischargeHeader.company_id == export_invoice.company_id,
+            DischargeHeader.status.in_([DischargeStatus.APPLIED, DischargeStatus.PARTIAL]),
+        )
+        .all()
+    )
+    if not headers:
+        return
+
+    op_date = (
+        export_invoice.invoice_date.date()
+        if hasattr(export_invoice.invoice_date, "date")
+        else export_invoice.invoice_date
+    )
+
+    for header in headers:
+        details: List[DischargeDetail] = (
+            db.query(DischargeDetail)
+            .filter(
+                DischargeDetail.discharge_header_id == header.id,
+                DischargeDetail.tenant_id == export_invoice.tenant_id,
+                DischargeDetail.company_id == export_invoice.company_id,
+            )
+            .all()
+        )
+
+        for detail in details:
+            # Reverse each consumed lot with a RETURN movement (append-only ledger)
+            ret_mov = BalanceMovement(
+                tenant_id=export_invoice.tenant_id,
+                company_id=export_invoice.company_id,
+                import_invoice_id=detail.import_line.invoice_id if detail.import_line else None,
+                import_item_line_id=detail.import_item_line_id,
+                part_number_id=None,
+                movement_type=MovementType.RETURN,
+                quantity=detail.quantity_discharged or Decimal(0),
+                value_me=detail.value_me,
+                value_mn=detail.value_mn,
+                net_weight=detail.net_weight,
+                source_invoice_id=export_invoice.id,
+                source_item_line_id=detail.export_item_line_id,
+                order_peps=0,  # set post-flush
+                operation_date=op_date,
+                notes=f"Reversa descargo factura {export_invoice.invoice_number} (header {header.id})",
+            )
+            db.add(ret_mov)
+            db.flush()
+            ret_mov.order_peps = ret_mov.id
+
+        header.status = DischargeStatus.CANCELLED
+        header.cancelled_by = cancelled_by
+        header.cancellation_reason = (
+            f"Des-actualización de factura de exportación {export_invoice.invoice_number}"
+        )
+
+
+def _set_invoice_unprocessed(invoice: InvoiceHeader, line_count: int) -> None:
+    """
+    Clarion mapping:
+      EqiFex:ComofueProcesada=''
+      EqiFex:Estatus='NA'
+      EqiFex:Cant_Partidas = Loc:PartidasExpo
+    """
+    invoice.process_log = None
+    invoice.status = InvoiceStatus.REVERSED
+    invoice.party_count = line_count
+
+
+def revert_process(
+    db: Session,
+    invoice: InvoiceHeader,
+    lines: List[LineItem],
+    tenant_id: str,
+    company_id: str,
+    errors: ErrorCollector,
+    cancelled_by: Optional[str] = None,
+) -> list:
+    """
+    Des-actualización de factura de exportación (paridad Clarion).
+
+    Notes:
+    - BEGIN/COMMIT/ROLLBACK SQL explícitos del Clarion se controlan con la
+      transacción de SQLAlchemy en el task (commit/rollback externo).
+    - Las rutinas Clarion invocadas con DO se dejan como TODO por ahora.
+    """
+    _ = (db, tenant_id, company_id)  # reserved for future TO DO implementations
+    sql_errors: list = []
+
+    # INICIALIZA QUEUES (Python: collector ya llega limpio por tarea)
+    # TODO: Compartir QSisGen / parámetros globales del Clarion.
+
+    # TODO: BEGIN TRAN (managed by SQLAlchemy session in task)
+    _todo_check_access_lock(invoice)
+
+    # VERIFICAR SI HAY PARTIDAS DE EXPORTACION
+    line_count = len(lines)
+
+    # VALIDACION DEL CAMBIO DE REGIMEN
+    _validate_regime_change_definitive_invoice_exists(db, invoice, errors)
+    errors.raise_if_errors()
+
+    # VALIDACIONES DE PARTIDAS NORMAL <> REPARACION
+    if (invoice.invoice_type or "").upper() != "NODES":
+        _return_discharged_quantities(db, invoice, lines)
+        _unmark_returned_series(db, invoice, lines)
+        _cancel_discharge_records(db, invoice, cancelled_by)
+
+    _set_invoice_unprocessed(invoice, line_count)
+
+    # TODO: COMMIT/ROLLBACK TRAN + QueueErrorSQL file handling + GBitacora
+    return sql_errors

backend/api/v1/modules/a76/invoices/exports/revert/pre_validators.py

@@ -0,0 +1,42 @@
+from typing import List
+
+from sqlalchemy.orm import Session
+
+from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceStatus
+from api.v1.modules.a76.items.models import LineItem
+from core.exceptions import ErrorCollector
+
+
+def pre_validators(
+    db: Session,
+    invoice: InvoiceHeader,
+    tenant_id: str,
+    company_id: str,
+    errors: ErrorCollector,
+) -> List[LineItem]:
+    """
+    Validaciones previas a la reversión de una factura de importación temporal.
+
+    - Verifica que la factura esté en estatus PROCESSED.
+    - Carga y retorna las partidas asociadas a la factura.
+    """
+    if invoice.status != InvoiceStatus.PROCESSED:
+        errors.add_error(
+            "status",
+            "La factura no fue procesada y no puede ser revertida",
+            solution=["Verifique el estatus de la factura antes de intentar deshacer el proceso"],
+            code="NOT_PROCESSED",
+            value=invoice.status,
+        )
+
+    lines: List[LineItem] = (
+        db.query(LineItem)
+        .filter(
+            LineItem.invoice_id == invoice.id,
+            LineItem.tenant_id == tenant_id,
+            LineItem.company_id == company_id,
+        )
+        .all()
+    )
+
+    return lines

backend/api/v1/modules/a76/invoices/exports/revert/routes.py

@@ -0,0 +1,83 @@
+from typing import Any, Dict
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from sqlalchemy.orm import Session
+
+from core.celery_app import celery_app
+from core.database import get_core_db
+from core.security import get_current_user, validate_access_to_resource
+
+from .task import revert_invoice_task
+
+router = APIRouter()
+
+
+@router.post("/invoices/{invoice_id}/revert")
+def trigger_invoice_revert(
+    invoice_id: int,
+    company_id: int = Query(..., description="Company ID"),
+    db: Session = Depends(get_core_db),
+    current_user: Dict[str, Any] = Depends(get_current_user),
+):
+    """
+    Inicia la des-actualización de una factura de importación temporal como
+    tarea Celery.
+    Retorna el task_id para hacer polling del progreso.
+    """
+    tenant_id = validate_access_to_resource(db, company_id, current_user)
+    cancelled_by = (
+        current_user.get("username")
+        or current_user.get("user_name")
+        or current_user.get("preferred_username")
+        or current_user.get("email")
+        or "system"
+    )
+
+    task = revert_invoice_task.apply_async(
+        args=[invoice_id, str(tenant_id), str(company_id), str(cancelled_by)]
+    )
+
+    return {"task_id": task.id}
+
+
+@router.get("/invoices/revert/{task_id}/status")
+def get_invoice_revert_status(task_id: str):
+    """
+    Consulta el estado de progreso de una tarea de des-actualización de
+    factura.
+
+    Retorna:
+      - state: 'PROCESSING' | 'SUCCESS' | 'FAILURE'
+      - info: { current: int, status: str }  (cuando state == 'PROCESSING')
+      - result: dict                          (cuando state == 'SUCCESS' o 'FAILURE')
+    """
+    task_result = celery_app.AsyncResult(task_id)
+
+    if task_result.state in ("PENDING", "STARTED"):
+        return {
+            "state": "PROCESSING",
+            "info": {"current": 0, "status": "Iniciando..."},
+        }
+
+    if task_result.state == "PROGRESS":
+        return {
+            "state": "PROCESSING",
+            "info": task_result.info or {"current": 0, "status": "Procesando..."},
+        }
+
+    if task_result.state == "SUCCESS":
+        return {
+            "state": "SUCCESS",
+            "result": task_result.result,
+        }
+
+    error_info = task_result.result
+    if isinstance(error_info, Exception):
+        error_msg = str(error_info)
+    else:
+        error_msg = str(error_info) if error_info else "Error desconocido"
+
+    return {
+        "state": "FAILURE",
+        "result": error_msg,
+    }

backend/api/v1/modules/a76/invoices/exports/revert/task.py

@@ -0,0 +1,89 @@
+from celery import Task
+
+from core.celery_app import celery_app
+from core.database import CoreSessionLocal
+from core.exceptions import ErrorCollector, ValidationException
+
+from api.v1.modules.a76.invoices.models import InvoiceHeader
+from .pre_validators import pre_validators
+from .main_process import revert_process
+
+
+def _progress(task: Task, current: int, status: str) -> None:
+    task.update_state(state="PROGRESS", meta={"current": current, "status": status})
+
+
+@celery_app.task(bind=True, name="revert_export_invoice_task")
+def revert_invoice_task(
+    self: Task,
+    invoice_id: int,
+    tenant_id: str,
+    company_id: str,
+    cancelled_by: str | None = None,
+) -> dict:
+    """
+    Des-actualiza una factura de importación temporal ejecutando todas las
+    validaciones y reversiones del proceso principal (revert/main_process) con
+    reporte de progreso.
+    """
+    db = CoreSessionLocal()
+    try:
+        # ── Paso 1: Cargar factura ────────────────────────────────────────────
+        _progress(self, 5, "Cargando factura...")
+        invoice: InvoiceHeader | None = db.get(InvoiceHeader, invoice_id)
+        if invoice is None:
+            return {
+                "status": "error",
+                "message": f"Factura con id {invoice_id} no encontrada.",
+                "errors": [],
+            }
+
+        errors = ErrorCollector()
+
+        # ── Paso 2: Pre-validaciones ──────────────────────────────────────────
+        _progress(self, 10, "Validando estatus de la factura...")
+        lines = pre_validators(db, invoice, tenant_id, company_id, errors)
+        if not lines:
+            errors.add_error(
+                field="line_items",
+                message="La factura no contiene partidas para revertir",
+                solution=["Verifique que la factura tenga partidas antes de intentar revertirla"],
+                code="NO_LINE_ITEMS",
+            )
+        errors.raise_if_errors()
+
+        # ── Paso 3: Validar cantidades y ejecutar reversión ───────────────────
+        _progress(self, 40, "Verificando saldos de partidas...")
+        sql_errors = revert_process(
+            db=db,
+            invoice=invoice,
+            lines=lines,
+            tenant_id=tenant_id,
+            company_id=company_id,
+            errors=errors,
+            cancelled_by=cancelled_by,
+        )
+
+        # ── Paso 4: Confirmar transacción ─────────────────────────────────────
+        _progress(self, 95, "Anulando saldos de inventario y confirmando...")
+        db.flush()
+        db.commit()
+
+        return {
+            "status": "success",
+            "invoice_id": invoice_id,
+            "sql_errors": sql_errors,
+        }
+
+    except ValidationException as exc:
+        db.rollback()
+        return {
+            "status": "validation_error",
+            "message": exc.message,
+            "errors": exc.errors,
+        }
+    except Exception as exc:
+        db.rollback()
+        raise exc
+    finally:
+        db.close()

backend/api/v1/modules/a76/invoices/imports/balance/void_balance_entries.py

@@ -80,8 +80,13 @@ def void_balance_entries(
         (BalanceMovement.movement_type.in_(NEGATIVE_MOVEMENTS), -1),
         else_=1,
     )
-    used_expr = case(
+    # "used" must be NET of returns:
+    #   + consumption/waste/scrap/destruction
+    #   - return
+    # so an already reverted discharge does not block import un-processing.
+    used_net_expr = case(
         (BalanceMovement.movement_type.in_(USED_MOVEMENTS), BalanceMovement.quantity),
+        (BalanceMovement.movement_type == MovementType.RETURN, -BalanceMovement.quantity),
         else_=Decimal(0),
     )
 
@@ -90,7 +95,7 @@ def void_balance_entries(
             select(
                 BalanceMovement.import_item_line_id,
                 func.sum(sign_expr * BalanceMovement.quantity).label("balance"),
-                func.sum(used_expr).label("used"),
+                func.sum(used_net_expr).label("used_net"),
             )
             .where(
                 BalanceMovement.import_item_line_id.in_(import_line_ids),
@@ -100,7 +105,7 @@ def void_balance_entries(
         .all()
     )
 
-    consumed_lots = [row for row in lot_summary if (row.used or 0) > 0]
+    consumed_lots = [row for row in lot_summary if (row.used_net or 0) > 0]
     if consumed_lots:
         lot_ids = ", ".join(str(r.import_item_line_id) for r in consumed_lots)
         raise ValueError(
@@ -109,13 +114,18 @@ def void_balance_entries(
             f"cancelarse primero (item_line ids: {lot_ids})."
         )
 
-    # ── 3. Build the ENTRY_VOID map: one void per ENTRY ──────────────────────
+    # ── 3. Build the ENTRY_VOID map: one void per LOT ────────────────────────
     # Map lot_id → open balance (should equal the original ENTRY qty since no
     # consumptions exist, but we use the actual net balance to be safe).
     balance_map: dict[int, Decimal] = {
         row.import_item_line_id: Decimal(str(row.balance or 0))
         for row in lot_summary
     }
+    # Pick one representative ENTRY per lot to copy informational fields.
+    entry_by_lot: dict[int, BalanceMovement] = {}
+    for e in entries:
+        if e.import_item_line_id not in entry_by_lot:
+            entry_by_lot[e.import_item_line_id] = e
 
     operation_date = (
         invoice.invoice_date.date()
@@ -124,8 +134,8 @@ def void_balance_entries(
     )
 
     voids: List[BalanceMovement] = []
-    for entry in entries:
-        open_qty = balance_map.get(entry.import_item_line_id, Decimal(0))
+    for lot_id, entry in entry_by_lot.items():
+        open_qty = balance_map.get(lot_id, Decimal(0))
         if open_qty <= 0:
             continue
 
@@ -133,7 +143,7 @@ def void_balance_entries(
             tenant_id=invoice.tenant_id,
             company_id=invoice.company_id,
             import_invoice_id=invoice.id,
-            import_item_line_id=entry.import_item_line_id,
+            import_item_line_id=lot_id,
             part_number_id=entry.part_number_id,
             movement_type=MovementType.ENTRY_VOID,
             quantity=open_qty,

backend/api/v1/modules/a76/invoices/imports/process/main_process.py

@@ -150,7 +150,7 @@ def _validate_sisimp_limits(
     pass
 
 
-def _update_invoice_totals(invoice: InvoiceHeader) -> None:
+def _update_invoice_totals(invoice: InvoiceHeader, lines: List[LineItem]) -> None:
     """
     Copia los totales calculados de financials/logistics al encabezado de la factura
     y calcula IVA, incrementables y valores de aduanas.
@@ -281,7 +281,7 @@ def main_process(db: Session, invoice: InvoiceHeader, tenant_id: str, company_id
         )
 
     # Paso 7: Actualizar totales, IVA e incrementables y marcar como procesada
-    _update_invoice_totals(invoice)
+    _update_invoice_totals(invoice, lines)
 
     # Paso 8: Generar saldos en a24.balance_movement (una entrada por partida)
     create_balance_entries(db, invoice, lines)

backend/api/v1/modules/a76/invoices/imports/process/sub_process/assing_values.py

@@ -15,8 +15,8 @@ def assign_values_lines(
     Calculates and assigns unit costs and values in all currency types for every
     line item, based on the invoice currency and exchange rates.
 
-    Also resets inventory counters (quantity_returned, quantity_returned_temp,
-    quantity_existence) to zero, as done in the legacy ASIGNAVALORES_PARTIDA routine.
+    Legacy returned/existence counters were removed from item_line_quantities;
+    balances are now derived from a24.balance_movement and discharges.
 
     Currency mapping (legacy -> current enum):
         ME (moneda extranjera / foreign)  -> Currency.FOREIGN
@@ -53,11 +53,7 @@ def assign_values_lines(
             line.financial.value_mxn = capture * tc_mm * tc * qty
             line.financial.value_mc = capture * qty
 
-        # Reset inventory counters
-        if line.quantity is not None:
-            line.quantity.quantity_returned = Decimal(0)
-            line.quantity.quantity_returned_temp = Decimal(0)
-            line.quantity.quantity_existence = Decimal(0)
+        # NOTE: no legacy returned/existence counters to reset.
 
 
 def assign_values_invoice(

backend/api/v1/modules/a76/invoices/imports/process/task.py

@@ -99,7 +99,7 @@ def process_invoice_task(self: Task, invoice_id: int, tenant_id: str, company_id
                 company_id=company_id,
                 sql_errors=sql_errors,
             )
-        _update_invoice_totals(invoice)
+        _update_invoice_totals(invoice, lines)
 
         # ── Paso 8: Generar saldos en a24.balance_movement ───────────────────
         _progress(self, 98, "Generando saldos de inventario...")

backend/api/v1/modules/a76/invoices/imports/revert/main_process.py

@@ -43,22 +43,12 @@ def _validate_returned_quantities(
     una factura de exportación procesada y se reporta qué factura debe
     desactualizarse primero.
     """
-    lines_with_balance = [
-        line for line in lines
-        if line.quantity is not None and (
-            (line.quantity.quantity_returned_temp or Decimal(0))
-            + (line.quantity.quantity_returned or Decimal(0))
-            + (line.quantity.quantity_existence or Decimal(0))
-        ) != Decimal(0)
-    ]
-
-    if not lines_with_balance:
-        return
-
-    for line in lines_with_balance:
-        qty_ret_temp = line.quantity.quantity_returned_temp or Decimal(0)
-        qty_ret      = line.quantity.quantity_returned      or Decimal(0)
-        qty_exist    = line.quantity.quantity_existence     or Decimal(0)
+    # Importante:
+    # Con la migración a tablas especializadas (discharge_* + balance_movement),
+    # los campos legacy de cantidades retornadas pueden quedar desfasados.
+    # Para bloquear una des-actualización solo debe considerarse descarga ACTIVA
+    # real (DischargeHeader.status=APPLIED y factura fuente procesada).
+    for line in lines:
 
         # Buscar DischargeDetail vinculados a esta partida de importación
         # cuya factura de exportación esté activa (PROCESSED).
@@ -110,22 +100,7 @@ def _validate_returned_quantities(
                     ],
                     code="LINE_HAS_ACTIVE_DISCHARGE",
                 )
-        else:
-            # La partida tiene saldo pero no hay descarga activa rastreable —
-            # reportar el saldo directamente para que el usuario lo investigue.
-            errors.add_error(
-                field=f"line[{line.line_number}].quantities",
-                message=(
-                    f"La Línea: {line.line_number} tiene saldos pendientes "
-                    f"(retornada: {qty_ret}, retornada temp: {qty_ret_temp}, "
-                    f"existencia: {qty_exist}) y no se puede desactualizar."
-                ),
-                solution=[
-                    "Verifique las exportaciones que afectan a esta partida "
-                    "y desactualícelas primero."
-                ],
-                code="LINE_HAS_BALANCE",
-            )
+        # Si no hay discharge activo, NO bloquear por contadores legacy.
 
 
 # ─────────────────────────────────────────────────────────────────────────────
@@ -169,11 +144,6 @@ def _reset_line_quantities(lines: List[LineItem]) -> None:
              ValorIVAMNUsado=0, ValorIVAMEUsado=0 (Clarion SCAII).
     """
     for line in lines:
-        if line.quantity is not None:
-            line.quantity.quantity_returned      = Decimal(0)
-            line.quantity.quantity_returned_temp = Decimal(0)
-            line.quantity.quantity_existence     = Decimal(0)
-
         if line.financial is not None:
             line.financial.value_returned_mxn = Decimal(0)
             line.financial.value_returned_usd = Decimal(0)

backend/api/v1/modules/a76/invoices/imports/revert/routes.py

@@ -7,7 +7,9 @@ from core.celery_app import celery_app
 from core.database import get_core_db
 from core.security import get_current_user, validate_access_to_resource
 
-from .task import revert_invoice_task
+from api.v1.modules.a76.invoices.models import InvoiceHeader, OperationType
+from .task import revert_invoice_task as revert_import_invoice_task
+from ...exports.revert.task import revert_invoice_task as revert_export_invoice_task
 
 router = APIRouter()
 
@@ -20,16 +22,32 @@ def trigger_invoice_revert(
     current_user: Dict[str, Any] = Depends(get_current_user),
 ):
     """
-    Inicia la des-actualización de una factura de importación temporal como
-    tarea Celery.
-    Retorna el task_id para hacer polling del progreso.
+    Endpoint unificado para des-actualizar facturas.
+    - operation_type=imp -> task de importación
+    - operation_type=exp -> task de exportación
     """
     tenant_id = validate_access_to_resource(db, company_id, current_user)
-
-    task = revert_invoice_task.apply_async(
-        args=[invoice_id, str(tenant_id), str(company_id)]
+    cancelled_by = (
+        current_user.get("username")
+        or current_user.get("user_name")
+        or current_user.get("preferred_username")
+        or current_user.get("email")
+        or "system"
     )
 
+    invoice: InvoiceHeader | None = db.get(InvoiceHeader, invoice_id)
+    if invoice is None:
+        raise HTTPException(status_code=404, detail=f"Factura {invoice_id} no encontrada.")
+
+    if invoice.operation_type == OperationType.EXP:
+        task = revert_export_invoice_task.apply_async(
+            args=[invoice_id, str(tenant_id), str(company_id), str(cancelled_by)]
+        )
+    else:
+        task = revert_import_invoice_task.apply_async(
+            args=[invoice_id, str(tenant_id), str(company_id), str(cancelled_by)]
+        )
+
     return {"task_id": task.id}
 
 

backend/api/v1/modules/a76/invoices/imports/revert/task.py

@@ -13,8 +13,14 @@ def _progress(task: Task, current: int, status: str) -> None:
     task.update_state(state="PROGRESS", meta={"current": current, "status": status})
 
 
-@celery_app.task(bind=True, name="revert_invoice_task")
-def revert_invoice_task(self: Task, invoice_id: int, tenant_id: str, company_id: str) -> dict:
+@celery_app.task(bind=True, name="revert_import_invoice_task")
+def revert_invoice_task(
+    self: Task,
+    invoice_id: int,
+    tenant_id: str,
+    company_id: str,
+    cancelled_by: str | None = None,
+) -> dict:
     """
     Des-actualiza una factura de importación temporal ejecutando todas las
     validaciones y reversiones del proceso principal (revert/main_process) con

backend/api/v1/modules/a76/items/line_quantities/models.py

@@ -30,9 +30,6 @@ class LineQuantity(Base):
     
     # Quantities - Special (SCAF specific)
     quantity_temp_export: Mapped[Optional[Decimal]] = mapped_column(Numeric(19, 8))  # CANTEXPOTEMP
-    quantity_existence: Mapped[Optional[Decimal]] = mapped_column(Numeric(19, 8))  # CANTEXISTENCIA
-    quantity_returned: Mapped[Optional[Decimal]] = mapped_column(Numeric(19, 8))  # CANTRETORNADA
-    quantity_returned_temp: Mapped[Optional[Decimal]] = mapped_column(Numeric(19, 8))  # CANTRETORNADATEMP
     serial_count: Mapped[Optional[int]] = mapped_column(Integer)  # CANT_SERIES/CANT_SERIESDEF
     
     # Weight

backend/api/v1/modules/a76/items/line_quantities/schemas.py

@@ -16,9 +16,6 @@ class LineQuantityBase(BaseModel):
     
     # Quantities - Special (SCAF specific)
     quantity_temp_export: Optional[Decimal] = Field(None, description="Temporary export quantity (CANTEXPOTEMP)")
-    quantity_existence: Optional[Decimal] = Field(None, description="Existence quantity (CANTEXISTENCIA)")
-    quantity_returned: Optional[Decimal] = Field(None, description="Returned quantity (CANTRETORNADA)")
-    quantity_returned_temp: Optional[Decimal] = Field(None, description="Returned temporary quantity (CANTRETORNADATEMP)")
     serial_count: Optional[int] = Field(None, description="Serial count (CANT_SERIES/CANT_SERIESDEF)")
     
     # Weight

backend/api/v1/modules/a76/items/service.py

@@ -44,6 +44,7 @@ from api.v1.modules.a76.invoices.models import InvoiceHeader
 from api.v1.modules.a76.parts.models import Part
 from api.v1.modules.a76.general_catalogs.identifiers.models import IdentifierDetail
 from api.v1.modules.a24.balance_movements.models import BalanceMovement, NEGATIVE_MOVEMENTS
+from api.v1.modules.a24.discharges.models import DischargeDetail, DischargeHeader, DischargeStatus
 
 logger = logging.getLogger(__name__)
 
@@ -733,12 +734,19 @@ class ItemService:
         )
 
         result = []
+        used_map = ItemService._used_quantities_by_procedure(
+            db=db,
+            import_line_ids=[line.id for line in lines],
+            as_of_date=as_of_date,
+        )
         for line in lines:
             available_balance = ItemService._compute_balance(db, line.id, as_of_date)
             qty = line.quantity
             desc = line.description
             fa = line.fa_data
             inv = line.invoice
+            qty_used_temp = used_map.get((line.id, "TEM"), Decimal(0))
+            qty_used_def = used_map.get((line.id, "DEF"), Decimal(0))
 
             # Count subitems (lines that reference this line as parent via subitem_number)
             subitem_count = 0
@@ -770,8 +778,8 @@ class ItemService:
                 "unit_of_measure_code": line.unit_of_measure_info.code if line.unit_of_measure_info else None,
                 # Quantities
                 "quantity": float(qty.quantity) if qty and qty.quantity is not None else None,
-                "quantity_returned_temp": float(qty.quantity_returned_temp) if qty and qty.quantity_returned_temp is not None else None,
-                "quantity_returned": float(qty.quantity_returned) if qty and qty.quantity_returned is not None else None,
+                "quantity_used_temp": float(qty_used_temp),
+                "quantity_used_def": float(qty_used_def),
                 # Balance
                 "available_balance": float(available_balance),
                 "has_balance": available_balance > Decimal(0),
@@ -811,3 +819,39 @@ class ItemService:
             )
         ).scalar()
         return Decimal(str(result or 0))
+
+    @staticmethod
+    def _used_quantities_by_procedure(
+        db: Session,
+        import_line_ids: List[int],
+        as_of_date: Optional[datetime.date],
+    ) -> dict[tuple[int, str], Decimal]:
+        """
+        Returns net used quantity by import line and procedence (TEM/DEF),
+        based on active discharge records only (new ledger logic).
+        """
+        if not import_line_ids:
+            return {}
+
+        query = (
+            select(
+                DischargeDetail.import_item_line_id,
+                DischargeDetail.procedence,
+                func.sum(DischargeDetail.quantity_discharged),
+            )
+            .join(DischargeHeader, DischargeHeader.id == DischargeDetail.discharge_header_id)
+            .where(
+                DischargeDetail.import_item_line_id.in_(import_line_ids),
+                DischargeHeader.status == DischargeStatus.APPLIED,
+                DischargeDetail.procedence.in_(["TEM", "DEF"]),
+            )
+            .group_by(DischargeDetail.import_item_line_id, DischargeDetail.procedence)
+        )
+        if as_of_date is not None:
+            query = query.where(DischargeHeader.discharge_date <= as_of_date)
+
+        rows = db.execute(query).all()
+        out: dict[tuple[int, str], Decimal] = {}
+        for line_id, procedence, qty in rows:
+            out[(int(line_id), str(procedence))] = Decimal(str(qty or 0))
+        return out

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

@@ -228,6 +228,7 @@ BASE_SELECT = """
     COALESCE(cl.material_key,'')            AS "C42",
     CONCAT(ped.year,'-',ped.customs_office,'-',ped.license,'-',ped.pedimento_number) AS "C43",
     COALESCE(ptc.payment_date_code, 'P')    AS "C44",
+    COALESCE(lm.last_movement_type, '')     AS "C_last_movement_type",
     COALESCE(ilc.octave_fraction,'')        AS "C45",
     ''                                      AS "C47",
     COALESCE(ped.pedimento_code,'')         AS "C48",
@@ -256,19 +257,63 @@ BASE_JOINS = """
     LEFT JOIN (
         SELECT DISTINCT ON (import_item_line_id)
             import_item_line_id,
-            quantity AS qty_impo,
-            value_me AS val_me_impo,
-            value_mn AS val_mn_impo
+            movement_type AS last_movement_type
         FROM a24.balance_movement
-        WHERE movement_type = 'entry' AND tenant_id = :tenant_id
+        WHERE tenant_id = :tenant_id
         ORDER BY import_item_line_id, id DESC
+    ) lm ON lm.import_item_line_id = il.id
+    LEFT JOIN (
+        SELECT
+            import_item_line_id,
+            SUM(
+                CASE
+                    WHEN movement_type = 'entry' THEN quantity
+                    WHEN movement_type = 'entry_void' THEN -1 * quantity
+                    ELSE 0
+                END
+            ) AS qty_impo,
+            SUM(
+                CASE
+                    WHEN movement_type = 'entry' THEN COALESCE(value_me, 0)
+                    WHEN movement_type = 'entry_void' THEN -1 * COALESCE(value_me, 0)
+                    ELSE 0
+                END
+            ) AS val_me_impo,
+            SUM(
+                CASE
+                    WHEN movement_type = 'entry' THEN COALESCE(value_mn, 0)
+                    WHEN movement_type = 'entry_void' THEN -1 * COALESCE(value_mn, 0)
+                    ELSE 0
+                END
+            ) AS val_mn_impo
+        FROM a24.balance_movement
+        WHERE tenant_id = :tenant_id
+        GROUP BY import_item_line_id
     ) ent ON ent.import_item_line_id = il.id
     LEFT JOIN (
         SELECT 
             import_item_line_id,
-            SUM(CASE WHEN movement_type IN ('consumption', 'waste', 'scrap', 'destruction') THEN quantity ELSE 0 END) as qty_used,
-            SUM(CASE WHEN movement_type IN ('consumption', 'waste', 'scrap', 'destruction') THEN value_me ELSE 0 END) as val_me_used,
-            SUM(CASE WHEN movement_type IN ('consumption', 'waste', 'scrap', 'destruction') THEN value_mn ELSE 0 END) as val_mn_used,
+            SUM(
+                CASE
+                    WHEN movement_type IN ('consumption', 'waste', 'scrap', 'destruction') THEN quantity
+                    WHEN movement_type = 'return' THEN -1 * quantity
+                    ELSE 0
+                END
+            ) as qty_used,
+            SUM(
+                CASE
+                    WHEN movement_type IN ('consumption', 'waste', 'scrap', 'destruction') THEN COALESCE(value_me, 0)
+                    WHEN movement_type = 'return' THEN -1 * COALESCE(value_me, 0)
+                    ELSE 0
+                END
+            ) as val_me_used,
+            SUM(
+                CASE
+                    WHEN movement_type IN ('consumption', 'waste', 'scrap', 'destruction') THEN COALESCE(value_mn, 0)
+                    WHEN movement_type = 'return' THEN -1 * COALESCE(value_mn, 0)
+                    ELSE 0
+                END
+            ) as val_mn_used,
             SUM(CASE WHEN movement_type IN ('consumption', 'waste', 'scrap', 'destruction', 'neg_adj', 'transfer_out', 'expiration', 'regime_chg_out', 'entry_void') 
                      THEN -1 * quantity ELSE quantity END) as qty_balance
         FROM a24.balance_movement
@@ -462,6 +507,10 @@ def _build_row(
     cant_used  = _d(row.get("C18"))
     cant_saldo = _d(row.get("C_balance"))
 
+    # Mostrar saldo 0, excepto lotes anulados (último movimiento ENTRY_VOID).
+    if str(row.get("C_last_movement_type") or "").lower() == "entry_void":
+        return None
+
     if filters.omit_low_balance and cant_saldo <= Decimal(0):
         return None