feature

backend/api/v1/modules/a76/layouts_csv/exportacion/routes.py

@@ -62,7 +62,7 @@ async def upload_import_file(
         "user_id": current_user.get("id"),
         "footer_config": footer_config,
         "operation_type": operation_type or "exp",
-        "template_id": template_id or ("exp_def_header" if model_target == "invoice_header" else "exp_def_details"),
+        "template_id": template_id or ("exp_def_header" if model_target == "invoice_header" else "exp_def_partidas"),
     }
 
     try:

backend/api/v1/modules/a76/layouts_csv/exportacion/tasks.py

@@ -38,7 +38,7 @@ def _norm_row(row: Dict[str, Any], template_id: str) -> Dict[str, Any]:
 def scan_file(self, job_id: str, model_target: str, config: str = None):
     """
     Para invoice_header: delega en facturas._do_scan_file con storage "exp" (validaciones FK, encabezados_expo).
-    Para invoice_details: scan sin validaciones (stub).
+    Para invoice_details: delega en facturas._do_scan_file con template exp_def_partidas (validaciones partidas expo).
     """
     logger.info("Exportación import: starting scan for job %s target %s", job_id, model_target)
 
@@ -46,45 +46,37 @@ def scan_file(self, job_id: str, model_target: str, config: str = None):
         from api.v1.modules.a76.layouts_csv.facturas.tasks import _do_scan_file
         return _do_scan_file(job_id, "invoice_header", config, job_type_override="exp")
 
-    # invoice_details: stub sin validaciones
+    if model_target == "invoice_details":
+        from api.v1.modules.a76.layouts_csv.facturas.tasks import _do_scan_file
+        return _do_scan_file(job_id, "invoice_details", config, job_type_override="exp")
+
+    # Fallback (e.g. unknown model_target)
     file_path = _ensure_file(job_id)
     if not file_path:
         return {"status": "failed", "error": "Archivo no encontrado (expirado o no subido). Sube de nuevo."}
     if os.path.getsize(file_path) == 0:
         return {"status": "failed", "error": "El archivo está vacío."}
     _ensure_meta(job_id, file_path)
-
     try:
         common_meta.require_tenant_context(file_path)
     except ValueError as e:
         return {"status": "failed", "error": str(e)}
-
     meta = common_meta.load_meta(file_path) or {}
-    template_id = meta.get("template_id") or (
-        "exp_def_header" if model_target == "invoice_header" else "exp_def_details"
-    )
-
-    total_rows = 0
-    processed_rows = 0
-
+    template_id = meta.get("template_id") or "exp_def_details"
     try:
         total_rows = common_csv_reader.count_csv_rows(file_path, has_header=True)
     except Exception as e:
         return {"status": "failed", "error": str(e)}
-
-    def on_progress(current: int, total: int) -> None:
-        self.update_state(state="PROGRESS", meta={"current": current, "total": total})
-
+    processed_rows = 0
     try:
         for i, row in common_csv_reader.iter_csv_rows(file_path, fieldnames=None):
             if i % 500 == 0:
-                on_progress(i, total_rows)
+                self.update_state(state="PROGRESS", meta={"current": i, "total": total_rows})
             _norm_row(row, template_id)
             processed_rows += 1
     except Exception as e:
         logger.error("Exportación import scan failed: %s", e)
         return {"status": "failed", "error": str(e)}
-
     return common_responses.scan_result(job_id, processed_rows, 0, [])
 
 
@@ -92,7 +84,7 @@ def scan_file(self, job_id: str, model_target: str, config: str = None):
 def insert_valid_rows(self, job_id: str, model_target: str):
     """
     Commit: para invoice_header delega en facturas (inserción real en BD con storage "exp").
-    Para invoice_details mantiene stub (sin inserción).
+    Para invoice_details delega en facturas (inserción partidas expo cuando esté implementada; mientras tanto mismo flujo).
     """
     logger.info("Exportación import: starting commit for job %s target %s", job_id, model_target)
 
@@ -100,7 +92,11 @@ def insert_valid_rows(self, job_id: str, model_target: str):
         from api.v1.modules.a76.layouts_csv.facturas.tasks import _do_insert_valid_rows
         return _do_insert_valid_rows(job_id, "invoice_header", job_type_override="exp")
 
-    # invoice_details: stub (sin inserción en BD)
+    if model_target == "invoice_details":
+        from api.v1.modules.a76.layouts_csv.facturas.tasks import _do_insert_valid_rows
+        return _do_insert_valid_rows(job_id, "invoice_details", job_type_override="exp")
+
+    # Fallback: stub sin inserción
     file_path = _ensure_file(job_id)
     if not file_path:
         alt_path = common_storage.file_path_for_job(JOB_TYPE, job_id)
@@ -121,7 +117,7 @@ def insert_valid_rows(self, job_id: str, model_target: str):
     error_lines = common_storage.get_error_lines(JOB_TYPE, job_id, error_path)
 
     template_id = meta.get("template_id") or (
-        "exp_def_header" if model_target == "invoice_header" else "exp_def_details"
+        "exp_def_header" if model_target == "invoice_header" else "exp_def_partidas"
     )
 
     try:

backend/api/v1/modules/a76/layouts_csv/exportacion/template_config.py

@@ -57,6 +57,36 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
         {"canonical": "FRACCION"},
         {"canonical": "ORDEN DE COMPRA", "aliases": ["ORDENCOMPRA"]},
     ],
+    # Partidas exportación definitiva (Clarion EstructuraParExpoCamReg: A–V)
+    "exp_def_partidas": [
+        {"canonical": "NUMERO FACTURA EXPO", "aliases": ["NUMERO FACTURA EXPO.", "NUM FACTURA EXPO", "FACTURA EXPO"]},
+        {"canonical": "LINEA EXPO", "aliases": ["LINEA EXPO.", "RENGLON EXPO"]},
+        {"canonical": "TIPO DE IMPO", "aliases": ["TIPO DE IMPO.", "TIPO IMPO", "PROCEDENCIA"]},
+        {"canonical": "FACTURA IMPO", "aliases": ["FACTURA IMPO.", "FACTURA IMPORTACION"]},
+        {"canonical": "LINEA IMPO", "aliases": ["LINEA IMPO.", "LINEA IMPORTACION"]},
+        {"canonical": "GENERA DESCARGA", "aliases": ["GENERA DESCARGA?", "DESCARGA"]},
+        {"canonical": "CANTIDAD EXPORTADA/DESCARGAR", "aliases": ["CANTIDAD EXPORTADA", "CANTIDAD EXPORTADA/DESCARGAR", "CANTIDAD"]},
+        {"canonical": "UNIDAD DE MEDIDA", "aliases": ["U.M.", "UNIDAD MEDIDA"]},
+        {"canonical": "COSTO UNITARIO", "aliases": ["COSTOUNITARIO"]},
+        {"canonical": "PESO NETO", "aliases": ["PESONETO"]},
+        {"canonical": "PESO BRUTO", "aliases": ["PESOBRUTO"]},
+        {"canonical": "SE PAGO IMPUESTO", "aliases": ["SE PAGO IMPUESTO? (SI o NO)", "SEPAGOIMPUESTO"]},
+        {"canonical": "FORMA DE PAGO", "aliases": ["FORMADEPAGO", "FORMA PAGO"]},
+        {"canonical": "DESCRIPCION EXTRA", "aliases": ["DESCRIPCION EXTRA", "DESCRIPCIONEXTRA"]},
+        {"canonical": "INFORMACION ADICIONAL", "aliases": ["INFORMACION ADICIONAL", "INFORMACIONADICIONAL"]},
+        {"canonical": "AGREGAR(A)/SUSTITUIR(S)", "aliases": ["AGREGAR(A)/SUSTITUIR(S)", "AGREGAR/SUSTITUIR", "SUSTITUIR"]},
+        {"canonical": "LOTE"},
+        {"canonical": "NUMERO ENTRADA", "aliases": ["NUMERO ENTRADA", "NUM ENTRADA"]},
+        {"canonical": "ES PARTIDA/SUBPARTIDA", "aliases": ["ES PARTIDA/SUBPARTIDA", "ESSUBPARTIDA", "ES PARTIDA O SUBPARTIDA"]},
+        {"canonical": "LINEA PRINCIPAL", "aliases": ["LINEAPRINCIPAL", "PARTIDA PRINCIPAL"]},
+        {"canonical": "FRACCION AMERICANA", "aliases": ["FRACCION AMERICANA", "FRACCIONAMERICANA"]},
+        {"canonical": "FRACCION ARANCELARIA", "aliases": ["FRACCION ARANCELARIA", "FRACCIONARANCELARIA"]},
+        {"canonical": "CANTIDAD BULTOS", "aliases": ["CANTIDADBULTOS"]},
+        {"canonical": "CLAVE BULTOS", "aliases": ["CLAVEBULTOS"]},
+        {"canonical": "PAIS ORIGEN", "aliases": ["PAISORIGEN", "PAIS"]},
+        {"canonical": "NUM. PARTE", "aliases": ["NUMPARTE", "NUMERO PARTE", "NUM PARTE"]},
+        {"canonical": "ORDEN DE COMPRA", "aliases": ["ORDENCOMPRA", "ORDEN DE VENTA"]},
+    ],
 }
 
 

backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py

@@ -222,9 +222,11 @@ def _do_scan_file(job_id: str, model_target: str, config: Optional[str] = None,
         "imp_temp_header" if model_target == "invoice_header" else
         "imp_temp_details" if model_target == "invoice_details" else "imp_temp_series"
     )
-    # Cuando el scan viene de Exportación (job_type_override "exp"), forzar exp_def_header para que corran las validaciones FK en el escaneo
+    # Cuando el scan viene de Exportación (job_type_override "exp"), forzar exp_def_header o exp_def_partidas
     if job_type_override == "exp" and model_target == "invoice_header":
         template_id = "exp_def_header"
+    if job_type_override == "exp" and model_target == "invoice_details":
+        template_id = "exp_def_partidas"
     inv_type_value = normalize_public_code(footer_config.get("invoice_type") or meta.get("invoice_type") or "TEM")
     if not inv_type_value:
         inv_type_value = "TEM"
@@ -1202,6 +1204,269 @@ def _do_scan_file(job_id: str, model_target: str, config: Optional[str] = None,
             logger.exception("Partidas importación definitiva scan failed: %s", e)
             return {"status": "failed", "error": str(e)}
 
+    # --- Partidas Exportación Definitiva: Clarion VALIDA_TODA_PAR_EXPO / VALIDA_PARCIAL_PAR_EXPO / VALIDACIONES_PAR_EXPO ---
+    if model_target == "invoice_details" and template_id == "exp_def_partidas":
+        try:
+            from api.v1.modules.a76.invoices.models import InvoiceHeader
+            from api.v1.modules.a76.items.models import LineItem
+            from api.v1.modules.a76.general_catalogs.units_of_measure.models import UnitOfMeasure
+            from api.v1.modules.a76.general_catalogs.packages.models import Package
+            from api.v1.modules.public.reference_data.payment_methods.models import PaymentMethod
+            from api.v1.modules.a76.general_catalogs.fractions.us_tariff_fractions.models import USTariffFraction
+            from api.v1.modules.a76.general_catalogs.company.models import Company
+            from api.v1.modules.a76.parts.models import Part
+            from .validators.partidas_expo import validate_row_partidas_expo
+
+            _fc = parse_footer_config(meta.get("footer_config"))
+            autonumerar = meta.get("autonumerar", True)
+            actualizar = meta.get("actualizar", False)
+            levantar_subpartidas = meta.get("levantar_subpartidas", False)
+            validar_decimales_pza = meta.get("validar_decimales_pza", False)
+            if _fc:
+                if "autonumerar" in _fc:
+                    autonumerar = bool(_fc["autonumerar"])
+                elif _fc.get("autonumber_partidas", "true") is not None:
+                    autonumerar = str(_fc.get("autonumber_partidas", "true")).lower() in ("true", "1", "si", "sí", "yes")
+                if "actualizar" in _fc:
+                    actualizar = bool(_fc["actualizar"])
+                if "levantar_subpartidas" in _fc:
+                    levantar_subpartidas = bool(_fc["levantar_subpartidas"])
+                if "validar_decimales_pza" in _fc:
+                    validar_decimales_pza = bool(_fc["validar_decimales_pza"])
+
+            RFC_EXCEPTION_UPDATED = set()
+            RFC_EXCEPTION_EGM = {"EGM0303257J1"}
+
+            with CoreSessionLocal() as session:
+                company = session.query(Company).filter(Company.id == company_id).first()
+                company_rfc = (company.rfc or "").strip().upper() if company else ""
+                if company_rfc in RFC_EXCEPTION_EGM:
+                    rfc_exception_egm = True
+                else:
+                    rfc_exception_egm = False
+
+                q_inv_expo = (
+                    session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.is_updated)
+                    .filter(
+                        InvoiceHeader.tenant_id == tenant_id,
+                        InvoiceHeader.company_id == company_id,
+                        InvoiceHeader.operation_type == "exp",
+                    )
+                )
+                invoice_id_by_number: Dict[str, int] = {}
+                invoice_updated_by_number: Dict[str, bool] = {}
+                for num, iid, is_upd in q_inv_expo.all():
+                    if num:
+                        invoice_id_by_number[str(num).strip()] = iid
+                        invoice_updated_by_number[str(num).strip()] = bool(is_upd)
+
+                existing_line_keys_by_invoice: Dict[str, Set[str]] = {}
+                q_li_expo = (
+                    session.query(InvoiceHeader.invoice_number, LineItem.line_number)
+                    .join(LineItem, LineItem.invoice_id == InvoiceHeader.id)
+                    .filter(
+                        InvoiceHeader.tenant_id == tenant_id,
+                        InvoiceHeader.company_id == company_id,
+                        InvoiceHeader.operation_type == "exp",
+                    )
+                )
+                for num, ln in q_li_expo.all():
+                    if num is not None:
+                        key = str(num).strip()
+                        if key not in existing_line_keys_by_invoice:
+                            existing_line_keys_by_invoice[key] = set()
+                        existing_line_keys_by_invoice[key].add(str(ln).strip())
+
+                partidas_principales_bd: Set[Tuple[str, str]] = set()
+                try:
+                    from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem
+                    q_pp = (
+                        session.query(InvoiceHeader.invoice_number, LineItem.line_number)
+                        .join(LineItem, LineItem.invoice_id == InvoiceHeader.id)
+                        .join(FaLineItem, FaLineItem.id == LineItem.id)
+                        .filter(
+                            InvoiceHeader.tenant_id == tenant_id,
+                            InvoiceHeader.company_id == company_id,
+                            InvoiceHeader.operation_type == "exp",
+                            FaLineItem.is_subitem == False,
+                            FaLineItem.contains_subitems == True,
+                        )
+                    )
+                    for num, ln in q_pp.all():
+                        if num is not None:
+                            partidas_principales_bd.add((str(num).strip(), str(ln).strip()))
+                except Exception:
+                    pass
+
+                factura_impo_tem_by_number: Dict[str, int] = {}
+                q_tem = session.query(InvoiceHeader.invoice_number, InvoiceHeader.id).filter(
+                    InvoiceHeader.tenant_id == tenant_id,
+                    InvoiceHeader.company_id == company_id,
+                    InvoiceHeader.operation_type == "imp",
+                    InvoiceHeader.invoice_type == "TEM",
+                )
+                for num, iid in q_tem.all():
+                    if num:
+                        factura_impo_tem_by_number[str(num).strip()] = iid
+
+                factura_impo_def_by_number: Dict[str, int] = {}
+                DEF_INVOICE_TYPES = ("DEF", "MATDE", "EXDEF")
+                q_def = session.query(InvoiceHeader.invoice_number, InvoiceHeader.id).filter(
+                    InvoiceHeader.tenant_id == tenant_id,
+                    InvoiceHeader.company_id == company_id,
+                    InvoiceHeader.operation_type == "imp",
+                    InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES),
+                )
+                for num, iid in q_def.all():
+                    if num:
+                        factura_impo_def_by_number[str(num).strip()] = iid
+
+                line_exists_tem: Set[Tuple[int, str]] = set()
+                q_li_tem = (
+                    session.query(LineItem.invoice_id, LineItem.line_number)
+                    .join(InvoiceHeader, InvoiceHeader.id == LineItem.invoice_id)
+                    .filter(
+                        InvoiceHeader.tenant_id == tenant_id,
+                        InvoiceHeader.company_id == company_id,
+                        InvoiceHeader.operation_type == "imp",
+                        InvoiceHeader.invoice_type == "TEM",
+                    )
+                )
+                for inv_id, ln in q_li_tem.all():
+                    if inv_id is not None and ln is not None:
+                        line_exists_tem.add((inv_id, str(ln).strip()))
+
+                line_exists_def: Set[Tuple[int, str]] = set()
+                q_li_def = (
+                    session.query(LineItem.invoice_id, LineItem.line_number)
+                    .join(InvoiceHeader, InvoiceHeader.id == LineItem.invoice_id)
+                    .filter(
+                        InvoiceHeader.tenant_id == tenant_id,
+                        InvoiceHeader.company_id == company_id,
+                        InvoiceHeader.operation_type == "imp",
+                        InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES),
+                    )
+                )
+                for inv_id, ln in q_li_def.all():
+                    if inv_id is not None and ln is not None:
+                        line_exists_def.add((inv_id, str(ln).strip()))
+
+                valid_uom_codes: Set[str] = set()
+                for u in session.query(UnitOfMeasure.code).filter(UnitOfMeasure.tenant_id == tenant_id, UnitOfMeasure.company_id == company_id).all():
+                    if u[0]:
+                        valid_uom_codes.add((u[0] or "").strip().upper())
+
+                valid_bulks_codes: Set[str] = set()
+                for p in session.query(Package.key).filter(Package.tenant_id == tenant_id, Package.company_id == company_id).all():
+                    if p[0]:
+                        valid_bulks_codes.add((p[0] or "").strip())
+
+                valid_payment_methods: Set[str] = set()
+                for row in session.query(PaymentMethod.key).all():
+                    if row[0] is not None:
+                        valid_payment_methods.add(str(row[0]).strip())
+
+                valid_fraction_ame: Set[str] = set()
+                for row in session.query(USTariffFraction.code).filter(USTariffFraction.tenant_id == tenant_id, USTariffFraction.company_id == company_id).all():
+                    if row[0]:
+                        valid_fraction_ame.add((row[0] or "").strip())
+
+                valid_part_numbers: Set[str] = set()
+                for row in session.query(Part.part_number).filter(Part.tenant_id == tenant_id, Part.company_id == company_id).all():
+                    if row[0]:
+                        valid_part_numbers.add((row[0] or "").strip().upper())
+
+            invoice_numbers_from_csv = set()
+            with open(file_path, "r", encoding="utf-8-sig") as f_in:
+                sample = f_in.read(2048)
+                f_in.seek(0)
+                try:
+                    dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
+                except Exception:
+                    dialect = "excel"
+                reader = csv.DictReader(f_in, dialect=dialect)
+                rows_list = list(reader)
+
+            for row in rows_list:
+                row_norm = row_from_template(row, "exp_def_partidas", normalize_header)
+                inv = (row_norm.get("NUMERO FACTURA EXPO") or row_norm.get("NUMERO FACTURA EXPO.") or row_norm.get("FACTURA EXPO") or "").strip()
+                if inv:
+                    invoice_numbers_from_csv.add(inv)
+            if company_rfc in RFC_EXCEPTION_EGM:
+                rfc_exception_updated = invoice_numbers_from_csv
+            else:
+                rfc_exception_updated = set()
+
+            line_counts_csv: Dict[Tuple[str, str], int] = {}
+            partidas_principales_csv: Set[Tuple[str, str]] = set()
+
+            def _get_row_expo(row_norm: Dict[str, Any], *keys: str) -> str:
+                for k in keys:
+                    v = row_norm.get(k)
+                    if v is not None and str(v).strip():
+                        return str(v).strip()
+                return ""
+
+            for row in rows_list:
+                row_norm = row_from_template(row, "exp_def_partidas", normalize_header)
+                inv = _get_row_expo(row_norm, "NUMERO FACTURA EXPO", "NUMERO FACTURA EXPO.", "FACTURA EXPO")
+                linea = _get_row_expo(row_norm, "LINEA EXPO", "LINEA EXPO.", "RENGLON EXPO")
+                if inv and linea:
+                    key = (inv, linea)
+                    line_counts_csv[key] = line_counts_csv.get(key, 0) + 1
+                u = _get_row_expo(row_norm, "ES PARTIDA/SUBPARTIDA", "ESSUBPARTIDA", "ES PARTIDA O SUBPARTIDA").upper()
+                if u == "P" and inv and linea:
+                    partidas_principales_csv.add((inv, linea))
+
+            error_count = 0
+            processed_rows = 0
+            error_lines_list = []
+            errors_detail = []
+
+            with open(error_path, "w", encoding="utf-8") as f_err:
+                for i, row in enumerate(rows_list, start=1):
+                    if i % 1000 == 0:
+                        self.update_state(state="PROGRESS", meta={"current": i, "total": total_rows, "errors": error_count})
+                    row_norm = row_from_template(row, "exp_def_partidas", normalize_header)
+                    err = validate_row_partidas_expo(
+                        row_norm,
+                        i,
+                        autonumerar=autonumerar,
+                        actualizar=actualizar,
+                        levantar_subpartidas=levantar_subpartidas,
+                        invoice_id_by_number=invoice_id_by_number,
+                        invoice_updated_by_number=invoice_updated_by_number,
+                        rfc_exception_updated=rfc_exception_updated,
+                        existing_line_keys_by_invoice=existing_line_keys_by_invoice,
+                        line_counts_csv=line_counts_csv,
+                        partidas_principales_csv=partidas_principales_csv,
+                        partidas_principales_bd=partidas_principales_bd,
+                        valid_uom_codes=valid_uom_codes,
+                        valid_bulks_codes=valid_bulks_codes,
+                        valid_payment_methods=valid_payment_methods,
+                        valid_fraction_ame=valid_fraction_ame,
+                        valid_part_numbers=valid_part_numbers,
+                        factura_impo_tem_by_number=factura_impo_tem_by_number,
+                        factura_impo_def_by_number=factura_impo_def_by_number,
+                        line_exists_tem=line_exists_tem,
+                        line_exists_def=line_exists_def,
+                        rfc_exception_egm=rfc_exception_egm,
+                        validar_decimales_pza=validar_decimales_pza,
+                    )
+                    if err:
+                        error_count += 1
+                        error_lines_list.append(err["line"])
+                        f_err.write(json.dumps({"line": err["line"], "col": err.get("col", ""), "msg": err.get("msg", "")}) + "\n")
+                        if len(errors_detail) < 500:
+                            errors_detail.append({"line": err["line"], "col": err.get("col", ""), "msg": err.get("msg", "")})
+                    processed_rows += 1
+
+            common_storage.store_error_lines(effective_job_type, job_id, error_lines_list)
+            return common_responses.scan_result(job_id, processed_rows, error_count, errors_detail)
+        except Exception as e:
+            logger.exception("Partidas exportación definitiva scan failed: %s", e)
+            return {"status": "failed", "error": str(e)}
+
     # --- Partidas Compras Mexicanas: misma lógica que Impo Def, facturas MEX ---
     if model_target == "invoice_details" and template_id == "cmex_details":
         try:
@@ -3308,6 +3573,24 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
     error_path = common_storage.error_path_for_job(effective_job_type, job_id)
     error_lines = common_storage.get_error_lines(effective_job_type, job_id, error_path)
 
+    # Partidas exportación definitiva: inserción pendiente (Phase 2); solo cleanup y respuesta
+    if model_target == "invoice_details" and meta.get("template_id") == "exp_def_partidas":
+        common_storage.cleanup_import_job(
+            effective_job_type, job_id,
+            file_path=file_path,
+            error_path=error_path,
+            meta_path=meta_path,
+        )
+        return {
+            "status": "finished",
+            "inserted": 0,
+            "skipped_invalid": len(error_lines),
+            "skipped_missing_fk": 0,
+            "skipped_duplicate": 0,
+            "skipped_details": [],
+            "message": "Validación de partidas de exportación completada. Inserción en BD pendiente de implementación (Phase 2).",
+        }
+
     # Si el upload fue de series (template_id imp_temp_series o imp_def_series), usar flujo series aunque model_target venga mal
     use_series_flow = (
         model_target == "invoice_series"

backend/api/v1/modules/a76/layouts_csv/facturas/template_config.py

@@ -188,6 +188,36 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
     # --- Partidas: Impo Def y Expo - misma estructura ---
     "imp_def_details": None,
     "exp_def_details": None,
+    # --- Partidas exportación definitiva (Clarion EstructuraParExpoCamReg A–V) ---
+    "exp_def_partidas": [
+        {"canonical": "NUMERO FACTURA EXPO", "aliases": ["NUMERO FACTURA EXPO.", "NUM FACTURA EXPO", "FACTURA EXPO"]},
+        {"canonical": "LINEA EXPO", "aliases": ["LINEA EXPO.", "RENGLON EXPO"]},
+        {"canonical": "TIPO DE IMPO", "aliases": ["TIPO DE IMPO.", "TIPO IMPO", "PROCEDENCIA"]},
+        {"canonical": "FACTURA IMPO", "aliases": ["FACTURA IMPO.", "FACTURA IMPORTACION"]},
+        {"canonical": "LINEA IMPO", "aliases": ["LINEA IMPO.", "LINEA IMPORTACION"]},
+        {"canonical": "GENERA DESCARGA", "aliases": ["GENERA DESCARGA?", "DESCARGA"]},
+        {"canonical": "CANTIDAD EXPORTADA/DESCARGAR", "aliases": ["CANTIDAD EXPORTADA", "CANTIDAD EXPORTADA/DESCARGAR", "CANTIDAD"]},
+        {"canonical": "UNIDAD DE MEDIDA", "aliases": ["U.M.", "UNIDAD MEDIDA"]},
+        {"canonical": "COSTO UNITARIO", "aliases": ["COSTOUNITARIO"]},
+        {"canonical": "PESO NETO", "aliases": ["PESONETO"]},
+        {"canonical": "PESO BRUTO", "aliases": ["PESOBRUTO"]},
+        {"canonical": "SE PAGO IMPUESTO", "aliases": ["SE PAGO IMPUESTO? (SI o NO)", "SEPAGOIMPUESTO"]},
+        {"canonical": "FORMA DE PAGO", "aliases": ["FORMADEPAGO", "FORMA PAGO"]},
+        {"canonical": "DESCRIPCION EXTRA", "aliases": ["DESCRIPCION EXTRA", "DESCRIPCIONEXTRA"]},
+        {"canonical": "INFORMACION ADICIONAL", "aliases": ["INFORMACION ADICIONAL", "INFORMACIONADICIONAL"]},
+        {"canonical": "AGREGAR(A)/SUSTITUIR(S)", "aliases": ["AGREGAR(A)/SUSTITUIR(S)", "AGREGAR/SUSTITUIR", "SUSTITUIR"]},
+        {"canonical": "LOTE"},
+        {"canonical": "NUMERO ENTRADA", "aliases": ["NUMERO ENTRADA", "NUM ENTRADA"]},
+        {"canonical": "ES PARTIDA/SUBPARTIDA", "aliases": ["ES PARTIDA/SUBPARTIDA", "ESSUBPARTIDA", "ES PARTIDA O SUBPARTIDA"]},
+        {"canonical": "LINEA PRINCIPAL", "aliases": ["LINEAPRINCIPAL", "PARTIDA PRINCIPAL"]},
+        {"canonical": "FRACCION AMERICANA", "aliases": ["FRACCION AMERICANA", "FRACCIONAMERICANA"]},
+        {"canonical": "FRACCION ARANCELARIA", "aliases": ["FRACCION ARANCELARIA", "FRACCIONARANCELARIA"]},
+        {"canonical": "CANTIDAD BULTOS", "aliases": ["CANTIDADBULTOS"]},
+        {"canonical": "CLAVE BULTOS", "aliases": ["CLAVEBULTOS"]},
+        {"canonical": "PAIS ORIGEN", "aliases": ["PAISORIGEN", "PAIS"]},
+        {"canonical": "NUM. PARTE", "aliases": ["NUMPARTE", "NUMERO PARTE", "NUM PARTE"]},
+        {"canonical": "ORDEN DE COMPRA", "aliases": ["ORDENCOMPRA", "ORDEN DE VENTA"]},
+    ],
     # --- Series de Importación Temporal (EstructuraSeriesFacImpoTemp.xls) ---
     # Clarion: NUMERO FACTURA, LINEA FACTURA, LINEA SERIE, SERIE, MODELO, NUM PARTE, SUB MODELO, NUMERO ID
     "imp_temp_series": [
@@ -224,6 +254,8 @@ def _resolve_template_columns(template_id: str) -> Optional[List[Dict[str, Any]]
         return TEMPLATE_COLUMNS.get("imp_temp_header")
     if template_id in ("imp_def_details", "exp_def_details"):
         return TEMPLATE_COLUMNS.get("imp_temp_details")
+    if template_id == "exp_def_partidas":
+        return TEMPLATE_COLUMNS.get("exp_def_partidas")
     if template_id == "cmex_details":
         return TEMPLATE_COLUMNS.get("imp_temp_details")
     if template_id == "cmex_series":

backend/api/v1/modules/a76/layouts_csv/facturas/validators/__init__.py

@@ -11,6 +11,7 @@ from .encabezados_impo_def import (
 )
 from .encabezados_cmex import validate_row_encabezados_cmex
 from .encabezados_expo import validate_row_encabezados_expo
+from .partidas_expo import validate_row_partidas_expo
 from .partidas_impo_def import validate_row_partidas_impo_def
 from .series_impo_def import (
     validate_row_series_impo_def,
@@ -22,6 +23,7 @@ __all__ = [
     "validate_row_encabezados_impo_def",
     "validate_row_encabezados_cmex",
     "validate_row_encabezados_expo",
+    "validate_row_partidas_expo",
     "validate_row_partidas_impo_def",
     "validate_row_series_impo_def",
     "row_to_series_normalized_def",

backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_expo.py

@@ -0,0 +1,558 @@
+"""
+Validaciones CSV para Partidas de Exportación Definitiva (y Cambio de Régimen).
+Paridad Clarion: VALIDA_TODA_PAR_EXPO, VALIDA_PARCIAL_PAR_EXPO, VALIDACIONES_PAR_EXPO.
+Estructura: NUMERO FACTURA EXPO, LINEA EXPO, TIPO DE IMPO, FACTURA IMPO, LINEA IMPO, GENERA DESCARGA, ...
+Variante RFC EGM0303257J1: columnas ES PARTIDA/SUBPARTIDA y LINEA PRINCIPAL en T y U.
+"""
+from decimal import Decimal, InvalidOperation
+from typing import Dict, Any, Optional, Set, Tuple, List
+
+from .partidas_impo_temp import _clip, _get
+
+# Longitudes Clarion partidas expo
+MAX_LEN_FACTURA_EXPO = 15
+MAX_LEN_LINEA_EXPO = 5
+MAX_LEN_TIPO_IMPO = 3
+MAX_LEN_ORDEN_COMPRA = 20
+MAX_LEN_NUM_PARTE = 30
+
+TIPO_IMPO_VALIDOS = frozenset({"TEM", "DEF"})
+GENERA_DESCARGA_VALIDOS = frozenset({"SI", "NO"})
+SE_PAGO_IMPUESTO_VALIDOS = frozenset({"SI", "NO"})
+
+
+def _err(
+    line_num: int,
+    col: str,
+    msg: str,
+    identifier: str = "ARCHIVO CSV",
+) -> Dict[str, Any]:
+    return {"line": line_num, "col": col, "msg": msg}
+
+
+def _parse_decimal(val: Any) -> Optional[Decimal]:
+    if val is None:
+        return None
+    s = _clip(val)
+    if not s:
+        return None
+    s = str(s).replace(",", "")
+    try:
+        return Decimal(s)
+    except (InvalidOperation, ValueError):
+        return None
+
+
+def _check_factura_expo_vacia(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
+    val = _get(row, "NUMERO FACTURA EXPO", "NUMERO FACTURA EXPO.", "NUM FACTURA EXPO", "FACTURA EXPO")
+    if not val:
+        return _err(
+            line_num,
+            "NUMERO FACTURA EXPO",
+            f"Error: (Celda A{line_num}) La Factura de Exportación está vacía y no se pueden hacer las validaciones. "
+            "Capturar en la Celda A un número de Factura existente al cual desee agregar o actualizar partidas.",
+        )
+    return None
+
+
+def _check_factura_expo_existe(
+    invoice_number: str,
+    line_num: int,
+    invoice_id_by_number: Dict[str, int],
+) -> Optional[Dict[str, Any]]:
+    if invoice_number not in invoice_id_by_number:
+        return _err(
+            line_num,
+            "NUMERO FACTURA EXPO",
+            f"Error: (Celda A{line_num}) La Factura de Exportación {invoice_number} no existe en SCAII y no se pueden hacer las validaciones. "
+            "Capturar en la Celda A un número de Factura existente al cual desee agregar o actualizar partidas.",
+        )
+    return None
+
+
+def _check_factura_expo_no_actualizada(
+    invoice_number: str,
+    line_num: int,
+    invoice_updated_by_number: Dict[str, bool],
+    rfc_exception_updated: Set[str],
+) -> Optional[Dict[str, Any]]:
+    if invoice_number in rfc_exception_updated:
+        return None
+    if invoice_updated_by_number.get(invoice_number, False):
+        return _err(
+            line_num,
+            "NUMERO FACTURA EXPO",
+            f"Error: (Celda A{line_num}) La Factura de Exportación: {invoice_number} ya existe y está Actualizada, no se puede hacer cambios a las facturas actualizadas. "
+            "Capturar otro número de Factura de Exportación o Desactualizar la factura.",
+        )
+    return None
+
+
+def _check_linea_expo_si_no_autonumerar(
+    row: Dict[str, Any], line_num: int, autonumerar: bool
+) -> Optional[Dict[str, Any]]:
+    if autonumerar:
+        return None
+    val = _get(row, "LINEA EXPO", "LINEA EXPO.", "RENGLON EXPO")
+    if not val:
+        return _err(
+            line_num,
+            "LINEA EXPO",
+            f"Error: (Celda B{line_num}) El campo de la línea de la partida está vacío y no se pueden hacer las validaciones, ya que se tiene la opción de Autonumerar como NO. "
+            "Capturar en la Celda B la línea de la partida al cual desee agregar o actualizar información.",
+        )
+    return None
+
+
+def _check_levantar_subpartidas_expo(
+    row: Dict[str, Any],
+    line_num: int,
+    levantar_subpartidas: bool,
+    rfc_exception_egm: bool,
+) -> Optional[Dict[str, Any]]:
+    """Clarion: si LevantarSubpartidas=S, obligatorios ES PARTIDA/SUBPARTIDA y LINEA PRINCIPAL (col S/T o T/U para EGM0303257J1)."""
+    if not levantar_subpartidas:
+        return None
+    es_sub = _get(row, "ES PARTIDA/SUBPARTIDA", "ESSUBPARTIDA", "ES PARTIDA O SUBPARTIDA")
+    if not es_sub:
+        col = "T" if rfc_exception_egm else "S"
+        return _err(
+            line_num,
+            "ES PARTIDA/SUBPARTIDA",
+            f"Error: (Celda {col}{line_num}) El campo del tipo de la partida (partida o subpartida) está vacío. "
+            "Capturar el tipo de la partida. [P] = Partida o [S] = Subpartida.",
+        )
+    linea_principal = _get(row, "LINEA PRINCIPAL", "LINEAPRINCIPAL", "PARTIDA PRINCIPAL")
+    if not linea_principal:
+        col = "U" if rfc_exception_egm else "T"
+        return _err(
+            line_num,
+            "LINEA PRINCIPAL",
+            f"Error: (Celda {col}{line_num}) El campo de la partida principal está vacío. "
+            f"Capturar en la Celda {col} la partida principal.",
+        )
+    return None
+
+
+def _valida_toda_obligatorios_expo(
+    row: Dict[str, Any],
+    line_num: int,
+    levantar_subpartidas: bool,
+    rfc_exception_egm: bool,
+) -> Optional[Dict[str, Any]]:
+    """VALIDA_TODA_PAR_EXPO obligatorios: C (tipo impo), F (descarga), G (cantidad); si descarga=SI: D, E; si subpartidas: S/T o T/U."""
+    obligatorios: List[str] = []
+    tipo_impo = _get(row, "TIPO DE IMPO", "TIPO DE IMPO.", "TIPO IMPO", "PROCEDENCIA")
+    if not tipo_impo:
+        obligatorios.append("(Col.C) Procedencia de la Importación.")
+    descarga = _get(row, "GENERA DESCARGA", "GENERA DESCARGA?", "DESCARGA")
+    if descarga == "SI":
+        if not _get(row, "FACTURA IMPO", "FACTURA IMPO.", "FACTURA IMPORTACION"):
+            obligatorios.append("(Col.D) Factura de Impo.")
+        if not _get(row, "LINEA IMPO", "LINEA IMPO.", "LINEA IMPORTACION"):
+            obligatorios.append("(Col.E) Línea de Impo.")
+    if not descarga:
+        obligatorios.append("(Col.F) Descarga? SI o NO")
+    if not _get(row, "CANTIDAD EXPORTADA/DESCARGAR", "CANTIDAD EXPORTADA", "CANTIDAD"):
+        obligatorios.append("(Col.G) Cantidad Expo.")
+    if levantar_subpartidas:
+        if rfc_exception_egm:
+            if not _get(row, "ES PARTIDA/SUBPARTIDA", "ESSUBPARTIDA", "ES PARTIDA O SUBPARTIDA"):
+                obligatorios.append("(Col.T) EsSubpartida?.")
+            if not _get(row, "LINEA PRINCIPAL", "LINEAPRINCIPAL", "PARTIDA PRINCIPAL"):
+                obligatorios.append("(Col.U) Partida Principal.")
+        else:
+            if not _get(row, "ES PARTIDA/SUBPARTIDA", "ESSUBPARTIDA", "ES PARTIDA O SUBPARTIDA"):
+                obligatorios.append("(Col.S) EsSubpartida?.")
+            if not _get(row, "LINEA PRINCIPAL", "LINEAPRINCIPAL", "PARTIDA PRINCIPAL"):
+                obligatorios.append("(Col.T) Partida Principal.")
+    if obligatorios:
+        return _err(
+            line_num,
+            "TIPO DE IMPO",
+            f"Existen campos vacíos que son obligatorios, es la {', '.join(obligatorios)}. "
+            "Revisar la línea del archivo y capturar los campos con la información correcta.",
+        )
+    return None
+
+
+def _valida_subpartidas_duplicados_expo(
+    factura_expo: str,
+    linea_expo: str,
+    line_num: int,
+    line_counts_csv: Dict[Tuple[str, str], int],
+) -> Optional[Dict[str, Any]]:
+    key = (factura_expo.strip(), _clip(linea_expo))
+    if line_counts_csv.get(key, 0) > 1:
+        return _err(
+            line_num,
+            "LINEA EXPO",
+            f"Error: (Celda B{line_num}) El campo de la partida está duplicado entre las partidas. "
+            f"Capturar en la Celda B{line_num} otro número de partida.",
+        )
+    return None
+
+
+def _valida_subpartida_tiene_principal_expo(
+    row: Dict[str, Any],
+    line_num: int,
+    factura_expo: str,
+    partidas_principales_csv: Set[Tuple[str, str]],
+    partidas_principales_bd: Set[Tuple[str, str]],
+    rfc_exception_egm: bool,
+) -> Optional[Dict[str, Any]]:
+    es_sub = _get(row, "ES PARTIDA/SUBPARTIDA", "ESSUBPARTIDA", "ES PARTIDA O SUBPARTIDA").upper()
+    linea_principal = _get(row, "LINEA PRINCIPAL", "LINEAPRINCIPAL", "PARTIDA PRINCIPAL")
+    if es_sub != "S" or not linea_principal or linea_principal == "0":
+        return None
+    key_principal = (factura_expo.strip(), _clip(linea_principal))
+    if key_principal in partidas_principales_csv or key_principal in partidas_principales_bd:
+        return None
+    col = "U" if rfc_exception_egm else "T"
+    return _err(
+        line_num,
+        "LINEA PRINCIPAL",
+        f"Error: (Celda {col}{line_num}) La partida principal {linea_principal} no existe. "
+        f"Capturar en la Celda {col} la partida principal y/o verificar que si permita contener subpartidas.",
+    )
+
+
+def _valida_subpartida_linea_principal_no_cero_expo(
+    row: Dict[str, Any], line_num: int, rfc_exception_egm: bool
+) -> Optional[Dict[str, Any]]:
+    es_sub = _get(row, "ES PARTIDA/SUBPARTIDA", "ESSUBPARTIDA", "ES PARTIDA O SUBPARTIDA").upper()
+    linea_principal = _get(row, "LINEA PRINCIPAL", "LINEAPRINCIPAL", "PARTIDA PRINCIPAL")
+    if es_sub == "S" and linea_principal == "0":
+        col = "U" if rfc_exception_egm else "T"
+        return _err(
+            line_num,
+            "LINEA PRINCIPAL",
+            f"Error: (Celda {col}{line_num}) La SubPartida no tiene asignada una partida principal. "
+            f"Capturar en la Celda {col} una partida principal.",
+        )
+    return None
+
+
+def _validaciones_par_expo(
+    row: Dict[str, Any],
+    line_num: int,
+    factura_expo: str,
+    valid_uom_codes: Set[str],
+    valid_bulks_codes: Set[str],
+    valid_payment_methods: Set[str],
+    valid_fraction_ame: Set[str],
+    valid_part_numbers: Optional[Set[str]],
+    factura_impo_tem_by_number: Dict[str, int],
+    factura_impo_def_by_number: Dict[str, int],
+    line_exists_tem: Set[Tuple[int, str]],
+    line_exists_def: Set[Tuple[int, str]],
+    validar_decimales_pza: bool = False,
+) -> Optional[Dict[str, Any]]:
+    """VALIDACIONES_PAR_EXPO: longitudes, TEM/DEF, FK factura impo + línea, descarga SI/NO, cantidad, U.M., bultos, forma pago, impuesto, fracción ame, num parte, decimales PZA."""
+    tipo_impo = _get(row, "TIPO DE IMPO", "TIPO DE IMPO.", "TIPO IMPO", "PROCEDENCIA").strip().upper()
+    if not tipo_impo:
+        tipo_impo = "TEM"
+    factura_impo = _get(row, "FACTURA IMPO", "FACTURA IMPO.", "FACTURA IMPORTACION")
+    linea_impo = _get(row, "LINEA IMPO", "LINEA IMPO.", "LINEA IMPORTACION")
+    descarga = _get(row, "GENERA DESCARGA", "GENERA DESCARGA?", "DESCARGA").upper()
+
+    # Longitud A
+    if factura_expo and len(factura_expo) > MAX_LEN_FACTURA_EXPO:
+        return _err(
+            line_num,
+            "NUMERO FACTURA EXPO",
+            f"Error: (Celda A{line_num}) La Factura de Exportación: {factura_expo} supera la longitud de caracteres. "
+            "Capturar en la Celda A el campo Factura de Exportación con formato ###############.",
+        )
+    # Longitud B
+    linea_expo = _get(row, "LINEA EXPO", "LINEA EXPO.", "RENGLON EXPO")
+    if linea_expo and len(linea_expo) > MAX_LEN_LINEA_EXPO:
+        return _err(
+            line_num,
+            "LINEA EXPO",
+            f"Error: (Celda B{line_num}) La Línea de Exportación: {linea_expo} supera la longitud de caracteres. "
+            "Capturar en la Celda B una Línea de Exportación con formato #####.",
+        )
+    # Longitud C
+    if tipo_impo and len(tipo_impo) > MAX_LEN_TIPO_IMPO:
+        return _err(
+            line_num,
+            "TIPO DE IMPO",
+            f"Error: (Celda C{line_num}) La Procedencia debe ser especificada como TEM o DEF. "
+            "Capturar en la Celda C una procedencia no mayor de 3 caracteres.",
+        )
+    # C: TEM o DEF
+    if tipo_impo and tipo_impo not in TIPO_IMPO_VALIDOS:
+        return _err(
+            line_num,
+            "TIPO DE IMPO",
+            f"Error: (Celda C{line_num}) El Tipo de Descargo: {tipo_impo} no es valido. Capturar uno valido como TEM o DEF.",
+        )
+    # D, E: factura impo + línea existen en catálogo TEM o DEF
+    if factura_impo and tipo_impo:
+        consec_tem = factura_impo_tem_by_number.get(factura_impo.strip())
+        consec_def = factura_impo_def_by_number.get(factura_impo.strip())
+        if tipo_impo == "TEM":
+            if consec_tem is None:
+                return _err(
+                    line_num,
+                    "FACTURA IMPO",
+                    f"Error: (Celda D{line_num}) La Factura: {factura_impo} de Importación Temporal no existe. "
+                    "Capturar un Número de Factura que exista en el Catálogo de Importaciones Temporales.",
+                )
+            key_line = (consec_tem, _clip(linea_impo))
+            if linea_impo and key_line not in line_exists_tem:
+                return _err(
+                    line_num,
+                    "LINEA IMPO",
+                    f"Error: (Celda D{line_num}, E{line_num}) La Factura: {factura_impo} con línea: {linea_impo} de Importación Temporal no existe. "
+                    "Capturar un Número de Factura con diferente línea que este en el Catálogo de Importaciones Temporales.",
+                )
+        elif tipo_impo == "DEF":
+            if consec_def is None:
+                return _err(
+                    line_num,
+                    "FACTURA IMPO",
+                    f"Error: (Celda D{line_num}) La Factura: {factura_impo} de Importación Definitiva no existe. "
+                    "Capturar un Número de Factura que exista en el Catálogo de Importaciones Definitivas.",
+                )
+            key_line = (consec_def, _clip(linea_impo))
+            if linea_impo and key_line not in line_exists_def:
+                return _err(
+                    line_num,
+                    "LINEA IMPO",
+                    f"Error: (Celda D{line_num}, E{line_num}) La Factura: {factura_impo} con línea: {linea_impo} de Importación Definitiva no existe. "
+                    "Capturar un Número de Factura con diferente línea que este en el Catálogo de Importaciones Definitivas.",
+                )
+    # F: SI o NO
+    if descarga and descarga not in GENERA_DESCARGA_VALIDOS:
+        return _err(
+            line_num,
+            "GENERA DESCARGA",
+            f"Error: (Celda F{line_num}) La captura: {descarga} no es valido para la opción de que la partida genere descarga. "
+            "Capturar un valor valido como SI o NO o dejar vacio y lo tomará como un SI.",
+        )
+    # G: cantidad no cero
+    cant_str = _get(row, "CANTIDAD EXPORTADA/DESCARGAR", "CANTIDAD EXPORTADA", "CANTIDAD")
+    if cant_str:
+        cant = _parse_decimal(cant_str)
+        if cant is not None and cant == 0:
+            return _err(
+                line_num,
+                "CANTIDAD EXPORTADA/DESCARGAR",
+                f"Error: (Celda G{line_num}) La Cantidad a Exportar: {cant_str} no puede ser cero. Capturar una cantidad a exportar valida.",
+            )
+    # H: U.M. en catálogo (si no se toma de impo)
+    um = _get(row, "UNIDAD DE MEDIDA", "U.M.", "UNIDAD MEDIDA")
+    if um and valid_uom_codes and um.upper() not in valid_uom_codes:
+        return _err(
+            line_num,
+            "UNIDAD DE MEDIDA",
+            f"Error: (Celda H{line_num}) La U.M.: {um} no existe en catálogo de Unidades de Medida. "
+            "Capturar una Clave de Unidad de Medida que exista en el Catálogo.",
+        )
+    # Bultos: cantidad + clave (Clarion Col L, M)
+    clave_bultos = _get(row, "CLAVE BULTOS", "CLAVEBULTOS")
+    cant_bultos = row.get("CANTIDAD BULTOS") or row.get("CANTIDADBULTOS")
+    if clave_bultos:
+        if valid_bulks_codes and clave_bultos not in valid_bulks_codes:
+            return _err(
+                line_num,
+                "CLAVE BULTOS",
+                f"Error: (Celda M{line_num}) La Clave de Bulto: {clave_bultos} no existe en el Catálogo de Claves de Bultos. "
+                "Darlo de alta en el Catálogo de Claves de Bultos o capturar uno ya existente.",
+            )
+        cant_bultos_val = _parse_decimal(cant_bultos)
+        if cant_bultos_val is None:
+            return _err(
+                line_num,
+                "CANTIDAD BULTOS",
+                f"Error: (Celda L{line_num}) La Cantidad de Bultos está vacía y en la Celda M{line_num} se tienen la Clave de Bulto: {clave_bultos}.",
+            )
+        if cant_bultos_val == 0:
+            return _err(
+                line_num,
+                "CANTIDAD BULTOS",
+                f"Error: (Celda L{line_num}) La Cantidad de Bultos es cero y en la Celda M{line_num} se tienen la Clave de Bulto: {clave_bultos}.",
+            )
+    else:
+        cant_bultos_val = _parse_decimal(cant_bultos)
+        if cant_bultos_val is not None and cant_bultos_val > 0:
+            return _err(
+                line_num,
+                "CANTIDAD BULTOS",
+                f"Error: (Celda L{line_num}) La Cantidad de Bultos es {cant_bultos} y en la Celda M{line_num} no se tienen la Clave de Bulto.",
+            )
+    # L: Se pagó impuesto SI/NO
+    se_pago = _get(row, "SE PAGO IMPUESTO", "SE PAGO IMPUESTO? (SI o NO)", "SEPAGOIMPUESTO")
+    if se_pago and se_pago.upper() not in SE_PAGO_IMPUESTO_VALIDOS:
+        return _err(
+            line_num,
+            "SE PAGO IMPUESTO",
+            f"Error: (Celda L{line_num}) El Valor Capturado para Se Pago Impuesto no es Válido. Capturar en la Celda L{line_num} SI o NO.",
+        )
+    # M: Forma de pago en catálogo
+    forma_pago = _get(row, "FORMA DE PAGO", "FORMADEPAGO", "FORMA PAGO")
+    if forma_pago and valid_payment_methods and forma_pago not in valid_payment_methods:
+        return _err(
+            line_num,
+            "FORMA DE PAGO",
+            f"Error: (Celda M{line_num}) La Forma de Pago Capturado no es Válido. "
+            "Capturar en la Celda M una Forma de Pago dentro del Catálogo General de Formas de Pago.",
+        )
+    # Fracción americana (Clarion Col R)
+    frac_ame = _get(row, "FRACCION AMERICANA", "FRACCIONAMERICANA")
+    if frac_ame and valid_fraction_ame and frac_ame not in valid_fraction_ame:
+        return _err(
+            line_num,
+            "FRACCION AMERICANA",
+            f"Error: (Celda R{line_num}) La Fracción Americana: {frac_ame} no está en el Catálogo de Fracciones Americanas. "
+            "Capturar en la Celda R una fracción que se encuentre en el catálogo o dar la de alta.",
+        )
+    # Orden de compra / orden venta (Clarion Col S) máx 20
+    orden = _get(row, "ORDEN DE COMPRA", "ORDENCOMPRA", "ORDEN DE VENTA")
+    if orden and len(orden) > MAX_LEN_ORDEN_COMPRA:
+        return _err(
+            line_num,
+            "ORDEN DE COMPRA",
+            f"Error: (Celda S{line_num}) La Orden de Venta: {orden} supera la cantidad de caracteres permitidos. "
+            "Capturar en la Celda S una orden de compra no mayor de 20 caracteres.",
+        )
+    # Número de parte (Clarion Col T): longitud y catálogo
+    num_parte = _get(row, "NUM. PARTE", "NUMPARTE", "NUMERO PARTE", "NUM PARTE")
+    if num_parte:
+        if len(num_parte) > MAX_LEN_NUM_PARTE:
+            return _err(
+                line_num,
+                "NUM. PARTE",
+                f"Error: (Celda T{line_num}) El Número de Parte: {num_parte} supera la longitud de caracteres. "
+                "Capturar en la Celda T el Número de Parte no mayor de 30 caracteres.",
+            )
+        if valid_part_numbers is not None and num_parte.upper() not in valid_part_numbers:
+            return _err(
+                line_num,
+                "NUM. PARTE",
+                f"Error: (Celda T{line_num}) El Número de Parte: {num_parte} no existe en el Catálogo de Partes. Darlo de alta en el Catálogo de Partes.",
+            )
+    # Decimales PZA
+    if validar_decimales_pza and um and um.upper() == "PZA" and cant_str:
+        d = _parse_decimal(cant_str)
+        if d is not None and d != int(d):
+            return _err(
+                line_num,
+                "CANTIDAD EXPORTADA/DESCARGAR",
+                f"Error: (Celda G{line_num}) La Unidad de Medida es PZA, Por lo Tanto no es Válida la Captura de Decimales. Asignar una Cantidad sin decimales.",
+            )
+    return None
+
+
+def validate_row_partidas_expo(
+    row: Dict[str, Any],
+    line_num: int,
+    autonumerar: bool,
+    actualizar: bool,
+    levantar_subpartidas: bool,
+    invoice_id_by_number: Dict[str, int],
+    invoice_updated_by_number: Dict[str, bool],
+    rfc_exception_updated: Set[str],
+    existing_line_keys_by_invoice: Dict[str, Set[str]],
+    line_counts_csv: Dict[Tuple[str, str], int],
+    partidas_principales_csv: Set[Tuple[str, str]],
+    partidas_principales_bd: Set[Tuple[str, str]],
+    valid_uom_codes: Set[str],
+    valid_bulks_codes: Set[str],
+    valid_payment_methods: Set[str],
+    valid_fraction_ame: Set[str],
+    valid_part_numbers: Optional[Set[str]],
+    factura_impo_tem_by_number: Dict[str, int],
+    factura_impo_def_by_number: Dict[str, int],
+    line_exists_tem: Set[Tuple[int, str]],
+    line_exists_def: Set[Tuple[int, str]],
+    rfc_exception_egm: bool = False,
+    validar_decimales_pza: bool = False,
+) -> Optional[Dict[str, Any]]:
+    """
+    Valida una fila de CSV de Partidas de Exportación Definitiva (o Cambio de Régimen).
+    Clarion: VALIDA_TODA_PAR_EXPO vs VALIDA_PARCIAL_PAR_EXPO según autonumerar, actualizar y si la partida existe.
+    """
+    err = _check_factura_expo_vacia(row, line_num)
+    if err:
+        return err
+
+    factura_expo = _get(row, "NUMERO FACTURA EXPO", "NUMERO FACTURA EXPO.", "NUM FACTURA EXPO", "FACTURA EXPO")
+    if not factura_expo:
+        return _err(line_num, "NUMERO FACTURA EXPO", "Requerido")
+
+    err = _check_factura_expo_existe(factura_expo, line_num, invoice_id_by_number)
+    if err:
+        return err
+
+    err = _check_factura_expo_no_actualizada(
+        factura_expo, line_num, invoice_updated_by_number, rfc_exception_updated
+    )
+    if err:
+        return err
+
+    err = _check_linea_expo_si_no_autonumerar(row, line_num, autonumerar)
+    if err:
+        return err
+
+    err = _check_levantar_subpartidas_expo(row, line_num, levantar_subpartidas, rfc_exception_egm)
+    if err:
+        return err
+
+    linea_expo = _get(row, "LINEA EXPO", "LINEA EXPO.", "RENGLON EXPO")
+    existing_lines = existing_line_keys_by_invoice.get(factura_expo.strip(), set())
+    partida_existe = bool(linea_expo and linea_expo in existing_lines)
+    use_partial = actualizar and not autonumerar and partida_existe
+
+    if use_partial:
+        return _validaciones_par_expo(
+            row,
+            line_num,
+            factura_expo,
+            valid_uom_codes=valid_uom_codes,
+            valid_bulks_codes=valid_bulks_codes,
+            valid_payment_methods=valid_payment_methods,
+            valid_fraction_ame=valid_fraction_ame,
+            valid_part_numbers=valid_part_numbers,
+            factura_impo_tem_by_number=factura_impo_tem_by_number,
+            factura_impo_def_by_number=factura_impo_def_by_number,
+            line_exists_tem=line_exists_tem,
+            line_exists_def=line_exists_def,
+            validar_decimales_pza=validar_decimales_pza,
+        )
+    else:
+        err = _valida_toda_obligatorios_expo(row, line_num, levantar_subpartidas, rfc_exception_egm)
+        if err:
+            return err
+        if levantar_subpartidas:
+            err = _valida_subpartidas_duplicados_expo(
+                factura_expo, linea_expo or "", line_num, line_counts_csv
+            )
+            if err:
+                return err
+            err = _valida_subpartida_tiene_principal_expo(
+                row, line_num, factura_expo,
+                partidas_principales_csv, partidas_principales_bd, rfc_exception_egm
+            )
+            if err:
+                return err
+            err = _valida_subpartida_linea_principal_no_cero_expo(row, line_num, rfc_exception_egm)
+            if err:
+                return err
+        return _validaciones_par_expo(
+            row,
+            line_num,
+            factura_expo,
+            valid_uom_codes=valid_uom_codes,
+            valid_bulks_codes=valid_bulks_codes,
+            valid_payment_methods=valid_payment_methods,
+            valid_fraction_ame=valid_fraction_ame,
+            valid_part_numbers=valid_part_numbers,
+            factura_impo_tem_by_number=factura_impo_tem_by_number,
+            factura_impo_def_by_number=factura_impo_def_by_number,
+            line_exists_tem=line_exists_tem,
+            line_exists_def=line_exists_def,
+            validar_decimales_pza=validar_decimales_pza,
+        )