From 71d3446351d7e1779dc7dc52f2476a77ad08c942 Mon Sep 17 00:00:00 2001 From: hreyes Date: Mon, 9 Mar 2026 12:26:23 -0600 Subject: [PATCH] feature/feature/clarion-validaciones-invoices-csv-partidas-def-impo --- .../modules/a76/layouts_csv/facturas/tasks.py | 262 ++++++++++++++++++ .../facturas/validators/__init__.py | 2 + .../facturas/validators/partidas_impo_def.py | 197 +++++++++++++ .../facturas/validators/partidas_impo_temp.py | 2 +- 4 files changed, 462 insertions(+), 1 deletion(-) create mode 100644 backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_impo_def.py 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 848e4181..b4efce49 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py @@ -607,6 +607,266 @@ def scan_file(self, job_id: str, model_target: str, config: str = None): logger.exception("Partidas import scan failed: %s", e) return {"status": "failed", "error": str(e)} + # --- Partidas de Importación Definitiva: flujo específico (Clarion VALIDA_TODA_PARIMPO_DEF / VALIDA_PARCIAL) --- + # Estructura de columnas: misma que partidas TEM (imp_def_details resuelve a imp_temp_details). Facturas DEF/MATDE/EXDEF. + if model_target == "invoice_details" and template_id == "imp_def_details": + try: + from api.v1.modules.a76.invoices.models import InvoiceHeader + from api.v1.modules.a76.items.models import LineItem + from api.v1.modules.a76.classes.models import Class + from api.v1.modules.a76.general_catalogs.units_of_measure.models import UnitOfMeasure + from api.v1.modules.a76.general_catalogs.packages.models import Package + from api.v1.modules.public.reference_data.countries.models import Country + from api.v1.modules.public.reference_data.sectors.models import Sector + from api.v1.modules.public.reference_data.valuation_methods.models import ValuationMethod + from api.v1.modules.public.reference_data.payment_methods.models import PaymentMethod + from api.v1.modules.a76.general_catalogs.fractions.us_tariff_fractions.models import USTariffFraction + from api.v1.modules.a76.general_catalogs.company.models import Company + from api.v1.modules.a76.parts.models import Part + from .validators.partidas_impo_def import validate_row_partidas_impo_def + + DEF_INVOICE_TYPES = ("DEF", "MATDE", "EXDEF") + _fc = parse_footer_config(meta.get("footer_config")) + autonumerar = meta.get("autonumerar", True) + actualizar = meta.get("actualizar", False) + levantar_subpartidas = meta.get("levantar_subpartidas", False) + calcular_costo_en_base_a_total = meta.get("calcular_costo_unitario_en_base_a_valor_total", False) + validar_decimales_pza = meta.get("validar_decimales_pza", False) + if _fc: + if "autonumerar" in _fc: + autonumerar = bool(_fc["autonumerar"]) + elif _fc.get("autonumber_partidas", "true") is not None: + autonumerar = str(_fc.get("autonumber_partidas", "true")).lower() in ("true", "1", "si", "sí", "yes") + if "actualizar" in _fc: + actualizar = bool(_fc["actualizar"]) + if "levantar_subpartidas" in _fc: + levantar_subpartidas = bool(_fc["levantar_subpartidas"]) + if "calcular_costo_unitario_en_base_a_valor_total" in _fc: + calcular_costo_en_base_a_total = bool(_fc["calcular_costo_unitario_en_base_a_valor_total"]) + if "validar_decimales_pza" in _fc: + validar_decimales_pza = bool(_fc["validar_decimales_pza"]) + + RFC_EXCEPTION_UPDATED = {"TPI121217SF6", "TCI170502858"} + RFC_EXCEPTION_NUM_PARTE = {"CTE980130518"} + + with CoreSessionLocal() as session: + q_inv = ( + session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.is_updated) + .filter( + InvoiceHeader.tenant_id == tenant_id, + InvoiceHeader.company_id == company_id, + InvoiceHeader.operation_type == "imp", + InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES), + ) + ) + invoice_id_by_number: Dict[str, int] = {} + invoice_updated_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_updated_by_number[str(num).strip()] = bool(is_upd) + + existing_line_keys_by_invoice: Dict[str, Set[str]] = {} + q_li = ( + session.query(InvoiceHeader.invoice_number, LineItem.line_number) + .join(LineItem, LineItem.invoice_id == InvoiceHeader.id) + .filter( + InvoiceHeader.tenant_id == tenant_id, + InvoiceHeader.company_id == company_id, + InvoiceHeader.operation_type == "imp", + InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES), + ) + ) + for num, ln in q_li.all(): + if num is not None: + key = str(num).strip() + if key not in existing_line_keys_by_invoice: + existing_line_keys_by_invoice[key] = set() + existing_line_keys_by_invoice[key].add(str(ln).strip()) + + partidas_principales_bd: Set[Tuple[str, str]] = set() + try: + from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem + q_pp = ( + session.query(InvoiceHeader.invoice_number, LineItem.line_number) + .join(LineItem, LineItem.invoice_id == InvoiceHeader.id) + .join(FaLineItem, FaLineItem.id == LineItem.id) + .filter( + InvoiceHeader.tenant_id == tenant_id, + InvoiceHeader.company_id == company_id, + InvoiceHeader.operation_type == "imp", + InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES), + FaLineItem.is_subitem == False, + FaLineItem.contains_subitems == True, + ) + ) + for num, ln in q_pp.all(): + if num is not None: + partidas_principales_bd.add((str(num).strip(), str(ln).strip())) + except Exception: + pass + + valid_class_codes: Set[str] = set() + class_um_by_code: Dict[str, str] = {} + class_fraction_by_code: Dict[str, str] = {} + class_desc_es_by_code: Dict[str, str] = {} + class_desc_en_by_code: Dict[str, str] = {} + for c in session.query(Class).filter(Class.tenant_id == tenant_id, Class.company_id == company_id).all(): + code = (c.class_code or "").strip().upper() + if code: + valid_class_codes.add(code) + class_um_by_code[code] = (c.unit_of_measure or "").strip().upper() + class_fraction_by_code[code] = (c.fraction or "").strip() + class_desc_es_by_code[code] = (c.description_es or "").strip() + class_desc_en_by_code[code] = (c.description_en or "").strip() + + valid_uom_codes: Set[str] = set() + for u in session.query(UnitOfMeasure.code).filter(UnitOfMeasure.tenant_id == tenant_id, UnitOfMeasure.company_id == company_id).all(): + if u[0]: + valid_uom_codes.add((u[0] or "").strip().upper()) + + valid_bulks_codes: Set[str] = set() + for p in session.query(Package.key).filter(Package.tenant_id == tenant_id, Package.company_id == company_id).all(): + if p[0]: + valid_bulks_codes.add((p[0] or "").strip()) + + valid_country_keys: Set[str] = set() + for row in session.query(Country.m3_key, Country.ame_key).all(): + if row[0]: + valid_country_keys.add((row[0] or "").strip().upper()) + if row[1]: + valid_country_keys.add((row[1] or "").strip().upper()) + + valid_fraction_ame: Set[str] = set() + for row in session.query(USTariffFraction.code).filter(USTariffFraction.tenant_id == tenant_id, USTariffFraction.company_id == company_id).all(): + if row[0]: + valid_fraction_ame.add((row[0] or "").strip()) + + authorized_sectors: Set[str] = set() + for row in session.query(Sector.key).filter(Sector.authorized == True).all(): + if row[0]: + authorized_sectors.add((row[0] or "").strip().upper()) + + valid_payment_methods: Set[str] = set() + for row in session.query(PaymentMethod.key).all(): + if row[0] is not None: + valid_payment_methods.add(str(row[0]).strip()) + + valid_valuation_methods: Set[str] = set() + for row in session.query(ValuationMethod.key).all(): + if row[0]: + valid_valuation_methods.add((row[0] or "").strip()) + + company = session.query(Company).filter(Company.id == company_id).first() + company_has_prosec = bool(company.prosec) if company else False + company_rfc = (company.rfc or "").strip().upper() if company else "" + + valid_part_numbers: Set[str] = set() + for row in session.query(Part.part_number).filter(Part.tenant_id == tenant_id, Part.company_id == company_id).all(): + if row[0]: + valid_part_numbers.add((row[0] or "").strip().upper()) + + rfc_exception_updated: Set[str] = set() + rfc_exception_num_parte: Set[str] = set() + + with open(file_path, "r", encoding="utf-8-sig") as f_in: + sample = f_in.read(2048) + f_in.seek(0) + try: + dialect = csv.Sniffer().sniff(sample, delimiters=",;\t") + except Exception: + dialect = "excel" + reader = csv.DictReader(f_in, dialect=dialect) + rows_list = list(reader) + + invoice_numbers_from_csv = set() + for row in rows_list: + inv = (row.get("NUMERO FACTURA") or row.get("NUM FACTURA") or row.get("FACTURA") or "").strip() + if inv: + invoice_numbers_from_csv.add(inv) + if company_rfc in RFC_EXCEPTION_UPDATED: + rfc_exception_updated = invoice_numbers_from_csv + if company_rfc in RFC_EXCEPTION_NUM_PARTE: + rfc_exception_num_parte = invoice_numbers_from_csv + + line_counts_csv: Dict[Tuple[str, str], int] = {} + partidas_principales_csv: Set[Tuple[str, str]] = set() + + def _get_row_def(row_norm: Dict[str, Any], *keys: str) -> str: + for k in keys: + v = row_norm.get(k) + if v is not None and str(v).strip(): + return str(v).strip() + return "" + + for row in rows_list: + row_norm = row_from_template(row, "imp_def_details", normalize_header) + inv = _get_row_def(row_norm, "NUMERO FACTURA", "NUM FACTURA", "FACTURA") + linea = _get_row_def(row_norm, "LINEA", "RENGLON", "PARTIDA") + if inv and linea: + key = (inv, linea) + line_counts_csv[key] = line_counts_csv.get(key, 0) + 1 + u = _get_row_def(row_norm, "ES PARTIDA O SUBPARTIDA", "ESSUBPARTIDA").upper() + if u == "P" and inv and linea: + partidas_principales_csv.add((inv, linea)) + + error_count = 0 + processed_rows = 0 + error_lines_list = [] + errors_detail = [] + + with open(error_path, "w", encoding="utf-8") as f_err: + for i, row in enumerate(rows_list, start=1): + 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_def_details", normalize_header) + err = validate_row_partidas_impo_def( + row_norm, + i, + autonumerar=autonumerar, + actualizar=actualizar, + levantar_subpartidas=levantar_subpartidas, + calcular_costo_en_base_a_total=calcular_costo_en_base_a_total, + validar_decimales_pza=validar_decimales_pza, + invoice_id_by_number=invoice_id_by_number, + invoice_updated_by_number=invoice_updated_by_number, + rfc_exception_updated=rfc_exception_updated, + existing_line_keys_by_invoice=existing_line_keys_by_invoice, + line_counts_csv=line_counts_csv, + partidas_principales_csv=partidas_principales_csv, + partidas_principales_bd=partidas_principales_bd, + valid_class_codes=valid_class_codes, + class_um_by_code=class_um_by_code, + class_fraction_by_code=class_fraction_by_code, + class_desc_es_by_code=class_desc_es_by_code, + class_desc_en_by_code=class_desc_en_by_code, + valid_uom_codes=valid_uom_codes, + valid_bulks_codes=valid_bulks_codes, + valid_country_keys=valid_country_keys, + valid_fraction_ame=valid_fraction_ame, + valid_payment_methods=valid_payment_methods, + valid_valuation_methods=valid_valuation_methods, + authorized_sectors=authorized_sectors, + company_has_prosec=company_has_prosec, + rfc_exception_num_parte=rfc_exception_num_parte or None, + valid_part_numbers=valid_part_numbers, + warnings=None, + ) + if err: + error_count += 1 + error_lines_list.append(err["line"]) + f_err.write(json.dumps({"line": err["line"], "col": err.get("col", ""), "msg": err.get("msg", "")}) + "\n") + if len(errors_detail) < 500: + errors_detail.append({"line": err["line"], "col": err.get("col", ""), "msg": err.get("msg", "")}) + processed_rows += 1 + + if error_lines_list: + common_storage.store_error_lines(JOB_TYPE, job_id, error_lines_list) + return common_responses.scan_result(job_id, processed_rows, error_count, errors_detail) + except Exception as e: + logger.exception("Partidas importación definitiva scan failed: %s", e) + return {"status": "failed", "error": str(e)} + # --- Encabezados de Importación Temporal: flujo específico (Clarion VALIDA_TODA / VALIDA_PARCIAL) --- if model_target == "invoice_header" and template_id == "imp_temp_header": try: @@ -2197,6 +2457,8 @@ def insert_valid_rows(self, job_id: str, model_target: str): ) if model_target == "invoice_header" and _template_id_insert == "imp_def_header": inv_type_value = "DEF" + if model_target == "invoice_details" and _template_id_insert == "imp_def_details": + inv_type_value = "DEF" logger.info(f"Processing CSV with operation_type={op_type_value}, invoice_type={inv_type_value}, date_format={date_format}") diff --git a/backend/api/v1/modules/a76/layouts_csv/facturas/validators/__init__.py b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/__init__.py index bc7ac06a..89fb2a9f 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/validators/__init__.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/__init__.py @@ -9,10 +9,12 @@ from .encabezados_impo_def import ( validate_row_encabezados_impo_def, parse_pedimento_col_a_impo_def, ) +from .partidas_impo_def import validate_row_partidas_impo_def __all__ = [ "validate_row_encabezados_impo_temp", "validate_row_encabezados_impo_def", + "validate_row_partidas_impo_def", "row_to_transport_type_clarion", "parse_pedimento_col_a", "parse_pedimento_col_a_impo_def", diff --git a/backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_impo_def.py b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_impo_def.py new file mode 100644 index 00000000..0c58884d --- /dev/null +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_impo_def.py @@ -0,0 +1,197 @@ +""" +Validaciones CSV para Partidas de Importación Definitiva. +Paridad Clarion: VALIDA_TODA_PARIMPO_DEF, VALIDA_PARCIAL_PARIMPO_DEF, VALIDACIONES_PARIMPO_DEF. +Reutiliza la lógica de partidas_impo_temp; solo cambia el mensaje cuando la factura no existe +(«no existe en el catálogo de Importación Definitiva») y el origen de facturas (DEF/MATDE/EXDEF en tasks.py). +Estructura de columnas: misma que partidas TEM (NUMERO FACTURA, LINEA, CLASE, ... ID TYPE). +""" +from typing import Dict, Any, Optional, Set, Tuple, List + +from .partidas_impo_temp import ( + _clip, + _get, + _check_factura_vacia, + _check_factura_no_actualizada, + _check_linea_si_no_autonumerar, + _check_levantar_subpartidas_uv, + _valida_toda_obligatorios, + _valida_toda_numericos, + _valida_subpartidas_duplicados, + _valida_subpartida_tiene_principal, + _valida_subpartida_v_no_cero, + _validaciones_parimpo_tem, + _warn_apostrofes_num_parte, +) + + +def _check_factura_existe_def( + invoice_number: str, + line_num: int, + invoice_id_by_number: Dict[str, int], +) -> Optional[Dict[str, Any]]: + """Misma lógica que _check_factura_existe; mensaje específico Importación Definitiva (Clarion).""" + if invoice_number not in invoice_id_by_number: + return { + "line": line_num, + "col": "NUMERO FACTURA", + "msg": ( + f"Error: (Celda A{line_num}) La Factura de Importación {invoice_number} " + "no existe en el catálogo de Importación Definitiva y no se pueden hacer las validaciones. " + ), + } + return None + + +def validate_row_partidas_impo_def( + row: Dict[str, Any], + line_num: int, + autonumerar: bool, + actualizar: bool, + levantar_subpartidas: bool, + calcular_costo_en_base_a_total: bool, + validar_decimales_pza: bool, + invoice_id_by_number: Dict[str, int], + invoice_updated_by_number: Dict[str, bool], + rfc_exception_updated: Set[str], + existing_line_keys_by_invoice: Dict[str, Set[str]], + line_counts_csv: Dict[Tuple[str, str], int], + partidas_principales_csv: Set[Tuple[str, str]], + partidas_principales_bd: Set[Tuple[str, str]], + valid_class_codes: Set[str], + class_um_by_code: Dict[str, str], + class_fraction_by_code: Dict[str, str], + class_desc_es_by_code: Dict[str, str], + class_desc_en_by_code: Dict[str, str], + valid_uom_codes: Set[str], + valid_bulks_codes: Set[str], + valid_country_keys: Set[str], + valid_fraction_ame: Set[str], + valid_payment_methods: Set[str], + valid_valuation_methods: Set[str], + authorized_sectors: Set[str], + company_has_prosec: bool, + rfc_exception_num_parte: Optional[Set[str]], + valid_part_numbers: Optional[Set[str]], + warnings: Optional[List[Dict[str, Any]]] = None, +) -> Optional[Dict[str, Any]]: + """ + Valida una fila de CSV de Partidas de Importación Definitiva. + Clarion: VALIDA_TODA_PARIMPO_DEF vs VALIDA_PARCIAL_PARIMPO_DEF según autonumerar, actualizar y si la partida existe. + Reutiliza todo de partidas_impo_temp salvo el check de factura existente (mensaje DEF). + """ + err = _check_factura_vacia(row, line_num) + if err: + return err + + invoice_number = _get(row, "NUMERO FACTURA", "NUM FACTURA", "FACTURA") + if not invoice_number: + return {"line": line_num, "col": "NUMERO FACTURA", "msg": "Requerido"} + + err = _check_factura_existe_def(invoice_number, line_num, invoice_id_by_number) + if err: + return err + + err = _check_factura_no_actualizada( + invoice_number, line_num, invoice_updated_by_number, rfc_exception_updated + ) + if err: + return err + + err = _check_linea_si_no_autonumerar(row, line_num, autonumerar) + if err: + return err + + err = _check_levantar_subpartidas_uv(row, line_num, levantar_subpartidas) + if err: + return err + + _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()) + partida_existe = bool(linea and linea in existing_lines) + use_partial = actualizar and not autonumerar and partida_existe + + if use_partial: + return _validaciones_parimpo_tem( + row, + line_num, + valid_class_codes=valid_class_codes, + class_um_by_code=class_um_by_code, + class_fraction_by_code=class_fraction_by_code, + class_desc_es_by_code=class_desc_es_by_code, + class_desc_en_by_code=class_desc_en_by_code, + valid_uom_codes=valid_uom_codes, + valid_bulks_codes=valid_bulks_codes, + valid_country_keys=valid_country_keys, + valid_fraction_ame=valid_fraction_ame, + valid_payment_methods=valid_payment_methods, + valid_valuation_methods=valid_valuation_methods, + authorized_sectors=authorized_sectors, + company_has_prosec=company_has_prosec, + validar_decimales_pza=validar_decimales_pza, + rfc_exception_num_parte=rfc_exception_num_parte, + invoice_number=invoice_number, + ) + else: + err = _valida_toda_obligatorios( + row, line_num, levantar_subpartidas, calcular_costo_en_base_a_total + ) + if err: + return err + # Importación Definitiva: NUM. PARTE es obligatorio en todas las partidas (el insert lo exige). + num_parte = _get(row, "NUM. PARTE", "NUMPARTE", "NUMERO PARTE", "NUM PARTE") + if not (num_parte and str(num_parte).strip()): + return { + "line": line_num, + "col": "NUM. PARTE", + "msg": "NUM. PARTE: Requerido (obligatorio para partidas de Importación Definitiva).", + } + err = _valida_toda_numericos(row, line_num, calcular_costo_en_base_a_total) + if err: + return err + if levantar_subpartidas: + err = _valida_subpartidas_duplicados( + invoice_number, linea, line_num, line_counts_csv + ) + if err: + return err + err = _valida_subpartida_tiene_principal( + row, line_num, partidas_principales_csv, partidas_principales_bd + ) + if err: + return err + err = _valida_subpartida_v_no_cero(row, line_num) + if err: + return err + err = _validaciones_parimpo_tem( + row, + line_num, + valid_class_codes=valid_class_codes, + class_um_by_code=class_um_by_code, + class_fraction_by_code=class_fraction_by_code, + class_desc_es_by_code=class_desc_es_by_code, + class_desc_en_by_code=class_desc_en_by_code, + valid_uom_codes=valid_uom_codes, + valid_bulks_codes=valid_bulks_codes, + valid_country_keys=valid_country_keys, + valid_fraction_ame=valid_fraction_ame, + valid_payment_methods=valid_payment_methods, + valid_valuation_methods=valid_valuation_methods, + authorized_sectors=authorized_sectors, + company_has_prosec=company_has_prosec, + validar_decimales_pza=validar_decimales_pza, + rfc_exception_num_parte=rfc_exception_num_parte, + invoice_number=invoice_number, + ) + if err: + return err + if rfc_exception_num_parte and invoice_number in rfc_exception_num_parte and 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: + return { + "line": line_num, + "col": "NUM. PARTE", + "msg": f"Error: (Celda W{line_num}) El número de parte Capturado: {num_parte} no existe.", + } + return None 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 c757e316..72b03ca7 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 @@ -214,7 +214,7 @@ def _valida_subpartida_tiene_principal( if not inv: return None key_principal = (inv.strip(), _clip(v)) - if key_principal in partidas_principales_en_csv or key_principal in partidas_principales_bd: + if key_principal in partidas_principales_en_csv or key_principal in partidas_principales_en_bd: return None return { "line": line_num,