From 23a9c1c40a9de09948816169fd89f02e04147820 Mon Sep 17 00:00:00 2001 From: hreyes Date: Mon, 9 Mar 2026 08:41:14 -0600 Subject: [PATCH] feature/clarion-invoices-temp-partidas-csv --- .../modules/a76/layouts_csv/facturas/tasks.py | 417 +++++++++++--- .../layouts_csv/facturas/template_config.py | 40 +- .../facturas/validators/partidas_impo_temp.py | 535 ++++++++++++++++++ 3 files changed, 915 insertions(+), 77 deletions(-) create mode 100644 backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_impo_temp.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 c3cf8d72..14d11aa4 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py @@ -301,7 +301,263 @@ def scan_file(self, job_id: str, model_target: str, config: str = None): except Exception as e: logger.exception("Series import scan failed: %s", e) return {"status": "failed", "error": str(e)} - + + # --- Partidas de Importación Temporal: flujo específico (Clarion VALIDA_TODA / VALIDA_PARCIAL) --- + if model_target == "invoice_details" and template_id == "imp_temp_details": + try: + from api.v1.modules.a76.invoices.models import InvoiceHeader + from api.v1.modules.a76.items.models import LineItem + from api.v1.modules.a76.classes.models import Class + from api.v1.modules.a76.general_catalogs.units_of_measure.models import UnitOfMeasure + from api.v1.modules.a76.general_catalogs.packages.models import Package + from api.v1.modules.public.reference_data.countries.models import Country + from api.v1.modules.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_temp import validate_row_partidas_impo_temp + + _fc = parse_footer_config(meta.get("footer_config")) + autonumerar = meta.get("autonumerar", True) + actualizar = meta.get("actualizar", False) + levantar_subpartidas = meta.get("levantar_subpartidas", False) + calcular_costo_en_base_a_total = meta.get("calcular_costo_unitario_en_base_a_valor_total", False) + validar_decimales_pza = meta.get("validar_decimales_pza", False) + if _fc: + if "autonumerar" in _fc: + autonumerar = bool(_fc["autonumerar"]) + elif _fc.get("autonumber_partidas", "true") is not None: + autonumerar = str(_fc.get("autonumber_partidas", "true")).lower() in ("true", "1", "si", "sí", "yes") + if "actualizar" in _fc: + actualizar = bool(_fc["actualizar"]) + if "levantar_subpartidas" in _fc: + levantar_subpartidas = bool(_fc["levantar_subpartidas"]) + if "calcular_costo_unitario_en_base_a_valor_total" in _fc: + calcular_costo_en_base_a_total = bool(_fc["calcular_costo_unitario_en_base_a_valor_total"]) + if "validar_decimales_pza" in _fc: + validar_decimales_pza = bool(_fc["validar_decimales_pza"]) + + RFC_EXCEPTION_UPDATED = {"TPI121217SF6", "TCI170502858"} + RFC_EXCEPTION_NUM_PARTE = {"CTE980130518"} + + with CoreSessionLocal() as session: + 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 == "TEM", + ) + ) + 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 == "TEM", + ) + ) + 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, + 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(row_norm: Dict[str, Any], *keys: str) -> str: + for k in keys: + v = row_norm.get(k) + if v is not None and str(v).strip(): + return str(v).strip() + return "" + + for row in rows_list: + row_norm = row_from_template(row, "imp_temp_details", normalize_header) + inv = _get_row(row_norm, "NUMERO FACTURA", "NUM FACTURA", "FACTURA") + linea = _get_row(row_norm, "LINEA", "RENGLON", "PARTIDA") + if inv and linea: + key = (inv, linea) + line_counts_csv[key] = line_counts_csv.get(key, 0) + 1 + u = _get_row(row_norm, "ES PARTIDA O SUBPARTIDA", "ESSUBPARTIDA").upper() + if u == "P" and inv and linea: + partidas_principales_csv.add((inv, linea)) + + error_count = 0 + processed_rows = 0 + error_lines_list = [] + errors_detail = [] + + with open(error_path, "w", encoding="utf-8") as f_err: + for i, row in enumerate(rows_list, start=1): + 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_details", normalize_header) + err = validate_row_partidas_impo_temp( + row_norm, + i, + autonumerar=autonumerar, + actualizar=actualizar, + levantar_subpartidas=levantar_subpartidas, + calcular_costo_en_base_a_total=calcular_costo_en_base_a_total, + validar_decimales_pza=validar_decimales_pza, + invoice_id_by_number=invoice_id_by_number, + invoice_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 import scan failed: %s", e) + return {"status": "failed", "error": str(e)} + try: from api.v1.modules.a76.invoices.models import InvoiceHeader from api.v1.modules.a76.clients_and_providers.models import ClientProvider @@ -1099,6 +1355,9 @@ def insert_valid_rows(self, job_id: str, model_target: str): from api.v1.modules.a76.items.line_customs.models import LineCustom from api.v1.modules.a76.items.line_descriptions.models import LineDescription from api.v1.modules.a76.parts.models import Part + from api.v1.modules.a76.classes.models import Class + from api.v1.modules.a76.general_catalogs.units_of_measure.models import UnitOfMeasure + from api.v1.modules.a76.general_catalogs.packages.models import Package footer_config = parse_footer_config(meta.get("footer_config")) @@ -1136,7 +1395,20 @@ def insert_valid_rows(self, job_id: str, model_target: str): currency_type_cache: Dict[str, Optional[str]] = {} customs_section_cache: Dict[str, Optional[str]] = {} part_cache: Dict[str, Optional[int]] = {} - + class_id_by_code: Dict[str, int] = {} + uom_id_by_code: Dict[str, int] = {} + package_id_by_key: Dict[str, int] = {} + if model_target == 'invoice_details': + for c in session.query(Class.id, Class.class_code).filter(Class.tenant_id == tenant_id, Class.company_id == company_id).all(): + if c[1]: + class_id_by_code[(c[1] or "").strip().upper()] = c[0] + for u in session.query(UnitOfMeasure.id, UnitOfMeasure.code).filter(UnitOfMeasure.tenant_id == tenant_id, UnitOfMeasure.company_id == company_id).all(): + if u[1]: + uom_id_by_code[(u[1] or "").strip().upper()] = u[0] + for p in session.query(Package.id, Package.key).filter(Package.tenant_id == tenant_id, Package.company_id == company_id).all(): + if p[1]: + package_id_by_key[(p[1] or "").strip()] = p[0] + validator = ForeignKeyValidator(session, tenant_id, company_id) with open(file_path, 'r', encoding='utf-8-sig') as f: @@ -1519,7 +1791,7 @@ def insert_valid_rows(self, job_id: str, model_target: str): skipped_missing_invoice += 1 continue - part_num = (row_norm.get('NUMPARTE') or row_norm.get('NUMERO PARTE') or '').strip() + 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: skipped_invalid += 1 reason = "NUMPARTE: Requerido" @@ -1533,66 +1805,56 @@ def insert_valid_rows(self, job_id: str, model_target: str): logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") continue - # --- Prevent Duplicates: Clear existing items for this invoice (Once per job) --- + # --- 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") - - # 1. Delete Items (Cascades to LineItem, LineFinancial, etc. if DB configured, check models) - # Checking Item model, we usually need to be careful. - # Assuming Cascade delete is set up on FKs or we rely on ORM cascade if using relationships. - # Here we use bulk delete. - session.query(Item).filter(Item.invoice_id == invoice_id).delete(synchronize_session=False) - - # 2. Delete InvoiceSalesDetails + 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) - cleared_invoices.add(invoice_id) - # --- NEW LOGIC: Expanded Anexo 76 Structure --- - - # A. Find/Cache Part - part_id = None - part_id = part_cache.get(part_num) - if part_id is None: - p = session.query(Part.id).filter( - Part.part_number == part_num, - Part.tenant_id == tenant_id, - Part.company_id == company_id - ).first() + # --- Partidas: LineItem with invoice_id (no Item parent) + full CSV mapping --- + part_num = (row_norm.get('NUM. PARTE') or row_norm.get('NUMPARTE') or row_norm.get('NUMERO PARTE') or row_norm.get('NUM PARTE') or '').strip() + part_id = part_cache.get(part_num) if part_num else None + if part_id is None and part_num: + p = session.query(Part.id).filter(Part.part_number == part_num, Part.tenant_id == tenant_id, Part.company_id == company_id).first() if p: part_id = p.id part_cache[part_num] = part_id line_num_val = (row_norm.get('LINEA') or row_norm.get('RENGLON') or row_norm.get('PARTIDA')) line_num = parse_int(line_num_val) or (len(details_to_insert) + 1) - - # 1. Parent Item - item = Item( + + class_code = (row_norm.get('CLASE') or '').strip().upper() + class_id = class_id_by_code.get(class_code) if class_code else None + uom_code = (row_norm.get('UNIDAD DE MEDIDA') or row_norm.get('UNIDAD MEDIDA') or '').strip().upper() + uom_id = uom_id_by_code.get(uom_code) if uom_code else None + bulk_key = (row_norm.get('CLAVE BULTOS') or row_norm.get('CLAVEBULTOS') or '').strip() + package_id = package_id_by_key.get(bulk_key) if bulk_key else None + + line = LineItem( invoice_id=invoice_id, + line_number=line_num, tenant_id=tenant_id, company_id=company_id, - item_type="N", # Default to Normal - system_origin="CSV" - ) - session.add(item) - session.flush() # Need item.id - - # 2. Main Line - line = LineItem( - item_id=item.id, - line_number=line_num, - part_number=part_id, - tenant_id=tenant_id, - company_id=company_id + part_number_id=part_id, + class_id=class_id, + unit_of_measure=uom_id, + order=(row_norm.get('ORDEN DE COMPRA') or row_norm.get('ORDENCOMPRA') or None), + material_type=(row_norm.get('ID TYPE') or row_norm.get('IDTYPE') or None), + tax_payment=(str(row_norm.get('SE PAGO IMPUESTO') or row_norm.get('SEPAGOIMPUESTO') or '').strip().upper() == 'SI'), + payment_method=(row_norm.get('FORMA DE PAGO') or row_norm.get('FORMADEPAGO') or row_norm.get('FORMA PAGO') or None), + valuation_method=(row_norm.get('METODO DE VALORACION') or row_norm.get('METODODEVALORACION') or row_norm.get('METODO VALORACION') or None), ) session.add(line) - session.flush() # Need line.id + session.flush() - # 3. Financial Data (vanilla: nulls from CSV -> 0) - price = parse_decimal(row_norm.get('PRECIO UNITARIO') or row_norm.get('PRECIOUNITARIO')) - val_com = parse_decimal(row_norm.get('VALOR COMERCIAL') or row_norm.get('VALORCOMERCIAL')) - qty = parse_decimal(row_norm.get('CANTIDAD')) - commercial_total = val_com or (price * qty if price and qty else None) + price = parse_decimal(row_norm.get('COSTO UNITARIO') or row_norm.get('PRECIO UNITARIO') or row_norm.get('PRECIOUNITARIO')) + if price is None: + total_val = parse_decimal(row_norm.get('TOTAL')) + qty = parse_decimal(row_norm.get('CANTIDAD IMPORTADA') or row_norm.get('CANTIDAD')) + price = (total_val / qty) if (total_val and qty and qty != 0) else None + qty = parse_decimal(row_norm.get('CANTIDAD IMPORTADA') or row_norm.get('CANTIDAD')) + commercial_total = (price * qty) if price and qty else parse_decimal(row_norm.get('TOTAL')) session.add(LineFinancial( item_line_id=line.id, @@ -1600,41 +1862,60 @@ def insert_valid_rows(self, job_id: str, model_target: str): total_commercial_value=decimal_or_zero(commercial_total), )) - # 4. Quantities (vanilla: nulls -> 0 so we always have a quantity row) + net_w = parse_decimal(row_norm.get('PESO NETO') or row_norm.get('PESONETO')) + gross_w = parse_decimal(row_norm.get('PESO BRUTO') or row_norm.get('PESOBRUTO')) session.add(LineQuantity( item_line_id=line.id, quantity=decimal_or_zero(qty), + net_weight=decimal_or_zero(net_w), + gross_weight=decimal_or_zero(gross_w), + package_quantity=int_or_zero(row_norm.get('CANTIDAD BULTOS') or row_norm.get('CANTIDADBULTOS')), + package_id=package_id, )) - # 5. Customs/Fraction - origin = row_norm.get('PAIS ORIGEN') or row_norm.get('PAISORIGEN') - fraction = row_norm.get('FRACCION') - if origin or fraction: - session.add(LineCustom( - item_line_id=line.id, - fraction=fraction, - origin_country=origin, - )) + origin = (row_norm.get('PAIS ORIGEN') or row_norm.get('PAISORIGEN') or row_norm.get('PAIS') or '').strip() + fraction = (row_norm.get('FRACCION ARANCELARIA') or row_norm.get('FRACCION') or row_norm.get('FRACCIONARANCELARIA') or '').strip() + fraction_type = (row_norm.get('PREFERENCIA ARANCELARIA') or row_norm.get('PREFERENCIA') or '').strip() + sector = (row_norm.get('SECTOR') or '').strip() + american_fraction = (row_norm.get('FRACCION AMERICANA') or row_norm.get('FRACCIONAMERICANA') or '').strip() + session.add(LineCustom( + item_line_id=line.id, + origin_country=origin or None, + fraction=fraction or None, + fraction_type=fraction_type or None, + sector=sector or None, + american_fraction=american_fraction or None, + )) - # 6. Description - desc = row_norm.get('DESCRIPCION') - if desc: - session.add(LineDescription( - item_line_id=line.id, - description_spanish=desc, - )) + desc_es = (row_norm.get('DESCRIPCION ESPAÑOL') or row_norm.get('DESCRIPCIONE') or row_norm.get('DESCRIPCION') or '').strip() + desc_en = (row_norm.get('DESCRIPCION INGLES') or row_norm.get('DESCRIPCIONI') or '').strip() + brand = (row_norm.get('MARCA') or '').strip() + model = (row_norm.get('MODELO') or '').strip() + extra_desc = (row_norm.get('DESCRIPCION EXTRA') or row_norm.get('DESCRIPCIONEXTRA') or '').strip() + additional_info = (row_norm.get('INFORMACION ADICIONAL') or row_norm.get('INFORMACIONADICIONAL') or '').strip() + lot = (row_norm.get('LOTE') or '').strip() + entry_number = (row_norm.get('NUMERO ENTRADA') or row_norm.get('NUMEROENTRADA') or row_norm.get('NUM ENTRADA') or '').strip() + session.add(LineDescription( + item_line_id=line.id, + description_spanish=desc_es or None, + description_english=desc_en or None, + brand=brand or None, + model=model or None, + extra_description=extra_desc or None, + additional_info_spanish=additional_info or None, + lot=lot or None, + entry_number=entry_number or None, + )) - # 7. Legacy Sales Details (For specific audit/UI fields; vanilla: nulls -> 0) - detail = InvoiceSalesDetails( + session.add(InvoiceSalesDetails( invoice_id=invoice_id, line_number=line_num, sales_order=(row_norm.get('ORDEN DE COMPRA') or row_norm.get('ORDENCOMPRA') or None), line_bundles=int_or_zero(row_norm.get('CANTIDAD BULTOS') or row_norm.get('CANTIDADBULTOS')), tenant_id=tenant_id, company_id=company_id, - ) - session.add(detail) - details_to_insert.append(item) # Use as counter/ref + )) + details_to_insert.append(line) # 3. Bulk Insert (ORM Transaction) try: diff --git a/backend/api/v1/modules/a76/layouts_csv/facturas/template_config.py b/backend/api/v1/modules/a76/layouts_csv/facturas/template_config.py index 2d6122a9..e6eb4a6a 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/template_config.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/template_config.py @@ -52,19 +52,41 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { "imp_def_header": None, # se resuelve igual que imp_temp_header # --- Encabezado factura: Expo (EstructuraEncFacExpoCamReg.xls) - misma estructura --- "exp_def_header": None, - # --- Partidas factura: Impo Temp (EstructuraParFacImpoTempAF.xls) --- + # --- Partidas factura: Impo Temp (EstructuraParFacImpoTemp - paridad Clarion A-AG) --- "imp_temp_details": [ {"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA"]}, {"canonical": "LINEA", "aliases": ["RENGLON", "PARTIDA"]}, - {"canonical": "NUMPARTE", "aliases": ["NUMERO PARTE"]}, - {"canonical": "PRECIO UNITARIO", "aliases": ["PRECIOUNITARIO"]}, - {"canonical": "VALOR COMERCIAL", "aliases": ["VALORCOMERCIAL"]}, - {"canonical": "CANTIDAD"}, + {"canonical": "CLASE"}, + {"canonical": "CANTIDAD IMPORTADA", "aliases": ["CANTIDAD"]}, + {"canonical": "UNIDAD DE MEDIDA", "aliases": ["UNIDAD MEDIDA"]}, + {"canonical": "COSTO UNITARIO", "aliases": ["COSTOUNITARIO", "PRECIO UNITARIO", "PRECIOUNITARIO"]}, + {"canonical": "PESO NETO", "aliases": ["PESONETO"]}, + {"canonical": "PESO BRUTO", "aliases": ["PESOBRUTO"]}, {"canonical": "CANTIDAD BULTOS", "aliases": ["CANTIDADBULTOS"]}, - {"canonical": "DESCRIPCION"}, - {"canonical": "PAIS ORIGEN", "aliases": ["PAISORIGEN"]}, - {"canonical": "FRACCION"}, - {"canonical": "ORDEN DE COMPRA", "aliases": ["ORDENCOMPRA"]}, + {"canonical": "CLAVE BULTOS", "aliases": ["CLAVEBULTOS"]}, + {"canonical": "PAIS ORIGEN", "aliases": ["PAISORIGEN", "PAIS"]}, + {"canonical": "FRACCION ARANCELARIA", "aliases": ["FRACCION", "FRACCIONARANCELARIA"]}, + {"canonical": "PREFERENCIA ARANCELARIA", "aliases": ["PREFERENCIA", "PREFERENCIAARANCELARIA"]}, + {"canonical": "SECTOR"}, + {"canonical": "FRACCION AMERICANA", "aliases": ["FRACCIONAMERICANA"]}, + {"canonical": "ORDEN DE COMPRA", "aliases": ["ORDENCOMPRA", "ORDEN COMPRA"]}, + {"canonical": "DESCRIPCION ESPAÑOL", "aliases": ["DESCRIPCIONE", "DESCRIPCION"]}, + {"canonical": "DESCRIPCION INGLES", "aliases": ["DESCRIPCION INGLES", "DESCRIPCIONI"]}, + {"canonical": "MARCA"}, + {"canonical": "MODELO"}, + {"canonical": "ES PARTIDA O SUBPARTIDA", "aliases": ["ES PARTIDA O SUBPARTIDA", "ESSUBPARTIDA"]}, + {"canonical": "LINEA PRINCIPAL", "aliases": ["LINEAPRINCIPAL", "PARTIDA PRINCIPAL"]}, + {"canonical": "NUM. PARTE", "aliases": ["NUMPARTE", "NUMERO PARTE", "NUM PARTE"]}, + {"canonical": "SE PAGO IMPUESTO", "aliases": ["SE PAGO IMPUESTO", "SEPAGOIMPUESTO"]}, + {"canonical": "FORMA DE PAGO", "aliases": ["FORMADEPAGO", "FORMA PAGO"]}, + {"canonical": "METODO DE VALORACION", "aliases": ["METODODEVALORACION", "METODO VALORACION"]}, + {"canonical": "DESCRIPCION EXTRA", "aliases": ["DESCRIPCIONEXTRA"]}, + {"canonical": "INFORMACION ADICIONAL", "aliases": ["INFORMACIONADICIONAL"]}, + {"canonical": "AGREGAR/SUSTITUIR", "aliases": ["AGREGAR SUSTITUIR", "SUSTITUIR"]}, + {"canonical": "TOTAL"}, + {"canonical": "NUMERO ENTRADA", "aliases": ["NUMEROENTRADA", "NUM ENTRADA"]}, + {"canonical": "LOTE"}, + {"canonical": "ID TYPE", "aliases": ["IDTYPE"]}, ], # --- Partidas: Impo Def y Expo - misma estructura --- "imp_def_details": 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 new file mode 100644 index 00000000..c757e316 --- /dev/null +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_impo_temp.py @@ -0,0 +1,535 @@ +""" +Validaciones CSV para Partidas de Importación Temporal. +Paridad Clarion: VALIDA_TODA_PARIMPO_TEM, VALIDA_PARCIAL_PARIMPO_TEM, VALIDACIONES_PARIMPO_TEM. +Estructura: NUMERO FACTURA, LINEA, CLASE, CANTIDAD IMPORTADA, ... hasta ID TYPE (columnas A-AG). +""" +from decimal import Decimal, InvalidOperation +from typing import Dict, Any, Optional, Set, Tuple, List + +# Longitudes máximas Clarion +MAX_LEN_FACTURA = 15 +MAX_LEN_LINEA = 5 +MAX_LEN_CLASE = 9 +MAX_LEN_CANTIDAD_STR = 19 +MAX_LEN_ORDEN_COMPRA = 20 + +PREFERENCIAS_VALIDAS = frozenset({"GENERAL", "TLCS", "PROSEC", "ALADI"}) +SE_PAGO_IMPUESTO_VALIDOS = frozenset({"SI", "NO"}) + +APOSTROFE = "'" + + +def _clip(val: Any) -> str: + if val is None: + return "" + return str(val).strip() + + +def _parse_decimal(val: Any) -> Optional[Decimal]: + if val is None: + return None + s = _clip(val) + if not s: + return None + s = s.replace(",", "") + try: + return Decimal(s) + except (InvalidOperation, ValueError): + return None + + +def _get(row: Dict[str, Any], *keys: str) -> str: + for k in keys: + v = row.get(k) + if v is not None and str(v).strip(): + return _clip(v) + return "" + + +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: + return { + "line": line_num, + "col": "NUMERO FACTURA", + "msg": "Error: (Celda A) La Factura de Importación está vacía y no se pueden hacer las validaciones.", + } + return None + + +def _check_factura_existe( + invoice_number: str, + line_num: int, + invoice_id_by_number: Dict[str, int], +) -> Optional[Dict[str, Any]]: + if invoice_number not in invoice_id_by_number: + return { + "line": line_num, + "col": "NUMERO FACTURA", + "msg": f"Error: (Celda A) La Factura de Importación {invoice_number} no existe en SCAII y no se pueden hacer las validaciones.", + } + return None + + +def _check_factura_no_actualizada( + invoice_number: str, + line_num: int, + invoice_updated_by_number: Dict[str, bool], + rfc_exception_updated: Set[str], +) -> Optional[Dict[str, Any]]: + if invoice_number in rfc_exception_updated: + return None + if invoice_updated_by_number.get(invoice_number, False): + return { + "line": line_num, + "col": "NUMERO FACTURA", + "msg": f"Error: (Celda A) La Factura de Importación: {invoice_number} ya existe y está Actualizada, no se puede hacer cambios a las facturas actualizadas.", + } + return None + + +def _check_linea_si_no_autonumerar(row: Dict[str, Any], line_num: int, autonumerar: bool) -> Optional[Dict[str, Any]]: + if autonumerar: + return None + val = _get(row, "LINEA", "RENGLON", "PARTIDA") + if not val: + return { + "line": line_num, + "col": "LINEA", + "msg": "Error: (Celda B) El campo de la línea de la partida está vacío y no se pueden hacer las validaciones, ya que se tiene la opción de Autonumerar como NO.", + } + return None + + +def _check_levantar_subpartidas_uv( + row: Dict[str, Any], line_num: int, levantar_subpartidas: bool +) -> Optional[Dict[str, Any]]: + if not levantar_subpartidas: + return None + u = _get(row, "ES PARTIDA O SUBPARTIDA", "ESSUBPARTIDA") + if not u: + return { + "line": line_num, + "col": "ES PARTIDA O SUBPARTIDA", + "msg": "Error: (Celda U) El campo del tipo de la partida (partida o subpartida) está vacío y no se pueden hacer las validaciones, ya que se tiene la opción de Levantar Subpartidas como Si.", + } + v = _get(row, "LINEA PRINCIPAL", "LINEAPRINCIPAL", "PARTIDA PRINCIPAL") + if not v: + return { + "line": line_num, + "col": "LINEA PRINCIPAL", + "msg": "Error: (Celda V) El campo de la partida principal está vacío y no se pueden hacer las validaciones, ya que se tiene la opción de Levantar Subpartidas como Si.", + } + return None + + +def _valida_toda_obligatorios( + row: Dict[str, Any], + line_num: int, + levantar_subpartidas: bool, + calcular_costo_en_base_a_total: bool, +) -> Optional[Dict[str, Any]]: + """Obligatorios vacíos: C, D, F (condicional), G, K, M; U, V si LevantarSubpartidas.""" + obligatorios: List[str] = [] + if not _get(row, "CLASE"): + obligatorios.append("(Col.C) Clases") + if not _get(row, "CANTIDAD IMPORTADA", "CANTIDAD"): + obligatorios.append("(Col.D) Cantidad Importada") + es_subpartida = _get(row, "ES PARTIDA O SUBPARTIDA", "ESSUBPARTIDA").upper() == "S" + if not calcular_costo_en_base_a_total and not es_subpartida: + if not _get(row, "COSTO UNITARIO", "COSTOUNITARIO", "PRECIO UNITARIO", "PRECIOUNITARIO"): + obligatorios.append("(Col.F) Costo Unitario") + if not _get(row, "PESO NETO", "PESONETO"): + obligatorios.append("(Col.G) Peso Neto") + if not _get(row, "PAIS ORIGEN", "PAISORIGEN", "PAIS"): + obligatorios.append("(Col.K) País") + if not _get(row, "PREFERENCIA ARANCELARIA", "PREFERENCIA", "PREFERENCIAARANCELARIA"): + obligatorios.append("(Col.M) Preferencia.") + if levantar_subpartidas: + if not _get(row, "ES PARTIDA O SUBPARTIDA", "ESSUBPARTIDA"): + obligatorios.append("(Col.U) EsSubpartida?.") + if not _get(row, "LINEA PRINCIPAL", "LINEAPRINCIPAL", "PARTIDA PRINCIPAL"): + obligatorios.append("(Col.V) Partida Principal.") + if obligatorios: + return { + "line": line_num, + "col": "CLASE", + "msg": f"Existen campos vacíos que son obligatorios: {', '.join(obligatorios)}.", + } + return None + + +def _valida_toda_numericos( + row: Dict[str, Any], + line_num: int, + calcular_costo_en_base_a_total: bool, +) -> Optional[Dict[str, Any]]: + """Costo unitario y peso neto no pueden ser cero (salvo subpartida / costo por total).""" + es_subpartida = _get(row, "ES PARTIDA O SUBPARTIDA", "ESSUBPARTIDA").upper() == "S" + if not calcular_costo_en_base_a_total and not es_subpartida: + costo = _parse_decimal(row.get("COSTO UNITARIO") or row.get("COSTOUNITARIO") or row.get("PRECIO UNITARIO") or row.get("PRECIOUNITARIO")) + if costo is not None and costo == 0: + return { + "line": line_num, + "col": "COSTO UNITARIO", + "msg": f"Error: (Celda F{line_num}) El Costo Unitario no puede ser cero.", + } + peso_neto = _parse_decimal(row.get("PESO NETO") or row.get("PESONETO")) + if peso_neto is not None and peso_neto == 0: + return { + "line": line_num, + "col": "PESO NETO", + "msg": f"Error: (Celda G{line_num}) El Peso Neto no puede ser cero.", + } + return None + + +def _valida_subpartidas_duplicados( + invoice_number: str, + linea: str, + line_num: int, + line_counts: Dict[Tuple[str, str], int], +) -> Optional[Dict[str, Any]]: + key = (invoice_number.strip(), _clip(linea)) + if line_counts.get(key, 0) > 1: + return { + "line": line_num, + "col": "LINEA", + "msg": f"Error: (Celda B{line_num}) El campo de la partida está duplicado entre las partidas.", + } + return None + + +def _valida_subpartida_tiene_principal( + row: Dict[str, Any], + line_num: int, + partidas_principales_en_csv: Set[Tuple[str, str]], + partidas_principales_en_bd: Set[Tuple[str, str]], +) -> Optional[Dict[str, Any]]: + u = _get(row, "ES PARTIDA O SUBPARTIDA", "ESSUBPARTIDA").upper() + v = _get(row, "LINEA PRINCIPAL", "LINEAPRINCIPAL", "PARTIDA PRINCIPAL") + if u != "S" or not v or v == "0": + return None + inv = _get(row, "NUMERO FACTURA", "NUM FACTURA", "FACTURA") + 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: + return None + return { + "line": line_num, + "col": "LINEA PRINCIPAL", + "msg": f"Error: (Celda V{line_num}) La partida principal {v} no existe.", + } + + +def _valida_subpartida_v_no_cero(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]: + u = _get(row, "ES PARTIDA O SUBPARTIDA", "ESSUBPARTIDA").upper() + v = _get(row, "LINEA PRINCIPAL", "LINEAPRINCIPAL", "PARTIDA PRINCIPAL") + if u == "S" and v == "0": + return { + "line": line_num, + "col": "LINEA PRINCIPAL", + "msg": f"Error: (Celda V{line_num}) La SubPartida no tiene asignada una partida principal.", + } + return None + + +def _validaciones_parimpo_tem( + row: Dict[str, Any], + line_num: int, + 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, + validar_decimales_pza: bool, + rfc_exception_num_parte: Optional[Set[str]], + invoice_number: str, +) -> Optional[Dict[str, Any]]: + """VALIDACIONES_PARIMPO_TEM: longitudes, catálogos, reglas de negocio.""" + + def err(col: str, msg: str) -> Dict[str, Any]: + return {"line": line_num, "col": col, "msg": msg} + + clase = _get(row, "CLASE") + um = _get(row, "UNIDAD DE MEDIDA", "UNIDAD MEDIDA") + + # Longitudes + factura = _get(row, "NUMERO FACTURA", "NUM FACTURA", "FACTURA") + if factura and len(factura) > MAX_LEN_FACTURA: + return err("NUMERO FACTURA", f"Error: (Celda A{line_num}) La Factura de Importación: {factura} supera la longitud de caracteres.") + linea = _get(row, "LINEA", "RENGLON", "PARTIDA") + if linea and len(linea) > MAX_LEN_LINEA: + return err("LINEA", f"Error: (Celda B{line_num}) La Línea de Importación: {linea} supera la longitud de caracteres.") + if clase: + if len(clase) > MAX_LEN_CLASE: + return err("CLASE", f"Error: (Celda C{line_num}) La Clase: {clase} supera la longitud de caracteres.") + if clase.upper() not in valid_class_codes: + return err("CLASE", f"Error: (Celda C{line_num}) La Clase: {clase} no existe en el Catálogo de Clases.") + if not um and not class_um_by_code.get(clase.upper()): + return err("UNIDAD DE MEDIDA", f"Error: (Celda E{line_num}) Debido a que esta celda es vacía, se asignará la unidad de medida de la Clase pero también está vacía.") + frac = _get(row, "FRACCION ARANCELARIA", "FRACCION", "FRACCIONARANCELARIA") + if not frac and not class_fraction_by_code.get(clase.upper()): + return err("FRACCION ARANCELARIA", f"Error: (Celda L{line_num}) Debido a que esta celda es vacía, se asignará la fracción de la Clase pero también está vacía.") + if not _get(row, "DESCRIPCION ESPAÑOL", "DESCRIPCIONE", "DESCRIPCION") and not class_desc_es_by_code.get(clase.upper()): + return err("DESCRIPCION ESPAÑOL", f"Error: (Celda Q{line_num}) Debido a que esta celda es vacía, se asignará la Descripción en Español de la Clase pero también está vacía.") + if not _get(row, "DESCRIPCION INGLES", "DESCRIPCIONI") and not class_desc_en_by_code.get(clase.upper()): + return err("DESCRIPCION INGLES", f"Error: (Celda R{line_num}) Debido a que esta celda es vacía, se asignará la Descripción en Inglés de la Clase pero también está vacía.") + + # D: Cantidad importada + cant_str = _get(row, "CANTIDAD IMPORTADA", "CANTIDAD") + if cant_str: + cant = _parse_decimal(cant_str) + if cant is not None and cant <= 0: + return err("CANTIDAD IMPORTADA", f"Error: (Celda D{line_num}) La Cantidad Importada: {cant_str} es cero.") + if len(cant_str) > MAX_LEN_CANTIDAD_STR: + return err("CANTIDAD IMPORTADA", f"Error: (Celda D{line_num}) La Cantidad Importada: {cant_str} supera la cantidad de caracteres permitidos.") + + # E: Unidad de medida en catálogo + if um and um.upper() not in valid_uom_codes: + return err("UNIDAD DE MEDIDA", f"Error: (Celda E{line_num}) La Unidad de Medida: {um} no existe en el Catálogo de Unidades de Medida.") + + # I, J: Bultos condicional + clave_bultos = _get(row, "CLAVE BULTOS", "CLAVEBULTOS") + cant_bultos = row.get("CANTIDAD BULTOS") or row.get("CANTIDADBULTOS") + if clave_bultos: + if clave_bultos not in valid_bulks_codes: + return err("CLAVE BULTOS", f"Error: (Celda J{line_num}) La Clave de Bulto: {clave_bultos} no existe en el Catálogo de Claves de Bultos.") + cant_bultos_val = _parse_decimal(cant_bultos) + if cant_bultos_val is None: + return err("CANTIDAD BULTOS", f"Error: (Celda I{line_num}) La Cantidad de Bultos está vacía y en la Celda J{line_num} se tiene la Clave de Bulto.") + if cant_bultos_val == 0: + return err("CANTIDAD BULTOS", f"Error: (Celda I{line_num}) La Cantidad de Bultos es cero y en la Celda J{line_num} se tiene la Clave de Bulto.") + else: + cant_bultos_val = _parse_decimal(cant_bultos) + if cant_bultos_val is not None and cant_bultos_val > 0: + return err("CANTIDAD BULTOS", f"Error: (Celda I{line_num}) La Cantidad de Bultos es {cant_bultos} y en la Celda J{line_num} no se tiene la Clave de Bulto.") + + # K: País SAAIM3 o americana + pais = _get(row, "PAIS ORIGEN", "PAISORIGEN", "PAIS") + if pais and pais.upper() not in valid_country_keys: + 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() + 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.") + 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.") + 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.") + + # O: Fracción americana + frac_ame = _get(row, "FRACCION AMERICANA", "FRACCIONAMERICANA") + if frac_ame and frac_ame not in valid_fraction_ame: + return err("FRACCION AMERICANA", f"Advertencia: (Celda O{line_num}) La Fracción Americana: {frac_ame} no existe en el Catálogo de Fracciones Americanas.") + + # P: Orden de compra máx 20 + orden = _get(row, "ORDEN DE COMPRA", "ORDENCOMPRA") + if orden and len(orden) > MAX_LEN_ORDEN_COMPRA: + return err("ORDEN DE COMPRA", f"Error: (Celda P{line_num}) La Orden de Compra: {orden} supera la cantidad de caracteres permitidos.") + + # Decimales PZA + if validar_decimales_pza: + um_code = (um or class_um_by_code.get(clase.upper() or "") or "").upper() + if um_code == "PZA" and cant_str: + d = _parse_decimal(cant_str) + if d is not None and d != int(d): + return err("CANTIDAD IMPORTADA", "Error: (Celda D) La Unidad de Medida es PZA, Por lo Tanto no es Válida la Captura de Decimales.") + + # Z: Método de valoración + met_val = _get(row, "METODO DE VALORACION", "METODODEVALORACION", "METODO VALORACION") + if met_val and met_val not in valid_valuation_methods: + return err("METODO DE VALORACION", f"Error: (Celda Z{line_num}) El Método de Valoración Capturado: {met_val} No es Válido.") + + # RFC excepción: W obligatorio + if rfc_exception_num_parte and invoice_number in rfc_exception_num_parte: + num_parte = _get(row, "NUM. PARTE", "NUMPARTE", "NUMERO PARTE", "NUM PARTE") + if not num_parte: + return err("NUM. PARTE", f"Error: (Celda W{line_num}) No está capturado el número de parte.") + + # X: Se pagó impuesto SI/NO + x = _get(row, "SE PAGO IMPUESTO", "SEPAGOIMPUESTO") + if x and x.upper() not in SE_PAGO_IMPUESTO_VALIDOS: + return err("SE PAGO IMPUESTO", f"Error: (Celda X{line_num}) El Valor Capturado para Se Pago Impuesto no es Válido. Capturar SI o NO.") + + # Y: Forma de pago en catálogo + forma_pago = _get(row, "FORMA DE PAGO", "FORMADEPAGO", "FORMA PAGO") + if forma_pago and forma_pago not in valid_payment_methods: + return err("FORMA DE PAGO", f"Error: (Celda Y{line_num}) La Forma de Pago Capturado no es Válido.") + + return None + + +def _warn_apostrofes_num_parte( + row: Dict[str, Any], line_num: int, warnings: Optional[List[Dict[str, Any]]] +) -> None: + val = _get(row, "NUM. PARTE", "NUMPARTE", "NUMERO PARTE", "NUM PARTE") + if val and APOSTROFE in val and warnings is not None: + warnings.append({ + "line": line_num, + "col": "NUM. PARTE", + "msg": f"Advertencia: El Número de Parte: {val} Contiene Apostrofes.", + "warning": True, + }) + + +def validate_row_partidas_impo_temp( + 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]]: + """ + Punto de entrada: valida una fila de CSV de Partidas de Importación Temporal. + Clarion: VALIDA_TODA vs VALIDA_PARCIAL según autonumerar, actualizar y si la partida existe. + """ + 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(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 + 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