From 79b2506af43ea3115d1d6dec20754deac5f35f71 Mon Sep 17 00:00:00 2001 From: hreyes Date: Tue, 24 Mar 2026 08:06:48 -0600 Subject: [PATCH] feature/fix-invoices-csv --- .../a76/layouts_csv/common/responses.py | 14 + .../line_item_enrichment/export_enrichment.py | 7 +- .../line_item_enrichment/import_enrichment.py | 6 +- .../modules/a76/layouts_csv/facturas/tasks.py | 619 +++++++++++++++--- .../facturas/validators/partidas_impo_temp.py | 59 +- .../a76/layouts_csv/pedmientos/tasks.py | 8 + 6 files changed, 591 insertions(+), 122 deletions(-) diff --git a/backend/api/v1/modules/a76/layouts_csv/common/responses.py b/backend/api/v1/modules/a76/layouts_csv/common/responses.py index 4d1f6d6d..848a2a3c 100644 --- a/backend/api/v1/modules/a76/layouts_csv/common/responses.py +++ b/backend/api/v1/modules/a76/layouts_csv/common/responses.py @@ -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 diff --git a/backend/api/v1/modules/a76/layouts_csv/facturas/line_item_enrichment/export_enrichment.py b/backend/api/v1/modules/a76/layouts_csv/facturas/line_item_enrichment/export_enrichment.py index 075210fa..354c7bf6 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/line_item_enrichment/export_enrichment.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/line_item_enrichment/export_enrichment.py @@ -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 = ( diff --git a/backend/api/v1/modules/a76/layouts_csv/facturas/line_item_enrichment/import_enrichment.py b/backend/api/v1/modules/a76/layouts_csv/facturas/line_item_enrichment/import_enrichment.py index e5f5fb77..afbf83c2 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/line_item_enrichment/import_enrichment.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/line_item_enrichment/import_enrichment.py @@ -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: diff --git a/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py b/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py index cdee8615..f1ad7d80 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py @@ -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 diff --git a/backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_impo_temp.py b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_impo_temp.py index 1180bf79..d90efaa5 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_impo_temp.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_impo_temp.py @@ -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({ diff --git a/backend/api/v1/modules/a76/layouts_csv/pedmientos/tasks.py b/backend/api/v1/modules/a76/layouts_csv/pedmientos/tasks.py index 58661da9..1ac7aa76 100644 --- a/backend/api/v1/modules/a76/layouts_csv/pedmientos/tasks.py +++ b/backend/api/v1/modules/a76/layouts_csv/pedmientos/tasks.py @@ -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, }