ADUANASOFT/plantillas-proyectos
13
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feature/fix-invoices-csv

Browse Source
This commit is contained in:
hreyes
2026-03-24 08:06:48 -06:00
parent 9cfc8b3215
commit 79b2506af4
6 changed files with 591 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
backend/api/v1/modules/a76/layouts_csv/common/responses.py
View File

@@ -5,6 +5,18 @@ from typing import Dict, Any, List, Optional
ERRORS_PREVIEW_LIMIT = 100
SUMMARY_LIMIT = 10
def _summarize_reasons(items: List[Dict[str, Any]], key_name: str) -> List[Dict[str, Any]]:
counts: Dict[str, int] = {}
for item in items:
reason = str(item.get(key_name, "")).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[:SUMMARY_LIMIT]]
def scan_result(
@@ -28,6 +40,7 @@ def scan_result(
# Para no saturar el front: devolvemos solo un preview.
# El detalle completo se descarga desde CSV usando el job_id.
"errors": errors_detail[:ERRORS_PREVIEW_LIMIT],
"error_summary": _summarize_reasons([e for e in errors_detail if not e.get("warning")], "msg"),
}
if message:
out["message"] = message
@@ -52,6 +65,7 @@ def commit_result(
"skipped_missing_fk": skipped_missing_fk,
"skipped_duplicate": skipped_duplicate,
"skipped_details": skipped_details,
"skipped_summary": _summarize_reasons(skipped_details, "reason"),
}
if message:
out["message"] = message

7
backend/api/v1/modules/a76/layouts_csv/facturas/line_item_enrichment/export_enrichment.py
View File

@@ -75,7 +75,12 @@ def apply_export_defaults_and_calculations_for_csv(
)
.first()
)
setattr(line_data, "part_info", part)
if part and line_data.description:
# Fallback explícito: priorizar descripciones de parte para paridad manual.
if not line_data.description.description_spanish and part.description_spanish:
line_data.description.description_spanish = part.description_spanish
if not line_data.description.description_english and part.description_english:
line_data.description.description_english = part.description_english
if not line_data.unit_of_measure and line_data.class_id:
class_info = (

6
backend/api/v1/modules/a76/layouts_csv/facturas/line_item_enrichment/import_enrichment.py
View File

@@ -82,7 +82,6 @@ def apply_import_defaults_and_calculations_for_csv(
)
.first()
)
setattr(line_data, "part_info", part)
if not line_data.fa_data:
line_data.fa_data = FaLineItemCreateDTO(
@@ -231,6 +230,11 @@ def apply_import_defaults_and_calculations_for_csv(
else:
line_data.customs.advalorem_american = us_fraction.ad_valorem
# Fallback explícito: primero descripción de parte (si existe), luego clase.
if not line_data.description.description_spanish and part and part.description_spanish:
line_data.description.description_spanish = part.description_spanish
if not line_data.description.description_english and part and part.description_english:
line_data.description.description_english = part.description_english
if not line_data.description.description_spanish and class_info:
line_data.description.description_spanish = class_info.description_es
if not line_data.description.description_english and class_info:

619
backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py
View File

@@ -304,6 +304,135 @@ def resolve_customs_broker_ref(
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."""
@@ -350,29 +479,17 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
except ValueError as e:
return {"status": "failed", "error": str(e)}
meta = common_meta.load_meta(file_path)
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"
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,
)
# Cuando el scan viene de Exportación (job_type_override "exp"), forzar exp_def_header o exp_def_partidas
if job_type_override == "exp" and model_target == "invoice_header":
template_id = "exp_def_header"
if job_type_override == "exp" and model_target == "invoice_details":
template_id = "exp_def_partidas"
if job_type_override == "exp" and model_target == "invoice_series":
template_id = "exp_def_series"
inv_type_value = normalize_public_code(footer_config.get("invoice_type") or meta.get("invoice_type") or "TEM")
if not inv_type_value:
inv_type_value = "TEM"
if model_target == "invoice_series" and inv_type_value in ("DEF", "MATDE", "EXDEF"):
template_id = "imp_def_series"
if meta.get("operation_type") == "exp" and model_target == "invoice_details":
template_id = "exp_def_partidas"
logger.info(
"Scan job %s template_id=%s model_target=%s job_type_override=%s",
job_id, template_id, model_target, job_type_override,
"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) ---
@@ -429,7 +546,7 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
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()] = bool(is_upd)
invoice_processed_by_number[str(num).strip()] = _is_invoice_processed_status(is_upd)
partida_max_series: Dict[Tuple[str, str], int] = {}
q_qty = (
@@ -606,7 +723,7 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
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()] = bool(is_upd)
invoice_processed_by_number[str(num).strip()] = _is_invoice_processed_status(is_upd)
partida_max_series: Dict[Tuple[str, str], int] = {}
q_qty = (
@@ -805,7 +922,7 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
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()] = bool(is_upd)
invoice_processed_by_number[str(num).strip()] = _is_invoice_processed_status(is_upd)
partida_max_series: Dict[Tuple[str, str], int] = {}
q_qty = (
@@ -985,7 +1102,7 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
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()] = bool(is_upd)
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()
@@ -1135,16 +1252,17 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
.filter(
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id,
InvoiceHeader.operation_type == "imp",
InvoiceHeader.invoice_type == "TEM",
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:
invoice_id_by_number[str(num).strip()] = iid
invoice_processed_by_number[str(num).strip()] = bool(is_upd)
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 = (
@@ -1153,13 +1271,13 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
.filter(
InvoiceHeader.tenant_id == tenant_id,
InvoiceHeader.company_id == company_id,
InvoiceHeader.operation_type == "imp",
InvoiceHeader.invoice_type == "TEM",
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 = str(num).strip()
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())
@@ -1180,7 +1298,7 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
)
for num, ln in q_pp.all():
if num is not None:
partidas_principales_bd.add((str(num).strip(), str(ln).strip()))
partidas_principales_bd.add((_normalize_invoice_number(num), str(ln).strip()))
except Exception:
pass
@@ -1259,18 +1377,86 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
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:
inv = (row.get("NUMERO FACTURA") or row.get("NUM FACTURA") or row.get("FACTURA") or "").strip()
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()
@@ -1283,7 +1469,7 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
for row in rows_list:
row_norm = row_from_template(row, "imp_temp_details", normalize_header)
inv = _get_row(row_norm, "NUMERO FACTURA", "NUM FACTURA", "FACTURA")
inv = _extract_invoice_number_from_row(row_norm)
linea = _get_row(row_norm, "LINEA", "RENGLON", "PARTIDA")
if inv and linea:
key = (inv, linea)
@@ -1363,8 +1549,35 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
)
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)
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)}
@@ -1426,7 +1639,7 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
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()] = bool(is_upd)
invoice_processed_by_number[str(num).strip()] = _is_invoice_processed_status(is_upd)
existing_line_keys_by_invoice: Dict[str, Set[str]] = {}
q_li = (
@@ -1708,7 +1921,7 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
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()] = bool(is_upd)
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 = (
@@ -1993,7 +2206,7 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
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()] = bool(is_upd)
invoice_processed_by_number[str(num).strip()] = _is_invoice_processed_status(is_upd)
existing_line_keys_by_invoice: Dict[str, Set[str]] = {}
q_li = (
@@ -2285,7 +2498,7 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
if num:
n = str(num).strip()
invoice_exists_by_number[n] = True
invoice_processed_by_number[n] = bool(is_upd)
invoice_processed_by_number[n] = _is_invoice_processed_status(is_upd)
pedimento_rows: List[Dict[str, Any]] = []
for p in (
@@ -2488,6 +2701,111 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
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,
}
)
if precheck_errors:
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"
)
unique_precheck_lines = sorted(precheck_lines)
common_storage.store_error_lines(effective_job_type, job_id, unique_precheck_lines)
return common_responses.scan_result(
job_id,
len(rows_list),
len(unique_precheck_lines),
precheck_errors,
total_rows_in_file=total_rows,
message="Precheck de referencias falló. Corrige catálogos/pedimentos antes de confirmar importación.",
)
error_count = 0
processed_rows = 0
error_lines_list: List[int] = []
@@ -2638,7 +2956,7 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
if num:
n = str(num).strip()
invoice_exists_by_number[n] = True
invoice_processed_by_number[n] = bool(is_upd)
invoice_processed_by_number[n] = _is_invoice_processed_status(is_upd)
pedimento_rows = []
for p in (
@@ -2994,7 +3312,7 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
if num:
n = str(num).strip()
invoice_exists_by_number[n] = True
invoice_processed_by_number[n] = bool(is_upd)
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:
@@ -3334,7 +3652,7 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
if num:
n = str(num).strip()
invoice_exists_by_number[n] = True
invoice_processed_by_number[n] = bool(is_upd)
invoice_processed_by_number[n] = _is_invoice_processed_status(is_upd)
valid_provider_ids = set()
valid_sold_to_ids = set()
@@ -3531,6 +3849,41 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] =
open(file_path, 'r', encoding='utf-8-sig') 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
@@ -4439,6 +4792,7 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
"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."
@@ -4502,7 +4856,7 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
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()] = bool(is_upd)
invoice_processed_by_number[str(num).strip()] = _is_invoice_processed_status(is_upd)
partida_max_series = {}
q_qty = (
@@ -4731,6 +5085,7 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
"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."
@@ -4788,7 +5143,7 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
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()] = bool(is_upd)
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]] = {}
@@ -4976,6 +5331,7 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
"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."
@@ -5028,6 +5384,7 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
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,
@@ -5048,42 +5405,19 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
else:
logger.info(f"Using date_format from config: {date_format}")
# Default types from config or fallback
op_type_value = OperationType(meta.get('operation_type', 'imp').lower())
inv_type_value = normalize_public_code(footer_config.get('invoice_type') or 'TEM') or 'TEM'
_template_id_insert = meta.get("template_id") or (
"imp_temp_header" if model_target == "invoice_header" else "imp_temp_details"
_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,
)
if job_type_override == "exp" and model_target == "invoice_details":
_template_id_insert = "exp_def_partidas"
if job_type_override == "exp" and model_target == "invoice_series":
_template_id_insert = "exp_def_series"
if model_target == "invoice_header" and _template_id_insert == "imp_def_header":
inv_type_value = "DEF"
if model_target == "invoice_header" and _template_id_insert == "cmex_header":
inv_type_value = "MEX"
if model_target == "invoice_header" and _template_id_insert == "exp_def_header":
op_type_value = OperationType("exp")
inv_type_value = normalize_public_code(
meta.get("tipo_factura") or footer_config.get("tipo_factura") or "AFIJO"
) or "AFIJO"
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"
if model_target == "invoice_details" and _template_id_insert == "imp_def_details":
inv_type_value = "DEF"
if model_target == "invoice_details" and _template_id_insert == "cmex_details":
inv_type_value = "MEX"
if model_target == "invoice_details" and _template_id_insert == "exp_def_partidas":
op_type_value = OperationType("exp")
inv_type_value = normalize_public_code(
meta.get("tipo_factura") or footer_config.get("tipo_factura") or "AFIJO"
) or "AFIJO"
if _template_id_insert == "cmex_series":
inv_type_value = "MEX"
logger.info(f"Processing CSV with operation_type={op_type_value}, invoice_type={inv_type_value}, date_format={date_format}")
@@ -5128,7 +5462,50 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
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,
@@ -5150,6 +5527,12 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
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)
@@ -5225,11 +5608,11 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
if i in error_lines:
skipped_invalid += 1
if model_target == 'invoice_header':
inv_for_detail = (row_norm.get('NUMERO FACTURA') or row_norm.get('NUM FACTURA') or row_norm.get('FACTURA') or '').strip()
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 = (row_norm.get('NUMERO FACTURA') or row_norm.get('NUM FACTURA') or row_norm.get('FACTURA') or '').strip()
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,
@@ -5240,7 +5623,7 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
# Mapping Logic (solo campos que acepta el modelo de facturas)
if model_target == 'invoice_header':
invoice_number = (row_norm.get('NUMERO FACTURA') or row_norm.get('NUM FACTURA') or row_norm.get('FACTURA') or '').strip()
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
@@ -5567,7 +5950,7 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
invoice_date=invoice_date,
operation_type=op_type_value,
status=InvoiceStatus.PENDING,
system="CSV",
system="fixed_asset",
capture_date=datetime.utcnow(),
capture_user=capture_user,
who_processed=capture_user,
@@ -5714,7 +6097,7 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
row_norm.get('NUMERO FACTURA EXPO') or row_norm.get('NUMERO FACTURA EXPO.') or row_norm.get('FACTURA EXPO') or ''
).strip()
else:
invoice_number = (row_norm.get('NUMERO FACTURA') or row_norm.get('NUM FACTURA') or '').strip()
invoice_number = _extract_invoice_number_from_row(row_norm)
if not invoice_number:
skipped_invalid += 1
continue
@@ -5737,22 +6120,16 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
)
invoice_id_cache[cache_key] = invoice_id
else:
cache_key = f"{invoice_number}|{inv_type_value}|{op_type_value.value}"
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.invoice_type == inv_type_value,
InvoiceHeader.operation_type == op_type_value,
)
.scalar()
)
invoice_id_cache[cache_key] = invoice_id
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(
@@ -5797,8 +6174,7 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
# 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})")
session.query(LineItem).filter(LineItem.invoice_id == invoice_id).delete(synchronize_session=False)
session.query(InvoiceSalesDetails).filter(InvoiceSalesDetails.invoice_id == invoice_id).delete(synchronize_session=False)
clear_existing_invoice_details(invoice_id)
cleared_invoices.add(invoice_id)
price = parse_decimal(row_norm.get('COSTO UNITARIO') or row_norm.get('COSTOUNITARIO'))
@@ -5864,8 +6240,19 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
continue
item_dict = line_data.model_dump(
exclude={"financial", "quantity", "customs", "description", "reference", "fa_data", "series"}
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
@@ -5886,13 +6273,14 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
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()
if not part_num:
require_part_number = company_rfc_insert in {"CTE980130518"}
if require_part_number and not part_num:
skipped_invalid += 1
reason = "NUMPARTE: Requerido"
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 not validator.check_exists(Part, part_num, field_name="part_number"):
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})
@@ -5902,8 +6290,7 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
# --- 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")
session.query(LineItem).filter(LineItem.invoice_id == invoice_id).delete(synchronize_session=False)
session.query(InvoiceSalesDetails).filter(InvoiceSalesDetails.invoice_id == invoice_id).delete(synchronize_session=False)
clear_existing_invoice_details(invoice_id)
cleared_invoices.add(invoice_id)
# --- Partidas importación: paridad con flujo normal (validators + ItemService) ---
@@ -5943,7 +6330,8 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
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 = (row_norm.get('PREFERENCIA ARANCELARIA') or row_norm.get('PREFERENCIA') or '').strip()
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()
@@ -6009,8 +6397,19 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
continue
item_dict = line_data.model_dump(
exclude={"financial", "quantity", "customs", "description", "reference", "fa_data", "series"}
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
@@ -6073,6 +6472,7 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
"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:
@@ -6084,7 +6484,8 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
"skipped_invalid": skipped_invalid,
"skipped_missing_invoice": skipped_missing_invoice,
"skipped_missing_fk": skipped_missing_fk,
"skipped_details": skipped_fk_details
"skipped_details": skipped_fk_details,
"skipped_summary": _summarize_top_reasons(skipped_fk_details),
}
else:
# Success case - at least some records were inserted
@@ -6094,7 +6495,8 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
"skipped_invalid": skipped_invalid,
"skipped_missing_invoice": skipped_missing_invoice,
"skipped_missing_fk": skipped_missing_fk,
"skipped_details": skipped_fk_details
"skipped_details": skipped_fk_details,
"skipped_summary": _summarize_top_reasons(skipped_fk_details),
}
except Exception as e:
@@ -6124,7 +6526,8 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt
"skipped_invalid": skipped_invalid,
"skipped_missing_invoice": skipped_missing_invoice,
"skipped_missing_fk": skipped_missing_fk,
"skipped_details": skipped_fk_details
"skipped_details": skipped_fk_details,
"skipped_summary": _summarize_top_reasons(skipped_fk_details),
}
return response

59
backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_impo_temp.py
View File

@@ -14,6 +14,13 @@ MAX_LEN_CANTIDAD_STR = 19
MAX_LEN_ORDEN_COMPRA = 20
PREFERENCIAS_VALIDAS = frozenset({"GENERAL", "TLCS", "PROSEC", "ALADI"})
PREFERENCIA_ALIASES = {
"N": "GENERAL",
"NORMAL": "GENERAL",
"GRAL": "GENERAL",
"TLC": "TLCS",
"PRO": "PROSEC",
}
SE_PAGO_IMPUESTO_VALIDOS = frozenset({"SI", "NO"})
APOSTROFE = "'"
@@ -46,6 +53,19 @@ def _get(row: Dict[str, Any], *keys: str) -> str:
return ""
def _norm_invoice(value: Any) -> str:
return _clip(value).upper()
def normalize_preferencia_arancelaria(value: Any) -> str:
"""Map CSV/manual textual variants to system canonical preference types."""
raw = _clip(value)
if not raw:
return ""
text = raw.upper()
return PREFERENCIA_ALIASES.get(text, text)
def _check_factura_vacia(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
val = _get(row, "NUMERO FACTURA", "NUM FACTURA", "FACTURA")
if not val:
@@ -62,7 +82,7 @@ def _check_factura_existe(
line_num: int,
invoice_id_by_number: Dict[str, int],
) -> Optional[Dict[str, Any]]:
if invoice_number not in invoice_id_by_number:
if _norm_invoice(invoice_number) not in invoice_id_by_number:
return {
"line": line_num,
"col": "NUMERO FACTURA",
@@ -77,9 +97,10 @@ def _check_factura_no_actualizada(
invoice_processed_by_number: Dict[str, bool],
rfc_exception_updated: Set[str],
) -> Optional[Dict[str, Any]]:
if invoice_number in rfc_exception_updated:
inv_norm = _norm_invoice(invoice_number)
if inv_norm in rfc_exception_updated:
return None
if invoice_processed_by_number.get(invoice_number, False):
if invoice_processed_by_number.get(inv_norm, False):
return {
"line": line_num,
"col": "NUMERO FACTURA",
@@ -190,7 +211,7 @@ def _valida_subpartidas_duplicados(
line_num: int,
line_counts: Dict[Tuple[str, str], int],
) -> Optional[Dict[str, Any]]:
key = (invoice_number.strip(), _clip(linea))
key = (_norm_invoice(invoice_number), _clip(linea))
if line_counts.get(key, 0) > 1:
return {
"line": line_num,
@@ -213,7 +234,7 @@ def _valida_subpartida_tiene_principal(
inv = _get(row, "NUMERO FACTURA", "NUM FACTURA", "FACTURA")
if not inv:
return None
key_principal = (inv.strip(), _clip(v))
key_principal = (_norm_invoice(inv), _clip(v))
if key_principal in partidas_principales_en_csv or key_principal in partidas_principales_en_bd:
return None
return {
@@ -320,19 +341,30 @@ def _validaciones_parimpo_tem(
return err("PAIS ORIGEN", f"Error: (Celda K{line_num}) El País: {pais} no se encontró como Clave SAAIM3 ni Clave Americana en el Catálogo de Paises.")
# M: Preferencia
pref = _get(row, "PREFERENCIA ARANCELARIA", "PREFERENCIA", "PREFERENCIAARANCELARIA").upper()
pref_raw = _get(row, "PREFERENCIA ARANCELARIA", "PREFERENCIA", "PREFERENCIAARANCELARIA")
pref = normalize_preferencia_arancelaria(pref_raw)
pref_ctx = f"{pref} (capturada: {pref_raw})" if pref_raw and pref_raw.strip().upper() != pref else pref
if pref and pref not in PREFERENCIAS_VALIDAS:
return err("PREFERENCIA ARANCELARIA", f"Error: (Celda M{line_num}) La Preferencia Arancelaria: {pref} no es correcta para el sistema SCAF.")
return err(
"PREFERENCIA ARANCELARIA",
f"Error: (Celda M{line_num}) La Preferencia Arancelaria: {pref_ctx} no es correcta para el sistema SCAF.",
)
sector = _get(row, "SECTOR")
if pref == "PROSEC":
if not sector:
return err("PREFERENCIA ARANCELARIA", f"Error: (Celda M{line_num} y N{line_num}) La Preferencia Arancelaria es: {pref} y en la columna N no tiene sector.")
return err(
"PREFERENCIA ARANCELARIA",
f"Error: (Celda M{line_num} y N{line_num}) La Preferencia Arancelaria es: {pref_ctx} y en la columna N no tiene sector.",
)
if sector not in authorized_sectors:
return err("SECTOR", f"Error: (Celda N{line_num}) El Sector: {sector} no existe en el Catálogo de Sectores.")
if not company_has_prosec:
return err("SECTOR", f"Error: (Celda N{line_num}) La empresa no cuenta con autorización PROSEC.")
elif pref and pref != "PROSEC" and sector:
return err("PREFERENCIA ARANCELARIA", f"Error: (Celda M{line_num} y N{line_num}) La Preferencia Arancelaria es: {pref} y en la columna N tiene sector.")
return err(
"PREFERENCIA ARANCELARIA",
f"Error: (Celda M{line_num} y N{line_num}) La Preferencia Arancelaria es: {pref_ctx} y en la columna N tiene sector.",
)
# O: Fracción americana
frac_ame = _get(row, "FRACCION AMERICANA", "FRACCIONAMERICANA")
@@ -454,7 +486,8 @@ def validate_row_partidas_impo_temp(
_warn_apostrofes_num_parte(row, line_num, warnings)
linea = _get(row, "LINEA", "RENGLON", "PARTIDA")
existing_lines = existing_line_keys_by_invoice.get(invoice_number.strip(), set())
invoice_number_norm = _norm_invoice(invoice_number)
existing_lines = existing_line_keys_by_invoice.get(invoice_number_norm, set())
partida_existe = bool(linea and linea in existing_lines)
use_partial = actualizar and not autonumerar and partida_existe
@@ -490,7 +523,7 @@ def validate_row_partidas_impo_temp(
errors.append(err)
if levantar_subpartidas:
for err in [
_valida_subpartidas_duplicados(invoice_number, linea, line_num, line_counts_csv),
_valida_subpartidas_duplicados(invoice_number_norm, linea, line_num, line_counts_csv),
_valida_subpartida_tiene_principal(row, line_num, partidas_principales_csv, partidas_principales_bd),
_valida_subpartida_v_no_cero(row, line_num),
]:
@@ -499,7 +532,9 @@ def validate_row_partidas_impo_temp(
err = _validaciones_parimpo_tem(row, line_num, **_validaciones_kwargs)
if err:
errors.append(err)
if rfc_exception_num_parte and invoice_number in rfc_exception_num_parte and valid_part_numbers is not None:
# Paridad con commit: si viene NUM. PARTE capturado, debe existir en catálogo.
# La excepción RFC solo controla obligatoriedad; no desactiva la verificación de existencia.
if valid_part_numbers is not None:
num_parte = _get(row, "NUM. PARTE", "NUMPARTE", "NUMERO PARTE", "NUM PARTE")
if num_parte and num_parte.upper() not in valid_part_numbers:
errors.append({

8
backend/api/v1/modules/a76/layouts_csv/pedmientos/tasks.py
View File

@@ -408,6 +408,8 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
)
total_skipped = skipped_invalid + skipped_missing_fk + skipped_duplicate
critical_reference_gaps = skipped_invalid + skipped_missing_fk
reference_state_ready = critical_reference_gaps == 0
if inserted_count == 0 and updated_count == 0 and total_skipped > 0:
reasons = "; ".join(
f"Línea {d.get('line', '?')}: {d.get('reason', '')}" for d in skipped_details[:5]
@@ -422,6 +424,8 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
"skipped_missing_fk": skipped_missing_fk,
"skipped_duplicate": skipped_duplicate,
"skipped_details": skipped_details,
"critical_reference_gaps": critical_reference_gaps,
"reference_state_ready": reference_state_ready,
"message": f"No se insertaron registros. {total_skipped} rechazados. Motivos: {reasons}",
}
if inserted_count == 0 and updated_count == 0:
@@ -434,6 +438,8 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
"skipped_missing_fk": skipped_missing_fk,
"skipped_duplicate": skipped_duplicate,
"skipped_details": skipped_details,
"critical_reference_gaps": critical_reference_gaps,
"reference_state_ready": reference_state_ready,
}
return {
"status": "finished",
@@ -443,6 +449,8 @@ def _do_commit(job_id: str) -> Dict[str, Any]:
"skipped_missing_fk": skipped_missing_fk,
"skipped_duplicate": skipped_duplicate,
"skipped_details": skipped_details,
"critical_reference_gaps": critical_reference_gaps,
"reference_state_ready": reference_state_ready,
}