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 4169392b..f2fda307 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py @@ -27,10 +27,20 @@ from ..common import storage as common_storage from ..common import meta as common_meta from ..common import responses as common_responses from .template_config import row_from_template +from .validators.encabezados_impo_temp import csv_tipo_moneda_es_me_mn_mc # Models are imported inside tasks to avoid circular dependencies and mapper initialization issues in the API process logger = logging.getLogger(__name__) + +def _merge_unique_invoice_scan_error_lines( + precheck_lines: Set[int], + row_validation_lines: List[int], +) -> List[int]: + """Unión deduplicada: líneas marcadas en precheck + líneas con error en validación por fila (imp_temp_header).""" + return sorted(set(precheck_lines) | set(row_validation_lines)) + + # Job type vacío para facturas (prefijo Redis "import_" sin tipo, ver common/storage.py) JOB_TYPE = "" @@ -2782,7 +2792,31 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] = } ) + # No return temprano: el commit siempre valida por fila (logística, etc.). Si solo se + # devolvía el precheck, valid_rows del scan quedaba inflado vs inserted/skipped del commit. + scan_precheck_message: Optional[str] = None if precheck_errors: + scan_precheck_message = ( + "Precheck de referencias falló. Corrige catálogos/pedimentos según los errores indicados. " + "Se aplicó también validación completa por fila (logística, etc.) para paridad con el commit." + ) + + processed_rows = 0 + error_lines_list: List[int] = [] + errors_detail: List[Dict[str, Any]] = [] + for e in precheck_errors: + if len(errors_detail) < 5000: + errors_detail.append( + { + "line": e["line"], + "col": e.get("col", ""), + "msg": e.get("msg", ""), + "solution": e.get("solution", ""), + "warning": False, + } + ) + + with CoreSessionLocal() as session: with open(error_path, "w", encoding="utf-8") as f_err: for e in precheck_errors: f_err.write( @@ -2796,99 +2830,96 @@ def _do_scan_file(self, job_id: str, model_target: str, config: Optional[str] = ) + "\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] = [] - errors_detail: List[Dict[str, Any]] = [] - - with open(error_path, "w", encoding="utf-8") as f_err: - for i, row in enumerate(rows_list, start=1): - if i % 1000 == 0: - self.update_state(state="PROGRESS", meta={"current": i, "total": total_rows, "errors": error_count}) - row_norm = row_from_template(row, "imp_temp_header", normalize_header) - warnings_row: List[Dict[str, Any]] = [] - row_errors = validate_row_encabezados_impo_temp( - row_norm, - i, - actualizar=actualizar, - invoice_exists_by_number=invoice_exists_by_number, - invoice_processed_by_number=invoice_processed_by_number, - pedimento_rows=pedimento_rows, - remesa_por_pedimento_bd=remesa_por_pedimento_bd, - remesa_por_pedimento_csv=remesa_por_pedimento_csv, - valid_provider_ids=valid_provider_ids, - valid_sold_to_ids=valid_sold_to_ids, - valid_shipped_to_ids=valid_shipped_to_ids, - valid_provider_short_names=valid_provider_short_names, - valid_sold_to_short_names=valid_sold_to_short_names, - valid_shipped_to_short_names=valid_shipped_to_short_names, - valid_broker_ids=valid_broker_ids, - valid_broker_claves=valid_broker_claves, - valid_transporter_keys=valid_transporter_keys, - valid_incoterms=valid_incoterms, - valid_aduana_codes=valid_aduana_codes, - valid_currency_codes=valid_currency_codes, - exchange_rate_by_date=exchange_rate_by_date, - invoice_has_partidas_by_number=invoice_has_partidas_by_number, - existing_tipo_moneda_by_number=existing_tipo_moneda_by_number, - autonumerar_remesas=autonumerar_remesas, - control_remesa=control_remesa, - remesa_inicio=remesa_inicio, - remesa_fin=remesa_fin, - date_format=date_format, - parse_date_fn=parse_date, - warnings=warnings_row, - ) - row_errors.extend( - logistics_scan_row_errors(row_norm, session, tenant_id, company_id, i) - ) - blocking = [e for e in row_errors if not e.get("warning")] - if blocking: - error_count += 1 - error_lines_list.append(i) - for e in blocking: - f_err.write( - json.dumps( + for i, row in enumerate(rows_list, start=1): + if i % 1000 == 0: + self.update_state( + state="PROGRESS", + meta={ + "current": i, + "total": total_rows, + "errors": len( + _merge_unique_invoice_scan_error_lines( + precheck_lines, error_lines_list + ) + ), + }, + ) + row_norm = row_from_template(row, "imp_temp_header", normalize_header) + warnings_row: List[Dict[str, Any]] = [] + row_errors = validate_row_encabezados_impo_temp( + row_norm, + i, + actualizar=actualizar, + invoice_exists_by_number=invoice_exists_by_number, + invoice_processed_by_number=invoice_processed_by_number, + pedimento_rows=pedimento_rows, + remesa_por_pedimento_bd=remesa_por_pedimento_bd, + remesa_por_pedimento_csv=remesa_por_pedimento_csv, + valid_provider_ids=valid_provider_ids, + valid_sold_to_ids=valid_sold_to_ids, + valid_shipped_to_ids=valid_shipped_to_ids, + valid_provider_short_names=valid_provider_short_names, + valid_sold_to_short_names=valid_sold_to_short_names, + valid_shipped_to_short_names=valid_shipped_to_short_names, + valid_broker_ids=valid_broker_ids, + valid_broker_claves=valid_broker_claves, + valid_transporter_keys=valid_transporter_keys, + valid_incoterms=valid_incoterms, + valid_aduana_codes=valid_aduana_codes, + valid_currency_codes=valid_currency_codes, + exchange_rate_by_date=exchange_rate_by_date, + invoice_has_partidas_by_number=invoice_has_partidas_by_number, + existing_tipo_moneda_by_number=existing_tipo_moneda_by_number, + autonumerar_remesas=autonumerar_remesas, + control_remesa=control_remesa, + remesa_inicio=remesa_inicio, + remesa_fin=remesa_fin, + date_format=date_format, + parse_date_fn=parse_date, + warnings=warnings_row, + ) + row_errors.extend( + logistics_scan_row_errors(row_norm, session, tenant_id, company_id, i) + ) + blocking = [e for e in row_errors if not e.get("warning")] + if blocking: + error_lines_list.append(i) + for e in blocking: + f_err.write( + json.dumps( + { + "line": e["line"], + "col": e.get("col", ""), + "msg": e.get("msg", ""), + "solution": e.get("solution", ""), + } + ) + + "\n" + ) + for e in row_errors + warnings_row: + if len(errors_detail) < 5000: + errors_detail.append( { "line": e["line"], "col": e.get("col", ""), "msg": e.get("msg", ""), "solution": e.get("solution", ""), + "warning": bool(e.get("warning", False)), } ) - + "\n" - ) - for e in row_errors + warnings_row: - if len(errors_detail) < 5000: - errors_detail.append( - { - "line": e["line"], - "col": e.get("col", ""), - "msg": e.get("msg", ""), - "solution": e.get("solution", ""), - "warning": bool(e.get("warning", False)), - } - ) - processed_rows += 1 + processed_rows += 1 - # El commit usa líneas únicas para omitir filas; el resumen preliminar - # debe usar la misma base para evitar discrepancias de válidos/errores. - unique_error_lines = sorted(set(error_lines_list)) + # El commit usa líneas únicas para omitir filas; unir precheck + validación por fila. + unique_error_lines = _merge_unique_invoice_scan_error_lines(precheck_lines, error_lines_list) error_count = len(unique_error_lines) common_storage.store_error_lines(effective_job_type, job_id, unique_error_lines) return common_responses.scan_result( - job_id, processed_rows, error_count, errors_detail, total_rows_in_file=total_rows + job_id, + processed_rows, + error_count, + errors_detail, + total_rows_in_file=total_rows, + message=scan_precheck_message, ) except Exception as e: logger.exception("Encabezados importación temporal scan failed: %s", e) @@ -4036,6 +4067,8 @@ def validate_row_phase_1( def check_currency(col_name): val = row.get(col_name) if val and str(val).strip(): + if csv_tipo_moneda_es_me_mn_mc(str(val)): + return None parsed_currency = parse_currency(val, None) val_norm = normalize_header(val) # parse_currency returns MANUAL if unknown, so if it wasn't explicitly MANUAL, it's invalid @@ -4430,6 +4463,13 @@ def parse_currency(value: Optional[str], currency_type: Optional[str]): return Currency.FOREIGN if "MANUAL" in normalized: return Currency.MANUAL + # MC (moneda por clave): paridad con encabezados_impo_temp / InvoiceFinancials + CLAVE MONEDA + if normalized.replace(" ", "") == "MC": + if currency_type and str(currency_type).strip().upper() == "MXN": + return Currency.LOCAL + if currency_type: + return Currency.FOREIGN + return Currency.FOREIGN if currency_type and str(currency_type).strip().upper() == "MXN": return Currency.LOCAL if currency_type: @@ -5870,14 +5910,15 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt currency_val = row_norm.get('TIPO MONEDA') if currency_val and str(currency_val).strip(): - parsed_currency = parse_currency(currency_val, None) - val_norm = normalize_header(currency_val) - if parsed_currency.value == "manual" and "MANUAL" not in val_norm: - skipped_invalid += 1 - reason = "TIPO MONEDA: Moneda invalida" - skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason}) - logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") - continue + if not csv_tipo_moneda_es_me_mn_mc(str(currency_val)): + parsed_currency = parse_currency(currency_val, None) + val_norm = normalize_header(currency_val) + if parsed_currency.value == "manual" and "MANUAL" not in val_norm: + skipped_invalid += 1 + reason = "TIPO MONEDA: Moneda invalida" + skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason}) + logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") + continue # 2. Client/Provider and broker checks are handled above diff --git a/backend/api/v1/modules/a76/layouts_csv/facturas/validators/encabezados_impo_temp.py b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/encabezados_impo_temp.py index d79df42e..bd6910ab 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/validators/encabezados_impo_temp.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/encabezados_impo_temp.py @@ -19,6 +19,16 @@ REGIMENES_VALIDOS = frozenset({"ITE", "ITR"}) TIPOS_MONEDA_VALIDOS = frozenset({"ME", "MN", "MC"}) TIPO_PESO_VALIDOS = frozenset({"KILOS", "LIBRAS"}) + +def csv_tipo_moneda_es_me_mn_mc(value: Any) -> bool: + """ + ME / MN / MC con la misma regla que _validaciones_moneda (strip + upper). + Compartido entre scan (encabezados) y commit / validate_row_strict en tasks. + """ + if value is None or not str(value).strip(): + return False + return str(value).strip().upper() in TIPOS_MONEDA_VALIDOS + # Clarion Col M → valor normalizado (minúscula para TransportType enum) TIPO_TRANSPORTE_CLARION_TO_NORM = { "NINGUNO": "none", diff --git a/backend/api/v1/modules/a76/layouts_csv/facturas/validators/transport_catalog.py b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/transport_catalog.py index 4597e479..9f6151e3 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/validators/transport_catalog.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/transport_catalog.py @@ -165,6 +165,16 @@ def validate_csv_invoice_logistics_transport( return None +def _csv_row_has_resolved_weight_unit(row_norm: dict) -> bool: + """ + Paridad con insert_valid_rows: has_logistics_data incluye parse_weight_unit(TIPO PESO). + Import diferido para evitar ciclo al cargar tasks. + """ + from api.v1.modules.a76.layouts_csv.facturas.tasks import parse_weight_unit + + return parse_weight_unit(row_norm.get("TIPO PESO")) is not None + + def logistics_scan_row_errors( row_norm: dict, session: Session, @@ -191,6 +201,7 @@ def logistics_scan_row_errors( or trailer_num_csv or _cell("CLAVE TRANSPORTISTA") or _cell("NOMBRE CONDUCTOR") + or _csv_row_has_resolved_weight_unit(row_norm) ) if not has_data: return [] diff --git a/backend/tests/unit/layouts_csv/test_fk_loader_state_mex_key.py b/backend/tests/unit/layouts_csv/test_fk_loader_state_mex_key.py deleted file mode 100644 index e28b42e5..00000000 --- a/backend/tests/unit/layouts_csv/test_fk_loader_state_mex_key.py +++ /dev/null @@ -1,130 +0,0 @@ -"""Smoke: loaders CSV incluyen State.mex_key en validación de ESTADO/PAÍS.""" -from unittest.mock import MagicMock, patch - - -def _fake_state(m3_key="MEX", description="Chihuahua", mex_key="CHH"): - s = MagicMock() - s.m3_key = m3_key - s.description = description - s.mex_key = mex_key - return s - - -def _fake_country(m3_key="MEX", ame_key="MX"): - c = MagicMock() - c.m3_key = m3_key - c.ame_key = ame_key - return c - - -def _session_with_state_queries(fake_state, fake_country): - """Simula session.query en el orden de transportistas/vehicles/trailers loaders.""" - n = [0] - - def query_side_effect(*_args, **_kwargs): - n[0] += 1 - m = MagicMock() - if n[0] == 1: - m.filter.return_value.all.return_value = [] - elif n[0] == 2: - m.all.return_value = [("MX",)] - elif n[0] == 3: - m.all.return_value = [fake_state] - elif n[0] == 4: - m.filter.return_value.first.return_value = fake_country - return m - - session = MagicMock() - session.query.side_effect = query_side_effect - return session - - -@patch("api.v1.modules.a76.layouts_csv.transportistas.common.fk_loader.CoreSessionLocal") -def test_transportistas_fk_sets_includes_mex_key(mock_session_local): - from api.v1.modules.a76.layouts_csv.transportistas.common.fk_loader import ( - load_transportistas_fk_sets, - ) - - st = _fake_state() - mock_session_local.return_value.__enter__.return_value = _session_with_state_queries( - st, _fake_country() - ) - - _, _, state_desc, state_cc = load_transportistas_fk_sets(1, 1) - - assert "CHH" in state_desc - assert "CHIHUAHUA" in state_desc - assert ("MX", "CHH") in state_cc - assert ("MX", "CHIHUAHUA") in state_cc - - -def _session_vehicles(fake_state, fake_country): - n = [0] - - def query_side_effect(*_args, **_kwargs): - n[0] += 1 - m = MagicMock() - if n[0] == 1: - m.all.return_value = [("VH",)] - elif n[0] == 2: - m.all.return_value = [("MX",)] - elif n[0] == 3: - m.all.return_value = [fake_state] - elif n[0] == 4: - m.filter.return_value.first.return_value = fake_country - return m - - session = MagicMock() - session.query.side_effect = query_side_effect - return session - - -@patch("api.v1.modules.a76.layouts_csv.vehicles.common.fk_loader.CoreSessionLocal") -def test_vehicles_fk_sets_includes_mex_key(mock_session_local): - from api.v1.modules.a76.layouts_csv.vehicles.common.fk_loader import load_vehicles_fk_sets - - mock_session_local.return_value.__enter__.return_value = _session_vehicles( - _fake_state(), _fake_country() - ) - - _, _, state_desc, state_cc = load_vehicles_fk_sets(1, 1) - - assert "CHH" in state_desc - assert ("MX", "CHH") in state_cc - - -def _session_trailers(fake_state, fake_country): - n = [0] - - def query_side_effect(*_args, **_kwargs): - n[0] += 1 - m = MagicMock() - if n[0] == 1: - m.all.return_value = [("BX",)] - elif n[0] == 2: - m.all.return_value = [("MX",)] - elif n[0] == 3: - fake_state.ame_key = None - m.all.return_value = [fake_state] - elif n[0] == 4: - m.filter.return_value.first.return_value = fake_country - return m - - session = MagicMock() - session.query.side_effect = query_side_effect - return session - - -@patch("api.v1.modules.a76.layouts_csv.trailers.common.fk_loader.CoreSessionLocal") -def test_trailers_fk_sets_includes_mex_key(mock_session_local): - from api.v1.modules.a76.layouts_csv.trailers.common.fk_loader import load_trailers_fk_sets - - mock_session_local.return_value.__enter__.return_value = _session_trailers( - _fake_state(), _fake_country() - ) - - _, _, state_desc, state_cc, ame_map = load_trailers_fk_sets(1, 1) - - assert "CHH" in state_desc - assert ("MX", "CHH") in state_cc - assert ame_map.get("CHH") == "CHIHUAHUA"