"""
Tareas Celery para importación CSV de facturas (encabezados, partidas, series).
Flujo: scan_file (validación) → insert_valid_rows (commit).

Objetivo en BD (paridad con flujo normal): al terminar el commit, los datos deben quedar
igual que por UI/API: encabezados con capture_user/who_processed; partidas con costos,
pesos y descripciones calculados/heredados según items/imports/validators; series con
campos no presentes en CSV en null. No se modifican plantillas CSV; no se inventan
datos sin fuente (p. ej. LineReference solo si hay fuente explícita).
"""
import os
from datetime import datetime, timedelta
from decimal import Decimal
import csv
import json
import logging
import re
import unicodedata
from typing import Dict, Any, Optional, List, Set, Tuple

from core.celery_app import celery_app
from core.database import CoreSessionLocal
from core.paths import layout_path
from sqlalchemy import func

from ..common import storage as common_storage
from ..common import meta as common_meta
from ..common import responses as common_responses
from ..common import csv_reader as common_csv_reader
from .template_config import row_from_template
from .validators.encabezados_impo_temp import csv_tipo_moneda_es_me_mn_mc
from api.v1.modules.a76.app_settings.service import AppSettingsService
# Models are imported inside tasks to avoid circular dependencies and mapper initialization issues in the API process

logger = logging.getLogger(__name__)


def _merge_unique_invoice_scan_error_lines(
    precheck_lines: Set[int],
    row_validation_lines: List[int],
) -> List[int]:
    """Unión deduplicada: líneas marcadas en precheck + líneas con error en validación por fila (imp_temp_header)."""
    return sorted(set(precheck_lines) | set(row_validation_lines))


# Job type vacío para facturas (prefijo Redis "import_" sin tipo, ver common/storage.py)
JOB_TYPE = ""

# Redis keys and TTL for import file/meta (exportados para routes; coinciden con common_storage cuando job_type="")
IMPORT_FILE_KEY_PREFIX = "import_file:"
IMPORT_META_KEY_PREFIX = "import_meta:"
IMPORT_ERROR_LINES_KEY_PREFIX = "import_error_lines:"
IMPORT_REDIS_TTL = common_storage.IMPORT_REDIS_TTL


def _ensure_worker_has_file_from_redis(job_id: str) -> Optional[str]:
    """Usa common storage con job_type vacío (prefijo import_)."""
    return common_storage.ensure_file_from_redis(JOB_TYPE, job_id, "Invoices import")


def _ensure_worker_has_meta_from_redis(job_id: str, file_path: str) -> bool:
    return common_storage.ensure_meta_from_redis(JOB_TYPE, job_id, file_path, "Invoices import")


def _delete_import_from_redis(job_id: str) -> None:
    common_storage.delete_import_from_redis(JOB_TYPE, job_id)


def _ensure_utf8_compatible_import_file(file_path: str) -> None:
    """
    Homogeneiza a UTF-8-SIG cuando el archivo venga en otro encoding
    para que todo el flujo legado de facturas lea exactamente lo mismo.
    """
    encoding = common_csv_reader.detect_text_encoding(file_path)
    if encoding in ("utf-8", "utf-8-sig"):
        return
    with open(file_path, "r", encoding=encoding) as src:
        content = src.read()
    with open(file_path, "w", encoding="utf-8-sig") as dst:
        dst.write(content)


def _facturas_csv_encoding(file_path: str) -> str:
    """
    Encapsula la detección para centralizar la política de lectura CSV en facturas.
    """
    return common_csv_reader.detect_text_encoding(file_path)

class ForeignKeyValidator:
    def __init__(self, session, tenant_id, company_id):
        self.session = session
        self.tenant_id = tenant_id
        self.company_id = company_id
        self.cache = {} # {(model_name, field_name, is_public, value): bool}

    def check_exists(self, model, value, field_name="id", is_public=False):
        if value is None:
            return True # Assume optional if None, or let DB handle not-null
        
        key = (model.__name__, field_name, is_public, value)
        if key in self.cache:
            return self.cache[key]
        
        col = getattr(model, field_name)
        if field_name == "short_name" and hasattr(model, "short_name"):
            query = self.session.query(col).filter(func.upper(col) == (value.upper() if isinstance(value, str) else value))
        else:
            query = self.session.query(col).filter(col == value)
        if not is_public:
            query = query.filter(model.tenant_id == self.tenant_id, model.company_id == self.company_id)
        
        exists = query.first() is not None
        self.cache[key] = exists
        return exists


TRANSPORT_TYPE_VALUES = {
    "none",
    "transport",
    "box",
    "licence plates",
    "truck",
    "vessel",
    "rail_barge",
    "container",
    "airplane",
    "gondola",
    "flatbed",
}


def normalize_public_code(value: Optional[str]) -> Optional[str]:
    if value is None:
        return None
    text = str(value).strip().upper()
    return text or None


def validate_public_code(
    validator: ForeignKeyValidator,
    model,
    value: Optional[str],
    line_num: int,
    col_name: str,
    field_name: str = "code",
    required: bool = False,
) -> Optional[Dict[str, Any]]:
    code = normalize_public_code(value)
    if not code:
        if required:
            return {
                "line": line_num,
                "col": col_name,
                "msg": "Requerido",
                "solution": f"Capturar el dato requerido en la columna {col_name}.",
            }
        return None
    if not validator.check_exists(model, code, field_name=field_name, is_public=True):
        return {
            "line": line_num,
            "col": col_name,
            "msg": "No existe en el catalogo",
            "solution": f"Capturar en la columna {col_name} un valor que exista en el catálogo.",
        }
    return None


def validate_tenant_fk_id(
    validator: ForeignKeyValidator,
    model,
    value: Optional[int],
    line_num: int,
    col_name: str,
    required: bool = False,
) -> Optional[Dict[str, Any]]:
    if value is None:
        if required:
            return {
                "line": line_num,
                "col": col_name,
                "msg": "Requerido",
                "solution": f"Capturar el dato requerido en la columna {col_name}.",
            }
        return None
    if not validator.check_exists(model, value):
        return {
            "line": line_num,
            "col": col_name,
            "msg": "No existe en el catalogo",
            "solution": f"Capturar en la columna {col_name} un valor que exista en el catálogo.",
        }
    return None


def _validate_client_provider_ref(
    validator: ForeignKeyValidator,
    model,
    raw_value: Any,
    line_num: int,
    col_name: str,
    required: bool,
) -> Optional[Dict[str, Any]]:
    """Valida CLAVE PROVEEDOR / VENDIDO A / ENVIADO A: acepta ID (entero) o short_name (texto)."""
    if raw_value is None or not str(raw_value).strip():
        if required:
            return {
                "line": line_num,
                "col": col_name,
                "msg": "Requerido",
                "solution": f"Capturar el dato requerido en la columna {col_name}.",
            }
        return None
    pid = parse_int(raw_value)
    if pid is not None:
        return validate_tenant_fk_id(validator, model, pid, line_num, col_name, required=False)
    short_norm = str(raw_value).strip().upper()
    if not validator.check_exists(model, short_norm, field_name="short_name"):
        return {
            "line": line_num,
            "col": col_name,
            "msg": "No existe en el catalogo",
            "solution": f"Capturar en la columna {col_name} un valor que exista en el catálogo.",
        }
    return None


def _validate_customs_broker_ref(
    validator: ForeignKeyValidator,
    model,
    raw_value: Any,
    line_num: int,
    col_name: str,
    required: bool = False,
    broker_lookup: Optional[Dict[str, Any]] = None,
) -> Optional[Dict[str, Any]]:
    """Valida AGENTE ADUANAL: acepta ID (entero) o clave broker_key (texto)."""
    if raw_value is None or not str(raw_value).strip():
        if required:
            return {
                "line": line_num,
                "col": col_name,
                "msg": "Requerido",
                "solution": f"Capturar el dato requerido en la columna {col_name}.",
            }
        return None
    if broker_lookup is not None:
        resolved_id, _mode, _normalized, _err = resolve_customs_broker_ref(raw_value, broker_lookup)
        if resolved_id is None:
            return {
                "line": line_num,
                "col": col_name,
                "msg": "No existe en el catalogo",
                "solution": f"Capturar en la columna {col_name} una clave que exista en el catálogo.",
            }
        return None
    pid = parse_int(raw_value)
    if pid is not None:
        # Si viene numérico primero intentamos como ID.
        # Si no existe ese ID, intentamos como broker_key (clave numérica).
        err = validate_tenant_fk_id(validator, model, pid, line_num, col_name, required=False)
        if err is None:
            return None
    clave = _normalize_broker_key(raw_value)
    if not validator.check_exists(model, clave, field_name="broker_key"):
        return {
            "line": line_num,
            "col": col_name,
            "msg": "No existe en el catalogo",
            "solution": f"Capturar en la columna {col_name} una clave que exista en el catálogo.",
        }
    return None


def _normalize_broker_key(value: Any) -> str:
    return str(value or "").strip().upper()


def _build_customs_broker_lookup(rows: List[Tuple[int, Optional[str]]]) -> Dict[str, Any]:
    """
    Construye lookup robusto para AGENTE ADUANAL:
    - id_set: IDs válidos
    - key_to_id: broker_key normalizado -> id
    - alias_to_id: alias P#### <-> #### solo cuando es inequívoco
    """
    id_set: Set[int] = set()
    key_to_id: Dict[str, int] = {}
    alias_votes: Dict[str, Set[int]] = {}

    for bid, bkey in rows:
        id_set.add(int(bid))
        norm = _normalize_broker_key(bkey)
        if not norm:
            continue
        key_to_id[norm] = int(bid)
        if norm.startswith("P") and norm[1:].isdigit():
            alias_votes.setdefault(norm[1:], set()).add(int(bid))
        elif norm.isdigit():
            alias_votes.setdefault(f"P{norm}", set()).add(int(bid))

    alias_to_id: Dict[str, int] = {}
    for alias, ids in alias_votes.items():
        if len(ids) == 1:
            alias_to_id[alias] = next(iter(ids))

    return {
        "id_set": id_set,
        "key_to_id": key_to_id,
        "alias_to_id": alias_to_id,
    }


def resolve_customs_broker_ref(
    raw_value: Any,
    broker_lookup: Dict[str, Any],
) -> Tuple[Optional[int], Optional[str], str, Optional[str]]:
    """
    Resuelve AGENTE ADUANAL con patrón robusto:
    id -> broker_key exacto -> alias P####/####.
    """
    text = str(raw_value or "").strip()
    if not text:
        return None, None, "", None

    id_set: Set[int] = broker_lookup.get("id_set", set())
    key_to_id: Dict[str, int] = broker_lookup.get("key_to_id", {})
    alias_to_id: Dict[str, int] = broker_lookup.get("alias_to_id", {})

    pid = parse_int(raw_value)
    if pid is not None and pid in id_set:
        return int(pid), "id", text, None

    norm = _normalize_broker_key(raw_value)
    if norm in key_to_id:
        return key_to_id[norm], "broker_key", norm, None
    if norm in alias_to_id:
        return alias_to_id[norm], "broker_key_alias", norm, None

    return None, None, norm, "No existe en el catalogo"


def _summarize_top_reasons(details: List[Dict[str, Any]], limit: int = 10) -> List[Dict[str, Any]]:
    counts: Dict[str, int] = {}
    for detail in details:
        reason = str(detail.get("reason", "")).strip()
        if not reason:
            continue
        counts[reason] = counts.get(reason, 0) + 1
    ordered = sorted(counts.items(), key=lambda kv: kv[1], reverse=True)
    return [{"reason": reason, "count": count} for reason, count in ordered[:limit]]


def _normalize_invoice_number(value: Any) -> str:
    return str(value or "").strip().upper()


def _extract_invoice_number_from_row(row: Dict[str, Any]) -> str:
    """Extrae y normaliza número de factura desde aliases de layouts."""
    return _normalize_invoice_number(
        row.get("NUMERO FACTURA")
        or row.get("NUM FACTURA")
        or row.get("FACTURA")
    )


def _resolve_import_context(
    model_target: str,
    meta: Dict[str, Any],
    footer_config: Dict[str, Any],
    job_type_override: Optional[str] = None,
) -> Tuple[str, str, str]:
    """
    Resuelve template/invoice_type/operation_type efectivos de forma consistente
    para scan y commit.
    """
    template_id = meta.get("template_id") or (
        "imp_temp_header" if model_target == "invoice_header" else
        "imp_temp_details" if model_target == "invoice_details" else "imp_temp_series"
    )
    op_type = str(meta.get("operation_type", "imp") or "imp").strip().lower()
    inv_type = normalize_public_code(
        footer_config.get("invoice_type") or meta.get("invoice_type") or "TEM"
    ) or "TEM"

    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"
    if job_type_override == "exp" and model_target == "invoice_series":
        template_id = "exp_def_series"

    if model_target == "invoice_details" and op_type == "exp":
        template_id = "exp_def_partidas"
    if model_target == "invoice_series" and inv_type in ("DEF", "MATDE", "EXDEF"):
        template_id = "imp_def_series"

    if model_target == "invoice_header" and template_id == "imp_def_header":
        inv_type = "DEF"
    if model_target == "invoice_header" and template_id == "cmex_header":
        inv_type = "MEX"
    if model_target == "invoice_header" and template_id == "exp_def_header":
        op_type = "exp"
        inv_type = normalize_public_code(
            meta.get("tipo_factura") or footer_config.get("tipo_factura") or "AFIJO"
        ) or "AFIJO"

    if model_target == "invoice_details" and template_id == "imp_def_details":
        inv_type = "DEF"
    if model_target == "invoice_details" and template_id == "cmex_details":
        inv_type = "MEX"
    if model_target == "invoice_details" and template_id == "exp_def_partidas":
        op_type = "exp"
        inv_type = normalize_public_code(
            meta.get("tipo_factura") or footer_config.get("tipo_factura") or "AFIJO"
        ) or "AFIJO"

    if template_id == "cmex_series":
        inv_type = "MEX"

    return template_id, inv_type, op_type


def _is_invoice_processed_status(value: Any) -> bool:
    """Mapea status de factura a bandera de procesado real."""
    if isinstance(value, bool):
        return value
    text = str(value or "").strip().lower()
    if not text:
        return False
    # pending equivale a no procesado.
    if text in {"pending", "pendiente", "false", "0", "no"}:
        return False
    # Solo estados explícitos de procesado/actualización bloquean modificaciones.
    return text in {"processed", "procesado", "updated", "actualizada", "actualizado", "true", "1", "yes", "si", "sí"}


def _resolve_invoice_id_for_details(
    session,
    InvoiceHeader,
    tenant_id: int,
    company_id: int,
    invoice_number_raw: Any,
    invoice_type: Optional[str],
    operation_type: Optional[Any],
    cache: Dict[str, Optional[int]],
) -> Optional[int]:
    invoice_norm = _normalize_invoice_number(invoice_number_raw)
    if not invoice_norm:
        return None
    op_text = str(operation_type.value if hasattr(operation_type, "value") else operation_type or "").strip().lower()
    cache_key = f"{invoice_norm}|{str(invoice_type or '').upper()}|{op_text}"
    if cache_key in cache:
        return cache[cache_key]
    q = (
        session.query(InvoiceHeader.id)
        .filter(
            InvoiceHeader.tenant_id == tenant_id,
            InvoiceHeader.company_id == company_id,
            func.upper(InvoiceHeader.invoice_number) == invoice_norm,
        )
    )
    if invoice_type:
        q = q.filter(InvoiceHeader.invoice_type == invoice_type)
    if operation_type:
        q = q.filter(InvoiceHeader.operation_type == operation_type)
    found = q.scalar()
    cache[cache_key] = found
    return found

@celery_app.task(bind=True)
def scan_file(self, job_id: str, model_target: str, config: str = None, job_type_override: Optional[str] = None):
    """Pass 1: Read CSV, Validate types, Write Errors to JSONL. Delegates to _do_scan_file."""
    return _do_scan_file(self, job_id, model_target, config, job_type_override)


def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] = None, job_type_override: Optional[str] = None) -> Dict[str, Any]:
    """Pass 1 body: load file/meta from storage, run validations, store error lines. Uses effective_job_type for storage."""
    effective_job_type = job_type_override if job_type_override is not None else JOB_TYPE
    log_prefix = "Exportación import" if effective_job_type else "Invoices import"

    logger.info(f"Starting scan for job {job_id} target {model_target}")

    # 1. Get file from Redis and write to worker local disk
    file_path = common_storage.ensure_file_from_redis(effective_job_type, job_id, log_prefix)
    if not file_path:
        return {"status": "failed", "error": "File not found (missing or expired in queue). Please upload again."}
    common_storage.ensure_meta_from_redis(effective_job_type, job_id, file_path, log_prefix)
    _ensure_utf8_compatible_import_file(file_path)

    error_path = common_storage.error_path_for_job(effective_job_type, job_id)

    total_rows = 0
    error_count = 0
    processed_rows = 0
    errors_detail: List[Dict[str, Any]] = []

    # 3. Count Total (Quick Pass) or just estimate
    try:
        with open(file_path, 'r', encoding=_facturas_csv_encoding(file_path)) as f:
            total_rows = sum(1 for _ in f) - 1  # Minus header
    except Exception as e:
        return {"status": "failed", "error": f"Cannot read file: {e}"}

    footer_config = parse_footer_config(config)
    date_format = footer_config.get("dateFormat")
    
    # Validate and set default date_format if not provided
    if not date_format:
        date_format = "yyyy-mm-dd"  # Default to ISO format
        logger.info(f"No date_format specified in config, using default: {date_format}")

    try:
        tenant_id, company_id = 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, inv_type_value, op_type_scan = _resolve_import_context(
        model_target=model_target,
        meta=meta,
        footer_config=footer_config,
        job_type_override=job_type_override,
    )

    logger.info(
        "Scan job %s template_id=%s model_target=%s job_type_override=%s op_type=%s inv_type=%s",
        job_id, template_id, model_target, job_type_override, op_type_scan, inv_type_value,
    )

    # --- Series de Importación Definitiva: flujo específico (Clarion VALIDA_TODA_SERIES_IMPO_DEF / VALIDA_PARCIAL) ---
    DEF_SERIES_TEMPLATE_OR_TYPE = (
        model_target == "invoice_series"
        and (
            template_id == "imp_def_series"
            or (inv_type_value in ("DEF", "MATDE", "EXDEF"))
        )
    )
    if DEF_SERIES_TEMPLATE_OR_TYPE:
        try:
            from api.v1.modules.a76.invoices.models import InvoiceHeader
            from api.v1.modules.a76.items.models import LineItem
            from api.v1.modules.a76.items.series.models import Serie
            from api.v1.modules.a76.items.line_quantities.models import LineQuantity
            from .validators.series_impo_def import (
                validate_row_series_impo_def,
            )

            DEF_INVOICE_TYPES = ("DEF", "MATDE", "EXDEF")
            actualizar = meta.get("actualizar", False)
            autonumerar = meta.get("autonumerar", True)
            validar_series_exception = meta.get("validar_series", False)
            _fc = parse_footer_config(meta.get("footer_config"))
            if _fc:
                if "actualizar" in _fc:
                    actualizar = bool(_fc["actualizar"])
                elif _fc.get("mode") == "update":
                    actualizar = True
                elif _fc.get("mode") == "replace":
                    actualizar = False
                if "autonumerar" in _fc:
                    autonumerar = bool(_fc["autonumerar"])
                else:
                    as_val = _fc.get("autonumber_series", "true")
                    autonumerar = str(as_val).lower() in ("true", "1", "si", "sí", "yes")
                if "validar_series" in _fc:
                    validar_series_exception = bool(_fc["validar_series"])

            with CoreSessionLocal() as session:
                q = (
                    session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.status)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == "imp",
                        InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES),
                    )
                )
                rows_inv = q.all()
                invoice_id_by_number: Dict[str, int] = {}
                invoice_processed_by_number: Dict[str, bool] = {}
                for num, iid, is_upd in rows_inv:
                    if num:
                        invoice_id_by_number[str(num).strip()] = iid
                        invoice_processed_by_number[str(num).strip()] = _is_invoice_processed_status(is_upd)

                partida_max_series: Dict[Tuple[str, str], int] = {}
                q_qty = (
                    session.query(
                        InvoiceHeader.invoice_number,
                        LineItem.line_number,
                        LineQuantity.quantity,
                    )
                    .join(LineItem, LineItem.invoice_id == InvoiceHeader.id)
                    .outerjoin(LineQuantity, LineQuantity.item_line_id == LineItem.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, ln, qty in q_qty.all():
                    if num is not None and ln is not None:
                        key = (str(num).strip(), str(ln).strip())
                        if qty is not None:
                            partida_max_series[key] = int(qty) if qty else 0
                        else:
                            partida_max_series[key] = 0

                existing_series_keys: Set[Tuple[str, str, str]] = set()
                existing_series_data: Dict[Tuple[str, str, str], Dict[str, Any]] = {}
                if actualizar and not autonumerar:
                    q_ser = (
                        session.query(
                            InvoiceHeader.invoice_number,
                            LineItem.line_number,
                            Serie.row,
                            Serie.serial_numbers,
                            Serie.model,
                            Serie.sub_model,
                            Serie.number_id,
                        )
                        .join(LineItem, LineItem.invoice_id == InvoiceHeader.id)
                        .join(Serie, Serie.line_item_id == LineItem.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, ln, rw, sn, md, sm, nid in q_ser.all():
                        if num is not None:
                            k = (str(num).strip(), str(ln).strip(), str(rw).strip())
                            existing_series_keys.add(k)
                            existing_series_data.setdefault(k, {
                                "serial_numbers": sn or "",
                                "model": md or "",
                                "sub_model": sm or "",
                                "number_id": nid or "",
                            })

                csv_series_count_so_far: Dict[Tuple[str, str], int] = {}

                with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in, open(error_path, "w", encoding="utf-8") as f_err:
                    sample = f_in.read(2048)
                    f_in.seek(0)
                    try:
                        dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                    except Exception:
                        dialect = "excel"
                    reader = csv.DictReader(f_in, dialect=dialect)
                    errors_detail = []
                    error_lines_list: List[int] = []
                    for i, row in enumerate(reader, start=1):
                        self.update_state(state="PROGRESS", meta={"current": i, "total": total_rows, "errors": len(error_lines_list)})
                        row_norm = row_from_template(row, "imp_def_series", normalize_header)
                        warnings_list: List[Dict[str, Any]] = []
                        row_errors = validate_row_series_impo_def(
                            row_norm,
                            i,
                            actualizar=actualizar,
                            autonumerar=autonumerar,
                            validar_series_exception=validar_series_exception,
                            invoice_id_by_number=invoice_id_by_number,
                            invoice_processed_by_number=invoice_processed_by_number,
                            partida_max_series=partida_max_series,
                            csv_series_count_so_far=csv_series_count_so_far,
                            existing_series_keys=existing_series_keys,
                            existing_series_data=existing_series_data,
                            warnings=warnings_list,
                        )
                        blocking = [e for e in row_errors if not e.get("warning")]
                        if blocking:
                            error_count += 1
                            error_lines_list.append(i)
                            for e in blocking:
                                f_err.write(
                                    json.dumps(
                                        {
                                            "line": e["line"],
                                            "col": e.get("col", ""),
                                            "msg": e.get("msg", ""),
                                            "solution": e.get("solution", ""),
                                        }
                                    )
                                    + "\n"
                                )
                        else:
                            inv_num = (row_norm.get("NUMERO FACTURA") or row_norm.get("NUM FACTURA") or "").strip()
                            line_fac = (row_norm.get("LINEA FACTURA") or row_norm.get("LINEA") or row_norm.get("PARTIDA") or "").strip()
                            if inv_num and line_fac:
                                key_csv = (inv_num, line_fac)
                                csv_series_count_so_far[key_csv] = csv_series_count_so_far.get(key_csv, 0) + 1
                        for e in row_errors + warnings_list:
                            if len(errors_detail) < 5000:
                                errors_detail.append(
                                    {
                                        "line": e["line"],
                                        "col": e.get("col", ""),
                                        "msg": e.get("msg", ""),
                                        "solution": e.get("solution", ""),
                                        "warning": bool(e.get("warning", False)),
                                    }
                                )
                        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, total_rows_in_file=total_rows
                )
        except Exception as e:
            logger.exception("Series importación definitiva scan failed: %s", e)
            return {"status": "failed", "error": str(e)}

    # --- Series de Exportación Definitiva: flujo exp_def_series (Clarion VALIDA_TODA_SERIES_EXPO / VALIDA_PARCIAL) ---
    if model_target == "invoice_series" and template_id == "exp_def_series":
        logger.info("Series expo scan: running validation for job %s", job_id)
        try:
            from api.v1.modules.a76.invoices.models import InvoiceHeader
            from api.v1.modules.a76.items.models import LineItem
            from api.v1.modules.a76.items.series.models import Serie
            from api.v1.modules.a76.items.line_quantities.models import LineQuantity
            from api.v1.modules.a76.general_catalogs.company.models import Company
            from .validators.series_expo import validate_row_series_expo

            _fc = parse_footer_config(meta.get("footer_config"))
            autonumerar = meta.get("autonumerar", True)
            actualizar = meta.get("actualizar", False)
            validar_series_exception = meta.get("validar_series", False)
            if _fc:
                if "autonumerar" in _fc:
                    autonumerar = bool(_fc["autonumerar"])
                elif _fc.get("autonumber_series", "true") is not None:
                    autonumerar = str(_fc.get("autonumber_series", "true")).lower() in ("true", "1", "si", "sí", "yes")
                if "actualizar" in _fc:
                    actualizar = bool(_fc["actualizar"])
                if "validar_series" in _fc:
                    validar_series_exception = bool(_fc["validar_series"])

            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 ""

                q_inv_expo = (
                    session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.status)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == "exp",
                    )
                )
                invoice_id_by_number: Dict[str, int] = {}
                invoice_processed_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_processed_by_number[str(num).strip()] = _is_invoice_processed_status(is_upd)

                partida_max_series: Dict[Tuple[str, str], int] = {}
                q_qty = (
                    session.query(
                        InvoiceHeader.invoice_number,
                        LineItem.line_number,
                        LineQuantity.quantity,
                    )
                    .join(LineItem, LineItem.invoice_id == InvoiceHeader.id)
                    .outerjoin(LineQuantity, LineQuantity.item_line_id == LineItem.id)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == "exp",
                    )
                )
                for num, ln, qty in q_qty.all():
                    if num is not None and ln is not None:
                        key = (str(num).strip(), str(ln).strip())
                        if qty is not None:
                            partida_max_series[key] = int(qty) if qty else 0
                        else:
                            partida_max_series[key] = 0

                existing_series_keys: Set[Tuple[str, str, str]] = set()
                existing_series_data: Dict[Tuple[str, str, str], Dict[str, Any]] = {}
                if actualizar and not autonumerar:
                    q_ser = (
                        session.query(
                            InvoiceHeader.invoice_number,
                            LineItem.line_number,
                            Serie.row,
                            Serie.serial_numbers,
                            Serie.model,
                            Serie.sub_model,
                            Serie.number_id,
                        )
                        .join(LineItem, LineItem.invoice_id == InvoiceHeader.id)
                        .join(Serie, Serie.line_item_id == LineItem.id)
                        .filter(
                            InvoiceHeader.tenant_id == tenant_id,
                            InvoiceHeader.company_id == company_id,
                            InvoiceHeader.operation_type == "exp",
                        )
                    )
                    for num, ln, rw, sn, md, sm, nid in q_ser.all():
                        if num is not None:
                            k = (str(num).strip(), str(ln).strip(), str(rw).strip())
                            existing_series_keys.add(k)
                            existing_series_data.setdefault(k, {
                                "serial_numbers": sn or "",
                                "model": md or "",
                                "sub_model": sm or "",
                                "number_id": nid or "",
                            })

                csv_series_count_so_far: Dict[Tuple[str, str], int] = {}

            invoice_numbers_from_csv: Set[str] = set()
            with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in:
                sample = f_in.read(2048)
                f_in.seek(0)
                try:
                    dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                except Exception:
                    dialect = "excel"
                reader = csv.DictReader(f_in, dialect=dialect)
                for row in reader:
                    row_norm = row_from_template(row, "exp_def_series", normalize_header)
                    inv = (row_norm.get("NUMERO FACTURA") or row_norm.get("NUM FACTURA") 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: Set[str] = invoice_numbers_from_csv
            else:
                rfc_exception_updated = set()

            with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in, open(error_path, "w", encoding="utf-8") as f_err:
                f_in.seek(0)
                try:
                    dialect = common_csv_reader.detect_csv_dialect(f_in.read(2048), delimiters=",;\t")
                except Exception:
                    dialect = "excel"
                f_in.seek(0)
                reader = csv.DictReader(f_in, dialect=dialect)
                errors_detail = []
                error_lines_list: List[int] = []
                for i, row in enumerate(reader, start=1):
                    self.update_state(state="PROGRESS", meta={"current": i, "total": total_rows, "errors": len(error_lines_list)})
                    row_norm = row_from_template(row, "exp_def_series", normalize_header)
                    warnings_list: List[Dict[str, Any]] = []
                    row_errors = validate_row_series_expo(
                        row_norm,
                        i,
                        actualizar=actualizar,
                        autonumerar=autonumerar,
                        validar_series_exception=validar_series_exception,
                        invoice_id_by_number=invoice_id_by_number,
                        invoice_processed_by_number=invoice_processed_by_number,
                        rfc_exception_updated=rfc_exception_updated,
                        partida_max_series=partida_max_series,
                        csv_series_count_so_far=csv_series_count_so_far,
                        existing_series_keys=existing_series_keys,
                        existing_series_data=existing_series_data,
                        warnings=warnings_list,
                    )
                    blocking = [e for e in row_errors if not e.get("warning")]
                    if blocking:
                        error_count += 1
                        error_lines_list.append(i)
                        for e in blocking:
                            f_err.write(
                                json.dumps(
                                    {
                                        "line": e["line"],
                                        "col": e.get("col", ""),
                                        "msg": e.get("msg", ""),
                                        "solution": e.get("solution", ""),
                                    }
                                )
                                + "\n"
                            )
                    else:
                        inv_num = (row_norm.get("NUMERO FACTURA") or row_norm.get("NUM FACTURA") or row_norm.get("FACTURA EXPO") or "").strip()
                        line_fac = (row_norm.get("LINEA FACTURA") or row_norm.get("LINEA") or row_norm.get("PARTIDA") or "").strip()
                        if inv_num and line_fac:
                            key_csv = (inv_num, line_fac)
                            csv_series_count_so_far[key_csv] = csv_series_count_so_far.get(key_csv, 0) + 1
                    for e in row_errors + warnings_list:
                        if len(errors_detail) < 5000:
                            errors_detail.append(
                                {
                                    "line": e["line"],
                                    "col": e.get("col", ""),
                                    "msg": e.get("msg", ""),
                                    "solution": e.get("solution", ""),
                                    "warning": bool(e.get("warning", False)),
                                }
                            )
                    processed_rows += 1

            # El commit usa líneas únicas para omitir filas; el resumen preliminar
            # debe usar la misma base para evitar discrepancias de válidos/errores.
            unique_error_lines = sorted(set(error_lines_list))
            error_count = len(unique_error_lines)
            common_storage.store_error_lines(effective_job_type, job_id, unique_error_lines)
            return common_responses.scan_result(
                job_id, processed_rows, error_count, errors_detail, total_rows_in_file=total_rows
            )
        except Exception as e:
            logger.exception("Series exportación definitiva scan failed: %s", e)
            return {"status": "failed", "error": str(e)}

    # --- Series Compras Mexicanas: misma lógica que Impo Def, facturas MEX ---
    if model_target == "invoice_series" and template_id == "cmex_series":
        try:
            from api.v1.modules.a76.invoices.models import InvoiceHeader
            from api.v1.modules.a76.items.models import LineItem
            from api.v1.modules.a76.items.series.models import Serie
            from api.v1.modules.a76.items.line_quantities.models import LineQuantity
            from .validators.series_impo_def import validate_row_series_impo_def

            actualizar = meta.get("actualizar", False)
            autonumerar = meta.get("autonumerar", True)
            validar_series_exception = meta.get("validar_series", False)
            _fc = parse_footer_config(meta.get("footer_config"))
            if _fc:
                if "actualizar" in _fc:
                    actualizar = bool(_fc["actualizar"])
                elif _fc.get("mode") == "update":
                    actualizar = True
                elif _fc.get("mode") == "replace":
                    actualizar = False
                if "autonumerar" in _fc:
                    autonumerar = bool(_fc["autonumerar"])
                else:
                    as_val = _fc.get("autonumber_series", "true")
                    autonumerar = str(as_val).lower() in ("true", "1", "si", "sí", "yes")
                if "validar_series" in _fc:
                    validar_series_exception = bool(_fc["validar_series"])

            with CoreSessionLocal() as session:
                q = (
                    session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.status)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == "imp",
                        InvoiceHeader.invoice_type == "MEX",
                    )
                )
                rows_inv = q.all()
                invoice_id_by_number: Dict[str, int] = {}
                invoice_processed_by_number: Dict[str, bool] = {}
                for num, iid, is_upd in rows_inv:
                    if num:
                        invoice_id_by_number[str(num).strip()] = iid
                        invoice_processed_by_number[str(num).strip()] = _is_invoice_processed_status(is_upd)

                partida_max_series: Dict[Tuple[str, str], int] = {}
                q_qty = (
                    session.query(
                        InvoiceHeader.invoice_number,
                        LineItem.line_number,
                        LineQuantity.quantity,
                    )
                    .join(LineItem, LineItem.invoice_id == InvoiceHeader.id)
                    .outerjoin(LineQuantity, LineQuantity.item_line_id == LineItem.id)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == "imp",
                        InvoiceHeader.invoice_type == "MEX",
                    )
                )
                for num, ln, qty in q_qty.all():
                    if num is not None and ln is not None:
                        key = (str(num).strip(), str(ln).strip())
                        if qty is not None:
                            partida_max_series[key] = int(qty) if qty else 0
                        else:
                            partida_max_series[key] = 0

                existing_series_keys: Set[Tuple[str, str, str]] = set()
                existing_series_data: Dict[Tuple[str, str, str], Dict[str, Any]] = {}
                if actualizar and not autonumerar:
                    q_ser = (
                        session.query(
                            InvoiceHeader.invoice_number,
                            LineItem.line_number,
                            Serie.row,
                            Serie.serial_numbers,
                            Serie.model,
                            Serie.sub_model,
                            Serie.number_id,
                        )
                        .join(LineItem, LineItem.invoice_id == InvoiceHeader.id)
                        .join(Serie, Serie.line_item_id == LineItem.id)
                        .filter(
                            InvoiceHeader.tenant_id == tenant_id,
                            InvoiceHeader.company_id == company_id,
                            InvoiceHeader.operation_type == "imp",
                            InvoiceHeader.invoice_type == "MEX",
                        )
                    )
                    for num, ln, rw, sn, md, sm, nid in q_ser.all():
                        if num is not None:
                            k = (str(num).strip(), str(ln).strip(), str(rw).strip())
                            existing_series_keys.add(k)
                            existing_series_data.setdefault(k, {
                                "serial_numbers": sn or "",
                                "model": md or "",
                                "sub_model": sm or "",
                                "number_id": nid or "",
                            })

                csv_series_count_so_far: Dict[Tuple[str, str], int] = {}

                with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in, open(error_path, "w", encoding="utf-8") as f_err:
                    sample = f_in.read(2048)
                    f_in.seek(0)
                    try:
                        dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                    except Exception:
                        dialect = "excel"
                    reader = csv.DictReader(f_in, dialect=dialect)
                    errors_detail = []
                    error_lines_list: List[int] = []
                    for i, row in enumerate(reader, start=1):
                        self.update_state(state="PROGRESS", meta={"current": i, "total": total_rows, "errors": len(error_lines_list)})
                        row_norm = row_from_template(row, "cmex_series", normalize_header)
                        warnings_list: List[Dict[str, Any]] = []
                        row_errors = validate_row_series_impo_def(
                            row_norm,
                            i,
                            actualizar=actualizar,
                            autonumerar=autonumerar,
                            validar_series_exception=validar_series_exception,
                            invoice_id_by_number=invoice_id_by_number,
                            invoice_processed_by_number=invoice_processed_by_number,
                            partida_max_series=partida_max_series,
                            csv_series_count_so_far=csv_series_count_so_far,
                            existing_series_keys=existing_series_keys,
                            existing_series_data=existing_series_data,
                            warnings=warnings_list,
                            catalog_label="Compras Mexicanas",
                        )
                        blocking = [e for e in row_errors if not e.get("warning")]
                        if blocking:
                            error_count += 1
                            error_lines_list.append(i)
                            for e in blocking:
                                f_err.write(
                                    json.dumps(
                                        {
                                            "line": e["line"],
                                            "col": e.get("col", ""),
                                            "msg": e.get("msg", ""),
                                            "solution": e.get("solution", ""),
                                        }
                                    )
                                    + "\n"
                                )
                        else:
                            inv_num = (row_norm.get("NUMERO FACTURA") or row_norm.get("NUM FACTURA") or "").strip()
                            line_fac = (row_norm.get("LINEA FACTURA") or row_norm.get("LINEA") or row_norm.get("PARTIDA") or "").strip()
                            if inv_num and line_fac:
                                key_csv = (inv_num, line_fac)
                                csv_series_count_so_far[key_csv] = csv_series_count_so_far.get(key_csv, 0) + 1
                        for e in row_errors + warnings_list:
                            if len(errors_detail) < 5000:
                                errors_detail.append(
                                    {
                                        "line": e["line"],
                                        "col": e.get("col", ""),
                                        "msg": e.get("msg", ""),
                                        "solution": e.get("solution", ""),
                                        "warning": bool(e.get("warning", False)),
                                    }
                                )
                        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, total_rows_in_file=total_rows
                )
        except Exception as e:
            logger.exception("Series Compras Mexicanas scan failed: %s", e)
            return {"status": "failed", "error": str(e)}

    # --- Series de Importación Temporal: flujo específico (Clarion VALIDA_TODA / VALIDA_PARCIAL) ---
    if model_target == "invoice_series":
        try:
            from api.v1.modules.a76.invoices.models import InvoiceHeader
            from api.v1.modules.a76.items.models import LineItem
            from api.v1.modules.a76.items.series.models import Serie
            from .validators.series_impo_temp import (
                validate_row_series_impo_temp,
            )

            actualizar = meta.get("actualizar", False)
            autonumerar = meta.get("autonumerar", True)
            validar_series_exception = meta.get("validar_series", False)
            _fc = parse_footer_config(meta.get("footer_config"))
            if _fc:
                if "actualizar" in _fc:
                    actualizar = bool(_fc["actualizar"])
                elif _fc.get("mode") == "update":
                    actualizar = True
                elif _fc.get("mode") == "replace":
                    actualizar = False
                if "autonumerar" in _fc:
                    autonumerar = bool(_fc["autonumerar"])
                else:
                    as_val = _fc.get("autonumber_series", "true")
                    autonumerar = str(as_val).lower() in ("true", "1", "si", "sí", "yes")
                if "validar_series" in _fc:
                    validar_series_exception = bool(_fc["validar_series"])

            with CoreSessionLocal() as session:
                # Invoice lookup: imp + TEM
                q = (
                    session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.status)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == "imp",
                        InvoiceHeader.invoice_type == "TEM",
                    )
                )
                rows_inv = q.all()
                invoice_id_by_number: Dict[str, int] = {}
                invoice_processed_by_number: Dict[str, bool] = {}
                for num, iid, is_upd in rows_inv:
                    if num:
                        invoice_id_by_number[str(num).strip()] = iid
                        invoice_processed_by_number[str(num).strip()] = _is_invoice_processed_status(is_upd)

                # Existing series keys: (invoice_number, linea_factura, linea_serie)
                existing_series_keys: Set[Tuple[str, str, str]] = set()
                existing_series_data: Dict[Tuple[str, str, str], Dict[str, Any]] = {}
                if actualizar and not autonumerar:
                    q_ser = (
                        session.query(
                            InvoiceHeader.invoice_number,
                            LineItem.line_number,
                            Serie.row,
                            Serie.serial_numbers,
                            Serie.model,
                            Serie.sub_model,
                            Serie.number_id,
                        )
                        .join(LineItem, LineItem.invoice_id == InvoiceHeader.id)
                        .join(Serie, Serie.line_item_id == LineItem.id)
                        .filter(
                            InvoiceHeader.tenant_id == tenant_id,
                            InvoiceHeader.company_id == company_id,
                            InvoiceHeader.operation_type == "imp",
                            InvoiceHeader.invoice_type == "TEM",
                        )
                    )
                    for num, ln, rw, sn, md, sm, nid in q_ser.all():
                        if num is not None:
                            k = (str(num).strip(), str(ln).strip(), str(rw).strip())
                            existing_series_keys.add(k)
                            existing_series_data.setdefault(k, {
                                "serial_numbers": sn or "",
                                "model": md or "",
                                "sub_model": sm or "",
                                "number_id": nid or "",
                            })

            with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in, open(error_path, "w", encoding="utf-8") as f_err:
                sample = f_in.read(2048)
                f_in.seek(0)
                try:
                    dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                except Exception:
                    dialect = "excel"
                reader = csv.DictReader(f_in, dialect=dialect)
                errors_detail: List[Dict[str, Any]] = []
                error_lines_list: List[int] = []
                for i, row in enumerate(reader, start=1):
                    self.update_state(state="PROGRESS", meta={"current": i, "total": total_rows, "errors": len(error_lines_list)})
                    row_norm = row_from_template(row, "imp_temp_series", normalize_header)
                    warnings_list: List[Dict[str, Any]] = []
                    row_errors = validate_row_series_impo_temp(
                        row_norm,
                        i,
                        actualizar=actualizar,
                        autonumerar=autonumerar,
                        validar_series_exception=validar_series_exception,
                        invoice_id_by_number=invoice_id_by_number,
                        invoice_processed_by_number=invoice_processed_by_number,
                        existing_series_keys=existing_series_keys,
                        existing_series_data=existing_series_data,
                        warnings=warnings_list,
                    )
                    blocking = [e for e in row_errors if not e.get("warning")]
                    if blocking:
                        error_count += 1
                        error_lines_list.append(i)
                        for e in blocking:
                            f_err.write(
                                json.dumps(
                                    {
                                        "line": e["line"],
                                        "col": e.get("col", ""),
                                        "msg": e.get("msg", ""),
                                        "solution": e.get("solution", ""),
                                    }
                                )
                                + "\n"
                            )
                    for e in row_errors + warnings_list:
                        if len(errors_detail) < 5000:
                            errors_detail.append(
                                {
                                    "line": e["line"],
                                    "col": e.get("col", ""),
                                    "msg": e.get("msg", ""),
                                    "solution": e.get("solution", ""),
                                    "warning": bool(e.get("warning", False)),
                                }
                            )
                    processed_rows += 1

            # El commit usa líneas únicas para omitir filas; el resumen preliminar
            # debe usar la misma base para evitar discrepancias de válidos/errores.
            unique_error_lines = sorted(set(error_lines_list))
            error_count = len(unique_error_lines)
            common_storage.store_error_lines(effective_job_type, job_id, unique_error_lines)
            return common_responses.scan_result(
                job_id, processed_rows, error_count, errors_detail, total_rows_in_file=total_rows
            )
        except Exception as e:
            logger.exception("Series import scan failed: %s", e)
            return {"status": "failed", "error": str(e)}

    # --- Partidas de Importación Temporal: flujo específico (Clarion VALIDA_TODA / VALIDA_PARCIAL) ---
    if model_target == "invoice_details" and template_id == "imp_temp_details":
        try:
            from api.v1.modules.a76.invoices.models import InvoiceHeader
            from api.v1.modules.a76.items.models import LineItem
            from api.v1.modules.a76.classes.models import Class
            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.countries.models import Country
            from api.v1.modules.a76.general_catalogs.sectors.models import Sector
            from api.v1.modules.public.reference_data.valuation_methods.models import ValuationMethod
            from api.v1.modules.public.reference_data.payment_methods.models import PaymentMethod
            from api.v1.modules.a76.general_catalogs.company.models import Company
            from api.v1.modules.a76.parts.models import Part
            from .validators.partidas_impo_temp import validate_row_partidas_impo_temp

            _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)
            calcular_costo_en_base_a_total = meta.get("calcular_costo_unitario_en_base_a_valor_total", 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 "calcular_costo_unitario_en_base_a_valor_total" in _fc:
                    calcular_costo_en_base_a_total = bool(_fc["calcular_costo_unitario_en_base_a_valor_total"])
                if "validar_decimales_pza" in _fc:
                    validar_decimales_pza = bool(_fc["validar_decimales_pza"])

            RFC_EXCEPTION_UPDATED = {"TPI121217SF6", "TCI170502858"}
            RFC_EXCEPTION_NUM_PARTE = {"CTE980130518"}

            with CoreSessionLocal() as session:
                # Resolver parámetros del sistema: validadecencant (decimales PZA) y validarseries (switch maestro)
                settings = AppSettingsService.get_resolved_settings(session, int(tenant_id), int(company_id))
                gen_params = settings.get("ssisgen", {}) or settings.get("qsisgen", {})
                if str(gen_params.get("validadecencant", "0")).strip() == "1":
                    validar_decimales_pza = True

                q_inv = (
                    session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.status)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == op_type_scan,
                        InvoiceHeader.invoice_type == inv_type_value,
                    )
                )
                invoice_id_by_number: Dict[str, int] = {}
                invoice_processed_by_number: Dict[str, bool] = {}
                for num, iid, is_upd in q_inv.all():
                    if num:
                        inv_norm = _normalize_invoice_number(num)
                        invoice_id_by_number[inv_norm] = iid
                        invoice_processed_by_number[inv_norm] = _is_invoice_processed_status(is_upd)

                existing_line_keys_by_invoice: Dict[str, Set[str]] = {}
                q_li = (
                    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 == op_type_scan,
                        InvoiceHeader.invoice_type == inv_type_value,
                    )
                )
                for num, ln in q_li.all():
                    if num is not None:
                        key = _normalize_invoice_number(num)
                        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,
                            FaLineItem.is_subitem == False,
                            FaLineItem.contains_subitems == True,
                        )
                    )
                    for num, ln in q_pp.all():
                        if num is not None:
                            partidas_principales_bd.add((_normalize_invoice_number(num), str(ln).strip()))
                except Exception:
                    pass

                valid_class_codes: Set[str] = set()
                class_um_by_code: Dict[str, str] = {}
                class_fraction_by_code: Dict[str, str] = {}
                class_desc_es_by_code: Dict[str, str] = {}
                class_desc_en_by_code: Dict[str, str] = {}
                for c in session.query(Class).filter(Class.tenant_id == tenant_id, Class.company_id == company_id).all():
                    code = (c.class_code or "").strip().upper()
                    if code:
                        valid_class_codes.add(code)
                        class_um_by_code[code] = (c.unit_of_measure or "").strip().upper()
                        class_fraction_by_code[code] = (c.fraction or "").strip()
                        class_desc_es_by_code[code] = (c.description_es or "").strip()
                        class_desc_en_by_code[code] = (c.description_en or "").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_country_keys: Set[str] = set()
                for row in session.query(Country.m3_key, Country.ame_key).all():
                    if row[0]:
                        valid_country_keys.add((row[0] or "").strip().upper())
                    if row[1]:
                        valid_country_keys.add((row[1] or "").strip().upper())

                valid_fraction_ame: Set[str] = set()  # unused; SITAR resolve per row in validators

                authorized_sectors: Set[str] = set()
                for row in session.query(Sector.key).filter(
                    Sector.authorized == True,
                    Sector.tenant_id == tenant_id,
                    Sector.company_id == company_id,
                ).all():
                    if row[0]:
                        authorized_sectors.add((row[0] or "").strip().upper())

                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_valuation_methods: Set[str] = set()
                for row in session.query(ValuationMethod.key).all():
                    if row[0]:
                        valid_valuation_methods.add((row[0] or "").strip())

                company = session.query(Company).filter(Company.id == company_id).first()
                company_has_prosec = bool(company.prosec) if company else False
                company_rfc = (company.rfc or "").strip().upper() if company else ""

                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())

            rfc_exception_updated: Set[str] = set()
            rfc_exception_num_parte: Set[str] = set()

            with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in:
                sample = f_in.read(2048)
                f_in.seek(0)
                try:
                    dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                except Exception:
                    dialect = "excel"
                reader = csv.DictReader(f_in, dialect=dialect)
                raw_fieldnames = list(reader.fieldnames or [])
                rows_list = list(reader)

            normalized_headers = {normalize_header(h or "") for h in raw_fieldnames}
            invoice_header_aliases = {"NUMERO FACTURA", "NUM FACTURA", "FACTURA"}
            has_invoice_header = any(h in normalized_headers for h in invoice_header_aliases)
            if not has_invoice_header:
                structure_error = {
                    "line": 1,
                    "col": "NUMERO FACTURA",
                    "msg": "Error estructural: no se encontró una columna de factura (NUMERO FACTURA / NUM FACTURA / FACTURA).",
                    "solution": "Corrige el encabezado de la columna A y vuelve a subir el archivo.",
                    "warning": False,
                }
                with open(error_path, "w", encoding="utf-8") as f_err:
                    f_err.write(
                        json.dumps(
                            {
                                "line": structure_error["line"],
                                "col": structure_error["col"],
                                "msg": structure_error["msg"],
                                "solution": structure_error["solution"],
                            }
                        ) + "\n"
                    )
                common_storage.store_error_lines(effective_job_type, job_id, [1])
                return common_responses.scan_result(
                    job_id=job_id,
                    processed_rows=len(rows_list),
                    error_count=1,
                    errors_detail=[structure_error],
                    total_rows_in_file=total_rows,
                    message="Precheck estructural falló: la columna de número de factura no fue detectada.",
                )

            invoice_numbers_from_csv = set()
            empty_invoice_lines = 0
            for row in rows_list:
                row_norm = row_from_template(row, "imp_temp_details", normalize_header)
                inv = _extract_invoice_number_from_row(row_norm)
                if inv:
                    invoice_numbers_from_csv.add(inv)
                else:
                    empty_invoice_lines += 1
            if company_rfc in RFC_EXCEPTION_UPDATED:
                rfc_exception_updated = invoice_numbers_from_csv
            if company_rfc in RFC_EXCEPTION_NUM_PARTE:
                rfc_exception_num_parte = invoice_numbers_from_csv

            if rows_list:
                empty_ratio = empty_invoice_lines / len(rows_list)
                if empty_ratio >= 0.8:
                    structure_error = {
                        "line": 1,
                        "col": "NUMERO FACTURA",
                        "msg": "Error estructural: la mayoría de filas no contiene número de factura utilizable.",
                        "solution": "Verifica delimitador/encabezados del CSV y que la columna de factura esté poblada.",
                        "warning": False,
                    }
                    with open(error_path, "w", encoding="utf-8") as f_err:
                        f_err.write(
                            json.dumps(
                                {
                                    "line": structure_error["line"],
                                    "col": structure_error["col"],
                                    "msg": structure_error["msg"],
                                    "solution": structure_error["solution"],
                                }
                            ) + "\n"
                        )
                    common_storage.store_error_lines(effective_job_type, job_id, [1])
                    return common_responses.scan_result(
                        job_id=job_id,
                        processed_rows=len(rows_list),
                        error_count=1,
                        errors_detail=[structure_error],
                        total_rows_in_file=total_rows,
                        message=f"Precheck estructural falló: {empty_invoice_lines}/{len(rows_list)} filas sin número de factura.",
                    )

            line_counts_csv: Dict[Tuple[str, str], int] = {}
            partidas_principales_csv: Set[Tuple[str, str]] = set()

            def _get_row(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, "imp_temp_details", normalize_header)
                inv = _extract_invoice_number_from_row(row_norm)
                linea = _get_row(row_norm, "LINEA", "RENGLON", "PARTIDA")
                if inv and linea:
                    key = (inv, linea)
                    line_counts_csv[key] = line_counts_csv.get(key, 0) + 1
                u = _get_row(row_norm, "ES PARTIDA O SUBPARTIDA", "ESSUBPARTIDA").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):
                    self.update_state(state="PROGRESS", meta={"current": i, "total": total_rows, "errors": error_count})
                    row_norm = row_from_template(row, "imp_temp_details", normalize_header)
                    row_errors = validate_row_partidas_impo_temp(
                        row_norm,
                        i,
                        autonumerar=autonumerar,
                        actualizar=actualizar,
                        levantar_subpartidas=levantar_subpartidas,
                        calcular_costo_en_base_a_total=calcular_costo_en_base_a_total,
                        validar_decimales_pza=validar_decimales_pza,
                        invoice_id_by_number=invoice_id_by_number,
                        invoice_processed_by_number=invoice_processed_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_class_codes=valid_class_codes,
                        class_um_by_code=class_um_by_code,
                        class_fraction_by_code=class_fraction_by_code,
                        class_desc_es_by_code=class_desc_es_by_code,
                        class_desc_en_by_code=class_desc_en_by_code,
                        valid_uom_codes=valid_uom_codes,
                        valid_bulks_codes=valid_bulks_codes,
                        valid_country_keys=valid_country_keys,
                        valid_fraction_ame=valid_fraction_ame,
                        valid_payment_methods=valid_payment_methods,
                        valid_valuation_methods=valid_valuation_methods,
                        authorized_sectors=authorized_sectors,
                        company_has_prosec=company_has_prosec,
                        rfc_exception_num_parte=rfc_exception_num_parte or None,
                        valid_part_numbers=valid_part_numbers,
                        warnings=None,
                    )
                    blocking = [e for e in row_errors if not e.get("warning")]
                    if blocking:
                        error_count += 1
                        error_lines_list.append(i)
                        for e in blocking:
                            f_err.write(
                                json.dumps(
                                    {
                                        "line": e["line"],
                                        "col": e.get("col", ""),
                                        "msg": e.get("msg", ""),
                                        "solution": e.get("solution", ""),
                                    }
                                )
                                + "\n"
                            )
                    for e in row_errors:
                        if len(errors_detail) < 5000:
                            errors_detail.append(
                                {
                                    "line": e["line"],
                                    "col": e.get("col", ""),
                                    "msg": e.get("msg", ""),
                                    "solution": e.get("solution", ""),
                                    "warning": bool(e.get("warning", False)),
                                }
                            )
                    processed_rows += 1

            unique_error_lines = sorted(set(error_lines_list))
            error_count = len(unique_error_lines)
            common_storage.store_error_lines(effective_job_type, job_id, unique_error_lines)
            scan_message = None
            if processed_rows > 0 and errors_detail:
                reason_counts: Dict[str, int] = {}
                for e in errors_detail:
                    if e.get("warning"):
                        continue
                    msg = str(e.get("msg", "")).strip()
                    if not msg:
                        continue
                    reason_counts[msg] = reason_counts.get(msg, 0) + 1
                if reason_counts:
                    top_reason, top_count = max(reason_counts.items(), key=lambda kv: kv[1])
                    if top_count / max(processed_rows, 1) >= 0.8:
                        scan_message = (
                            f"Causa dominante de rechazo ({top_count}/{processed_rows}): {top_reason}. "
                            f"contexto invoice_type={inv_type_value}, operation_type={op_type_scan}, template_id={template_id}."
                        )
                        logger.warning("Scan partidas causa dominante: %s", scan_message)
            return common_responses.scan_result(
                job_id,
                processed_rows,
                error_count,
                errors_detail,
                total_rows_in_file=total_rows,
                message=scan_message,
            )
        except Exception as e:
            logger.exception("Partidas import scan failed: %s", e)
            return {"status": "failed", "error": str(e)}

    # --- Partidas de Importación Definitiva: flujo específico (Clarion VALIDA_TODA_PARIMPO_DEF / VALIDA_PARCIAL) ---
    # Estructura de columnas: misma que partidas TEM (imp_def_details resuelve a imp_temp_details). Facturas DEF/MATDE/EXDEF.
    if model_target == "invoice_details" and template_id == "imp_def_details":
        try:
            from api.v1.modules.a76.invoices.models import InvoiceHeader
            from api.v1.modules.a76.items.models import LineItem
            from api.v1.modules.a76.classes.models import Class
            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.countries.models import Country
            from api.v1.modules.a76.general_catalogs.sectors.models import Sector
            from api.v1.modules.public.reference_data.valuation_methods.models import ValuationMethod
            from api.v1.modules.public.reference_data.payment_methods.models import PaymentMethod
            from api.v1.modules.a76.general_catalogs.company.models import Company
            from api.v1.modules.a76.parts.models import Part
            from .validators.partidas_impo_def import validate_row_partidas_impo_def

            DEF_INVOICE_TYPES = ("DEF", "MATDE", "EXDEF")
            _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)
            calcular_costo_en_base_a_total = meta.get("calcular_costo_unitario_en_base_a_valor_total", 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 "calcular_costo_unitario_en_base_a_valor_total" in _fc:
                    calcular_costo_en_base_a_total = bool(_fc["calcular_costo_unitario_en_base_a_valor_total"])
                if "validar_decimales_pza" in _fc:
                    validar_decimales_pza = bool(_fc["validar_decimales_pza"])

            RFC_EXCEPTION_UPDATED = {"TPI121217SF6", "TCI170502858"}
            RFC_EXCEPTION_NUM_PARTE = {"CTE980130518"}

            with CoreSessionLocal() as session:
                # Resolver parámetros del sistema: validadecencant (decimales PZA) y validarseries (switch maestro)
                settings = AppSettingsService.get_resolved_settings(session, int(tenant_id), int(company_id))
                gen_params = settings.get("ssisgen", {}) or settings.get("qsisgen", {})
                if str(gen_params.get("validadecencant", "0")).strip() == "1":
                    validar_decimales_pza = True

                q_inv = (
                    session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.status)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == "imp",
                        InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES),
                    )
                )
                invoice_id_by_number: Dict[str, int] = {}
                invoice_processed_by_number: Dict[str, bool] = {}
                for num, iid, is_upd in q_inv.all():
                    if num:
                        invoice_id_by_number[str(num).strip()] = iid
                        invoice_processed_by_number[str(num).strip()] = _is_invoice_processed_status(is_upd)

                existing_line_keys_by_invoice: Dict[str, Set[str]] = {}
                q_li = (
                    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 == "imp",
                        InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES),
                    )
                )
                for num, ln in q_li.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 == "imp",
                            InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES),
                            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

                valid_class_codes: Set[str] = set()
                class_um_by_code: Dict[str, str] = {}
                class_fraction_by_code: Dict[str, str] = {}
                class_desc_es_by_code: Dict[str, str] = {}
                class_desc_en_by_code: Dict[str, str] = {}
                for c in session.query(Class).filter(Class.tenant_id == tenant_id, Class.company_id == company_id).all():
                    code = (c.class_code or "").strip().upper()
                    if code:
                        valid_class_codes.add(code)
                        class_um_by_code[code] = (c.unit_of_measure or "").strip().upper()
                        class_fraction_by_code[code] = (c.fraction or "").strip()
                        class_desc_es_by_code[code] = (c.description_es or "").strip()
                        class_desc_en_by_code[code] = (c.description_en or "").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_country_keys: Set[str] = set()
                for row in session.query(Country.m3_key, Country.ame_key).all():
                    if row[0]:
                        valid_country_keys.add((row[0] or "").strip().upper())
                    if row[1]:
                        valid_country_keys.add((row[1] or "").strip().upper())

                valid_fraction_ame: Set[str] = set()  # unused; SITAR resolve per row in validators

                authorized_sectors: Set[str] = set()
                for row in session.query(Sector.key).filter(
                    Sector.authorized == True,
                    Sector.tenant_id == tenant_id,
                    Sector.company_id == company_id,
                ).all():
                    if row[0]:
                        authorized_sectors.add((row[0] or "").strip().upper())

                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_valuation_methods: Set[str] = set()
                for row in session.query(ValuationMethod.key).all():
                    if row[0]:
                        valid_valuation_methods.add((row[0] or "").strip())

                company = session.query(Company).filter(Company.id == company_id).first()
                company_has_prosec = bool(company.prosec) if company else False
                company_rfc = (company.rfc or "").strip().upper() if company else ""

                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())

            rfc_exception_updated: Set[str] = set()
            rfc_exception_num_parte: Set[str] = set()

            with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in:
                sample = f_in.read(2048)
                f_in.seek(0)
                try:
                    dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                except Exception:
                    dialect = "excel"
                reader = csv.DictReader(f_in, dialect=dialect)
                rows_list = list(reader)

            invoice_numbers_from_csv = set()
            for row in rows_list:
                inv = (row.get("NUMERO FACTURA") or row.get("NUM FACTURA") or row.get("FACTURA") or "").strip()
                if inv:
                    invoice_numbers_from_csv.add(inv)
            if company_rfc in RFC_EXCEPTION_UPDATED:
                rfc_exception_updated = invoice_numbers_from_csv
            if company_rfc in RFC_EXCEPTION_NUM_PARTE:
                rfc_exception_num_parte = invoice_numbers_from_csv

            line_counts_csv: Dict[Tuple[str, str], int] = {}
            partidas_principales_csv: Set[Tuple[str, str]] = set()

            def _get_row_def(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, "imp_def_details", normalize_header)
                inv = _get_row_def(row_norm, "NUMERO FACTURA", "NUM FACTURA", "FACTURA")
                linea = _get_row_def(row_norm, "LINEA", "RENGLON", "PARTIDA")
                if inv and linea:
                    key = (inv, linea)
                    line_counts_csv[key] = line_counts_csv.get(key, 0) + 1
                u = _get_row_def(row_norm, "ES PARTIDA O SUBPARTIDA", "ESSUBPARTIDA").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):
                    self.update_state(state="PROGRESS", meta={"current": i, "total": total_rows, "errors": error_count})
                    row_norm = row_from_template(row, "imp_def_details", normalize_header)
                    row_errors = validate_row_partidas_impo_def(
                        row_norm,
                        i,
                        autonumerar=autonumerar,
                        actualizar=actualizar,
                        levantar_subpartidas=levantar_subpartidas,
                        calcular_costo_en_base_a_total=calcular_costo_en_base_a_total,
                        validar_decimales_pza=validar_decimales_pza,
                        invoice_id_by_number=invoice_id_by_number,
                        invoice_processed_by_number=invoice_processed_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_class_codes=valid_class_codes,
                        class_um_by_code=class_um_by_code,
                        class_fraction_by_code=class_fraction_by_code,
                        class_desc_es_by_code=class_desc_es_by_code,
                        class_desc_en_by_code=class_desc_en_by_code,
                        valid_uom_codes=valid_uom_codes,
                        valid_bulks_codes=valid_bulks_codes,
                        valid_country_keys=valid_country_keys,
                        valid_fraction_ame=valid_fraction_ame,
                        valid_payment_methods=valid_payment_methods,
                        valid_valuation_methods=valid_valuation_methods,
                        authorized_sectors=authorized_sectors,
                        company_has_prosec=company_has_prosec,
                        rfc_exception_num_parte=rfc_exception_num_parte or None,
                        valid_part_numbers=valid_part_numbers,
                        warnings=None,
                    )
                    blocking = [e for e in row_errors if not e.get("warning")]
                    if blocking:
                        error_count += 1
                        error_lines_list.append(i)
                        for e in blocking:
                            f_err.write(
                                json.dumps(
                                    {
                                        "line": e["line"],
                                        "col": e.get("col", ""),
                                        "msg": e.get("msg", ""),
                                        "solution": e.get("solution", ""),
                                    }
                                )
                                + "\n"
                            )
                    for e in row_errors:
                        if len(errors_detail) < 5000:
                            errors_detail.append(
                                {
                                    "line": e["line"],
                                    "col": e.get("col", ""),
                                    "msg": e.get("msg", ""),
                                    "solution": e.get("solution", ""),
                                    "warning": bool(e.get("warning", False)),
                                }
                            )
                    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 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":
        logger.info("Partidas expo scan: running validation for job %s", job_id)
        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.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:
                # Resolver parámetros del sistema: validadecencant (decimales PZA) y validarseries (switch maestro)
                settings = AppSettingsService.get_resolved_settings(session, int(tenant_id), int(company_id))
                gen_params = settings.get("ssisgen", {}) or settings.get("qsisgen", {})
                if str(gen_params.get("validadecencant", "0")).strip() == "1":
                    validar_decimales_pza = True

                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.status)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == "exp",
                    )
                )
                invoice_id_by_number: Dict[str, int] = {}
                invoice_processed_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_processed_by_number[str(num).strip()] = _is_invoice_processed_status(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()  # unused; SITAR resolve per row in validators

                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=_facturas_csv_encoding(file_path)) as f_in:
                sample = f_in.read(2048)
                f_in.seek(0)
                try:
                    dialect = common_csv_reader.detect_csv_dialect(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):
                    row_norm = row_from_template(row, "exp_def_partidas", normalize_header)
                    row_errors = validate_row_partidas_expo(
                        row_norm,
                        i,
                        autonumerar=autonumerar,
                        actualizar=actualizar,
                        levantar_subpartidas=levantar_subpartidas,
                        invoice_id_by_number=invoice_id_by_number,
                        invoice_processed_by_number=invoice_processed_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,
                    )
                    blocking = [e for e in row_errors if not e.get("warning")]
                    if blocking:
                        error_count += 1
                        error_lines_list.append(i)
                        for e in blocking:
                            f_err.write(
                                json.dumps(
                                    {
                                        "line": e["line"],
                                        "col": e.get("col", ""),
                                        "msg": e.get("msg", ""),
                                        "solution": e.get("solution", ""),
                                    }
                                )
                                + "\n"
                            )
                    for e in row_errors:
                        if len(errors_detail) < 5000:
                            errors_detail.append(
                                {
                                    "line": e["line"],
                                    "col": e.get("col", ""),
                                    "msg": e.get("msg", ""),
                                    "solution": e.get("solution", ""),
                                    "warning": bool(e.get("warning", False)),
                                }
                            )
                    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, total_rows_in_file=total_rows
            )
        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:
            from api.v1.modules.a76.invoices.models import InvoiceHeader
            from api.v1.modules.a76.items.models import LineItem
            from api.v1.modules.a76.classes.models import Class
            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.countries.models import Country
            from api.v1.modules.a76.general_catalogs.sectors.models import Sector
            from api.v1.modules.public.reference_data.valuation_methods.models import ValuationMethod
            from api.v1.modules.public.reference_data.payment_methods.models import PaymentMethod
            from api.v1.modules.a76.general_catalogs.company.models import Company
            from api.v1.modules.a76.parts.models import Part
            from .validators.partidas_impo_def import validate_row_partidas_impo_def

            _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)
            calcular_costo_en_base_a_total = meta.get("calcular_costo_unitario_en_base_a_valor_total", 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 "calcular_costo_unitario_en_base_a_valor_total" in _fc:
                    calcular_costo_en_base_a_total = bool(_fc["calcular_costo_unitario_en_base_a_valor_total"])
                if "validar_decimales_pza" in _fc:
                    validar_decimales_pza = bool(_fc["validar_decimales_pza"])

            RFC_EXCEPTION_UPDATED = {"TPI121217SF6", "TCI170502858"}
            RFC_EXCEPTION_NUM_PARTE = {"CTE980130518"}

            with CoreSessionLocal() as session:
                # Resolver parámetros del sistema: validadecencant (decimales PZA) y validarseries (switch maestro)
                settings = AppSettingsService.get_resolved_settings(session, int(tenant_id), int(company_id))
                gen_params = settings.get("ssisgen", {}) or settings.get("qsisgen", {})
                if str(gen_params.get("validadecencant", "0")).strip() == "1":
                    validar_decimales_pza = True

                q_inv = (
                    session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.status)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == "imp",
                        InvoiceHeader.invoice_type == "MEX",
                    )
                )
                invoice_id_by_number: Dict[str, int] = {}
                invoice_processed_by_number: Dict[str, bool] = {}
                for num, iid, is_upd in q_inv.all():
                    if num:
                        invoice_id_by_number[str(num).strip()] = iid
                        invoice_processed_by_number[str(num).strip()] = _is_invoice_processed_status(is_upd)

                existing_line_keys_by_invoice: Dict[str, Set[str]] = {}
                q_li = (
                    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 == "imp",
                        InvoiceHeader.invoice_type == "MEX",
                    )
                )
                for num, ln in q_li.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 == "imp",
                            InvoiceHeader.invoice_type == "MEX",
                            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

                valid_class_codes: Set[str] = set()
                class_um_by_code: Dict[str, str] = {}
                class_fraction_by_code: Dict[str, str] = {}
                class_desc_es_by_code: Dict[str, str] = {}
                class_desc_en_by_code: Dict[str, str] = {}
                for c in session.query(Class).filter(Class.tenant_id == tenant_id, Class.company_id == company_id).all():
                    code = (c.class_code or "").strip().upper()
                    if code:
                        valid_class_codes.add(code)
                        class_um_by_code[code] = (c.unit_of_measure or "").strip().upper()
                        class_fraction_by_code[code] = (c.fraction or "").strip()
                        class_desc_es_by_code[code] = (c.description_es or "").strip()
                        class_desc_en_by_code[code] = (c.description_en or "").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_country_keys: Set[str] = set()
                for row in session.query(Country.m3_key, Country.ame_key).all():
                    if row[0]:
                        valid_country_keys.add((row[0] or "").strip().upper())
                    if row[1]:
                        valid_country_keys.add((row[1] or "").strip().upper())

                valid_fraction_ame: Set[str] = set()  # unused; SITAR resolve per row in validators

                authorized_sectors: Set[str] = set()
                for row in session.query(Sector.key).filter(
                    Sector.authorized == True,
                    Sector.tenant_id == tenant_id,
                    Sector.company_id == company_id,
                ).all():
                    if row[0]:
                        authorized_sectors.add((row[0] or "").strip().upper())

                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_valuation_methods: Set[str] = set()
                for row in session.query(ValuationMethod.key).all():
                    if row[0]:
                        valid_valuation_methods.add((row[0] or "").strip())

                company = session.query(Company).filter(Company.id == company_id).first()
                company_has_prosec = bool(company.prosec) if company else False
                company_rfc = (company.rfc or "").strip().upper() if company else ""

                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())

            rfc_exception_updated: Set[str] = set()
            rfc_exception_num_parte: Set[str] = set()

            with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in:
                sample = f_in.read(2048)
                f_in.seek(0)
                try:
                    dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                except Exception:
                    dialect = "excel"
                reader = csv.DictReader(f_in, dialect=dialect)
                rows_list = list(reader)

            invoice_numbers_from_csv = set()
            for row in rows_list:
                inv = (row.get("NUMERO FACTURA") or row.get("NUM FACTURA") or row.get("FACTURA") or "").strip()
                if inv:
                    invoice_numbers_from_csv.add(inv)
            if company_rfc in RFC_EXCEPTION_UPDATED:
                rfc_exception_updated = invoice_numbers_from_csv
            if company_rfc in RFC_EXCEPTION_NUM_PARTE:
                rfc_exception_num_parte = invoice_numbers_from_csv

            line_counts_csv: Dict[Tuple[str, str], int] = {}
            partidas_principales_csv: Set[Tuple[str, str]] = set()

            def _get_row_cmex(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, "cmex_details", normalize_header)
                inv = _get_row_cmex(row_norm, "NUMERO FACTURA", "NUM FACTURA", "FACTURA")
                linea = _get_row_cmex(row_norm, "LINEA", "RENGLON", "PARTIDA")
                if inv and linea:
                    key = (inv, linea)
                    line_counts_csv[key] = line_counts_csv.get(key, 0) + 1
                u = _get_row_cmex(row_norm, "ES PARTIDA O SUBPARTIDA", "ESSUBPARTIDA").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):
                    self.update_state(state="PROGRESS", meta={"current": i, "total": total_rows, "errors": error_count})
                    row_norm = row_from_template(row, "cmex_details", normalize_header)
                    row_errors = validate_row_partidas_impo_def(
                        row_norm,
                        i,
                        autonumerar=autonumerar,
                        actualizar=actualizar,
                        levantar_subpartidas=levantar_subpartidas,
                        calcular_costo_en_base_a_total=calcular_costo_en_base_a_total,
                        validar_decimales_pza=validar_decimales_pza,
                        invoice_id_by_number=invoice_id_by_number,
                        invoice_processed_by_number=invoice_processed_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_class_codes=valid_class_codes,
                        class_um_by_code=class_um_by_code,
                        class_fraction_by_code=class_fraction_by_code,
                        class_desc_es_by_code=class_desc_es_by_code,
                        class_desc_en_by_code=class_desc_en_by_code,
                        valid_uom_codes=valid_uom_codes,
                        valid_bulks_codes=valid_bulks_codes,
                        valid_country_keys=valid_country_keys,
                        valid_fraction_ame=valid_fraction_ame,
                        valid_payment_methods=valid_payment_methods,
                        valid_valuation_methods=valid_valuation_methods,
                        authorized_sectors=authorized_sectors,
                        company_has_prosec=company_has_prosec,
                        rfc_exception_num_parte=rfc_exception_num_parte or None,
                        valid_part_numbers=valid_part_numbers,
                        warnings=None,
                        catalog_label="Compras Mexicanas",
                    )
                    blocking = [e for e in row_errors if not e.get("warning")]
                    if blocking:
                        error_count += 1
                        error_lines_list.append(i)
                        for e in blocking:
                            f_err.write(
                                json.dumps(
                                    {
                                        "line": e["line"],
                                        "col": e.get("col", ""),
                                        "msg": e.get("msg", ""),
                                        "solution": e.get("solution", ""),
                                    }
                                )
                                + "\n"
                            )
                    for e in row_errors:
                        if len(errors_detail) < 5000:
                            errors_detail.append(
                                {
                                    "line": e["line"],
                                    "col": e.get("col", ""),
                                    "msg": e.get("msg", ""),
                                    "solution": e.get("solution", ""),
                                    "warning": bool(e.get("warning", False)),
                                }
                            )
                    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 Compras Mexicanas scan failed: %s", e)
            return {"status": "failed", "error": str(e)}

    # --- Encabezados de Importación Temporal: flujo específico (Clarion VALIDA_TODA / VALIDA_PARCIAL) ---
    if model_target == "invoice_header" and template_id == "imp_temp_header":
        try:
            from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceComplianceMx, InvoiceFinancials
            from api.v1.modules.a76.items.models import LineItem
            from api.v1.modules.a76.pedmientos.models.pedimentos import Pedimentos
            from api.v1.modules.a76.pedmientos.models.pedimento_dates import PedimentoDates
            from api.v1.modules.a76.clients_and_providers.models import ClientProvider
            from api.v1.modules.a76.customs_brokers.models import CustomsBroker
            from api.v1.modules.public.reference_data.customs_sections.models import CustomsSection
            from api.v1.modules.public.reference_data.incoterms.models import Incoterm
            from api.v1.modules.public.reference_data.currency_types.models import CurrencyType
            from api.v1.modules.a76.general_catalogs.exchange_rate.models import ExchangeRate
            from api.v1.modules.a76.transportation.transporters.models import Transporter
            from .validators.encabezados_impo_temp import (
                validate_row_encabezados_impo_temp,
                parse_pedimento_col_a,
                _pedimento_key_from_parsed,
            )
            from .validators.transport_catalog import logistics_scan_row_errors

            def _ped_key_from_row(ped_str: str) -> Optional[str]:
                parsed = parse_pedimento_col_a(ped_str)
                if not parsed:
                    return None
                return _pedimento_key_from_parsed(parsed[0], parsed[1], parsed[2])

            _fc = parse_footer_config(meta.get("footer_config"))
            actualizar = meta.get("actualizar", False)
            autonumerar_remesas = meta.get("autonumerar_remesas", False)
            control_remesa = bool(_fc.get("control_remesa", False))
            remesa_inicio = _fc.get("remesa_inicio")
            remesa_fin = _fc.get("remesa_fin")
            if remesa_inicio is not None:
                try:
                    remesa_inicio = int(remesa_inicio)
                except (TypeError, ValueError):
                    remesa_inicio = None
            if remesa_fin is not None:
                try:
                    remesa_fin = int(remesa_fin)
                except (TypeError, ValueError):
                    remesa_fin = None
            if _fc:
                if "actualizar" in _fc:
                    actualizar = bool(_fc["actualizar"])
                if "autonumerar_remesas" in _fc:
                    autonumerar_remesas = bool(_fc["autonumerar_remesas"])

            with CoreSessionLocal() as session:
                q_inv = (
                    session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.status)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == "imp",
                        InvoiceHeader.invoice_type == "TEM",
                    )
                )
                invoice_exists_by_number: Dict[str, bool] = {}
                invoice_processed_by_number: Dict[str, bool] = {}
                for num, iid, is_upd in q_inv.all():
                    if num:
                        n = str(num).strip()
                        invoice_exists_by_number[n] = True
                        invoice_processed_by_number[n] = _is_invoice_processed_status(is_upd)

                pedimento_rows: List[Dict[str, Any]] = []
                for p in (
                    session.query(
                        Pedimentos.id,
                        Pedimentos.year,
                        Pedimentos.customs_office,
                        Pedimentos.license,
                        Pedimentos.pedimento_number,
                        Pedimentos.operation_type,
                        Pedimentos.regime,
                        Pedimentos.pedimento_type,
                    )
                    .filter(
                        Pedimentos.tenant_id == tenant_id,
                        Pedimentos.company_id == company_id,
                    )
                    .all()
                ):
                    co = (p.customs_office or "").strip()
                    lic = (p.license or "").strip()
                    num = (p.pedimento_number or "").strip()
                    if not lic or not num:
                        continue
                    entry_date = None
                    end_date = None
                    pd = (
                        session.query(PedimentoDates.entry_date, PedimentoDates.end_date)
                        .filter(PedimentoDates.pedimento_id == p.id).first()
                    )
                    if pd:
                        entry_date = pd[0]
                        end_date = pd[1]
                    info = {
                        "id": p.id,
                        "customs_office": co,
                        "license": lic,
                        "pedimento_number": num,
                        "regime": (p.regime or "").strip(),
                        "operation_type": (p.operation_type or "").strip(),
                        "pedimento_type": (p.pedimento_type or "").strip(),
                        "entry_date": entry_date,
                        "end_date": end_date,
                    }
                    pedimento_rows.append(info)

                remesa_por_pedimento_bd: Dict[str, Set[int]] = {}
                q_rem = (
                    session.query(
                        InvoiceComplianceMx.remesa,
                        Pedimentos.year,
                        Pedimentos.customs_office,
                        Pedimentos.license,
                        Pedimentos.pedimento_number,
                    )
                    .join(Pedimentos, Pedimentos.id == InvoiceComplianceMx.pedimento_id)
                    .filter(
                        InvoiceComplianceMx.tenant_id == tenant_id,
                        InvoiceComplianceMx.company_id == company_id,
                        InvoiceComplianceMx.pedimento_id.isnot(None),
                        InvoiceComplianceMx.remesa.isnot(None),
                    )
                )
                for rem, y, co, lic, num in q_rem.all():
                    if co and lic and num and rem is not None:
                        key = _pedimento_key_from_parsed(
                            (co or "").strip(),
                            (lic or "").strip(),
                            (num or "").strip(),
                        )
                        if key not in remesa_por_pedimento_bd:
                            remesa_por_pedimento_bd[key] = set()
                        remesa_por_pedimento_bd[key].add(int(rem))

                valid_provider_ids: Set[int] = set()
                valid_sold_to_ids: Set[int] = set()
                valid_shipped_to_ids: Set[int] = set()
                valid_provider_short_names: Set[str] = set()
                valid_sold_to_short_names: Set[str] = set()
                valid_shipped_to_short_names: Set[str] = set()
                for cp in session.query(ClientProvider.id, ClientProvider.short_name).filter(
                    ClientProvider.tenant_id == tenant_id,
                    ClientProvider.company_id == company_id,
                ).all():
                    valid_provider_ids.add(cp[0])
                    valid_sold_to_ids.add(cp[0])
                    valid_shipped_to_ids.add(cp[0])
                    if cp[1] and str(cp[1]).strip():
                        sn_upper = str(cp[1]).strip().upper()
                        valid_provider_short_names.add(sn_upper)
                        valid_sold_to_short_names.add(sn_upper)
                        valid_shipped_to_short_names.add(sn_upper)

                broker_rows = session.query(CustomsBroker.id, CustomsBroker.broker_key).filter(
                    CustomsBroker.tenant_id == tenant_id,
                    CustomsBroker.company_id == company_id,
                ).all()
                broker_lookup_scan = _build_customs_broker_lookup(broker_rows)
                valid_broker_ids: Set[int] = set(broker_lookup_scan["id_set"])
                valid_broker_claves: Set[str] = set(broker_lookup_scan["key_to_id"].keys()) | set(
                    broker_lookup_scan["alias_to_id"].keys()
                )

                valid_transporter_keys: Set[str] = set()
                for t in session.query(Transporter.transporter_key).filter(
                    Transporter.tenant_id == tenant_id,
                    Transporter.company_id == company_id,
                ).all():
                    if t[0]:
                        valid_transporter_keys.add((t[0] or "").strip().upper())

                valid_incoterms: Set[str] = set()
                for inc in session.query(Incoterm.code).all():
                    if inc[0]:
                        valid_incoterms.add((inc[0] or "").strip().upper())

                valid_aduana_codes: Set[str] = set()
                for cs in session.query(CustomsSection.customs_code).all():
                    if cs[0]:
                        valid_aduana_codes.add((cs[0] or "").strip())

                valid_currency_codes: Set[str] = set()
                for ct in session.query(CurrencyType.code).all():
                    if ct[0]:
                        valid_currency_codes.add((ct[0] or "").strip().upper())

                exchange_rate_by_date: Dict[str, Any] = {}
                for er in session.query(ExchangeRate.date, ExchangeRate.value).filter(
                    ExchangeRate.tenant_id == tenant_id,
                    ExchangeRate.company_id == company_id,
                ).all():
                    if er[0] and er[1] is not None:
                        dk = er[0].strftime("%Y-%m-%d") if hasattr(er[0], "strftime") else str(er[0])[:10]
                        exchange_rate_by_date[dk] = er[1]

                invoice_has_partidas_by_number: Dict[str, bool] = {}
                existing_tipo_moneda_by_number: Dict[str, str] = {}
                q_li_count = (
                    session.query(InvoiceHeader.invoice_number, func.count(LineItem.id))
                    .join(LineItem, LineItem.invoice_id == InvoiceHeader.id)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == "imp",
                        InvoiceHeader.invoice_type == "TEM",
                    )
                    .group_by(InvoiceHeader.invoice_number)
                )
                for num, cnt in q_li_count.all():
                    if num:
                        invoice_has_partidas_by_number[str(num).strip()] = cnt > 0
                q_fin = (
                    session.query(InvoiceHeader.invoice_number, InvoiceFinancials.currency)
                    .join(InvoiceFinancials, InvoiceFinancials.invoice_id == InvoiceHeader.id)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == "imp",
                        InvoiceHeader.invoice_type == "TEM",
                    )
                )
                for num, cur in q_fin.all():
                    if num and cur:
                        cur_str = (cur or "").strip().lower()
                        if cur_str == "foreign":
                            existing_tipo_moneda_by_number[str(num).strip()] = "ME"
                        elif cur_str == "local":
                            existing_tipo_moneda_by_number[str(num).strip()] = "MN"
                        else:
                            existing_tipo_moneda_by_number[str(num).strip()] = cur_str.upper()[:2]

            with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in:
                sample = f_in.read(2048)
                f_in.seek(0)
                try:
                    dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                except Exception:
                    dialect = "excel"
                reader = csv.DictReader(f_in, dialect=dialect)
                rows_list = list(reader)

            remesa_por_pedimento_csv: Dict[str, Dict[int, str]] = {}
            for row in rows_list:
                row_norm = row_from_template(row, "imp_temp_header", normalize_header)
                ped = (row_norm.get("PEDIMENTO") or "").strip()
                rem = row_norm.get("REMESA")
                factura = (row_norm.get("NUMERO FACTURA") or row_norm.get("NUM FACTURA") or row_norm.get("FACTURA") or "").strip()
                if not ped or not factura:
                    continue
                key = _ped_key_from_row(ped)
                if not key:
                    continue
                try:
                    rem_int = int(rem) if rem is not None and str(rem).strip() else None
                except (TypeError, ValueError):
                    rem_int = None
                if rem_int is not None:
                    if key not in remesa_por_pedimento_csv:
                        remesa_por_pedimento_csv[key] = {}
                    if rem_int not in remesa_por_pedimento_csv[key]:
                        remesa_por_pedimento_csv[key][rem_int] = factura

            # Precheck de referencias críticas para detectar desalineaciones antes del scan completo.
            pedimento_keys = {
                _pedimento_key_from_parsed(
                    (p.get("customs_office") or "").strip(),
                    (p.get("license") or "").strip(),
                    (p.get("pedimento_number") or "").strip(),
                )
                for p in pedimento_rows
                if p.get("customs_office") and p.get("license") and p.get("pedimento_number")
            }
            precheck_errors: List[Dict[str, Any]] = []
            precheck_lines: Set[int] = set()
            for i, row in enumerate(rows_list, start=1):
                row_norm = row_from_template(row, "imp_temp_header", normalize_header)
                ped_raw = (row_norm.get("PEDIMENTO") or "").strip()
                inc_raw = (row_norm.get("CLAVE INCOTERM") or row_norm.get("INCOTERM") or "").strip().upper()
                aduana_cruce = (row_norm.get("ADUANA DE CRUCE") or "").strip()
                invoice_number = (row_norm.get("NUMERO FACTURA") or row_norm.get("NUM FACTURA") or row_norm.get("FACTURA") or "").strip()
                invoice_date = parse_date(row_norm.get("FECHA FACTURA") or row_norm.get("FECHA"), date_format)

                if not invoice_number:
                    precheck_lines.add(i)
                    precheck_errors.append(
                        {
                            "line": i,
                            "col": "NUMERO FACTURA",
                            "msg": "Error: Número de factura faltante/inválido.",
                            "solution": "Completa NUMERO FACTURA para poder insertar/actualizar.",
                            "warning": False,
                        }
                    )
                elif not invoice_date and not (actualizar and invoice_exists_by_number.get(invoice_number)):
                    precheck_lines.add(i)
                    precheck_errors.append(
                        {
                            "line": i,
                            "col": "FECHA FACTURA",
                            "msg": "Error: Fecha de factura faltante/inválida.",
                            "solution": "Corrige FECHA FACTURA o usa update sobre una factura existente con fecha válida.",
                            "warning": False,
                        }
                    )

                if inc_raw and inc_raw not in valid_incoterms:
                    precheck_lines.add(i)
                    precheck_errors.append(
                        {
                            "line": i,
                            "col": "CLAVE INCOTERM",
                            "msg": f"Error: La clave en CLAVE INCOTERM no existe en el Catálogo de INCOTERM ({inc_raw}).",
                            "solution": "Corrige la clave de INCOTERM o completa el catálogo antes de importar.",
                            "warning": False,
                        }
                    )

                if ped_raw:
                    key = _ped_key_from_row(ped_raw)
                    if not key or key not in pedimento_keys:
                        precheck_lines.add(i)
                        precheck_errors.append(
                            {
                                "line": i,
                                "col": "PEDIMENTO",
                                "msg": f"Error: Número de Pedimento: {ped_raw} no existe en el Catálogo de Pedimentos.",
                                "solution": "Carga/corrige el pedimento antes de volver a importar facturas.",
                                "warning": False,
                            }
                        )
                    if not aduana_cruce:
                        precheck_lines.add(i)
                        precheck_errors.append(
                            {
                                "line": i,
                                "col": "ADUANA DE CRUCE",
                                "msg": "Error: Existen campos vacíos que son obligatorios: (Col.AB) Aduana de Cruce.",
                                "solution": "Completa ADUANA DE CRUCE en el CSV para filas con pedimento.",
                                "warning": False,
                            }
                        )

            # No return temprano: el commit siempre valida por fila (logística, etc.). Si solo se
            # devolvía el precheck, valid_rows del scan quedaba inflado vs inserted/skipped del commit.
            scan_precheck_message: Optional[str] = None
            if precheck_errors:
                scan_precheck_message = (
                    "Precheck de referencias falló. Corrige catálogos/pedimentos según los errores indicados. "
                    "Se aplicó también validación completa por fila (logística, etc.) para paridad con el commit."
                )

            processed_rows = 0
            error_lines_list: List[int] = []
            errors_detail: List[Dict[str, Any]] = []
            for e in precheck_errors:
                if len(errors_detail) < 5000:
                    errors_detail.append(
                        {
                            "line": e["line"],
                            "col": e.get("col", ""),
                            "msg": e.get("msg", ""),
                            "solution": e.get("solution", ""),
                            "warning": False,
                        }
                    )

            with CoreSessionLocal() as session:
                with open(error_path, "w", encoding="utf-8") as f_err:
                    for e in precheck_errors:
                        f_err.write(
                            json.dumps(
                                {
                                    "line": e["line"],
                                    "col": e.get("col", ""),
                                    "msg": e.get("msg", ""),
                                    "solution": e.get("solution", ""),
                                }
                            )
                            + "\n"
                        )
                    for i, row in enumerate(rows_list, start=1):
                        self.update_state(
                                state="PROGRESS",
                                meta={
                                    "current": i,
                                    "total": total_rows,
                                    "errors": len(
                                        _merge_unique_invoice_scan_error_lines(
                                            precheck_lines, error_lines_list
                                        )
                                    ),
                                },
                            )
                        row_norm = row_from_template(row, "imp_temp_header", normalize_header)
                        warnings_row: List[Dict[str, Any]] = []
                        row_errors = validate_row_encabezados_impo_temp(
                            row_norm,
                            i,
                            actualizar=actualizar,
                            invoice_exists_by_number=invoice_exists_by_number,
                            invoice_processed_by_number=invoice_processed_by_number,
                            pedimento_rows=pedimento_rows,
                            remesa_por_pedimento_bd=remesa_por_pedimento_bd,
                            remesa_por_pedimento_csv=remesa_por_pedimento_csv,
                            valid_provider_ids=valid_provider_ids,
                            valid_sold_to_ids=valid_sold_to_ids,
                            valid_shipped_to_ids=valid_shipped_to_ids,
                            valid_provider_short_names=valid_provider_short_names,
                            valid_sold_to_short_names=valid_sold_to_short_names,
                            valid_shipped_to_short_names=valid_shipped_to_short_names,
                            valid_broker_ids=valid_broker_ids,
                            valid_broker_claves=valid_broker_claves,
                            valid_transporter_keys=valid_transporter_keys,
                            valid_incoterms=valid_incoterms,
                            valid_aduana_codes=valid_aduana_codes,
                            valid_currency_codes=valid_currency_codes,
                            exchange_rate_by_date=exchange_rate_by_date,
                            invoice_has_partidas_by_number=invoice_has_partidas_by_number,
                            existing_tipo_moneda_by_number=existing_tipo_moneda_by_number,
                            autonumerar_remesas=autonumerar_remesas,
                            control_remesa=control_remesa,
                            remesa_inicio=remesa_inicio,
                            remesa_fin=remesa_fin,
                            date_format=date_format,
                            parse_date_fn=parse_date,
                            warnings=warnings_row,
                        )
                        row_errors.extend(
                            logistics_scan_row_errors(row_norm, session, tenant_id, company_id, i)
                        )
                        blocking = [e for e in row_errors if not e.get("warning")]
                        if blocking:
                            error_lines_list.append(i)
                            for e in blocking:
                                f_err.write(
                                    json.dumps(
                                        {
                                            "line": e["line"],
                                            "col": e.get("col", ""),
                                            "msg": e.get("msg", ""),
                                            "solution": e.get("solution", ""),
                                        }
                                    )
                                    + "\n"
                                )
                        for e in row_errors + warnings_row:
                            if len(errors_detail) < 5000:
                                errors_detail.append(
                                    {
                                        "line": e["line"],
                                        "col": e.get("col", ""),
                                        "msg": e.get("msg", ""),
                                        "solution": e.get("solution", ""),
                                        "warning": bool(e.get("warning", False)),
                                    }
                                )
                        processed_rows += 1

            # El commit usa líneas únicas para omitir filas; unir precheck + validación por fila.
            unique_error_lines = _merge_unique_invoice_scan_error_lines(precheck_lines, error_lines_list)
            error_count = len(unique_error_lines)
            common_storage.store_error_lines(effective_job_type, job_id, unique_error_lines)
            return common_responses.scan_result(
                job_id,
                processed_rows,
                error_count,
                errors_detail,
                total_rows_in_file=total_rows,
                message=scan_precheck_message,
            )
        except Exception as e:
            logger.exception("Encabezados importación temporal scan failed: %s", e)
            return {"status": "failed", "error": str(e)}

    # --- Encabezados de Importación Definitiva: flujo específico (Clarion VALIDA_TODA_FACIMPO_DEF / VALIDA_PARCIAL) ---
    if model_target == "invoice_header" and template_id == "imp_def_header":
        try:
            from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceComplianceMx, InvoiceFinancials
            from api.v1.modules.a76.items.models import LineItem
            from api.v1.modules.a76.pedmientos.models.pedimentos import Pedimentos
            from api.v1.modules.a76.pedmientos.models.pedimento_dates import PedimentoDates
            from api.v1.modules.a76.clients_and_providers.models import ClientProvider
            from api.v1.modules.a76.customs_brokers.models import CustomsBroker
            from api.v1.modules.public.reference_data.customs_sections.models import CustomsSection
            from api.v1.modules.public.reference_data.incoterms.models import Incoterm
            from api.v1.modules.public.reference_data.currency_types.models import CurrencyType
            from api.v1.modules.a76.general_catalogs.exchange_rate.models import ExchangeRate
            from api.v1.modules.a76.transportation.transporters.models import Transporter
            from .validators.encabezados_impo_def import (
                validate_row_encabezados_impo_def,
                parse_pedimento_col_a_impo_def,
            )
            from .validators.encabezados_impo_temp import _pedimento_key_from_parsed
            from .validators.transport_catalog import logistics_scan_row_errors

            DEF_INVOICE_TYPES = ("DEF", "MATDE", "EXDEF")

            def _ped_key_from_row_def(ped_str: str) -> Optional[str]:
                parsed = parse_pedimento_col_a_impo_def(ped_str)
                if not parsed:
                    return None
                return _pedimento_key_from_parsed(parsed[0], parsed[1], parsed[2])

            _fc = parse_footer_config(meta.get("footer_config"))
            actualizar = meta.get("actualizar", False)
            autonumerar_remesas = meta.get("autonumerar_remesas", False)
            control_remesa = bool(_fc.get("control_remesa", False))
            remesa_inicio = _fc.get("remesa_inicio")
            remesa_fin = _fc.get("remesa_fin")
            if remesa_inicio is not None:
                try:
                    remesa_inicio = int(remesa_inicio)
                except (TypeError, ValueError):
                    remesa_inicio = None
            if remesa_fin is not None:
                try:
                    remesa_fin = int(remesa_fin)
                except (TypeError, ValueError):
                    remesa_fin = None
            if _fc:
                if "actualizar" in _fc:
                    actualizar = bool(_fc["actualizar"])
                if "autonumerar_remesas" in _fc:
                    autonumerar_remesas = bool(_fc["autonumerar_remesas"])

            with CoreSessionLocal() as session:
                q_inv = (
                    session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.status)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == "imp",
                        InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES),
                    )
                )
                invoice_exists_by_number = {}
                invoice_processed_by_number = {}
                for num, iid, is_upd in q_inv.all():
                    if num:
                        n = str(num).strip()
                        invoice_exists_by_number[n] = True
                        invoice_processed_by_number[n] = _is_invoice_processed_status(is_upd)

                pedimento_rows = []
                for p in (
                    session.query(
                        Pedimentos.id,
                        Pedimentos.year,
                        Pedimentos.customs_office,
                        Pedimentos.license,
                        Pedimentos.pedimento_number,
                        Pedimentos.operation_type,
                        Pedimentos.regime,
                        Pedimentos.pedimento_type,
                    )
                    .filter(
                        Pedimentos.tenant_id == tenant_id,
                        Pedimentos.company_id == company_id,
                        Pedimentos.operation_type == "imp",
                        Pedimentos.regime == "IMD",
                    )
                    .all()
                ):
                    co = (p.customs_office or "").strip()
                    lic = (p.license or "").strip()
                    num = (p.pedimento_number or "").strip()
                    if not lic or not num:
                        continue
                    entry_date = None
                    end_date = None
                    pd = (
                        session.query(PedimentoDates.entry_date, PedimentoDates.end_date)
                        .filter(PedimentoDates.pedimento_id == p.id).first()
                    )
                    if pd:
                        entry_date = pd[0]
                        end_date = pd[1]
                    info = {
                        "id": p.id,
                        "customs_office": co,
                        "license": lic,
                        "pedimento_number": num,
                        "regime": (p.regime or "").strip(),
                        "operation_type": (p.operation_type or "").strip(),
                        "pedimento_type": (p.pedimento_type or "").strip(),
                        "entry_date": entry_date,
                        "end_date": end_date,
                    }
                    pedimento_rows.append(info)

                remesa_por_pedimento_bd = {}
                q_rem = (
                    session.query(
                        InvoiceComplianceMx.remesa,
                        Pedimentos.customs_office,
                        Pedimentos.license,
                        Pedimentos.pedimento_number,
                    )
                    .join(InvoiceHeader, InvoiceHeader.id == InvoiceComplianceMx.invoice_id)
                    .join(Pedimentos, Pedimentos.id == InvoiceComplianceMx.pedimento_id)
                    .filter(
                        InvoiceComplianceMx.tenant_id == tenant_id,
                        InvoiceComplianceMx.company_id == company_id,
                        InvoiceComplianceMx.pedimento_id.isnot(None),
                        InvoiceComplianceMx.remesa.isnot(None),
                        InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES),
                    )
                )
                for rem, co, lic, num in q_rem.all():
                    if co and lic and num and rem is not None:
                        key = _pedimento_key_from_parsed(
                            (co or "").strip(),
                            (lic or "").strip(),
                            (num or "").strip(),
                        )
                        if key not in remesa_por_pedimento_bd:
                            remesa_por_pedimento_bd[key] = set()
                        remesa_por_pedimento_bd[key].add(int(rem))

                valid_provider_ids = set()
                valid_sold_to_ids = set()
                valid_shipped_to_ids = set()
                valid_provider_short_names = set()
                valid_sold_to_short_names = set()
                valid_shipped_to_short_names = set()
                for cp in session.query(ClientProvider.id, ClientProvider.short_name).filter(
                    ClientProvider.tenant_id == tenant_id,
                    ClientProvider.company_id == company_id,
                ).all():
                    valid_provider_ids.add(cp[0])
                    valid_sold_to_ids.add(cp[0])
                    valid_shipped_to_ids.add(cp[0])
                    if cp[1] and str(cp[1]).strip():
                        sn_upper = str(cp[1]).strip().upper()
                        valid_provider_short_names.add(sn_upper)
                        valid_sold_to_short_names.add(sn_upper)
                        valid_shipped_to_short_names.add(sn_upper)

                broker_rows = session.query(CustomsBroker.id, CustomsBroker.broker_key).filter(
                    CustomsBroker.tenant_id == tenant_id,
                    CustomsBroker.company_id == company_id,
                ).all()
                broker_lookup_scan = _build_customs_broker_lookup(broker_rows)
                valid_broker_ids = set(broker_lookup_scan["id_set"])
                valid_broker_claves = set(broker_lookup_scan["key_to_id"].keys()) | set(
                    broker_lookup_scan["alias_to_id"].keys()
                )

                valid_transporter_keys = set()
                for t in session.query(Transporter.transporter_key).filter(
                    Transporter.tenant_id == tenant_id,
                    Transporter.company_id == company_id,
                ).all():
                    if t[0]:
                        valid_transporter_keys.add((t[0] or "").strip().upper())

                valid_incoterms = set()
                for inc in session.query(Incoterm.code).all():
                    if inc[0]:
                        valid_incoterms.add((inc[0] or "").strip().upper())

                valid_aduana_codes = set()
                for cs in session.query(CustomsSection.customs_code).all():
                    if cs[0]:
                        valid_aduana_codes.add((cs[0] or "").strip())

                valid_currency_codes = set()
                for ct in session.query(CurrencyType.code).all():
                    if ct[0]:
                        valid_currency_codes.add((ct[0] or "").strip().upper())

                exchange_rate_by_date = {}
                for er in session.query(ExchangeRate.date, ExchangeRate.value).filter(
                    ExchangeRate.tenant_id == tenant_id,
                    ExchangeRate.company_id == company_id,
                ).all():
                    if er[0] and er[1] is not None:
                        dk = er[0].strftime("%Y-%m-%d") if hasattr(er[0], "strftime") else str(er[0])[:10]
                        exchange_rate_by_date[dk] = er[1]

                invoice_has_partidas_by_number = {}
                existing_tipo_moneda_by_number = {}
                q_li_count = (
                    session.query(InvoiceHeader.invoice_number, func.count(LineItem.id))
                    .join(LineItem, LineItem.invoice_id == InvoiceHeader.id)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == "imp",
                        InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES),
                    )
                    .group_by(InvoiceHeader.invoice_number)
                )
                for num, cnt in q_li_count.all():
                    if num:
                        invoice_has_partidas_by_number[str(num).strip()] = cnt > 0
                q_fin = (
                    session.query(InvoiceHeader.invoice_number, InvoiceFinancials.currency)
                    .join(InvoiceFinancials, InvoiceFinancials.invoice_id == 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, cur in q_fin.all():
                    if num and cur:
                        cur_str = (cur or "").strip().lower()
                        if cur_str == "foreign":
                            existing_tipo_moneda_by_number[str(num).strip()] = "ME"
                        elif cur_str == "local":
                            existing_tipo_moneda_by_number[str(num).strip()] = "MN"
                        else:
                            existing_tipo_moneda_by_number[str(num).strip()] = (cur_str or "").upper()[:2]

            with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in:
                sample = f_in.read(2048)
                f_in.seek(0)
                try:
                    dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                except Exception:
                    dialect = "excel"
                reader = csv.DictReader(f_in, dialect=dialect)
                rows_list = list(reader)

            remesa_por_pedimento_csv = {}
            for row in rows_list:
                row_norm = row_from_template(row, "imp_def_header", normalize_header)
                ped = (row_norm.get("PEDIMENTO") or "").strip()
                rem = row_norm.get("REMESA")
                factura = (row_norm.get("NUMERO FACTURA") or row_norm.get("NUM FACTURA") or row_norm.get("FACTURA") or "").strip()
                if not ped or not factura:
                    continue
                key = _ped_key_from_row_def(ped)
                if not key:
                    continue
                try:
                    rem_int = int(rem) if rem is not None and str(rem).strip() else None
                except (TypeError, ValueError):
                    rem_int = None
                if rem_int is not None:
                    if key not in remesa_por_pedimento_csv:
                        remesa_por_pedimento_csv[key] = {}
                    if rem_int not in remesa_por_pedimento_csv[key]:
                        remesa_por_pedimento_csv[key][rem_int] = factura

            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):
                    self.update_state(state="PROGRESS", meta={"current": i, "total": total_rows, "errors": error_count})
                    row_norm = row_from_template(row, "imp_def_header", normalize_header)
                    warnings_row = []
                    row_errors = validate_row_encabezados_impo_def(
                        row_norm,
                        i,
                        actualizar=actualizar,
                        invoice_exists_by_number=invoice_exists_by_number,
                        invoice_processed_by_number=invoice_processed_by_number,
                        pedimento_rows=pedimento_rows,
                        remesa_por_pedimento_bd=remesa_por_pedimento_bd,
                        remesa_por_pedimento_csv=remesa_por_pedimento_csv,
                        valid_provider_ids=valid_provider_ids,
                        valid_sold_to_ids=valid_sold_to_ids,
                        valid_shipped_to_ids=valid_shipped_to_ids,
                        valid_provider_short_names=valid_provider_short_names,
                        valid_sold_to_short_names=valid_sold_to_short_names,
                        valid_shipped_to_short_names=valid_shipped_to_short_names,
                        valid_broker_ids=valid_broker_ids,
                        valid_broker_claves=valid_broker_claves,
                        valid_transporter_keys=valid_transporter_keys,
                        valid_incoterms=valid_incoterms,
                        valid_aduana_codes=valid_aduana_codes,
                        valid_currency_codes=valid_currency_codes,
                        exchange_rate_by_date=exchange_rate_by_date,
                        invoice_has_partidas_by_number=invoice_has_partidas_by_number,
                        existing_tipo_moneda_by_number=existing_tipo_moneda_by_number,
                        autonumerar_remesas=autonumerar_remesas,
                        control_remesa=control_remesa,
                        remesa_inicio=remesa_inicio,
                        remesa_fin=remesa_fin,
                        date_format=date_format,
                        parse_date_fn=parse_date,
                        warnings=warnings_row,
                    )
                    row_errors.extend(
                        logistics_scan_row_errors(row_norm, session, tenant_id, company_id, i)
                    )
                    blocking = [e for e in row_errors if not e.get("warning")]
                    if blocking:
                        error_count += 1
                        error_lines_list.append(i)
                        for e in blocking:
                            f_err.write(
                                json.dumps(
                                    {
                                        "line": e["line"],
                                        "col": e.get("col", ""),
                                        "msg": e.get("msg", ""),
                                        "solution": e.get("solution", ""),
                                    }
                                )
                                + "\n"
                            )
                    for e in row_errors + warnings_row:
                        if len(errors_detail) < 5000:
                            errors_detail.append(
                                {
                                    "line": e["line"],
                                    "col": e.get("col", ""),
                                    "msg": e.get("msg", ""),
                                    "solution": e.get("solution", ""),
                                    "warning": bool(e.get("warning", False)),
                                }
                            )
                    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, total_rows_in_file=total_rows
            )
        except Exception as e:
            logger.exception("Encabezados importación definitiva scan failed: %s", e)
            return {"status": "failed", "error": str(e)}

    # --- Encabezados Exportación (Expo Def) y Cambio de Régimen: flujo específico (Clarion VALIDA_TODA_FAC_EXPO / VALIDA_PARCIAL) ---
    if model_target == "invoice_header" and template_id == "exp_def_header":
        try:
            from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceComplianceMx, InvoiceFinancials
            from api.v1.modules.a76.items.models import LineItem
            from api.v1.modules.a76.pedmientos.models.pedimentos import Pedimentos
            from api.v1.modules.a76.pedmientos.models.pedimento_dates import PedimentoDates
            from api.v1.modules.a76.clients_and_providers.models import ClientProvider
            from api.v1.modules.a76.customs_brokers.models import CustomsBroker
            from api.v1.modules.public.reference_data.customs_sections.models import CustomsSection
            from api.v1.modules.public.reference_data.incoterms.models import Incoterm
            from api.v1.modules.public.reference_data.currency_types.models import CurrencyType
            from api.v1.modules.a76.general_catalogs.exchange_rate.models import ExchangeRate
            from api.v1.modules.a76.transportation.transporters.models import Transporter
            from api.v1.modules.a76.manifests.manifest.models import Manifest
            from .validators.encabezados_expo import validate_row_encabezados_expo
            from .validators.encabezados_impo_temp import _pedimento_key_from_parsed
            from .validators.encabezados_impo_def import parse_pedimento_col_a_impo_def
            from .validators.transport_catalog import logistics_scan_row_errors

            _fc = parse_footer_config(meta.get("footer_config"))
            actualizar = meta.get("actualizar", False)
            autonumerar_remesas = meta.get("autonumerar_remesas", False)
            recalcular_fecha_pedimentos = meta.get("recalcular_fecha_pedimentos", False)
            if _fc:
                if "actualizar" in _fc:
                    actualizar = bool(_fc["actualizar"])
                if "autonumerar_remesas" in _fc:
                    autonumerar_remesas = bool(_fc["autonumerar_remesas"])
                if "recalcular_fecha_pedimentos" in _fc:
                    recalcular_fecha_pedimentos = bool(_fc["recalcular_fecha_pedimentos"])

            cambio_regimen_raw = (meta.get("cambio_regimen") or _fc.get("cambio_regimen") or "NO").strip().upper()
            cambio_regimen = cambio_regimen_raw == "SI"
            tipo_factura = (meta.get("tipo_factura") or _fc.get("tipo_factura") or "AFIJO").strip().upper()

            TIPOS_FACTURA_EXPO_VALIDOS = frozenset({"NODES", "AFIJO", "DONAC", "SCRAP"})
            if tipo_factura not in TIPOS_FACTURA_EXPO_VALIDOS:
                return {
                    "status": "failed",
                    "error": f"Tipo de factura '{tipo_factura}' no válido para Exportación. Debe ser uno de: NODES, AFIJO, DONAC, SCRAP.",
                }

            if cambio_regimen and tipo_factura != "AFIJO":
                return {
                    "status": "failed",
                    "error": "Este tipo de factura no es compatible para Cambio de Régimen, seleccionar AFIJO.",
                }

            def _ped_key_from_row_expo(ped_str: str):
                parsed = parse_pedimento_col_a_impo_def(ped_str)
                if not parsed:
                    return None
                return _pedimento_key_from_parsed(parsed[0], parsed[1], parsed[2])

            with CoreSessionLocal() as session:
                q_inv = (
                    session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.status, InvoiceHeader.status_rep)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == "exp",
                    )
                )
                invoice_exists_by_number = {}
                invoice_processed_by_number = {}
                invoice_in_report_by_number = {}
                for num, iid, is_upd, is_rep in q_inv.all():
                    if num:
                        n = str(num).strip()
                        invoice_exists_by_number[n] = True
                        invoice_processed_by_number[n] = _is_invoice_processed_status(is_upd)
                        invoice_in_report_by_number[n] = bool(is_rep) if is_rep is not None else False

                if cambio_regimen:
                    ped_filter_op = "imp"
                    ped_filter_regimes = ["IMD"]
                else:
                    ped_filter_op = "exp"
                    ped_filter_regimes = ["EXD", "ETE", "ETR"]

                pedimento_rows = []
                for p in (
                    session.query(
                        Pedimentos.id,
                        Pedimentos.customs_office,
                        Pedimentos.license,
                        Pedimentos.pedimento_number,
                        Pedimentos.operation_type,
                        Pedimentos.regime,
                        Pedimentos.pedimento_type,
                        Pedimentos.pedimento_code,
                    )
                    .filter(
                        Pedimentos.tenant_id == tenant_id,
                        Pedimentos.company_id == company_id,
                        Pedimentos.operation_type == ped_filter_op,
                        Pedimentos.regime.in_(ped_filter_regimes),
                    )
                    .all()
                ):
                    co = (p.customs_office or "").strip()
                    lic = (p.license or "").strip()
                    num = (p.pedimento_number or "").strip()
                    if not lic or not num:
                        continue
                    entry_date = None
                    end_date = None
                    pd = (
                        session.query(PedimentoDates.entry_date, PedimentoDates.end_date)
                        .filter(PedimentoDates.pedimento_id == p.id).first()
                    )
                    if pd:
                        entry_date = pd[0]
                        end_date = pd[1]
                    info = {
                        "id": p.id,
                        "customs_office": co,
                        "license": lic,
                        "pedimento_number": num,
                        "regime": (p.regime or "").strip(),
                        "operation_type": (p.operation_type or "").strip().upper()[:3],
                        "pedimento_type": (p.pedimento_type or "").strip(),
                        "pedimento_code": (p.pedimento_code or "").strip().upper(),
                        "entry_date": entry_date,
                        "end_date": end_date,
                    }
                    pedimento_rows.append(info)

                remesa_por_pedimento_bd = {}
                q_rem = (
                    session.query(
                        InvoiceComplianceMx.remesa,
                        Pedimentos.customs_office,
                        Pedimentos.license,
                        Pedimentos.pedimento_number,
                    )
                    .join(InvoiceHeader, InvoiceHeader.id == InvoiceComplianceMx.invoice_id)
                    .join(Pedimentos, Pedimentos.id == InvoiceComplianceMx.pedimento_id)
                    .filter(
                        InvoiceComplianceMx.tenant_id == tenant_id,
                        InvoiceComplianceMx.company_id == company_id,
                        InvoiceComplianceMx.pedimento_id.isnot(None),
                        InvoiceComplianceMx.remesa.isnot(None),
                        InvoiceHeader.operation_type == "exp",
                    )
                )
                for rem, co, lic, num in q_rem.all():
                    if co and lic and num and rem is not None:
                        key = _pedimento_key_from_parsed(
                            (co or "").strip()[:2],
                            (lic or "").strip(),
                            (num or "").strip(),
                        )
                        if key not in remesa_por_pedimento_bd:
                            remesa_por_pedimento_bd[key] = set()
                        remesa_por_pedimento_bd[key].add(int(rem))

                valid_provider_ids = set()
                valid_sold_to_ids = set()
                valid_shipped_to_ids = set()
                valid_provider_short_names = set()
                valid_sold_to_short_names = set()
                valid_shipped_to_short_names = set()
                for cp in session.query(ClientProvider.id, ClientProvider.short_name).filter(
                    ClientProvider.tenant_id == tenant_id,
                    ClientProvider.company_id == company_id,
                ).all():
                    valid_provider_ids.add(cp[0])
                    valid_sold_to_ids.add(cp[0])
                    valid_shipped_to_ids.add(cp[0])
                    if cp[1] and str(cp[1]).strip():
                        sn_upper = str(cp[1]).strip().upper()
                        valid_provider_short_names.add(sn_upper)
                        valid_sold_to_short_names.add(sn_upper)
                        valid_shipped_to_short_names.add(sn_upper)

                broker_rows = session.query(CustomsBroker.id, CustomsBroker.broker_key).filter(
                    CustomsBroker.tenant_id == tenant_id,
                    CustomsBroker.company_id == company_id,
                ).all()
                broker_lookup_scan = _build_customs_broker_lookup(broker_rows)
                valid_broker_ids = set(broker_lookup_scan["id_set"])
                valid_broker_claves = set(broker_lookup_scan["key_to_id"].keys()) | set(
                    broker_lookup_scan["alias_to_id"].keys()
                )

                valid_transporter_keys = set()
                for t in session.query(Transporter.transporter_key).filter(
                    Transporter.tenant_id == tenant_id,
                    Transporter.company_id == company_id,
                ).all():
                    if t[0]:
                        valid_transporter_keys.add((t[0] or "").strip().upper())

                valid_incoterms = set()
                for inc in session.query(Incoterm.code).all():
                    if inc[0]:
                        valid_incoterms.add((inc[0] or "").strip().upper())

                valid_aduana_codes = set()
                for cs in session.query(CustomsSection.customs_code).all():
                    if cs[0]:
                        valid_aduana_codes.add((cs[0] or "").strip())

                valid_currency_codes = set()
                for ct in session.query(CurrencyType.code).all():
                    if ct[0]:
                        valid_currency_codes.add((ct[0] or "").strip().upper())

                valid_manifiesto_codes = set()
                for m in session.query(Manifest.manifest_number).filter(
                    Manifest.tenant_id == tenant_id,
                    Manifest.company_id == company_id,
                ).all():
                    if m[0] and str(m[0]).strip():
                        valid_manifiesto_codes.add(str(m[0]).strip())

                exchange_rate_by_date = {}
                for er in session.query(ExchangeRate.date, ExchangeRate.value).filter(
                    ExchangeRate.tenant_id == tenant_id,
                    ExchangeRate.company_id == company_id,
                ).all():
                    if er[0] and er[1] is not None:
                        dk = er[0].strftime("%Y-%m-%d") if hasattr(er[0], "strftime") else str(er[0])[:10]
                        exchange_rate_by_date[dk] = er[1]

                invoice_has_partidas_by_number = {}
                existing_tipo_moneda_by_number = {}
                q_li_count = (
                    session.query(InvoiceHeader.invoice_number, func.count(LineItem.id))
                    .join(LineItem, LineItem.invoice_id == InvoiceHeader.id)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == "exp",
                    )
                    .group_by(InvoiceHeader.invoice_number)
                )
                for num, cnt in q_li_count.all():
                    if num:
                        invoice_has_partidas_by_number[str(num).strip()] = cnt > 0
                q_fin = (
                    session.query(InvoiceHeader.invoice_number, InvoiceFinancials.currency)
                    .join(InvoiceFinancials, InvoiceFinancials.invoice_id == InvoiceHeader.id)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == "exp",
                    )
                )
                for num, cur in q_fin.all():
                    if num and cur:
                        cur_str = (cur or "").strip().lower()
                        if cur_str == "foreign":
                            existing_tipo_moneda_by_number[str(num).strip()] = "ME"
                        elif cur_str == "local":
                            existing_tipo_moneda_by_number[str(num).strip()] = "MN"
                        else:
                            existing_tipo_moneda_by_number[str(num).strip()] = (cur_str or "").upper()[:2]

            date_format = _fc.get("dateFormat") or meta.get("date_format")

            with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in:
                sample = f_in.read(2048)
                f_in.seek(0)
                try:
                    dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                except Exception:
                    dialect = "excel"
                reader = csv.DictReader(f_in, dialect=dialect)
                rows_list = list(reader)

            remesa_por_pedimento_csv = {}
            for row in rows_list:
                row_norm = row_from_template(row, "exp_def_header", normalize_header)
                ped = (row_norm.get("PEDIMENTO") or "").strip()
                rem = row_norm.get("REMESA")
                factura = (row_norm.get("NUMERO FACTURA") or row_norm.get("NUM FACTURA") or row_norm.get("FACTURA") or "").strip()
                if not ped or not factura:
                    continue
                key = _ped_key_from_row_expo(ped)
                if not key:
                    continue
                try:
                    rem_int = int(rem) if rem is not None and str(rem).strip() else None
                except (TypeError, ValueError):
                    rem_int = None
                if rem_int is not None:
                    if key not in remesa_por_pedimento_csv:
                        remesa_por_pedimento_csv[key] = {}
                    if rem_int not in remesa_por_pedimento_csv[key]:
                        remesa_por_pedimento_csv[key][rem_int] = factura

            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):
                    self.update_state(state="PROGRESS", meta={"current": i, "total": total_rows, "errors": error_count})
                    row_norm = row_from_template(row, "exp_def_header", normalize_header)
                    warnings_row = []
                    row_errors = validate_row_encabezados_expo(
                        row_norm,
                        i,
                        actualizar=actualizar,
                        cambio_regimen=cambio_regimen,
                        tipo_factura=tipo_factura,
                        invoice_exists_by_number=invoice_exists_by_number,
                        invoice_processed_by_number=invoice_processed_by_number,
                        invoice_in_report_by_number=invoice_in_report_by_number,
                        pedimento_rows=pedimento_rows,
                        remesa_por_pedimento_bd=remesa_por_pedimento_bd,
                        remesa_por_pedimento_csv=remesa_por_pedimento_csv,
                        valid_provider_ids=valid_provider_ids,
                        valid_sold_to_ids=valid_sold_to_ids,
                        valid_shipped_to_ids=valid_shipped_to_ids,
                        valid_broker_ids=valid_broker_ids,
                        valid_broker_claves=valid_broker_claves,
                        valid_transporter_keys=valid_transporter_keys,
                        valid_incoterms=valid_incoterms,
                        valid_aduana_codes=valid_aduana_codes,
                        valid_currency_codes=valid_currency_codes,
                        valid_provider_short_names=valid_provider_short_names,
                        valid_sold_to_short_names=valid_sold_to_short_names,
                        valid_shipped_to_short_names=valid_shipped_to_short_names,
                        valid_manifiesto_codes=valid_manifiesto_codes,
                        valid_enviado_por_ids=valid_provider_ids,
                        valid_enviado_por_short_names=valid_provider_short_names,
                        exchange_rate_by_date=exchange_rate_by_date,
                        invoice_has_partidas_by_number=invoice_has_partidas_by_number,
                        existing_tipo_moneda_by_number=existing_tipo_moneda_by_number,
                        autonumerar_remesas=autonumerar_remesas,
                        recalcular_fecha_pedimentos=recalcular_fecha_pedimentos,
                        date_format=date_format,
                        parse_date_fn=parse_date,
                        warnings=warnings_row,
                    )
                    row_errors.extend(
                        logistics_scan_row_errors(row_norm, session, tenant_id, company_id, i)
                    )
                    blocking = [e for e in row_errors if not e.get("warning")]
                    if blocking:
                        error_count += 1
                        error_lines_list.append(i)
                        for e in blocking:
                            f_err.write(
                                json.dumps(
                                    {
                                        "line": e["line"],
                                        "col": e.get("col", ""),
                                        "msg": e.get("msg", ""),
                                        "solution": e.get("solution", ""),
                                    }
                                )
                                + "\n"
                            )
                    for e in row_errors + warnings_row:
                        if len(errors_detail) < 5000:
                            errors_detail.append(
                                {
                                    "line": e["line"],
                                    "col": e.get("col", ""),
                                    "msg": e.get("msg", ""),
                                    "solution": e.get("solution", ""),
                                    "warning": bool(e.get("warning", False)),
                                }
                            )
                    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, total_rows_in_file=total_rows
            )
        except Exception as e:
            logger.exception("Encabezados exportación scan failed: %s", e)
            return {"status": "failed", "error": str(e)}

    # --- Encabezados Compras Mexicanas: flujo específico (Clarion VALIDA_TODA_FAC_COM_MEX / VALIDA_PARCIAL) ---
    if model_target == "invoice_header" and template_id == "cmex_header":
        try:
            from api.v1.modules.a76.invoices.models import InvoiceHeader, InvoiceFinancials
            from api.v1.modules.a76.items.models import LineItem
            from api.v1.modules.a76.clients_and_providers.models import ClientProvider
            from api.v1.modules.public.reference_data.incoterms.models import Incoterm
            from api.v1.modules.public.reference_data.currency_types.models import CurrencyType
            from api.v1.modules.a76.general_catalogs.exchange_rate.models import ExchangeRate
            from api.v1.modules.a76.transportation.transporters.models import Transporter
            from .validators.encabezados_cmex import validate_row_encabezados_cmex
            from .validators.transport_catalog import logistics_scan_row_errors

            _fc = parse_footer_config(meta.get("footer_config"))
            actualizar = meta.get("actualizar", False)
            if _fc and "actualizar" in _fc:
                actualizar = bool(_fc["actualizar"])

            with CoreSessionLocal() as session:
                q_inv = (
                    session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.status)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == "imp",
                        InvoiceHeader.invoice_type == "MEX",
                    )
                )
                invoice_exists_by_number = {}
                invoice_processed_by_number = {}
                for num, iid, is_upd in q_inv.all():
                    if num:
                        n = str(num).strip()
                        invoice_exists_by_number[n] = True
                        invoice_processed_by_number[n] = _is_invoice_processed_status(is_upd)

                valid_provider_ids = set()
                valid_sold_to_ids = set()
                valid_shipped_to_ids = set()
                valid_provider_short_names = set()
                valid_sold_to_short_names = set()
                valid_shipped_to_short_names = set()
                for cp in session.query(ClientProvider.id, ClientProvider.short_name).filter(
                    ClientProvider.tenant_id == tenant_id,
                    ClientProvider.company_id == company_id,
                ).all():
                    valid_provider_ids.add(cp[0])
                    valid_sold_to_ids.add(cp[0])
                    valid_shipped_to_ids.add(cp[0])
                    if cp[1] and str(cp[1]).strip():
                        sn_upper = str(cp[1]).strip().upper()
                        valid_provider_short_names.add(sn_upper)
                        valid_sold_to_short_names.add(sn_upper)
                        valid_shipped_to_short_names.add(sn_upper)

                valid_transporter_keys = set()
                for t in session.query(Transporter.transporter_key).filter(
                    Transporter.tenant_id == tenant_id,
                    Transporter.company_id == company_id,
                ).all():
                    if t[0]:
                        valid_transporter_keys.add((t[0] or "").strip().upper())

                valid_incoterms = set()
                for inc in session.query(Incoterm.code).all():
                    if inc[0]:
                        valid_incoterms.add((inc[0] or "").strip().upper())

                valid_currency_codes = set()
                for ct in session.query(CurrencyType.code).all():
                    if ct[0]:
                        valid_currency_codes.add((ct[0] or "").strip().upper())

                exchange_rate_by_date = {}
                for er in session.query(ExchangeRate.date, ExchangeRate.value).filter(
                    ExchangeRate.tenant_id == tenant_id,
                    ExchangeRate.company_id == company_id,
                ).all():
                    if er[0] and er[1] is not None:
                        dk = er[0].strftime("%Y-%m-%d") if hasattr(er[0], "strftime") else str(er[0])[:10]
                        exchange_rate_by_date[dk] = er[1]

                invoice_has_partidas_by_number = {}
                existing_tipo_moneda_by_number = {}
                q_li_count = (
                    session.query(InvoiceHeader.invoice_number, func.count(LineItem.id))
                    .join(LineItem, LineItem.invoice_id == InvoiceHeader.id)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == "imp",
                        InvoiceHeader.invoice_type == "MEX",
                    )
                    .group_by(InvoiceHeader.invoice_number)
                )
                for num, cnt in q_li_count.all():
                    if num:
                        invoice_has_partidas_by_number[str(num).strip()] = cnt > 0
                q_fin = (
                    session.query(InvoiceHeader.invoice_number, InvoiceFinancials.currency)
                    .join(InvoiceFinancials, InvoiceFinancials.invoice_id == InvoiceHeader.id)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == "imp",
                        InvoiceHeader.invoice_type == "MEX",
                    )
                )
                for num, cur in q_fin.all():
                    if num and cur:
                        cur_str = (cur or "").strip().lower()
                        if cur_str == "foreign":
                            existing_tipo_moneda_by_number[str(num).strip()] = "ME"
                        elif cur_str == "local":
                            existing_tipo_moneda_by_number[str(num).strip()] = "MN"
                        else:
                            existing_tipo_moneda_by_number[str(num).strip()] = (cur_str or "").upper()[:2]

            with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f_in:
                sample = f_in.read(2048)
                f_in.seek(0)
                try:
                    dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                except Exception:
                    dialect = "excel"
                reader = csv.DictReader(f_in, dialect=dialect)
                rows_list = list(reader)

            error_count = 0
            processed_rows = 0
            error_lines_list = []
            errors_detail = []

            with CoreSessionLocal() as session:
                with open(error_path, "w", encoding="utf-8") as f_err:
                    for i, row in enumerate(rows_list, start=1):
                        self.update_state(
                                state="PROGRESS",
                                meta={"current": i, "total": total_rows, "errors": error_count},
                            )
                        row_norm = row_from_template(row, "cmex_header", normalize_header)
                        warnings_row = []
                        row_errors = validate_row_encabezados_cmex(
                            row_norm,
                            i,
                            actualizar=actualizar,
                            invoice_exists_by_number=invoice_exists_by_number,
                            invoice_processed_by_number=invoice_processed_by_number,
                            valid_provider_ids=valid_provider_ids,
                            valid_sold_to_ids=valid_sold_to_ids,
                            valid_shipped_to_ids=valid_shipped_to_ids,
                            valid_transporter_keys=valid_transporter_keys,
                            valid_incoterms=valid_incoterms,
                            valid_currency_codes=valid_currency_codes,
                            exchange_rate_by_date=exchange_rate_by_date,
                            invoice_has_partidas_by_number=invoice_has_partidas_by_number,
                            existing_tipo_moneda_by_number=existing_tipo_moneda_by_number,
                            valid_provider_short_names=valid_provider_short_names,
                            valid_sold_to_short_names=valid_sold_to_short_names,
                            valid_shipped_to_short_names=valid_shipped_to_short_names,
                            date_format=date_format,
                            parse_date_fn=parse_date,
                            warnings=warnings_row,
                        )
                        row_errors.extend(
                            logistics_scan_row_errors(row_norm, session, tenant_id, company_id, i)
                        )
                        blocking = [e for e in row_errors if not e.get("warning")]
                        if blocking:
                            error_count += 1
                            error_lines_list.append(i)
                            for e in blocking:
                                f_err.write(
                                    json.dumps(
                                        {
                                            "line": e["line"],
                                            "col": e.get("col", ""),
                                            "msg": e.get("msg", ""),
                                            "solution": e.get("solution", ""),
                                        }
                                    )
                                    + "\n"
                                )
                        for e in row_errors + warnings_row:
                            if len(errors_detail) < 5000:
                                errors_detail.append(
                                    {
                                        "line": e["line"],
                                        "col": e.get("col", ""),
                                        "msg": e.get("msg", ""),
                                        "solution": e.get("solution", ""),
                                        "warning": bool(e.get("warning", False)),
                                    }
                                )
                        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, total_rows_in_file=total_rows
            )
        except Exception as e:
            logger.exception("Encabezados Compras Mexicanas scan failed: %s", e)
            return {"status": "failed", "error": str(e)}

    try:
        from api.v1.modules.a76.invoices.models import InvoiceHeader
        from api.v1.modules.a76.clients_and_providers.models import ClientProvider
        from api.v1.modules.a76.customs_brokers.models import CustomsBroker
        from api.v1.modules.public.reference_data.currency_types.models import CurrencyType
        from api.v1.modules.public.reference_data.pedimento_codes.models import PedimentoCode
        from api.v1.modules.public.reference_data.code_pedimento_regimens.models import (
            CodePedimentoRegimen,
        )
        from api.v1.modules.public.reference_data.pedimento_regimens.models import RegimenPedimento
        from api.v1.modules.public.reference_data.invoice_types.models import InvoiceType
        from api.v1.modules.public.reference_data.customs_sections.models import CustomsSection
        from api.v1.modules.public.reference_data.incoterms.models import Incoterm
        from api.v1.modules.a76.parts.models import Part

        models = {
            "InvoiceHeader": InvoiceHeader,
            "InvoiceType": InvoiceType,
            "ClientProvider": ClientProvider,
            "CustomsBroker": CustomsBroker,
            "RegimenPedimento": RegimenPedimento,
            "CodePedimentoRegimen": CodePedimentoRegimen,
            "PedimentoCode": PedimentoCode,
            "CurrencyType": CurrencyType,
            "CustomsSection": CustomsSection,
            "Incoterm": Incoterm,
            "Part": Part,
        }

        with CoreSessionLocal() as session, \
             open(file_path, 'r', encoding=_facturas_csv_encoding(file_path)) as f_in, \
             open(error_path, 'w', encoding='utf-8') as f_err:
            validator = ForeignKeyValidator(session, tenant_id, company_id)

            def clear_existing_invoice_details(invoice_id: int) -> None:
                """
                Reemplazo seguro por factura: evita bulk delete directo sobre item_lines
                para que ORM aplique cascadas y no rompa FK en tablas hijas.
                """
                existing_line_ids = [
                    line_id
                    for (line_id,) in session.query(LineItem.id)
                    .filter(LineItem.invoice_id == invoice_id)
                    .all()
                ]
                if not existing_line_ids:
                    session.query(InvoiceSalesDetails).filter(
                        InvoiceSalesDetails.invoice_id == invoice_id
                    ).delete(synchronize_session=False)
                    return

                # Dependencia cross-schema fuera de cascada DB; se limpia explícitamente.
                session.query(FaLineItem).filter(
                    FaLineItem.id.in_(existing_line_ids)
                ).delete(synchronize_session=False)

                # Borrado por ORM para activar cascadas definidas en LineItem.
                existing_lines = (
                    session.query(LineItem)
                    .filter(LineItem.id.in_(existing_line_ids))
                    .all()
                )
                for existing_line in existing_lines:
                    session.delete(existing_line)

                session.query(InvoiceSalesDetails).filter(
                    InvoiceSalesDetails.invoice_id == invoice_id
                ).delete(synchronize_session=False)
            invoice_id_cache: Dict[str, Optional[int]] = {}
            
            # Detect Delimiter
            sample = f_in.read(2048)
            f_in.seek(0)
            try:
                dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
            except:
                dialect = 'excel'
                
            reader = csv.DictReader(f_in, dialect=dialect)
            
            for i, row in enumerate(reader, start=1):
                # Check for Progress Update
                self.update_state(state='PROGRESS', meta={
                    'current': i,
                    'total': total_rows,
                    'errors': error_count
                })
                
                # Solo columnas de la plantilla (respetar plantilla tal cual)
                row_norm = row_from_template(row, template_id, normalize_header)
                errors = validate_row_strict(
                    row_norm,
                    model_target,
                    i,
                    date_format,
                    validator,
                    inv_type_value,
                    invoice_id_cache,
                    models,
                )
                
                if errors:
                    error_count += 1
                    f_err.write(json.dumps(errors) + "\n")
                    if len(errors_detail) < 5000:
                        errors_detail.append({
                            "line": errors.get("line", i),
                            "col": errors.get("col", ""),
                            "msg": errors.get("msg", ""),
                            "solution": errors.get("solution", ""),
                            "warning": bool(errors.get("warning", False)),
                        })
                
                processed_rows += 1
                
    except Exception as e:
        logger.error(f"Scan failed: {e}")
        return {"status": "failed", "error": str(e)}

    # 4. Store error line numbers in Redis so insert_valid_rows can skip them (any worker)
    error_lines_list = []
    try:
        if os.path.exists(error_path):
            with open(error_path, "r", encoding="utf-8") as f:
                for line in f:
                    try:
                        err = json.loads(line)
                        if "line" in err:
                            error_lines_list.append(err["line"])
                    except Exception:
                        pass
        common_storage.store_error_lines(effective_job_type, job_id, error_lines_list)
    except Exception as e:
        logger.warning(f"Failed to store error lines in Redis: {e}")

    return common_responses.scan_result(job_id, processed_rows, error_count, errors_detail)

def validate_row_phase_1(
    row: Dict[str, Any],
    target: str,
    line_num: int,
    date_format: Optional[str],
) -> Optional[Dict[str, Any]]:
    """
    Validation: Unique IDs, Dates, and Numeric constraint checks.
    Target: 'invoice_header' or 'invoice_details'
    """
    def check_decimal(col_name):
        val = row.get(col_name)
        if val and str(val).strip():
            if parse_decimal(val) is None:
                return {
                    "line": line_num,
                    "col": col_name,
                    "msg": "Debe ser un número decimal válido",
                    "solution": f"Capturar en la columna {col_name} un número decimal válido (ej. 12.34).",
                }
        return None

    def check_int(col_name):
        val = row.get(col_name)
        if val and str(val).strip():
            if parse_int(val) is None:
                return {
                    "line": line_num,
                    "col": col_name,
                    "msg": "Debe ser un número entero válido",
                    "solution": f"Capturar en la columna {col_name} un número entero válido (solo números, sin letras).",
                }
        return None
        
    def check_date(col_name):
        date_str = row.get(col_name)
        if date_str and str(date_str).strip():
            if not is_valid_date(date_str, date_format):
                expected = display_date_format(date_format)
                return {
                    "line": line_num,
                    "col": col_name,
                    "msg": f"Formato de fecha inválido ({expected})",
                    "solution": f"Capturar una fecha con el formato esperado ({expected}).",
                }
        return None

    def check_weight(col_name):
        val = row.get(col_name)
        if val and str(val).strip():
            if parse_weight_unit(val) is None:
                return {
                    "line": line_num,
                    "col": col_name,
                    "msg": "Unidad de peso inválida (ej. KGS, LBS)",
                    "solution": f"Capturar en la columna {col_name} una unidad de peso válida (ej. KGS o LBS).",
                }
        return None

    def check_currency(col_name):
        val = row.get(col_name)
        if val and str(val).strip():
            if csv_tipo_moneda_es_me_mn_mc(str(val)):
                return None
            parsed_currency = parse_currency(val, None)
            val_norm = normalize_header(val)
            # parse_currency returns MANUAL if unknown, so if it wasn't explicitly MANUAL, it's invalid
            if parsed_currency.value == "manual" and "MANUAL" not in val_norm:
                return {
                    "line": line_num,
                    "col": col_name,
                    "msg": "Moneda inválida (ej. MN, ME, USD, PESOS)",
                    "solution": f"Capturar en la columna {col_name} una moneda válida (ej. MN, ME, USD, PESOS) o usar 'MANUAL' si aplica.",
                }
        return None

    def check_transport_type(col_name):
        val = row.get(col_name)
        if val and str(val).strip():
            if str(val).strip().lower() not in TRANSPORT_TYPE_VALUES:
                return {
                    "line": line_num,
                    "col": col_name,
                    "msg": "Tipo de transporte inválido (ej. box, truck, container)",
                    "solution": f"Capturar en la columna {col_name} un tipo de transporte válido (ej. box, truck, container).",
                }
        return None

    # A. Invoice Header
    if target == 'invoice_header':
        # 1. Unique ID
        if not row.get('NUMERO FACTURA') and not row.get('NUM FACTURA') and not row.get('ID'):
            return {
                "line": line_num,
                "col": "NUMERO FACTURA",
                "msg": "Requerido",
                "solution": "Capturar el número de Factura en la columna NUMERO FACTURA.",
            }
        
        # 2. Date Format
        date_str = row.get('FECHA FACTURA') or row.get('FECHA')
        if not date_str or not str(date_str).strip():
            return {
                "line": line_num,
                "col": "FECHA FACTURA",
                "msg": "Requerido",
                "solution": "Capturar la FECHA FACTURA en la columna FECHA FACTURA.",
            }
            
        err = check_date('FECHA FACTURA') or check_date('FECHA')
        if err: return err
        
        err = check_date('FECHA EMISION')
        if err: return err

        # 3. Numeric Fields
        for col in ['TIPO DE CAMBIO', 'FLETES', 'VALOR SEGUROS', 'SEGUROS', 'EMBALAJES', 'OTROS INCREMENTABLES']:
            err = check_decimal(col)
            if err: return err
            
        # 4. Integer FKs (CLAVE PROVEEDOR/VENDIDO/ENVIADO aceptan short_name; AGENTE ADUANAL acepta clave; solo REMESA exige entero)
        int_fk_cols = ['CLAVE PROVEEDOR', 'CLAVE VENDIDO A', 'CLAVE ENVIADO A', 'AGENTE ADUANAL', 'REMESA']
        if target == 'invoice_header':
            int_fk_cols = ['REMESA']  # proveedor/vendido/enviado por short_name; agente aduanal por clave
        for col in int_fk_cols:
            err = check_int(col)
            if err: return err
            
        # 5. Enums
        for col in ['TIPO PESO']:
            err = check_weight(col)
            if err: return err
            
        for col in ['TIPO MONEDA']:
            err = check_currency(col)
            if err: return err
            
        for col in ['TIPO TRANSPORTE']:
            err = check_transport_type(col)
            if err: return err

    # B. Invoice Details (Parts)
    elif target == 'invoice_details':
        # 1. Line Number
        if not row.get('LINEA') and not row.get('RENGLON') and not row.get('PARTIDA'):
            return {
                "line": line_num,
                "col": "LINEA",
                "msg": "Requerido",
                "solution": "Capturar el número de línea en la columna LINEA.",
            }

        # 2. Parent Link (Invoice Number)
        if not (row.get('NUMERO FACTURA') or row.get('NUM FACTURA') or row.get('FACTURA')):
            return {
                "line": line_num,
                "col": "NUMERO FACTURA",
                "msg": "Requerido",
                "solution": "Capturar el número de Factura en la columna NUMERO FACTURA.",
            }
            
        # 3. Numeric Fields
        for col in ['PRECIO UNITARIO', 'PRECIOUNITARIO', 'VALOR COMERCIAL', 'VALORCOMERCIAL', 'CANTIDAD']:
            err = check_decimal(col)
            if err: return err

        for col in ['CANTIDAD BULTOS', 'CANTIDADBULTOS', 'LINEA', 'RENGLON', 'PARTIDA']:
            err = check_int(col)
            if err: return err
    
    return None


def validate_row_strict(
    row: Dict[str, Any],
    target: str,
    line_num: int,
    date_format: Optional[str],
    validator: ForeignKeyValidator,
    inv_type_value: str,
    invoice_id_cache: Dict[str, Optional[int]],
    models: Dict[str, Any],
) -> Optional[Dict[str, Any]]:
    err = validate_row_phase_1(row, target, line_num, date_format)
    if err:
        return err

    InvoiceHeader = models["InvoiceHeader"]
    InvoiceType = models["InvoiceType"]
    ClientProvider = models["ClientProvider"]
    CustomsBroker = models["CustomsBroker"]
    RegimenPedimento = models["RegimenPedimento"]
    CurrencyType = models["CurrencyType"]
    CustomsSection = models["CustomsSection"]
    Incoterm = models["Incoterm"]
    Part = models["Part"]

    if target == "invoice_header":
        if not validator.check_exists(InvoiceType, inv_type_value, field_name="key", is_public=True):
            return {
                "line": line_num,
                "col": "TIPO FACTURA",
                "msg": "No existe en el catalogo",
                "solution": "Capturar un TIPO FACTURA válido que exista en el catálogo.",
            }

        err = _validate_client_provider_ref(
            validator, ClientProvider, row.get("CLAVE PROVEEDOR"), line_num, "CLAVE PROVEEDOR", required=True
        )
        if err:
            return err

        err = _validate_client_provider_ref(
            validator, ClientProvider, row.get("CLAVE VENDIDO A"), line_num, "CLAVE VENDIDO A", required=True
        )
        if err:
            return err

        err = _validate_client_provider_ref(
            validator, ClientProvider, row.get("CLAVE ENVIADO A"), line_num, "CLAVE ENVIADO A", required=True
        )
        if err:
            return err

        err = _validate_customs_broker_ref(
            validator, CustomsBroker, row.get("AGENTE ADUANAL"), line_num, "AGENTE ADUANAL", required=False
        )
        if err:
            return err

        err = validate_public_code(
            validator,
            RegimenPedimento,
            row.get("REGIMEN") or row.get("CLAVEDOCUMENTO"),
            line_num,
            "CLAVEDOCUMENTO",
        )
        if err:
            return err

        err = validate_public_code(
            validator,
            CustomsSection,
            row.get("ADUANA DE CRUCE"),
            line_num,
            "ADUANA DE CRUCE",
            field_name="customs_code",
        )
        if err:
            return err

        err = validate_public_code(
            validator,
            CurrencyType,
            row.get("CLAVE MONEDA"),
            line_num,
            "CLAVE MONEDA",
        )
        if err:
            return err

        err = validate_public_code(
            validator,
            Incoterm,
            row.get("CLAVE INCOTERM"),
            line_num,
            "CLAVE INCOTERM",
        )
        if err:
            return err

        from .validators.transport_catalog import logistics_scan_row_errors

        log_errs = logistics_scan_row_errors(
            row,
            validator.session,
            validator.tenant_id,
            validator.company_id,
            line_num,
        )
        if log_errs:
            return log_errs[0]

    elif target == "invoice_details":
        invoice_number = (row.get("NUMERO FACTURA") or row.get("NUM FACTURA") or row.get("FACTURA") or "").strip()
        if not invoice_number:
            return {
                "line": line_num,
                "col": "NUMERO FACTURA",
                "msg": "Requerido",
                "solution": "Capturar el número de Factura en la columna NUMERO FACTURA.",
            }

        cache_key = f"{invoice_number}|{inv_type_value}"
        if cache_key in invoice_id_cache:
            invoice_id = invoice_id_cache[cache_key]
        else:
            invoice_id = (
                validator.session.query(InvoiceHeader.id)
                .filter(
                    InvoiceHeader.tenant_id == validator.tenant_id,
                    InvoiceHeader.company_id == validator.company_id,
                    InvoiceHeader.invoice_number == invoice_number,
                    InvoiceHeader.invoice_type == inv_type_value,
                )
                .scalar()
            )
            invoice_id_cache[cache_key] = invoice_id
        if not invoice_id:
            return {
                "line": line_num,
                "col": "NUMERO FACTURA",
                "msg": "Factura no existe",
                "solution": "Capturar un número de Factura que exista en el sistema (verificar NUMERO FACTURA).",
            }

        part_num = (row.get("NUMPARTE") or row.get("NUMERO PARTE") or "").strip()
        if not part_num:
            return {
                "line": line_num,
                "col": "NUMPARTE",
                "msg": "Requerido",
                "solution": "Capturar el número de parte en la columna NUMPARTE.",
            }
        if not validator.check_exists(Part, part_num, field_name="part_number"):
            return {
                "line": line_num,
                "col": "NUMPARTE",
                "msg": "No existe en el catalogo",
                "solution": "Capturar un número de parte que exista en el catálogo (verificar NUMPARTE).",
            }

    return None

def parse_footer_config(config: Optional[str]) -> Dict[str, Any]:
    if not config:
        return {}
    try:
        if isinstance(config, str):
            return json.loads(config)
        if isinstance(config, dict):
            return config
    except Exception:
        return {}
    return {}


def display_date_format(date_format: Optional[str]) -> str:
    if not date_format:
        return "YYYY-MM-DD"
    return date_format.upper()


def parse_date(date_text: Optional[str], date_format: Optional[str]) -> Optional[datetime.date]:
    if not date_text:
        return None
    raw = str(date_text).strip()
    # Soporte para fechas numéricas (layouts legacy):
    # - Clarion: días desde 1800-12-28 (ej. 75519)
    # - Excel serial: días desde 1899-12-30
    if re.fullmatch(r"\d+(\.\d+)?", raw):
        try:
            serial = int(float(raw))
            if serial > 0:
                clarion_date = datetime(1800, 12, 28).date() + timedelta(days=serial)
                if 1900 <= clarion_date.year <= 2200:
                    return clarion_date
                excel_date = datetime(1899, 12, 30).date() + timedelta(days=serial)
                if 1900 <= excel_date.year <= 2200:
                    return excel_date
        except Exception:
            pass
    candidates = []
    fmt_map = {
        "dd/mm/yyyy": "%d/%m/%Y",
        "mm/dd/yyyy": "%m/%d/%Y",
        "yyyy-mm-dd": "%Y-%m-%d",
    }
    if date_format and date_format in fmt_map:
        candidates.append(fmt_map[date_format])
    candidates.extend(["%Y-%m-%d", "%d/%m/%Y", "%m/%d/%Y"])
    for fmt in candidates:
        try:
            return datetime.strptime(raw, fmt).date()
        except ValueError:
            continue
    return None


def is_valid_date(date_text: Optional[str], date_format: Optional[str]) -> bool:
    return parse_date(date_text, date_format) is not None


def normalize_header(name: Optional[str]) -> str:
    if not name:
        return ""
    name = unicodedata.normalize("NFKD", str(name)).upper()
    name = "".join(ch for ch in name if not unicodedata.combining(ch))
    name = re.sub(r"[^A-Z0-9]+", " ", name)
    return re.sub(r"\s+", " ", name).strip()


def normalize_row(row: Dict[str, Any]) -> Dict[str, Any]:
    return {normalize_header(k): v for k, v in row.items()}


def parse_int(value: Any) -> Optional[int]:
    if value is None:
        return None
    text = str(value).strip()
    if not text:
        return None
    try:
        return int(text)
    except ValueError:
        pass
    try:
        f = float(text.replace(",", ""))
        if f == int(f):
            return int(f)
        return None
    except ValueError:
        return None


def parse_decimal(value: Any) -> Optional[Decimal]:
    if value is None:
        return None
    text = str(value).strip()
    if not text:
        return None
    text = text.replace(",", "")
    try:
        return Decimal(text)
    except Exception:
        return None


def decimal_or_zero(value: Any) -> Decimal:
    """Return parsed decimal or Decimal('0') for CSV nulls/empty (vanilla default)."""
    return parse_decimal(value) or Decimal("0")


def int_or_zero(value: Any) -> int:
    """Return parsed int or 0 for CSV nulls/empty (vanilla default)."""
    return parse_int(value) if parse_int(value) is not None else 0


def parse_currency(value: Optional[str], currency_type: Optional[str]):
    from api.v1.modules.a76.invoices.models import Currency
    if value:
        normalized = normalize_header(value)
        if normalized in {"MN", "M N", "NACIONAL", "LOCAL", "PESOS", "PESO"}:
            return Currency.LOCAL
        if normalized in {"ME", "M E", "EXTRANJERA", "EXTRANJERO", "FOREIGN", "USD", "DOLAR", "DOLARES"}:
            return Currency.FOREIGN
        if "MANUAL" in normalized:
            return Currency.MANUAL
        # MC (moneda por clave): paridad con encabezados_impo_temp / InvoiceFinancials + CLAVE MONEDA
        if normalized.replace(" ", "") == "MC":
            if currency_type and str(currency_type).strip().upper() == "MXN":
                return Currency.LOCAL
            if currency_type:
                return Currency.FOREIGN
            return Currency.FOREIGN
    if currency_type and str(currency_type).strip().upper() == "MXN":
        return Currency.LOCAL
    if currency_type:
        return Currency.FOREIGN
    return Currency.MANUAL


def parse_weight_unit(value: Optional[str]):
    from api.v1.modules.a76.invoices.models import WeightUnit
    if not value:
        return None
    normalized = normalize_header(value)
    if normalized in {"KG", "KGS", "KILOS", "KILOGRAMOS"}:
        return WeightUnit.KGS
    if normalized in {"LB", "LBS", "LIBRAS"}:
        return WeightUnit.LBS
    return None


def resolve_tenant_fk_id(
    session: CoreSessionLocal,
    model,
    value: Optional[int],
    tenant_id: int,
    company_id: int,
    cache: Dict[int, Optional[int]],
) -> Optional[int]:
    if value is None:
        return None
    if value in cache:
        return cache[value]
    exists = (
        session.query(model.id)
        .filter(
            model.id == value,
            model.tenant_id == tenant_id,
            model.company_id == company_id,
        )
        .scalar()
    )
    cache[value] = value if exists is not None else None
    return cache[value]


def resolve_client_provider_id(
    session: CoreSessionLocal,
    model,
    value: Any,
    tenant_id: int,
    company_id: int,
    cache: Dict[Any, Optional[int]],
) -> Optional[int]:
    """Resuelve ID de ClientProvider por id (entero) o por short_name (texto). value puede ser int o str."""
    if value is None or (isinstance(value, str) and not value.strip()):
        return None
    if value in cache:
        return cache[value]
    pid = parse_int(value)
    if pid is not None:
        found = (
            session.query(model.id)
            .filter(
                model.id == pid,
                model.tenant_id == tenant_id,
                model.company_id == company_id,
            )
            .scalar()
        )
        cache[value] = found
        return found
    short_norm = str(value).strip().upper()
    if short_norm in cache:
        return cache[short_norm]
    found = (
        session.query(model.id)
        .filter(
            func.upper(model.short_name) == short_norm,
            model.tenant_id == tenant_id,
            model.company_id == company_id,
        )
        .scalar()
    )
    cache[value] = found
    cache[short_norm] = found
    return found


def resolve_customs_broker_id(
    session: CoreSessionLocal,
    model,
    value: Any,
    tenant_id: int,
    company_id: int,
    cache: Dict[Any, Optional[int]],
    broker_lookup: Optional[Dict[str, Any]] = None,
) -> Optional[int]:
    """Resuelve ID de CustomsBroker por id (entero) o por broker_key (clave). value puede ser int o str."""
    if value is None or (isinstance(value, str) and not value.strip()):
        return None
    if broker_lookup is not None:
        resolved_id, _mode, norm, _err = resolve_customs_broker_ref(value, broker_lookup)
        cache[value] = resolved_id
        if norm:
            cache[norm] = resolved_id
        return resolved_id
    if value in cache:
        return cache[value]
    pid = parse_int(value)
    if pid is not None:
        # Primero resolvemos como ID; si no existe, seguimos como broker_key.
        found = (
            session.query(model.id)
            .filter(
                model.id == pid,
                model.tenant_id == tenant_id,
                model.company_id == company_id,
            )
            .scalar()
        )
        if found is not None:
            cache[value] = found
            return found
    clave = str(value).strip()
    if clave in cache:
        return cache[clave]
    found = (
        session.query(model.id)
        .filter(
            model.broker_key == clave,
            model.tenant_id == tenant_id,
            model.company_id == company_id,
        )
        .scalar()
    )
    cache[value] = found
    cache[clave] = found
    return found


def resolve_public_code(
    session: CoreSessionLocal,
    model,
    column,
    value: Optional[str],
    cache: Dict[str, Optional[str]],
) -> Optional[str]:
    if not value:
        return None
    normalized = str(value).strip().upper()
    if not normalized:
        return None
    if normalized in cache:
        return cache[normalized]
    exists = session.query(column).filter(column == normalized).scalar()
    cache[normalized] = normalized if exists is not None else None
    return cache[normalized]

@celery_app.task(bind=True)
def insert_valid_rows(self, job_id: str, model_target: str, job_type_override: Optional[str] = None):
    """Pass 2: Re-read CSV, Skip Errors, Bulk Insert. Delegates to _do_insert_valid_rows."""
    return _do_insert_valid_rows(job_id, model_target, job_type_override)


def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Optional[str] = None) -> Dict[str, Any]:
    """
    Pass 2: Re-read CSV, Skip Errors, Bulk Insert.
    File and meta are loaded from Redis if present (same as scan_file), so worker does not need shared filesystem.
    When job_type_override is set (e.g. "exp" for Exportación), storage keys use that prefix.
    """
    effective_job_type = job_type_override if job_type_override is not None else JOB_TYPE
    log_prefix = "Exportación import" if effective_job_type else "Invoices import"

    logger.info(f"Starting Commit for {job_id} target {model_target}")

    # Ensure we have the file on this worker: prefer Redis (so any worker can run commit)
    file_path = common_storage.ensure_file_from_redis(effective_job_type, job_id, log_prefix)
    if not file_path:
        alt_path = common_storage.file_path_for_job(effective_job_type, job_id)
        if not os.path.exists(alt_path):
            return {"status": "failed", "error": "File not found (missing or expired). Please upload and confirm again."}
        file_path = alt_path
    else:
        common_storage.ensure_meta_from_redis(effective_job_type, job_id, file_path, log_prefix)
    _ensure_utf8_compatible_import_file(file_path)

    try:
        tenant_id, company_id = common_meta.require_tenant_context(file_path)
    except ValueError as e:
        return {"status": "failed", "error": str(e)}

    meta = common_meta.load_meta(file_path)
    meta_path = common_meta.get_meta_path(file_path)
    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)

    # 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"
        or meta.get("template_id") in ("imp_temp_series", "imp_def_series", "cmex_series", "exp_def_series")
    )

    _footer_for_series = parse_footer_config(meta.get("footer_config")) or {}
    _inv_type_series = normalize_public_code(_footer_for_series.get("invoice_type") or meta.get("invoice_type") or "")
    use_def_series_commit = (
        use_series_flow
        and (
            meta.get("template_id") == "imp_def_series"
            or meta.get("template_id") == "cmex_series"
            or _inv_type_series in ("DEF", "MATDE", "EXDEF")
        )
    )
    use_expo_series_commit = use_series_flow and meta.get("template_id") == "exp_def_series"

    # --- Series de Exportación Definitiva: commit (INSERT/UPDATE item_line_series para facturas exp) ---
    if use_expo_series_commit:
        try:
            from api.v1.modules.a76.invoices.models import InvoiceHeader
            from api.v1.modules.a76.items.models import LineItem
            from api.v1.modules.a76.items.series.models import Serie
            from .validators.series_expo import row_to_series_normalized_expo

            actualizar = meta.get("actualizar", False)
            autonumerar = meta.get("autonumerar", True)
            _fc = parse_footer_config(meta.get("footer_config"))
            if _fc:
                if "actualizar" in _fc:
                    actualizar = bool(_fc["actualizar"])
                if "autonumerar" in _fc:
                    autonumerar = bool(_fc["autonumerar"])
                else:
                    as_val = _fc.get("autonumber_series", "true")
                    autonumerar = str(as_val).lower() in ("true", "1", "si", "sí", "yes")

            with CoreSessionLocal() as session:
                q_inv = (
                    session.query(InvoiceHeader.invoice_number, InvoiceHeader.id)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == "exp",
                    )
                )
                invoice_id_by_number: Dict[str, int] = {}
                for num, iid in q_inv.all():
                    if num:
                        invoice_id_by_number[str(num).strip()] = iid

                inserted_count = 0
                updated_count = 0
                skipped_invalid = 0
                skipped_details: List[Dict[str, Any]] = []

                with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f:
                    sample = f.read(2048)
                    f.seek(0)
                    try:
                        dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                    except Exception:
                        dialect = "excel"
                    reader = csv.DictReader(f, dialect=dialect)

                    for i, row in enumerate(reader, start=1):
                        if i in error_lines:
                            skipped_invalid += 1
                            continue
                        row_norm = row_from_template(row, "exp_def_series", normalize_header)
                        data = row_to_series_normalized_expo(row_norm)
                        invoice_number = data["NUMERO FACTURA"]
                        linea_factura = data["LINEA FACTURA"]
                        linea_serie = data["LINEA SERIE"]

                        if not invoice_number or invoice_number not in invoice_id_by_number:
                            skipped_invalid += 1
                            skipped_details.append({
                                "line": i,
                                "invoice": invoice_number or "(vacío)",
                                "reason": "Factura de exportación no encontrada.",
                                "solution": "Capturar en el CSV un número de Factura de exportación que exista en el sistema (columna NUMERO FACTURA).",
                            })
                            continue
                        invoice_id = invoice_id_by_number[invoice_number]
                        line_number_val = parse_int(linea_factura)
                        if line_number_val is None:
                            skipped_invalid += 1
                            skipped_details.append({
                                "line": i,
                                "invoice": invoice_number,
                                "reason": "LINEA FACTURA debe ser numérico.",
                                "solution": "Capturar en el CSV la LINEA FACTURA solo con números enteros.",
                            })
                            continue
                        line_item = (
                            session.query(LineItem)
                            .filter(
                                LineItem.invoice_id == invoice_id,
                                LineItem.line_number == line_number_val,
                                LineItem.tenant_id == tenant_id,
                                LineItem.company_id == company_id,
                            )
                            .first()
                        )
                        if not line_item:
                            line_numbers = [
                                r[0] for r in
                                session.query(LineItem.line_number)
                                .filter(
                                    LineItem.invoice_id == invoice_id,
                                    LineItem.tenant_id == tenant_id,
                                    LineItem.company_id == company_id,
                                )
                                .order_by(LineItem.line_number)
                                .all()
                            ]
                            existing_str = ", ".join(str(n) for n in line_numbers) if line_numbers else "ninguna"
                            skipped_invalid += 1
                            skipped_details.append({
                                "line": i,
                                "invoice": invoice_number,
                                "reason": f"Partida línea {linea_factura} no existe en la factura. Partidas existentes: {existing_str}.",
                                "solution": "Revisar que la partida (LINEA FACTURA) exista en la factura indicada en el CSV.",
                            })
                            continue

                        if autonumerar:
                            max_row = (
                                session.query(Serie.row)
                                .filter(Serie.line_item_id == line_item.id)
                                .order_by(Serie.row.desc())
                                .limit(1)
                                .scalar()
                            )
                            row_num = (max_row or 0) + 1
                        else:
                            row_num = parse_int(linea_serie)
                            if row_num is None:
                                skipped_invalid += 1
                                skipped_details.append({
                                    "line": i,
                                    "invoice": invoice_number,
                                    "reason": "LINEA SERIE debe ser numérico.",
                                    "solution": "Capturar en el CSV la LINEA SERIE solo con números enteros.",
                                })
                                continue

                        existing_serie = (
                            session.query(Serie)
                            .filter(
                                Serie.line_item_id == line_item.id,
                                Serie.row == row_num,
                            )
                            .first()
                        )

                        if existing_serie:
                            if actualizar:
                                existing_serie.serial_numbers = data["SERIE"] or existing_serie.serial_numbers
                                existing_serie.model = data["MODELO"] or existing_serie.model
                                existing_serie.sub_model = data["SUB MODELO"] or existing_serie.sub_model
                                existing_serie.number_id = data["NUMERO ID"] or existing_serie.number_id
                                session.add(existing_serie)
                                updated_count += 1
                            else:
                                skipped_invalid += 1
                                skipped_details.append({
                                    "line": i,
                                    "invoice": invoice_number,
                                    "reason": "Serie ya existe (use actualizar).",
                                    "solution": "La serie ya existe: activa la opción de Actualizar o ajusta tu carga para que actualice en lugar de crear.",
                                })
                        else:
                            new_serie = Serie(
                                tenant_id=tenant_id,
                                company_id=company_id,
                                line_item_id=line_item.id,
                                row=row_num,
                                serial_numbers=data["SERIE"] or None,
                                model=data["MODELO"] or None,
                                sub_model=data["SUB MODELO"] or None,
                                number_id=data["NUMERO ID"] or None,
                            )
                            session.add(new_serie)
                            inserted_count += 1

                session.commit()

            common_storage.cleanup_import_job(effective_job_type, job_id, file_path=file_path, error_path=error_path, meta_path=meta_path)
            status = "finished" if (inserted_count + updated_count) > 0 else ("warning" if skipped_invalid else "failed")
            out = {
                "status": status,
                "inserted": inserted_count,
                "updated": updated_count,
                "skipped_invalid": skipped_invalid,
                "skipped_missing_fk": 0,
                "skipped_duplicate": 0,
                "skipped_details": skipped_details,
                "skipped_summary": _summarize_top_reasons(skipped_details),
            }
            if status == "failed":
                out["error"] = "No hay registros válidos en el archivo CSV."
            elif status == "warning" and skipped_invalid:
                out["message"] = f"No se insertaron registros. {skipped_invalid} fueron rechazados."
            return out
        except Exception as e:
            logger.exception("Series exportación definitiva commit failed: %s", e)
            return {"status": "failed", "error": str(e)}

    # --- Series de Importación Definitiva: commit (INSERT/UPDATE item_line_series para facturas DEF o MEX) ---
    if use_def_series_commit:
        try:
            from api.v1.modules.a76.invoices.models import InvoiceHeader
            from api.v1.modules.a76.items.models import LineItem
            from api.v1.modules.a76.items.series.models import Serie
            from api.v1.modules.a76.items.line_quantities.models import LineQuantity
            from .validators.series_impo_def import (
                validate_row_series_impo_def,
                row_to_series_normalized_def,
            )

            _series_template_id = meta.get("template_id")
            is_cmex_series = _series_template_id == "cmex_series"
            SERIES_INV_TYPES = ("MEX",) if is_cmex_series else ("DEF", "MATDE", "EXDEF")
            series_row_template_id = "cmex_series" if is_cmex_series else "imp_def_series"

            DEF_INVOICE_TYPES = ("DEF", "MATDE", "EXDEF")  # keep for any legacy reference
            actualizar = meta.get("actualizar", False)
            autonumerar = meta.get("autonumerar", True)
            validar_series_exception = meta.get("validar_series", False)
            _fc = parse_footer_config(meta.get("footer_config"))
            if _fc:
                if "actualizar" in _fc:
                    actualizar = bool(_fc["actualizar"])
                elif _fc.get("mode") == "update":
                    actualizar = True
                elif _fc.get("mode") == "replace":
                    actualizar = False
                if "autonumerar" in _fc:
                    autonumerar = bool(_fc["autonumerar"])
                else:
                    as_val = _fc.get("autonumber_series", "true")
                    autonumerar = str(as_val).lower() in ("true", "1", "si", "sí", "yes")
                if "validar_series" in _fc:
                    validar_series_exception = bool(_fc["validar_series"])

            with CoreSessionLocal() as session:
                # Switch maestro: validarseries desactiva toda la validación de series cuando = 0
                _sys_settings = AppSettingsService.get_resolved_settings(session, int(tenant_id), int(company_id))
                _gen_p = _sys_settings.get("ssisgen", {}) or _sys_settings.get("qsisgen", {})
                if str(_gen_p.get("validarseries", "0")).strip() != "1":
                    validar_series_exception = False

                q = (
                    session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.status)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == "imp",
                        InvoiceHeader.invoice_type.in_(SERIES_INV_TYPES),
                    )
                )
                rows_inv = q.all()
                invoice_id_by_number: Dict[str, int] = {}
                invoice_processed_by_number: Dict[str, bool] = {}
                for num, iid, is_upd in rows_inv:
                    if num:
                        invoice_id_by_number[str(num).strip()] = iid
                        invoice_processed_by_number[str(num).strip()] = _is_invoice_processed_status(is_upd)

                partida_max_series = {}
                q_qty = (
                    session.query(
                        InvoiceHeader.invoice_number,
                        LineItem.line_number,
                        LineQuantity.quantity,
                    )
                    .join(LineItem, LineItem.invoice_id == InvoiceHeader.id)
                    .outerjoin(LineQuantity, LineQuantity.item_line_id == LineItem.id)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == "imp",
                        InvoiceHeader.invoice_type.in_(SERIES_INV_TYPES),
                    )
                )
                for num, ln, qty in q_qty.all():
                    if num is not None and ln is not None:
                        key = (str(num).strip(), str(ln).strip())
                        partida_max_series[key] = int(qty) if qty else 0

                existing_series_keys = set()
                existing_series_data: Dict[Tuple[str, str, str], Dict[str, Any]] = {}
                if actualizar and not autonumerar:
                    q_ser = (
                        session.query(
                            InvoiceHeader.invoice_number,
                            LineItem.line_number,
                            Serie.row,
                            Serie.serial_numbers,
                            Serie.model,
                            Serie.sub_model,
                            Serie.number_id,
                        )
                        .join(LineItem, LineItem.invoice_id == InvoiceHeader.id)
                        .join(Serie, Serie.line_item_id == LineItem.id)
                        .filter(
                            InvoiceHeader.tenant_id == tenant_id,
                            InvoiceHeader.company_id == company_id,
                            InvoiceHeader.operation_type == "imp",
                            InvoiceHeader.invoice_type.in_(SERIES_INV_TYPES),
                        )
                    )
                    for num, ln, rw, sn, md, sm, nid in q_ser.all():
                        if num is not None:
                            k = (str(num).strip(), str(ln).strip(), str(rw).strip())
                            existing_series_keys.add(k)
                            existing_series_data.setdefault(k, {
                                "serial_numbers": sn or "",
                                "model": md or "",
                                "sub_model": sm or "",
                                "number_id": nid or "",
                            })

                csv_series_count_so_far: Dict[Tuple[str, str], int] = {}
                inserted_count = 0
                updated_count = 0
                skipped_invalid = 0
                skipped_details: List[Dict[str, Any]] = []

                with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f:
                    sample = f.read(2048)
                    f.seek(0)
                    try:
                        dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                    except Exception:
                        dialect = "excel"
                    reader = csv.DictReader(f, dialect=dialect)

                    for i, row in enumerate(reader, start=1):
                        if i in error_lines:
                            continue
                        row_norm = row_from_template(row, series_row_template_id, normalize_header)
                        row_errors = validate_row_series_impo_def(
                            row_norm,
                            i,
                            actualizar=actualizar,
                            autonumerar=autonumerar,
                            validar_series_exception=validar_series_exception,
                            invoice_id_by_number=invoice_id_by_number,
                            invoice_processed_by_number=invoice_processed_by_number,
                            partida_max_series=partida_max_series,
                            csv_series_count_so_far=csv_series_count_so_far,
                            existing_series_keys=existing_series_keys,
                            existing_series_data=existing_series_data,
                            warnings=None,
                            catalog_label="Compras Mexicanas" if is_cmex_series else "Importación Definitiva",
                        )
                        _blocking = [e for e in row_errors if not e.get("warning")]
                        if _blocking:
                            skipped_invalid += 1
                            for blocking in _blocking:
                                skipped_details.append({
                                    "line": i,
                                    "invoice": (row_norm.get("NUMERO FACTURA") or row_norm.get("NUM FACTURA") or "").strip(),
                                    "reason": blocking.get("msg", ""),
                                    "solution": blocking.get("solution", ""),
                                })
                            continue

                        data = row_to_series_normalized_def(row_norm)
                        invoice_number = data["NUMERO FACTURA"]
                        linea_factura = data["LINEA FACTURA"]
                        linea_serie = data["LINEA SERIE"]

                        if not invoice_number or invoice_number not in invoice_id_by_number:
                            skipped_invalid += 1
                            continue
                        invoice_id = invoice_id_by_number[invoice_number]
                        line_number_val = parse_int(linea_factura)
                        if line_number_val is None:
                            skipped_invalid += 1
                            skipped_details.append({
                                "line": i,
                                "invoice": invoice_number,
                                "reason": "LINEA FACTURA debe ser numérico.",
                                "solution": "Capturar en el CSV la LINEA FACTURA solo con números enteros.",
                            })
                            continue
                        line_item = (
                            session.query(LineItem)
                            .filter(
                                LineItem.invoice_id == invoice_id,
                                LineItem.line_number == line_number_val,
                                LineItem.tenant_id == tenant_id,
                                LineItem.company_id == company_id,
                            )
                            .first()
                        )
                        if not line_item:
                            line_numbers = [
                                r[0] for r in
                                session.query(LineItem.line_number)
                                .filter(
                                    LineItem.invoice_id == invoice_id,
                                    LineItem.tenant_id == tenant_id,
                                    LineItem.company_id == company_id,
                                )
                                .order_by(LineItem.line_number)
                                .all()
                            ]
                            existing_str = ", ".join(str(n) for n in line_numbers) if line_numbers else "ninguna"
                            skipped_invalid += 1
                            skipped_details.append({
                                "line": i,
                                "invoice": invoice_number,
                                "reason": f"Partida línea {linea_factura} no existe en la factura. Partidas existentes en la factura: {existing_str}.",
                                "solution": "Revisar que la partida (línea) exista en la factura indicada y corregir el valor en el CSV.",
                            })
                            continue

                        if autonumerar:
                            max_row = (
                                session.query(Serie.row)
                                .filter(Serie.line_item_id == line_item.id)
                                .order_by(Serie.row.desc())
                                .limit(1)
                                .scalar()
                            )
                            row_num = (max_row or 0) + 1
                        else:
                            row_num = parse_int(linea_serie)
                            if row_num is None:
                                skipped_invalid += 1
                                skipped_details.append({
                                    "line": i,
                                    "invoice": invoice_number,
                                    "reason": "LINEA SERIE debe ser numérico.",
                                    "solution": "Capturar en el CSV la LINEA SERIE solo con números enteros.",
                                })
                                continue

                        existing_serie = (
                            session.query(Serie)
                            .filter(
                                Serie.line_item_id == line_item.id,
                                Serie.row == row_num,
                            )
                            .first()
                        )

                        if existing_serie:
                            if actualizar:
                                existing_serie.serial_numbers = data["SERIE"] or existing_serie.serial_numbers
                                existing_serie.model = data["MODELO"] or existing_serie.model
                                existing_serie.sub_model = data["SUB MODELO"] or existing_serie.sub_model
                                existing_serie.number_id = data["NUMERO ID"] or existing_serie.number_id
                                session.add(existing_serie)
                                updated_count += 1
                            else:
                                skipped_invalid += 1
                                skipped_details.append({
                                    "line": i,
                                    "invoice": invoice_number,
                                    "reason": "Serie ya existe (use actualizar).",
                                    "solution": "Si la serie ya existe, usa el modo Actualizar para que el archivo haga update.",
                                })
                        else:
                            new_serie = Serie(
                                tenant_id=tenant_id,
                                company_id=company_id,
                                line_item_id=line_item.id,
                                row=row_num,
                                serial_numbers=data["SERIE"] or None,
                                model=data["MODELO"] or None,
                                sub_model=data["SUB MODELO"] or None,
                                number_id=data["NUMERO ID"] or None,
                            )
                            session.add(new_serie)
                            inserted_count += 1

                        key_csv = (invoice_number.strip(), (linea_factura or "").strip())
                        if key_csv[0] and key_csv[1]:
                            csv_series_count_so_far[key_csv] = csv_series_count_so_far.get(key_csv, 0) + 1

                session.commit()

            common_storage.cleanup_import_job(effective_job_type, job_id, file_path=file_path, error_path=error_path, meta_path=meta_path)
            status = "finished" if (inserted_count + updated_count) > 0 else ("warning" if skipped_invalid else "failed")
            out = {
                "status": status,
                "inserted": inserted_count,
                "updated": updated_count,
                "skipped_invalid": skipped_invalid,
                "skipped_missing_fk": 0,
                "skipped_duplicate": 0,
                "skipped_details": skipped_details,
                "skipped_summary": _summarize_top_reasons(skipped_details),
            }
            if status == "failed":
                out["error"] = "No hay registros válidos en el archivo CSV."
            elif status == "warning" and skipped_invalid:
                out["message"] = f"No se insertaron registros. {skipped_invalid} fueron rechazados."
            return out
        except Exception as e:
            logger.exception("Series importación definitiva commit failed: %s", e)
            return {"status": "failed", "error": str(e)}

    # --- Series de Importación Temporal: commit (INSERT/UPDATE item_line_series) ---
    # Paridad CSV: campos no presentes en CSV se persisten como null; no se exigen campos que no están en la plantilla.
    if use_series_flow:
        try:
            from api.v1.modules.a76.invoices.models import InvoiceHeader
            from api.v1.modules.a76.items.models import LineItem
            from api.v1.modules.a76.items.series.models import Serie
            from .validators.series_impo_temp import (
                validate_row_series_impo_temp,
                row_to_series_normalized,
            )

            actualizar = meta.get("actualizar", False)
            autonumerar = meta.get("autonumerar", True)
            validar_series_exception = meta.get("validar_series", False)
            _fc = parse_footer_config(meta.get("footer_config"))
            if _fc:
                if "actualizar" in _fc:
                    actualizar = bool(_fc["actualizar"])
                elif _fc.get("mode") == "update":
                    actualizar = True
                elif _fc.get("mode") == "replace":
                    actualizar = False
                if "autonumerar" in _fc:
                    autonumerar = bool(_fc["autonumerar"])
                else:
                    as_val = _fc.get("autonumber_series", "true")
                    autonumerar = str(as_val).lower() in ("true", "1", "si", "sí", "yes")
                if "validar_series" in _fc:
                    validar_series_exception = bool(_fc["validar_series"])

            with CoreSessionLocal() as session:
                # Switch maestro: validarseries desactiva toda la validación de series cuando = 0
                _sys_settings = AppSettingsService.get_resolved_settings(session, int(tenant_id), int(company_id))
                _gen_p = _sys_settings.get("ssisgen", {}) or _sys_settings.get("qsisgen", {})
                if str(_gen_p.get("validarseries", "0")).strip() != "1":
                    validar_series_exception = False

                q = (
                    session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.status)
                    .filter(
                        InvoiceHeader.tenant_id == tenant_id,
                        InvoiceHeader.company_id == company_id,
                        InvoiceHeader.operation_type == "imp",
                        InvoiceHeader.invoice_type == "TEM",
                    )
                )
                rows_inv = q.all()
                invoice_id_by_number: Dict[str, int] = {}
                invoice_processed_by_number: Dict[str, bool] = {}
                for num, iid, is_upd in rows_inv:
                    if num:
                        invoice_id_by_number[str(num).strip()] = iid
                        invoice_processed_by_number[str(num).strip()] = _is_invoice_processed_status(is_upd)

                existing_series_keys: Set[Tuple[str, str, str]] = set()
                existing_series_data: Dict[Tuple[str, str, str], Dict[str, Any]] = {}
                if actualizar and not autonumerar:
                    q_ser = (
                        session.query(
                            InvoiceHeader.invoice_number,
                            LineItem.line_number,
                            Serie.row,
                            Serie.serial_numbers,
                            Serie.model,
                            Serie.sub_model,
                            Serie.number_id,
                        )
                        .join(LineItem, LineItem.invoice_id == InvoiceHeader.id)
                        .join(Serie, Serie.line_item_id == LineItem.id)
                        .filter(
                            InvoiceHeader.tenant_id == tenant_id,
                            InvoiceHeader.company_id == company_id,
                            InvoiceHeader.operation_type == "imp",
                            InvoiceHeader.invoice_type == "TEM",
                        )
                    )
                    for num, ln, rw, sn, md, sm, nid in q_ser.all():
                        if num is not None:
                            k = (str(num).strip(), str(ln).strip(), str(rw).strip())
                            existing_series_keys.add(k)
                            existing_series_data.setdefault(k, {
                                "serial_numbers": sn or "",
                                "model": md or "",
                                "sub_model": sm or "",
                                "number_id": nid or "",
                            })

                inserted_count = 0
                updated_count = 0
                skipped_invalid = 0
                skipped_details: List[Dict[str, Any]] = []

                with open(file_path, "r", encoding=_facturas_csv_encoding(file_path)) as f:
                    sample = f.read(2048)
                    f.seek(0)
                    try:
                        dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                    except Exception:
                        dialect = "excel"
                    reader = csv.DictReader(f, dialect=dialect)

                    for i, row in enumerate(reader, start=1):
                        if i in error_lines:
                            continue
                        row_norm = row_from_template(row, "imp_temp_series", normalize_header)
                        row_errors = validate_row_series_impo_temp(
                            row_norm,
                            i,
                            actualizar=actualizar,
                            autonumerar=autonumerar,
                            validar_series_exception=validar_series_exception,
                            invoice_id_by_number=invoice_id_by_number,
                            invoice_processed_by_number=invoice_processed_by_number,
                            existing_series_keys=existing_series_keys,
                            existing_series_data=existing_series_data,
                            warnings=None,
                        )
                        _blocking = [e for e in row_errors if not e.get("warning")]
                        if _blocking:
                            skipped_invalid += 1
                            for blocking in _blocking:
                                skipped_details.append({
                                    "line": i,
                                    "invoice": (row_norm.get("NUMERO FACTURA") or row_norm.get("NUM FACTURA") or "").strip(),
                                    "reason": blocking.get("msg", ""),
                                    "solution": blocking.get("solution", ""),
                                })
                            continue

                        data = row_to_series_normalized(row_norm)
                        invoice_number = data["NUMERO FACTURA"]
                        linea_factura = data["LINEA FACTURA"]
                        linea_serie = data["LINEA SERIE"]

                        if not invoice_number or invoice_number not in invoice_id_by_number:
                            skipped_invalid += 1
                            continue
                        invoice_id = invoice_id_by_number[invoice_number]
                        line_number_val = parse_int(linea_factura)
                        if line_number_val is None:
                            skipped_invalid += 1
                            skipped_details.append({
                                "line": i,
                                "invoice": invoice_number,
                                "reason": "LINEA FACTURA debe ser numérico.",
                                "solution": "Capturar en el CSV la LINEA FACTURA solo con números enteros.",
                            })
                            continue
                        line_item = (
                            session.query(LineItem)
                            .filter(
                                LineItem.invoice_id == invoice_id,
                                LineItem.line_number == line_number_val,
                                LineItem.tenant_id == tenant_id,
                                LineItem.company_id == company_id,
                            )
                            .first()
                        )
                        if not line_item:
                            skipped_invalid += 1
                            skipped_details.append({
                                "line": i,
                                "invoice": invoice_number,
                                "reason": f"Partida línea {linea_factura} no existe en la factura.",
                                "solution": "Revisar que la partida (línea) exista en la factura indicada.",
                            })
                            continue

                        if autonumerar:
                            max_row = (
                                session.query(Serie.row)
                                .filter(Serie.line_item_id == line_item.id)
                                .order_by(Serie.row.desc())
                                .limit(1)
                                .scalar()
                            )
                            row_num = (max_row or 0) + 1
                        else:
                            row_num = parse_int(linea_serie)
                            if row_num is None:
                                skipped_invalid += 1
                                skipped_details.append({
                                    "line": i,
                                    "invoice": invoice_number,
                                    "reason": "LINEA SERIE debe ser numérico.",
                                    "solution": "Capturar en el CSV la LINEA SERIE solo con números enteros.",
                                })
                                continue

                        existing_serie = (
                            session.query(Serie)
                            .filter(
                                Serie.line_item_id == line_item.id,
                                Serie.row == row_num,
                            )
                            .first()
                        )

                        if existing_serie:
                            if actualizar:
                                existing_serie.serial_numbers = data["SERIE"] or existing_serie.serial_numbers
                                existing_serie.model = data["MODELO"] or existing_serie.model
                                existing_serie.sub_model = data["SUB MODELO"] or existing_serie.sub_model
                                existing_serie.number_id = data["NUMERO ID"] or existing_serie.number_id
                                session.add(existing_serie)
                                updated_count += 1
                            else:
                                skipped_invalid += 1
                                skipped_details.append({
                                    "line": i,
                                    "invoice": invoice_number,
                                    "reason": "Serie ya existe (use actualizar).",
                                    "solution": "Si la serie ya existe, usa el modo Actualizar para que el sistema haga update.",
                                })
                        else:
                            new_serie = Serie(
                                tenant_id=tenant_id,
                                company_id=company_id,
                                line_item_id=line_item.id,
                                row=row_num,
                                serial_numbers=data["SERIE"] or None,
                                model=data["MODELO"] or None,
                                sub_model=data["SUB MODELO"] or None,
                                number_id=data["NUMERO ID"] or None,
                            )
                            session.add(new_serie)
                            inserted_count += 1

                session.commit()

            common_storage.cleanup_import_job(effective_job_type, job_id, file_path=file_path, error_path=error_path, meta_path=meta_path)
            status = "finished" if (inserted_count + updated_count) > 0 else ("warning" if skipped_invalid else "failed")
            out = {
                "status": status,
                "inserted": inserted_count,
                "updated": updated_count,
                "skipped_invalid": skipped_invalid,
                "skipped_missing_fk": 0,
                "skipped_duplicate": 0,
                "skipped_details": skipped_details,
                "skipped_summary": _summarize_top_reasons(skipped_details),
            }
            if status == "failed":
                out["error"] = "No hay registros válidos en el archivo CSV."
            elif status == "warning" and skipped_invalid:
                out["message"] = f"No se insertaron registros. {skipped_invalid} fueron rechazados."
            return out
        except Exception as e:
            logger.exception("Series import commit failed: %s", e)
            return {"status": "failed", "error": str(e)}

    try:
        from api.v1.modules.a76.invoices.models import (
            InvoiceHeader,
            InvoiceComplianceMx,
            InvoiceFinancials,
            InvoiceLogistics,
            InvoiceSalesDetails,
            InvoiceStatus,
            OperationType,
            TransportType,
            WeightUnit,
        )
        from api.v1.modules.a76.clients_and_providers.models import ClientProvider
        from api.v1.modules.a76.customs_brokers.models import CustomsBroker
        from api.v1.modules.a76.pedmientos.models.pedimentos import Pedimentos
        from api.v1.modules.public.reference_data.currency_types.models import CurrencyType
        from api.v1.modules.public.reference_data.pedimento_codes.models import PedimentoCode
        from api.v1.modules.public.reference_data.code_pedimento_regimens.models import CodePedimentoRegimen
        from api.v1.modules.public.reference_data.pedimento_regimens.models import RegimenPedimento
        from api.v1.modules.public.reference_data.invoice_types.models import InvoiceType
        from api.v1.modules.public.reference_data.customs_sections.models import CustomsSection
        from api.v1.modules.public.reference_data.incoterms.models import Incoterm
        from .validators.encabezados_impo_temp import (
            parse_pedimento_col_a,
            _pedimento_key_from_parsed,
            row_to_transport_type_clarion,
            _patente_from_agente_aduanal,
        )
        from .validators.transport_catalog import (
            normalize_transport_cell,
            resolve_logistics_transport_fields_for_commit,
            resolve_transport_csv_fields,
            validate_csv_invoice_logistics_transport,
        )
        from .validators.encabezados_impo_def import parse_pedimento_col_a_impo_def
        from .validators.pedimento_resolution import resolve_pedimento_candidates

        from api.v1.modules.a76.items.models import LineItem
        from api.v1.modules.a76.items.line_financials.models import LineFinancial
        from api.v1.modules.a76.items.line_quantities.models import LineQuantity
        from api.v1.modules.a76.items.line_customs.models import LineCustom
        from api.v1.modules.a76.items.line_descriptions.models import LineDescription
        from api.v1.modules.a76.items.schemas import LineItemCreate
        from api.v1.modules.a76.items.line_financials.schemas import LineFinancialCreate
        from api.v1.modules.a76.items.line_quantities.schemas import LineQuantityCreate
        from api.v1.modules.a76.items.line_customs.schemas import LineCustomCreate
        from api.v1.modules.a76.items.line_descriptions.schemas import LineDescriptionCreate
        from api.v1.modules.a24.fa.fa_item_lines.dto import FaLineItemCreateDTO
        from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem
        from api.v1.modules.a76.layouts_csv.facturas.line_item_enrichment import (
            apply_import_defaults_and_calculations_for_csv,
            apply_export_defaults_and_calculations_for_csv,
        )
        from api.v1.modules.a76.items.service import ItemService
        from api.v1.modules.a76.parts.models import Part
        from api.v1.modules.a76.classes.models import Class
        from api.v1.modules.a76.general_catalogs.units_of_measure.models import UnitOfMeasure
        from api.v1.modules.a76.general_catalogs.packages.models import Package

        footer_config = parse_footer_config(meta.get("footer_config"))

        date_format = footer_config.get("dateFormat")
        # Validate and set default date_format if not provided
        if not date_format:
            date_format = "yyyy-mm-dd"  # Default to ISO format
            logger.info(f"No date_format specified in config, using default: {date_format}")
        else:
            logger.info(f"Using date_format from config: {date_format}")
        
        _template_id_insert, inv_type_value, op_type_raw = _resolve_import_context(
            model_target=model_target,
            meta=meta,
            footer_config=footer_config,
            job_type_override=job_type_override,
        )
        op_type_value = OperationType(op_type_raw)
        _es_cambio_regimen = None
        if model_target == "invoice_header" and _template_id_insert == "exp_def_header":
            cambio_regimen_raw = (
                str(meta.get("cambio_regimen") or footer_config.get("cambio_regimen") or "NO").strip().upper()
            )
            _es_cambio_regimen = "S" if cambio_regimen_raw == "SI" else "N"

        logger.info(f"Processing CSV with operation_type={op_type_value}, invoice_type={inv_type_value}, date_format={date_format}")

        headers_to_insert = []
        details_to_insert = []
        skipped_invalid = 0
        skipped_missing_invoice = 0
        skipped_missing_fk = 0
        skipped_fk_details = []
        inserted_count = 0
        response = None

        with CoreSessionLocal() as session:
            invoice_id_cache = {}
            cleared_invoices = set() # Track invoices where we've already cleared items in this job
            provider_cache: Dict[Any, Optional[int]] = {}
            sold_to_cache: Dict[Any, Optional[int]] = {}
            shipped_to_cache: Dict[Any, Optional[int]] = {}
            broker_cache: Dict[Any, Optional[int]] = {}
            regimen_cache: Dict[str, Optional[str]] = {}
            currency_type_cache: Dict[str, Optional[str]] = {}
            customs_section_cache: Dict[str, Optional[str]] = {}
            part_cache: Dict[str, Optional[int]] = {}
            pedimento_id_cache: Dict[str, Optional[int]] = {}
            pedimento_rows_insert: List[Dict[str, Any]] = []
            broker_lookup: Optional[Dict[str, Any]] = None
            shipped_by_cache: Dict[Any, Optional[int]] = {}
            _fc_insert = parse_footer_config(meta.get("footer_config"))
            autonumerar_remesas_insert = _fc_insert.get("autonumerar_remesas", False)
            actualizar_insert = bool(meta.get("actualizar", False))
            if isinstance(_fc_insert, dict):
                if "actualizar" in _fc_insert:
                    actualizar_insert = bool(_fc_insert.get("actualizar"))
                elif _fc_insert.get("mode") == "update":
                    actualizar_insert = True
                elif _fc_insert.get("mode") == "replace":
                    actualizar_insert = False
            class_id_by_code: Dict[str, int] = {}
            class_uom_by_code: Dict[str, Optional[str]] = {}
            class_fraction_by_code: Dict[str, Optional[str]] = {}
            class_desc_es_by_code: Dict[str, Optional[str]] = {}
            class_desc_en_by_code: Dict[str, Optional[str]] = {}
            uom_id_by_code: Dict[str, int] = {}
            package_id_by_key: Dict[str, int] = {}
            company_rfc_insert = ""
            normalize_preferencia_arancelaria_commit = lambda value: (str(value or "").strip().upper())

            def clear_existing_invoice_details(invoice_id: int) -> None:
                """
                Reemplazo seguro por factura en commit de invoice_details.
                Evita bulk delete directo sobre item_lines para no romper FK.
                """
                existing_line_ids = [
                    line_id
                    for (line_id,) in session.query(LineItem.id)
                    .filter(LineItem.invoice_id == invoice_id)
                    .all()
                ]
                if not existing_line_ids:
                    session.query(InvoiceSalesDetails).filter(
                        InvoiceSalesDetails.invoice_id == invoice_id
                    ).delete(synchronize_session=False)
                    return

                session.query(FaLineItem).filter(
                    FaLineItem.id.in_(existing_line_ids)
                ).delete(synchronize_session=False)

                existing_lines = (
                    session.query(LineItem)
                    .filter(LineItem.id.in_(existing_line_ids))
                    .all()
                )
                for existing_line in existing_lines:
                    session.delete(existing_line)

                session.query(InvoiceSalesDetails).filter(
                    InvoiceSalesDetails.invoice_id == invoice_id
                ).delete(synchronize_session=False)

            if model_target == 'invoice_details':
                try:
                    from .validators.partidas_impo_temp import (
                        normalize_preferencia_arancelaria as normalize_preferencia_arancelaria_commit,
                    )
                except Exception:
                    # Fallback conservador: comportamiento previo (trim + upper).
                    normalize_preferencia_arancelaria_commit = lambda value: (str(value or "").strip().upper())
                for c in session.query(
                    Class.id,
                    Class.class_code,
                    Class.unit_of_measure,
                    Class.fraction,
                    Class.description_es,
                    Class.description_en,
                ).filter(Class.tenant_id == tenant_id, Class.company_id == company_id).all():
                    if c[1]:
                        class_code_key = (c[1] or "").strip().upper()
                        class_id_by_code[class_code_key] = c[0]
                        class_uom_by_code[class_code_key] = (c[2] or "").strip().upper() or None
                        class_fraction_by_code[class_code_key] = (c[3] or "").strip() or None
                        class_desc_es_by_code[class_code_key] = (c[4] or "").strip() or None
                        class_desc_en_by_code[class_code_key] = (c[5] or "").strip() or None
                for u in session.query(UnitOfMeasure.id, UnitOfMeasure.code).filter(UnitOfMeasure.tenant_id == tenant_id, UnitOfMeasure.company_id == company_id).all():
                    if u[1]:
                        uom_id_by_code[(u[1] or "").strip().upper()] = u[0]
                for p in session.query(Package.id, Package.key).filter(Package.tenant_id == tenant_id, Package.company_id == company_id).all():
                    if p[1]:
                        package_id_by_key[(p[1] or "").strip()] = p[0]
                try:
                    from api.v1.modules.a76.general_catalogs.company.models import Company
                    company = session.query(Company).filter(Company.id == company_id).first()
                    company_rfc_insert = (company.rfc or "").strip().upper() if company else ""
                except Exception:
                    company_rfc_insert = ""

            validator = ForeignKeyValidator(session, tenant_id, company_id)

            if model_target == "invoice_header":
                broker_rows = session.query(CustomsBroker.id, CustomsBroker.broker_key).filter(
                    CustomsBroker.tenant_id == tenant_id,
                    CustomsBroker.company_id == company_id,
                ).all()
                broker_lookup = _build_customs_broker_lookup(broker_rows)
                for p in (
                    session.query(
                        Pedimentos.id,
                        Pedimentos.customs_office,
                        Pedimentos.license,
                        Pedimentos.pedimento_number,
                        Pedimentos.operation_type,
                        Pedimentos.regime,
                    )
                    .filter(
                        Pedimentos.tenant_id == tenant_id,
                        Pedimentos.company_id == company_id,
                    )
                    .all()
                ):
                    co = (p.customs_office or "").strip()
                    lic = (p.license or "").strip()
                    num = (p.pedimento_number or "").strip()
                    if not lic or not num:
                        continue
                    info = {
                        "id": p.id,
                        "customs_office": co,
                        "license": lic,
                        "pedimento_number": num,
                        "operation_type": (p.operation_type or "").strip(),
                        "regime": (p.regime or "").strip(),
                    }
                    pedimento_rows_insert.append(info)

            error_msg_by_line: Dict[int, str] = {}
            if error_path and os.path.exists(error_path):
                try:
                    with open(error_path, "r", encoding="utf-8") as f_err:
                        for line in f_err:
                            line = line.strip()
                            if not line:
                                continue
                            try:
                                rec = json.loads(line)
                                if "line" in rec and "msg" in rec:
                                    error_msg_by_line[int(rec["line"])] = str(rec["msg"]).strip()
                            except (json.JSONDecodeError, ValueError, TypeError):
                                pass
                except Exception:
                    pass

            with open(file_path, 'r', encoding=_facturas_csv_encoding(file_path)) as f:
                # Detect Delimiter
                sample = f.read(2048)
                f.seek(0)
                try:
                    dialect = common_csv_reader.detect_csv_dialect(sample, delimiters=",;\t")
                except:
                    dialect = 'excel'
                
                reader = csv.DictReader(f, dialect=dialect)

                template_id = _template_id_insert

                for i, row in enumerate(reader, start=1):
                    row_norm = row_from_template(row, template_id, normalize_header)
                    customs_broker_id_resolved: Optional[int] = None
                    if i in error_lines:
                        skipped_invalid += 1
                        if model_target == 'invoice_header':
                            inv_for_detail = _extract_invoice_number_from_row(row_norm)
                        elif _template_id_insert == "exp_def_partidas":
                            inv_for_detail = (row_norm.get('NUMERO FACTURA EXPO') or row_norm.get('NUMERO FACTURA EXPO.') or row_norm.get('FACTURA EXPO') or '').strip()
                        else:
                            inv_for_detail = _extract_invoice_number_from_row(row_norm)
                        reason = error_msg_by_line.get(i, "Línea marcada con error en el escaneo previo (revisar reporte de validación).")
                        skipped_fk_details.append({
                            "line": i,
                            "invoice": inv_for_detail or "(vacío)",
                            "reason": reason,
                        })
                        continue

                    # Mapping Logic (solo campos que acepta el modelo de facturas)
                    if model_target == 'invoice_header':
                        invoice_number = _extract_invoice_number_from_row(row_norm)
                        invoice_date = parse_date(row_norm.get('FECHA FACTURA') or row_norm.get('FECHA'), date_format)
                        existing_header = None

                        if not invoice_number:
                            skipped_invalid += 1
                            reason = "Número de factura faltante/inválido"
                            skipped_fk_details.append({"line": i, "invoice": invoice_number or "(vacío)", "reason": reason})
                            logger.debug(f"Row {i}: Skipped - missing invoice_number. Invoice: {invoice_number}")
                            continue

                        if not invoice_date and actualizar_insert:
                            # En modo update, si la fecha no viene parseable en CSV,
                            # reutilizamos la fecha existente de la factura.
                            existing_header = (
                                session.query(InvoiceHeader)
                                .filter(
                                    InvoiceHeader.tenant_id == tenant_id,
                                    InvoiceHeader.company_id == company_id,
                                    InvoiceHeader.invoice_number == invoice_number,
                                    InvoiceHeader.invoice_type == inv_type_value,
                                    InvoiceHeader.operation_type == op_type_value,
                                )
                                .first()
                            )
                            if existing_header and existing_header.invoice_date:
                                invoice_date = existing_header.invoice_date

                        if not invoice_date:
                            skipped_invalid += 1
                            reason = "Fecha de factura faltante/inválida"
                            skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
                            logger.debug(
                                f"Row {i}: Skipped - invalid invoice_date. Invoice: {invoice_number}, "
                                f"Date: {row_norm.get('FECHA FACTURA') or row_norm.get('FECHA')}"
                            )
                            continue

                        # --- NEW: Foreign Key Validations ---
                        # 1. Invoice Type (Public)
                        if not validator.check_exists(InvoiceType, inv_type_value, field_name="key", is_public=True):
                            skipped_missing_fk += 1
                            reason = f"Tipo de factura '{inv_type_value}' no existe"
                            skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
                            logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
                            continue

                        err = _validate_client_provider_ref(
                            validator,
                            ClientProvider,
                            row_norm.get('CLAVE PROVEEDOR'),
                            i,
                            "CLAVE PROVEEDOR",
                            required=True,
                        )
                        if err:
                            skipped_invalid += 1
                            reason = f"{err['col']}: {err['msg']}"
                            skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
                            logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
                            continue

                        err = _validate_client_provider_ref(
                            validator,
                            ClientProvider,
                            row_norm.get('CLAVE VENDIDO A'),
                            i,
                            "CLAVE VENDIDO A",
                            required=True,
                        )
                        if err:
                            skipped_invalid += 1
                            reason = f"{err['col']}: {err['msg']}"
                            skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
                            logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
                            continue

                        err = _validate_client_provider_ref(
                            validator,
                            ClientProvider,
                            row_norm.get('CLAVE ENVIADO A'),
                            i,
                            "CLAVE ENVIADO A",
                            required=True,
                        )
                        if err:
                            skipped_invalid += 1
                            reason = f"{err['col']}: {err['msg']}"
                            skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
                            logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
                            continue

                        if inv_type_value != "MEX":
                            if broker_lookup is not None:
                                raw_agent = row_norm.get('AGENTE ADUANAL')
                                resolved_id, mode, normalized_agent, resolve_err = resolve_customs_broker_ref(raw_agent, broker_lookup)
                                if resolve_err:
                                    skipped_missing_fk += 1
                                    reason = (
                                        f"AGENTE ADUANAL: {resolve_err} "
                                        f"(valor='{str(raw_agent or '').strip()}', normalizado='{normalized_agent}')"
                                    )
                                    skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
                                    logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
                                    continue
                                customs_broker_id_resolved = resolved_id
                                # Sustitución interna: una vez resuelto, usar el ID real para todo el pipeline.
                                if customs_broker_id_resolved is not None:
                                    row_norm['AGENTE ADUANAL'] = customs_broker_id_resolved

                            err = _validate_customs_broker_ref(
                                validator,
                                CustomsBroker,
                                row_norm.get('AGENTE ADUANAL'),
                                i,
                                "AGENTE ADUANAL",
                                required=False,
                                broker_lookup=broker_lookup,
                            )
                            if err:
                                skipped_missing_fk += 1
                                reason = f"{err['col']}: {err['msg']}"
                                skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
                                logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
                                continue

                            err = validate_public_code(
                                validator,
                                RegimenPedimento,
                                row_norm.get('REGIMEN') or row_norm.get('CLAVEDOCUMENTO'),
                                i,
                                "CLAVEDOCUMENTO",
                            )
                            if err:
                                skipped_missing_fk += 1
                                reason = f"{err['col']}: {err['msg']}"
                                skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
                                logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
                                continue

                            err = validate_public_code(
                                validator,
                                CustomsSection,
                                row_norm.get('ADUANA DE CRUCE'),
                                i,
                                "ADUANA DE CRUCE",
                                field_name="customs_code",
                            )
                            if err:
                                skipped_missing_fk += 1
                                reason = f"{err['col']}: {err['msg']}"
                                skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
                                logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
                                continue

                        if inv_type_value == "MEX":
                            err = validate_public_code(
                                validator,
                                CurrencyType,
                                row_norm.get('CLAVE MONEDA'),
                                i,
                                "CLAVE MONEDA",
                            )
                            if err and (row_norm.get('TIPO MONEDA') or '').strip().upper() == 'MC':
                                skipped_missing_fk += 1
                                reason = f"{err['col']}: {err['msg']}"
                                skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
                                logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
                                continue
                        else:
                            err = validate_public_code(
                                validator,
                                CurrencyType,
                                row_norm.get('CLAVE MONEDA'),
                                i,
                                "CLAVE MONEDA",
                            )
                            if err:
                                skipped_missing_fk += 1
                                reason = f"{err['col']}: {err['msg']}"
                                skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
                                logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
                                continue

                        err = validate_public_code(
                            validator,
                            Incoterm,
                            row_norm.get('CLAVE INCOTERM'),
                            i,
                            "CLAVE INCOTERM",
                        )
                        if err:
                            skipped_missing_fk += 1
                            reason = f"{err['col']}: {err['msg']}"
                            skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
                            logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
                            continue

                        transport_type_val = row_norm.get('TIPO TRANSPORTE')
                        transport_str_normalized = (row_to_transport_type_clarion(transport_type_val) or str(transport_type_val or "").strip().lower() or "none")
                        if transport_type_val and transport_str_normalized not in TRANSPORT_TYPE_VALUES:
                            skipped_invalid += 1
                            reason = "TIPO TRANSPORTE: Tipo de transporte invalido"
                            skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
                            logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
                            continue

                        currency_val = row_norm.get('TIPO MONEDA')
                        if currency_val and str(currency_val).strip():
                            if not csv_tipo_moneda_es_me_mn_mc(str(currency_val)):
                                parsed_currency = parse_currency(currency_val, None)
                                val_norm = normalize_header(currency_val)
                                if parsed_currency.value == "manual" and "MANUAL" not in val_norm:
                                    skipped_invalid += 1
                                    reason = "TIPO MONEDA: Moneda invalida"
                                    skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
                                    logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
                                    continue
                        
                        # 2. Client/Provider and broker checks are handled above

                        # --- 4. Check for Existing Invoice (Upsert Logic) ---
                        if invoice_number and existing_header is None:
                            existing_header = (
                                session.query(InvoiceHeader)
                                .filter(
                                    InvoiceHeader.tenant_id == tenant_id,
                                    InvoiceHeader.company_id == company_id,
                                    InvoiceHeader.invoice_number == invoice_number,
                                    InvoiceHeader.invoice_type == inv_type_value,
                                    InvoiceHeader.operation_type == op_type_value,
                                )
                                .first()
                            )

                        # --- Resolve PEDIMENTO (Col A) to pedimento_id and REMESA (Col B) ---
                        pedimento_id = None
                        remesa_val = parse_int(row_norm.get('REMESA')) if inv_type_value != "MEX" else None
                        ped_str = (row_norm.get('PEDIMENTO') or '').strip() if inv_type_value != "MEX" else ''
                        if ped_str:
                            if template_id == "exp_def_header":
                                parsed = parse_pedimento_col_a_impo_def(ped_str)
                            else:
                                parsed = parse_pedimento_col_a(ped_str)
                            if parsed:
                                customs_office_p, license_p, num_p = (x.strip() if x else "" for x in parsed)
                                patente_lookup = _patente_from_agente_aduanal(row_norm, license_p)
                                cache_key_p = f"{template_id}|{_es_cambio_regimen}|{customs_office_p}|{license_p}|{num_p}|{patente_lookup}"
                                if cache_key_p not in pedimento_id_cache:
                                    ped_resolved = resolve_pedimento_candidates(
                                        customs_office=customs_office_p,
                                        license_val=license_p,
                                        pedimento_number=num_p,
                                        patente_lookup=patente_lookup,
                                        pedimento_rows=pedimento_rows_insert,
                                    )
                                    resolved_id: Optional[int] = None
                                    if ped_resolved["status"] == "ok" and ped_resolved["pedimento"]:
                                        ped_info = ped_resolved["pedimento"]
                                        op_type = str(ped_info.get("operation_type") or "").strip().lower()
                                        regimen = str(ped_info.get("regime") or "").strip().upper()
                                        if template_id == "exp_def_header":
                                            if _es_cambio_regimen == "S":
                                                if op_type == "imp" and regimen == "IMD":
                                                    resolved_id = ped_info.get("id")
                                            else:
                                                if op_type == "exp" and regimen in {"EXD", "ETE", "ETR"}:
                                                    resolved_id = ped_info.get("id")
                                        else:
                                            resolved_id = ped_info.get("id")
                                    pedimento_id_cache[cache_key_p] = resolved_id
                                pedimento_id = pedimento_id_cache[cache_key_p]
                                if pedimento_id is not None and remesa_val is None and autonumerar_remesas_insert:
                                    max_rem = (
                                        session.query(func.max(InvoiceComplianceMx.remesa))
                                        .filter(InvoiceComplianceMx.pedimento_id == pedimento_id)
                                        .scalar()
                                    )
                                    remesa_val = (max_rem or 0) + 1

                        if existing_header:
                            # UPDATE existing header (paridad con InvoiceService.update)
                            header = existing_header
                            header.invoice_date = invoice_date
                            header.operation_type = op_type_value
                            # CSV import only captures data; "processed" is set by the manual/import processing flow.
                            # Legacy CSV imports may have persisted booleans ('True'/'False') into status.
                            if header.status not in (InvoiceStatus.PENDING, InvoiceStatus.PROCESSED, InvoiceStatus.REVERSED):
                                header.status = InvoiceStatus.PENDING
                            header.updated_date = datetime.utcnow()
                            capture_user = meta.get("capture_user") or "CSV"
                            header.who_processed = capture_user
                            # Backfill capture_user if missing or generic (paridad con service)
                            if not header.capture_user or header.capture_user == "System":
                                if capture_user != "CSV":
                                    header.capture_user = capture_user
                            header.document_type = (
                                None if inv_type_value == "MEX" else
                                resolve_public_code(
                                    session,
                                    RegimenPedimento,
                                    RegimenPedimento.code,
                                    (row_norm.get('REGIMEN') or row_norm.get('CLAVEDOCUMENTO')),
                                    regimen_cache,
                                )
                            )
                            header.project_number = (row_norm.get('NUM PROYECTO') or row_norm.get('NUMPROYECTO') or None)
                            header.purchase_order = (row_norm.get('ORDEN COMPRA') or row_norm.get('ORDENCOMPRA') or None)
                            header.alternate_invoice = (row_norm.get('FACTURA ALTERNA') or None)
                            header.invoice_ref = (row_norm.get('FACTURA EXPO REF') or row_norm.get('FACTURAEXPOREF') or None)
                            header.emission_date = parse_date(row_norm.get('FECHA EMISION'), date_format)
                            header.observation_es = (row_norm.get('OBSERVACIONES E') or None)
                            header.observation_en = (row_norm.get('OBSERVACIONES I') or None)
                            logger.info(f"Row {i}: Updating existing invoice {invoice_number}")
                            
                            # Clean up related data that will be re-inserted/updated
                            # Note: compliance, financials, logistics are 1-to-1 relationships and will be updated by assignment below
                            # but we might want to be explicit if ORM doesn't handle replace well.
                            # SQLAlchemy relationship assignment usually handles 1-to-1 updates correctly.
                            
                        else:
                            # CREATE new header (paridad con InvoiceService.create: capture_user, who_processed)
                            capture_user = meta.get("capture_user") or "CSV"
                            header = InvoiceHeader(
                                invoice_number=invoice_number,
                                invoice_date=invoice_date,
                                operation_type=op_type_value,
                                status=InvoiceStatus.PENDING,
                                system="fixed_asset",
                                capture_date=datetime.utcnow(),
                                capture_user=capture_user,
                                who_processed=capture_user,
                                invoice_type=inv_type_value,
                                document_type=(
                                    None if inv_type_value == "MEX" else
                                    resolve_public_code(
                                        session,
                                        RegimenPedimento,
                                        RegimenPedimento.code,
                                        (row_norm.get('REGIMEN') or row_norm.get('CLAVEDOCUMENTO')),
                                        regimen_cache,
                                    )
                                ),
                                project_number=(row_norm.get('NUM PROYECTO') or row_norm.get('NUMPROYECTO') or None),
                                purchase_order=(row_norm.get('ORDEN COMPRA') or row_norm.get('ORDENCOMPRA') or None),
                                alternate_invoice=(row_norm.get('FACTURA ALTERNA') or None),
                                invoice_ref=(row_norm.get('FACTURA EXPO REF') or row_norm.get('FACTURAEXPOREF') or None),
                                emission_date=parse_date(row_norm.get('FECHA EMISION'), date_format),
                                observation_es=(row_norm.get('OBSERVACIONES E') or None),
                                observation_en=(row_norm.get('OBSERVACIONES I') or None),
                                tenant_id=tenant_id,
                                company_id=company_id,
                            )

                        compliance = InvoiceComplianceMx(
                            pedimento_id=pedimento_id if inv_type_value != "MEX" else None,
                            remesa=remesa_val if inv_type_value != "MEX" else None,
                            aduana=(
                                None if inv_type_value == "MEX" else
                                resolve_public_code(
                                    session,
                                    CustomsSection,
                                    CustomsSection.customs_code,
                                    row_norm.get('ADUANA DE CRUCE'),
                                    customs_section_cache,
                                )
                            ),
                            provider_id=resolve_client_provider_id(
                                session,
                                ClientProvider,
                                row_norm.get('CLAVE PROVEEDOR'),
                                tenant_id,
                                company_id,
                                provider_cache,
                            ),
                            sold_to_id=resolve_client_provider_id(
                                session,
                                ClientProvider,
                                row_norm.get('CLAVE VENDIDO A'),
                                tenant_id,
                                company_id,
                                sold_to_cache,
                            ),
                            shipped_to_id=resolve_client_provider_id(
                                session,
                                ClientProvider,
                                row_norm.get('CLAVE ENVIADO A'),
                                tenant_id,
                                company_id,
                                shipped_to_cache,
                            ),
                            customs_broker_id=(
                                None if inv_type_value == "MEX" else
                                customs_broker_id_resolved
                                if customs_broker_id_resolved is not None
                                else resolve_customs_broker_id(
                                    session,
                                    CustomsBroker,
                                    row_norm.get('AGENTE ADUANAL'),
                                    tenant_id,
                                    company_id,
                                    broker_cache,
                                    broker_lookup=broker_lookup,
                                )
                            ),
                            edocument=(row_norm.get('E DOCUMENT') or None),
                            vucem_operation_num=(row_norm.get('NUM OPERACION') or None),
                            manifest_number=(row_norm.get('MANIFIESTO') or None),
                            shipped_by_id=resolve_client_provider_id(
                                session,
                                ClientProvider,
                                row_norm.get('ENVIADO POR'),
                                tenant_id,
                                company_id,
                                shipped_by_cache,
                            ) if row_norm.get('ENVIADO POR') else None,
                            tenant_id=tenant_id,
                            company_id=company_id,
                        )

                        financials_currency_type = resolve_public_code(
                            session,
                            CurrencyType,
                            CurrencyType.code,
                            row_norm.get('CLAVE MONEDA'),
                            currency_type_cache,
                        )
                        financials = InvoiceFinancials(
                            currency=parse_currency(row_norm.get('TIPO MONEDA'), financials_currency_type),
                            currency_type=financials_currency_type,
                            exchange_rate=decimal_or_zero(row_norm.get('TIPO DE CAMBIO')),
                            freight=decimal_or_zero(row_norm.get('FLETES')),
                            insurance_value=decimal_or_zero(row_norm.get('VALOR SEGUROS')),
                            insurance=decimal_or_zero(row_norm.get('SEGUROS')),
                            packaging=decimal_or_zero(row_norm.get('EMBALAJES')),
                            other_increments=decimal_or_zero(row_norm.get('OTROS INCREMENTABLES')),
                            tenant_id=tenant_id,
                            company_id=company_id,
                        )

                        weight_type = parse_weight_unit(row_norm.get('TIPO PESO'))
                        logistics = None
                        transport_key_csv, trailer_num_csv, transport_num_csv = resolve_transport_csv_fields(row_norm)
                        has_logistics_data = bool(
                            weight_type
                            or row_norm.get('TIPO TRANSPORTE')
                            or row_norm.get('NUMERO TRANSPORTE')
                            or transport_key_csv
                            or trailer_num_csv
                            or row_norm.get('CLAVE TRANSPORTISTA')
                            or row_norm.get('NOMBRE CONDUCTOR')
                        )
                        if has_logistics_data:
                            raw_transport = row_norm.get('TIPO TRANSPORTE')
                            transport_str = (row_to_transport_type_clarion(raw_transport) or str(raw_transport or "").strip().lower() or "none")
                            try:
                                transport_type = TransportType(transport_str)
                            except ValueError:
                                transport_type = TransportType.NONE
                            err_transport = validate_csv_invoice_logistics_transport(
                                row_norm=row_norm,
                                session=session,
                                tenant_id=tenant_id,
                                company_id=company_id,
                                transport_type=transport_type,
                            )
                            if err_transport:
                                skipped_invalid += 1
                                reason = err_transport
                                skipped_fk_details.append(
                                    {"line": i, "invoice": invoice_number, "reason": reason}
                                )
                                logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
                                continue

                            transport_num_effective = (transport_num_csv or None)
                            if transport_type == TransportType.NONE:
                                transport_num_effective = None

                            carrier_code = normalize_transport_cell(row_norm.get('CLAVE TRANSPORTISTA')) or None
                            carrier_int_id = None
                            if carrier_code:
                                from api.v1.modules.a76.transportation.transporters.models import (
                                    Transporter,
                                )

                                carrier_obj = (
                                    session.query(Transporter)
                                    .filter(
                                        Transporter.transporter_key == carrier_code,
                                        Transporter.tenant_id == tenant_id,
                                        Transporter.company_id == company_id,
                                    )
                                    .first()
                                )
                                if carrier_obj:
                                    carrier_int_id = carrier_obj.transporter_id

                            (
                                transport_id_value,
                                trailer_num_value,
                                transport_int_id,
                                trailer_int_id,
                            ) = resolve_logistics_transport_fields_for_commit(
                                session=session,
                                tenant_id=tenant_id,
                                company_id=company_id,
                                clave_transporte_raw=transport_key_csv,
                                numero_caja_raw=trailer_num_csv,
                            )
                            logistics = InvoiceLogistics(
                                carrier_id=carrier_code,
                                carrier_int_id=carrier_int_id,
                                driver_name=(row_norm.get('NOMBRE CONDUCTOR') or None),
                                transport_type=transport_type,
                                transport_num=transport_num_effective,
                                transport_id=transport_id_value,
                                trailer_num=trailer_num_value,
                                transport_int_id=transport_int_id,
                                trailer_int_id=trailer_int_id,
                                weight_type=weight_type or WeightUnit.KGS,
                                seal_number=(row_norm.get('PRECINTO') or None),
                                incoterm=(row_norm.get('CLAVE INCOTERM') or None),
                                entry_exit_date=parse_date(row_norm.get('FECHA EMISION'), date_format),
                                tenant_id=tenant_id,
                                company_id=company_id,
                            )

                        header.compliance_mx = compliance
                        header.financials = financials
                        if logistics:
                            header.logistics = logistics

                        headers_to_insert.append(header)

                    elif model_target == 'invoice_details':
                        if _template_id_insert == "exp_def_partidas":
                            invoice_number = (
                                row_norm.get('NUMERO FACTURA EXPO') or row_norm.get('NUMERO FACTURA EXPO.') or row_norm.get('FACTURA EXPO') or ''
                            ).strip()
                        else:
                            invoice_number = _extract_invoice_number_from_row(row_norm)
                        if not invoice_number:
                            skipped_invalid += 1
                            continue

                        if _template_id_insert == "exp_def_partidas":
                            # Expo: lookup by invoice_number + operation_type only (no invoice_type filter, matching scan behavior)
                            cache_key = f"{invoice_number}|exp"
                            if cache_key in invoice_id_cache:
                                invoice_id = invoice_id_cache[cache_key]
                            else:
                                invoice_id = (
                                    session.query(InvoiceHeader.id)
                                    .filter(
                                        InvoiceHeader.tenant_id == tenant_id,
                                        InvoiceHeader.company_id == company_id,
                                        InvoiceHeader.invoice_number == invoice_number,
                                        InvoiceHeader.operation_type == "exp",
                                    )
                                    .scalar()
                                )
                                invoice_id_cache[cache_key] = invoice_id
                        else:
                            invoice_id = _resolve_invoice_id_for_details(
                                session=session,
                                InvoiceHeader=InvoiceHeader,
                                tenant_id=tenant_id,
                                company_id=company_id,
                                invoice_number_raw=invoice_number,
                                invoice_type=inv_type_value,
                                operation_type=op_type_value,
                                cache=invoice_id_cache,
                            )

                        if not invoice_id:
                            logger.warning(
                                "Invoice not found for details row %s (invoice_number=%s)",
                                i,
                                invoice_number,
                            )
                            skipped_missing_invoice += 1
                            continue

                        # --- Partidas Exportación Definitiva: paridad con flujo normal (validators + ItemService) ---
                        if _template_id_insert == "exp_def_partidas":
                            part_num = (row_norm.get('NUM. PARTE') or row_norm.get('NUMPARTE') or row_norm.get('NUMERO PARTE') or row_norm.get('NUM PARTE') or '').strip()
                            part_id = part_cache.get(part_num) if part_num else None
                            if part_id is None and part_num:
                                p = session.query(Part.id).filter(Part.part_number == part_num, Part.tenant_id == tenant_id, Part.company_id == company_id).first()
                                if p:
                                    part_id = p.id
                                    part_cache[part_num] = p.id

                            line_num_val = (row_norm.get('LINEA EXPO') or row_norm.get('LINEA EXPO.') or row_norm.get('RENGLON EXPO'))
                            line_num = parse_int(line_num_val) or (len(details_to_insert) + 1)

                            uom_code = (row_norm.get('UNIDAD DE MEDIDA') or row_norm.get('U.M.') or row_norm.get('UNIDAD MEDIDA') or '').strip().upper()
                            uom_id = uom_id_by_code.get(uom_code) if uom_code else None
                            bulk_key = (row_norm.get('CLAVE BULTOS') or row_norm.get('CLAVEBULTOS') or '').strip()
                            package_id = package_id_by_key.get(bulk_key) if bulk_key else None

                            descarga_val = (row_norm.get('GENERA DESCARGA') or row_norm.get('GENERA DESCARGA?') or row_norm.get('DESCARGA') or 'SI').strip().upper()
                            tipo_impo = (row_norm.get('TIPO DE IMPO') or row_norm.get('TIPO DE IMPO.') or row_norm.get('TIPO IMPO') or row_norm.get('PROCEDENCIA') or '').strip().upper()
                            factura_impo = (row_norm.get('FACTURA IMPO') or row_norm.get('FACTURA IMPO.') or row_norm.get('FACTURA IMPORTACION') or '').strip()
                            linea_impo_val = (row_norm.get('LINEA IMPO') or row_norm.get('LINEA IMPO.') or row_norm.get('LINEA IMPORTACION') or '').strip()

                            se_pago = (row_norm.get('SE PAGO IMPUESTO') or row_norm.get('SE PAGO IMPUESTO? (SI o NO)') or row_norm.get('SEPAGOIMPUESTO') or '').strip().upper()
                            forma_pago = (row_norm.get('FORMA DE PAGO') or row_norm.get('FORMADEPAGO') or row_norm.get('FORMA PAGO') or '').strip() or None

                            es_sub_raw = (row_norm.get('ES PARTIDA/SUBPARTIDA') or row_norm.get('ESSUBPARTIDA') or row_norm.get('ES PARTIDA O SUBPARTIDA') or '').strip().upper()
                            linea_principal_val = (row_norm.get('LINEA PRINCIPAL') or row_norm.get('LINEAPRINCIPAL') or row_norm.get('PARTIDA PRINCIPAL') or '').strip()
                            is_subitem = (es_sub_raw == 'S')
                            contains_subitems = (es_sub_raw == 'P')

                            # Clear existing line items once per invoice
                            if invoice_id not in cleared_invoices:
                                logger.info(f"Clearing existing details for Expo Invoice {invoice_number} (ID: {invoice_id})")
                                clear_existing_invoice_details(invoice_id)
                                cleared_invoices.add(invoice_id)

                            price = parse_decimal(row_norm.get('COSTO UNITARIO') or row_norm.get('COSTOUNITARIO'))
                            qty = parse_decimal(row_norm.get('CANTIDAD EXPORTADA/DESCARGAR') or row_norm.get('CANTIDAD EXPORTADA') or row_norm.get('CANTIDAD'))
                            commercial_total = (price * qty) if price and qty else None
                            net_w = parse_decimal(row_norm.get('PESO NETO') or row_norm.get('PESONETO'))
                            gross_w = parse_decimal(row_norm.get('PESO BRUTO') or row_norm.get('PESOBRUTO'))
                            origin = (row_norm.get('PAIS ORIGEN') or row_norm.get('PAISORIGEN') or row_norm.get('PAIS') or '').strip()
                            fraction = (row_norm.get('FRACCION ARANCELARIA') or row_norm.get('FRACCIONARANCELARIA') or '').strip()
                            american_fraction = (row_norm.get('FRACCION AMERICANA') or row_norm.get('FRACCIONAMERICANA') or '').strip()
                            extra_desc = (row_norm.get('DESCRIPCION EXTRA') or row_norm.get('DESCRIPCIONEXTRA') or '').strip()
                            additional_info = (row_norm.get('INFORMACION ADICIONAL') or row_norm.get('INFORMACIONADICIONAL') or '').strip()
                            lot = (row_norm.get('LOTE') or '').strip()
                            entry_number = (row_norm.get('NUMERO ENTRADA') or row_norm.get('NUM ENTRADA') or '').strip()
                            order_compra = (row_norm.get('ORDEN DE COMPRA') or row_norm.get('ORDENCOMPRA') or row_norm.get('ORDEN DE VENTA') or None)

                            line_data = LineItemCreate(
                                invoice_id=invoice_id,
                                line_number=line_num,
                                part_number_id=part_id,
                                class_id=None,
                                unit_of_measure=uom_id,
                                order=order_compra,
                                tax_payment=(se_pago == 'SI'),
                                payment_method=forma_pago,
                                financial=LineFinancialCreate(
                                    unit_cost_capture=decimal_or_zero(price),
                                    total_commercial_value=decimal_or_zero(commercial_total),
                                ),
                                quantity=LineQuantityCreate(
                                    quantity=decimal_or_zero(qty),
                                    net_weight=decimal_or_zero(net_w),
                                    gross_weight=decimal_or_zero(gross_w),
                                    package_quantity=int_or_zero(row_norm.get('CANTIDAD BULTOS') or row_norm.get('CANTIDADBULTOS')),
                                    package_id=package_id,
                                ),
                                customs=LineCustomCreate(
                                    origin_country=origin or None,
                                    fraction=fraction or None,
                                    american_fraction=american_fraction or None,
                                ),
                                description=LineDescriptionCreate(
                                    extra_description=extra_desc or None,
                                    additional_info_spanish=additional_info or None,
                                    lot=lot or None,
                                    entry_number=entry_number or None,
                                ),
                                fa_data=FaLineItemCreateDTO(
                                    search_invoice=factura_impo or None,
                                    search_line=parse_int(linea_impo_val),
                                    search_type=tipo_impo or None,
                                    download=(descarga_val == 'SI'),
                                    is_subitem=is_subitem,
                                    contains_subitems=contains_subitems,
                                    subitem_number=parse_int(linea_principal_val) if is_subitem else 0,
                                ),
                            )
                            if not apply_export_defaults_and_calculations_for_csv(
                                session, line_data, tenant_id, company_id, line_num
                            ):
                                skipped_invalid += 1
                                skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": "Factura sin datos financieros para enriquecer partida de exportación."})
                                continue

                            item_dict = line_data.model_dump(
                                exclude={
                                    "financial",
                                    "quantity",
                                    "customs",
                                    "description",
                                    "reference",
                                    "fa_data",
                                    "series",
                                    "identifiers",
                                },
                                exclude_none=True,
                            )
                            item_dict.pop("identifiers", None)
                            item_dict["tenant_id"] = tenant_id
                            item_dict["company_id"] = company_id
                            item_dict["line_number"] = line_num
                            line = LineItem(**item_dict)
                            session.add(line)
                            session.flush()
                            ItemService._create_line_nested_data(session, line, line_data, tenant_id, company_id)

                            session.add(InvoiceSalesDetails(
                                invoice_id=invoice_id,
                                line_number=line_num,
                                sales_order=order_compra,
                                line_bundles=int_or_zero(row_norm.get('CANTIDAD BULTOS') or row_norm.get('CANTIDADBULTOS')),
                                tenant_id=tenant_id,
                                company_id=company_id,
                            ))
                            details_to_insert.append(line)
                            continue

                        part_num = (row_norm.get('NUM. PARTE') or row_norm.get('NUMPARTE') or row_norm.get('NUMERO PARTE') or row_norm.get('NUM PARTE') or '').strip()
                        require_part_number = company_rfc_insert in {"CTE980130518"}
                        if require_part_number and not part_num:
                            skipped_invalid += 1
                            reason = "NUMPARTE: Requerido por regla de RFC"
                            skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
                            logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
                            continue
                        if part_num and not validator.check_exists(Part, part_num, field_name="part_number"):
                            skipped_missing_fk += 1
                            reason = f"NUMPARTE '{part_num}' no existe"
                            skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason})
                            logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}")
                            continue

                        # --- Prevent Duplicates: Clear existing line items for this invoice (Once per job) ---
                        if invoice_id not in cleared_invoices:
                            logger.info(f"Clearing existing details for Invoice {invoice_number} (ID: {invoice_id}) to prevent duplicates")
                            clear_existing_invoice_details(invoice_id)
                            cleared_invoices.add(invoice_id)

                        # --- Partidas importación: paridad con flujo normal (validators + ItemService) ---
                        # LineReference: solo se crea si line_data.reference viene informado; no inventar datos sin fuente (plan paridad CSV).
                        # Build LineItemCreate from CSV, apply import defaults/calculations, then persist.
                        part_num = (row_norm.get('NUM. PARTE') or row_norm.get('NUMPARTE') or row_norm.get('NUMERO PARTE') or row_norm.get('NUM PARTE') or '').strip()
                        part_id = part_cache.get(part_num) if part_num else None
                        if part_id is None and part_num:
                            p = session.query(Part.id).filter(Part.part_number == part_num, Part.tenant_id == tenant_id, Part.company_id == company_id).first()
                            if p:
                                part_id = p.id
                                part_cache[part_num] = part_id

                        line_num_val = (row_norm.get('LINEA') or row_norm.get('RENGLON') or row_norm.get('PARTIDA'))
                        line_num = parse_int(line_num_val) or (len(details_to_insert) + 1)

                        class_code = (row_norm.get('CLASE') or '').strip().upper()
                        class_id = class_id_by_code.get(class_code) if class_code else None
                        uom_code = (row_norm.get('UNIDAD DE MEDIDA') or row_norm.get('UNIDAD MEDIDA') or '').strip().upper()
                        if not uom_code and class_code:
                            uom_code = class_uom_by_code.get(class_code) or ''
                        uom_id = uom_id_by_code.get(uom_code) if uom_code else None
                        bulk_key = (row_norm.get('CLAVE BULTOS') or row_norm.get('CLAVEBULTOS') or '').strip()
                        package_id = package_id_by_key.get(bulk_key) if bulk_key else None

                        price = parse_decimal(row_norm.get('COSTO UNITARIO') or row_norm.get('PRECIO UNITARIO') or row_norm.get('PRECIOUNITARIO'))
                        if price is None:
                            total_val = parse_decimal(row_norm.get('TOTAL'))
                            qty = parse_decimal(row_norm.get('CANTIDAD IMPORTADA') or row_norm.get('CANTIDAD'))
                            price = (total_val / qty) if (total_val and qty and qty != 0) else None
                        qty = parse_decimal(row_norm.get('CANTIDAD IMPORTADA') or row_norm.get('CANTIDAD'))
                        commercial_total = (price * qty) if price and qty else parse_decimal(row_norm.get('TOTAL'))
                        net_w = parse_decimal(row_norm.get('PESO NETO') or row_norm.get('PESONETO'))
                        gross_w = parse_decimal(row_norm.get('PESO BRUTO') or row_norm.get('PESOBRUTO'))

                        origin = (row_norm.get('PAIS ORIGEN') or row_norm.get('PAISORIGEN') or row_norm.get('PAIS') or '').strip()
                        fraction = (row_norm.get('FRACCION ARANCELARIA') or row_norm.get('FRACCION') or row_norm.get('FRACCIONARANCELARIA') or '').strip()
                        if not fraction and class_code:
                            fraction = class_fraction_by_code.get(class_code) or ''
                        fraction_type_raw = row_norm.get('PREFERENCIA ARANCELARIA') or row_norm.get('PREFERENCIA') or ''
                        fraction_type = normalize_preferencia_arancelaria_commit(fraction_type_raw)
                        sector = (row_norm.get('SECTOR') or '').strip()
                        american_fraction = (row_norm.get('FRACCION AMERICANA') or row_norm.get('FRACCIONAMERICANA') or '').strip()

                        desc_es = (row_norm.get('DESCRIPCION ESPAÑOL') or row_norm.get('DESCRIPCIONE') or row_norm.get('DESCRIPCION') or '').strip()
                        if not desc_es and class_code:
                            desc_es = class_desc_es_by_code.get(class_code) or ''
                        desc_en = (row_norm.get('DESCRIPCION INGLES') or row_norm.get('DESCRIPCIONI') or '').strip()
                        if not desc_en and class_code:
                            desc_en = class_desc_en_by_code.get(class_code) or ''
                        brand = (row_norm.get('MARCA') or '').strip()
                        model = (row_norm.get('MODELO') or '').strip()
                        extra_desc = (row_norm.get('DESCRIPCION EXTRA') or row_norm.get('DESCRIPCIONEXTRA') or '').strip()
                        additional_info = (row_norm.get('INFORMACION ADICIONAL') or row_norm.get('INFORMACIONADICIONAL') or '').strip()
                        lot = (row_norm.get('LOTE') or '').strip()
                        entry_number = (row_norm.get('NUMERO ENTRADA') or row_norm.get('NUMEROENTRADA') or row_norm.get('NUM ENTRADA') or '').strip()

                        line_data = LineItemCreate(
                            invoice_id=invoice_id,
                            line_number=line_num,
                            part_number_id=part_id,
                            class_id=class_id,
                            unit_of_measure=uom_id,
                            order=(row_norm.get('ORDEN DE COMPRA') or row_norm.get('ORDENCOMPRA') or None),
                            material_type=(row_norm.get('ID TYPE') or row_norm.get('IDTYPE') or None),
                            tax_payment=(str(row_norm.get('SE PAGO IMPUESTO') or row_norm.get('SEPAGOIMPUESTO') or '').strip().upper() == 'SI'),
                            payment_method=(row_norm.get('FORMA DE PAGO') or row_norm.get('FORMADEPAGO') or row_norm.get('FORMA PAGO') or None),
                            valuation_method=(row_norm.get('METODO DE VALORACION') or row_norm.get('METODODEVALORACION') or row_norm.get('METODO VALORACION') or None),
                            financial=LineFinancialCreate(
                                unit_cost_capture=decimal_or_zero(price),
                                total_commercial_value=decimal_or_zero(commercial_total),
                            ),
                            quantity=LineQuantityCreate(
                                quantity=decimal_or_zero(qty),
                                net_weight=decimal_or_zero(net_w),
                                gross_weight=decimal_or_zero(gross_w),
                                package_quantity=int_or_zero(row_norm.get('CANTIDAD BULTOS') or row_norm.get('CANTIDADBULTOS')),
                                package_id=package_id,
                            ),
                            customs=LineCustomCreate(
                                origin_country=origin or None,
                                fraction=fraction or None,
                                fraction_type=fraction_type or None,
                                sector=sector or None,
                                american_fraction=american_fraction or None,
                            ),
                            description=LineDescriptionCreate(
                                description_spanish=desc_es or None,
                                description_english=desc_en or None,
                                brand=brand or None,
                                model=model or None,
                                extra_description=extra_desc or None,
                                additional_info_spanish=additional_info or None,
                                lot=lot or None,
                                entry_number=entry_number or None,
                            ),
                            fa_data=FaLineItemCreateDTO(is_subitem=False, contains_subitems=False),
                        )
                        if not apply_import_defaults_and_calculations_for_csv(
                            session, line_data, tenant_id, company_id, line_num
                        ):
                            skipped_invalid += 1
                            skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": "Factura sin datos financieros/logísticos para enriquecer partida."})
                            continue

                        item_dict = line_data.model_dump(
                            exclude={
                                "financial",
                                "quantity",
                                "customs",
                                "description",
                                "reference",
                                "fa_data",
                                "series",
                                "identifiers",
                            },
                            exclude_none=True,
                        )
                        item_dict.pop("identifiers", None)
                        item_dict["tenant_id"] = tenant_id
                        item_dict["company_id"] = company_id
                        item_dict["line_number"] = line_num
                        line = LineItem(**item_dict)
                        session.add(line)
                        session.flush()
                        ItemService._create_line_nested_data(session, line, line_data, tenant_id, company_id)
                        session.add(InvoiceSalesDetails(
                            invoice_id=invoice_id,
                            line_number=line_num,
                            sales_order=(row_norm.get('ORDEN DE COMPRA') or row_norm.get('ORDENCOMPRA') or None),
                            line_bundles=int_or_zero(row_norm.get('CANTIDAD BULTOS') or row_norm.get('CANTIDADBULTOS')),
                            tenant_id=tenant_id,
                            company_id=company_id,
                        ))
                        details_to_insert.append(line)

            # 3. Bulk Insert (ORM Transaction)
            try:
                if model_target == 'invoice_header':
                    if headers_to_insert:
                        logger.info(f"Attempting to commit {len(headers_to_insert)} headers")
                        session.add_all(headers_to_insert)
                        session.commit()
                        inserted_count = len(headers_to_insert)
                        logger.info(f"Headers commit successful. Inserted: {inserted_count}")
                    else:
                        logger.warning(f"No headers to insert for job {job_id}")
                else:
                    if details_to_insert:
                        logger.info(f"Attempting to commit {len(details_to_insert)} items and related data")
                        session.commit() # Everything was already added with session.add()
                        inserted_count = len(details_to_insert)
                        logger.info(f"Details commit successful. Inserted: {inserted_count}")
                    else:
                        logger.warning(f"No details to insert for job {job_id}")

            except Exception as db_err:
                session.rollback()
                logger.error(f"DB Error during {model_target} commit: {db_err}")
                import traceback
                logger.error(traceback.format_exc())
                return {"status": "failed", "error": str(db_err)}
            
            # 4. Determine final status and prepare response (inside session block to access variables)
            total_skipped = skipped_invalid + skipped_missing_fk + skipped_missing_invoice
            
            # Log summary
            logger.info(f"Job {job_id} completed. Inserted: {inserted_count}, Skipped: {total_skipped} "
                        f"(invalid: {skipped_invalid}, missing_fk: {skipped_missing_fk}, missing_invoice: {skipped_missing_invoice})")
            
            # Prepare response based on results
            if inserted_count == 0:
                if total_skipped > 0:
                    logger.warning(f"No valid records to insert for job {job_id}. All {total_skipped} records were rejected.")
                    response = {
                        "status": "warning",
                        "inserted": 0,
                        "skipped_invalid": skipped_invalid,
                        "skipped_missing_invoice": skipped_missing_invoice,
                        "skipped_missing_fk": skipped_missing_fk,
                        "skipped_details": skipped_fk_details,
                        "skipped_summary": _summarize_top_reasons(skipped_fk_details),
                        "message": f"No se insertaron registros. {total_skipped} fueron rechazados. Revisa el detalle por línea a continuación.",
                    }
                else:
                    logger.error(f"No valid records found in CSV for job {job_id}")
                    response = {
                        "status": "failed",
                        "error": "No hay registros válidos en el archivo CSV",
                        "inserted": 0,
                        "skipped_invalid": skipped_invalid,
                        "skipped_missing_invoice": skipped_missing_invoice,
                        "skipped_missing_fk": skipped_missing_fk,
                        "skipped_details": skipped_fk_details,
                        "skipped_summary": _summarize_top_reasons(skipped_fk_details),
                    }
            else:
                # Success case - at least some records were inserted
                response = {
                    "status": "finished",
                    "inserted": inserted_count,
                    "skipped_invalid": skipped_invalid,
                    "skipped_missing_invoice": skipped_missing_invoice,
                    "skipped_missing_fk": skipped_missing_fk,
                    "skipped_details": skipped_fk_details,
                    "skipped_summary": _summarize_top_reasons(skipped_fk_details),
                }
    
    except Exception as e:
        logger.error(f"Task failed: {e}")
        import traceback
        logger.error(traceback.format_exc())
        return {"status": "failed", "error": str(e)}

    # 5. Cleanup: remove temp files and Redis keys so data is not kept indefinitely
    try:
        common_storage.cleanup_import_job(
            effective_job_type, job_id,
            file_path=file_path,
            error_path=error_path,
            meta_path=meta_path,
        )
    except Exception as cleanup_err:
        logger.warning("Failed to cleanup temp files or Redis: %s", cleanup_err)

    # Ensure response is defined (fallback in case of unexpected errors)
    if response is None:
        logger.error(f"Unexpected error: response not set for job {job_id}")
        response = {
            "status": "failed",
            "error": "Error inesperado durante el procesamiento",
            "inserted": 0,
            "skipped_invalid": skipped_invalid,
            "skipped_missing_invoice": skipped_missing_invoice,
            "skipped_missing_fk": skipped_missing_fk,
            "skipped_details": skipped_fk_details,
            "skipped_summary": _summarize_top_reasons(skipped_fk_details),
        }
    
    return response
