diff --git a/backend/api/v1/modules/a76/csv_templates/registry.py b/backend/api/v1/modules/a76/csv_templates/registry.py index af15b48d..b14ae3a6 100644 --- a/backend/api/v1/modules/a76/csv_templates/registry.py +++ b/backend/api/v1/modules/a76/csv_templates/registry.py @@ -59,15 +59,19 @@ def _canonicals_from_columns(cols: Optional[List[Dict]]) -> List[str]: def _build_registry() -> Dict[str, List[str]]: registry: Dict[str, List[str]] = {} - # a76/imports (facturas): imp_temp_header, imp_temp_details, imp_def_*, exp_def_*, cmex_header + # a76/imports (facturas): imp_temp_header, imp_temp_details, imp_def_*, exp_def_*, cmex_*, series for tid in ( "imp_temp_header", "imp_temp_details", + "imp_temp_series", "imp_def_header", "imp_def_details", + "imp_def_series", "exp_def_header", "exp_def_details", "cmex_header", + "cmex_details", + "cmex_series", ): cols = resolve_imports_template(tid) registry[tid] = _canonicals_from_columns(cols) @@ -145,9 +149,13 @@ TEMPLATE_FILENAMES: Dict[str, str] = { "transporters": "EstructuraCatTransportistas.csv", "imp_temp_header": "EstructuraEncFacImpoTemp.csv", "imp_temp_details": "EstructuraParFacImpoTempAF.csv", + "imp_temp_series": "EstructuraSeriesFacImpoTemp.csv", "imp_def_header": "EstructuraEncFacImpoDef.csv", "imp_def_details": "EstructuraParFacImpoDefAF.csv", + "imp_def_series": "EstructuraSeriesFacImpoDef.csv", "cmex_header": "EstructuraEncFacComprasMex.csv", + "cmex_details": "EstructuraParFacComprasMex.csv", + "cmex_series": "EstructuraSeriesFacComprasMex.csv", "exp_def_header": "EstructuraEncFacExpoCamReg.csv", "exp_def_details": "EstructuraParExpoCamReg.csv", } 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 7971749c..b877c1bc 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py @@ -393,6 +393,167 @@ def scan_file(self, job_id: str, model_target: str, config: str = None): logger.exception("Series importación definitiva scan failed: %s", e) return {"status": "failed", "error": str(e)} + # --- Series Compras Mexicanas: misma lógica que Impo Def, facturas MEX --- + if model_target == "invoice_series" and template_id == "cmex_series": + try: + from api.v1.modules.a76.invoices.models import InvoiceHeader + from api.v1.modules.a76.items.models import LineItem + from api.v1.modules.a76.items.series.models import Serie + from api.v1.modules.a76.items.line_quantities.models import LineQuantity + from .validators.series_impo_def import validate_row_series_impo_def + + actualizar = meta.get("actualizar", False) + autonumerar = meta.get("autonumerar", True) + validar_series_exception = meta.get("validar_series", False) + _fc = parse_footer_config(meta.get("footer_config")) + if _fc: + if "actualizar" in _fc: + actualizar = bool(_fc["actualizar"]) + elif _fc.get("mode") == "update": + actualizar = True + elif _fc.get("mode") == "replace": + actualizar = False + if "autonumerar" in _fc: + autonumerar = bool(_fc["autonumerar"]) + else: + as_val = _fc.get("autonumber_series", "true") + autonumerar = str(as_val).lower() in ("true", "1", "si", "sí", "yes") + if "validar_series" in _fc: + validar_series_exception = bool(_fc["validar_series"]) + + with CoreSessionLocal() as session: + q = ( + 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 == "MEX", + ) + ) + rows_inv = q.all() + invoice_id_by_number: Dict[str, int] = {} + invoice_updated_by_number: Dict[str, bool] = {} + for num, iid, is_upd in rows_inv: + if num: + invoice_id_by_number[str(num).strip()] = iid + invoice_updated_by_number[str(num).strip()] = bool(is_upd) + + partida_max_series: Dict[Tuple[str, str], int] = {} + q_qty = ( + session.query( + InvoiceHeader.invoice_number, + LineItem.line_number, + LineQuantity.quantity, + ) + .join(LineItem, LineItem.invoice_id == InvoiceHeader.id) + .outerjoin(LineQuantity, LineQuantity.item_line_id == LineItem.id) + .filter( + InvoiceHeader.tenant_id == tenant_id, + InvoiceHeader.company_id == company_id, + InvoiceHeader.operation_type == "imp", + InvoiceHeader.invoice_type == "MEX", + ) + ) + for num, ln, qty in q_qty.all(): + if num is not None and ln is not None: + key = (str(num).strip(), str(ln).strip()) + if qty is not None: + partida_max_series[key] = int(qty) if qty else 0 + else: + partida_max_series[key] = 0 + + existing_series_keys: Set[Tuple[str, str, str]] = set() + existing_series_data: Dict[Tuple[str, str, str], Dict[str, Any]] = {} + if actualizar and not autonumerar: + q_ser = ( + session.query( + InvoiceHeader.invoice_number, + LineItem.line_number, + Serie.row, + Serie.serial_numbers, + Serie.model, + Serie.sub_model, + Serie.number_id, + ) + .join(LineItem, LineItem.invoice_id == InvoiceHeader.id) + .join(Serie, Serie.line_item_id == LineItem.id) + .filter( + InvoiceHeader.tenant_id == tenant_id, + InvoiceHeader.company_id == company_id, + InvoiceHeader.operation_type == "imp", + InvoiceHeader.invoice_type == "MEX", + ) + ) + for num, ln, rw, sn, md, sm, nid in q_ser.all(): + if num is not None: + k = (str(num).strip(), str(ln).strip(), str(rw).strip()) + existing_series_keys.add(k) + existing_series_data.setdefault(k, { + "serial_numbers": sn or "", + "model": md or "", + "sub_model": sm or "", + "number_id": nid or "", + }) + + csv_series_count_so_far: Dict[Tuple[str, str], int] = {} + + with open(file_path, "r", encoding="utf-8-sig") as f_in, open(error_path, "w", encoding="utf-8") as f_err: + 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) + errors_detail = [] + error_lines_list: List[int] = [] + for i, row in enumerate(reader, start=1): + if i % 1000 == 0: + self.update_state(state="PROGRESS", meta={"current": i, "total": total_rows, "errors": len(error_lines_list)}) + row_norm = row_from_template(row, "cmex_series", normalize_header) + warnings_list: List[Dict[str, Any]] = [] + err = validate_row_series_impo_def( + row_norm, + i, + actualizar=actualizar, + autonumerar=autonumerar, + validar_series_exception=validar_series_exception, + invoice_id_by_number=invoice_id_by_number, + invoice_updated_by_number=invoice_updated_by_number, + partida_max_series=partida_max_series, + csv_series_count_so_far=csv_series_count_so_far, + existing_series_keys=existing_series_keys, + existing_series_data=existing_series_data, + warnings=warnings_list, + catalog_label="Compras Mexicanas", + ) + if err and not err.get("warning"): + error_count += 1 + error_lines_list.append(err["line"]) + f_err.write(json.dumps(err) + "\n") + if len(errors_detail) < 500: + errors_detail.append({"line": err["line"], "col": err.get("col", ""), "msg": err.get("msg", "")}) + else: + inv_num = (row_norm.get("NUMERO FACTURA") or row_norm.get("NUM FACTURA") or "").strip() + line_fac = (row_norm.get("LINEA FACTURA") or row_norm.get("LINEA") or row_norm.get("PARTIDA") or "").strip() + if inv_num and line_fac: + key_csv = (inv_num, line_fac) + csv_series_count_so_far[key_csv] = csv_series_count_so_far.get(key_csv, 0) + 1 + for w in warnings_list: + if len(errors_detail) < 500: + errors_detail.append({"line": w["line"], "col": w.get("col", ""), "msg": w.get("msg", ""), "warning": True}) + 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, total_rows_in_file=total_rows + ) + except Exception as e: + logger.exception("Series Compras Mexicanas scan failed: %s", e) + return {"status": "failed", "error": str(e)} + # --- Series de Importación Temporal: flujo específico (Clarion VALIDA_TODA / VALIDA_PARCIAL) --- if model_target == "invoice_series": try: @@ -1038,6 +1199,265 @@ def scan_file(self, job_id: str, model_target: str, config: str = None): logger.exception("Partidas importación definitiva scan failed: %s", e) return {"status": "failed", "error": str(e)} + # --- Partidas Compras Mexicanas: misma lógica que Impo Def, facturas MEX --- + if model_target == "invoice_details" and template_id == "cmex_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 + + _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 == "MEX", + ) + ) + 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 == "MEX", + ) + ) + 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 == "MEX", + 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_cmex(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, "cmex_details", normalize_header) + inv = _get_row_cmex(row_norm, "NUMERO FACTURA", "NUM FACTURA", "FACTURA") + linea = _get_row_cmex(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_cmex(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, "cmex_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, + catalog_label="Compras Mexicanas", + ) + 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 Compras Mexicanas 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: @@ -2525,7 +2945,7 @@ def insert_valid_rows(self, job_id: str, model_target: str): # Si el upload fue de series (template_id imp_temp_series o imp_def_series), usar flujo series aunque model_target venga mal use_series_flow = ( model_target == "invoice_series" - or meta.get("template_id") in ("imp_temp_series", "imp_def_series") + or meta.get("template_id") in ("imp_temp_series", "imp_def_series", "cmex_series") ) _footer_for_series = parse_footer_config(meta.get("footer_config")) or {} @@ -2534,11 +2954,12 @@ def insert_valid_rows(self, job_id: str, model_target: str): use_series_flow and ( meta.get("template_id") == "imp_def_series" + or meta.get("template_id") == "cmex_series" or _inv_type_series in ("DEF", "MATDE", "EXDEF") ) ) - # --- Series de Importación Definitiva: commit (INSERT/UPDATE item_line_series para facturas DEF) --- + # --- Series de Importación Definitiva: commit (INSERT/UPDATE item_line_series para facturas DEF o MEX) --- if use_def_series_commit: try: from api.v1.modules.a76.invoices.models import InvoiceHeader @@ -2550,7 +2971,12 @@ def insert_valid_rows(self, job_id: str, model_target: str): row_to_series_normalized_def, ) - DEF_INVOICE_TYPES = ("DEF", "MATDE", "EXDEF") + _series_template_id = meta.get("template_id") + is_cmex_series = _series_template_id == "cmex_series" + SERIES_INV_TYPES = ("MEX",) if is_cmex_series else ("DEF", "MATDE", "EXDEF") + series_row_template_id = "cmex_series" if is_cmex_series else "imp_def_series" + + DEF_INVOICE_TYPES = ("DEF", "MATDE", "EXDEF") # keep for any legacy reference actualizar = meta.get("actualizar", False) autonumerar = meta.get("autonumerar", True) validar_series_exception = meta.get("validar_series", False) @@ -2577,7 +3003,7 @@ def insert_valid_rows(self, job_id: str, model_target: str): InvoiceHeader.tenant_id == tenant_id, InvoiceHeader.company_id == company_id, InvoiceHeader.operation_type == "imp", - InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES), + InvoiceHeader.invoice_type.in_(SERIES_INV_TYPES), ) ) rows_inv = q.all() @@ -2601,7 +3027,7 @@ def insert_valid_rows(self, job_id: str, model_target: str): InvoiceHeader.tenant_id == tenant_id, InvoiceHeader.company_id == company_id, InvoiceHeader.operation_type == "imp", - InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES), + InvoiceHeader.invoice_type.in_(SERIES_INV_TYPES), ) ) for num, ln, qty in q_qty.all(): @@ -2628,7 +3054,7 @@ def insert_valid_rows(self, job_id: str, model_target: str): InvoiceHeader.tenant_id == tenant_id, InvoiceHeader.company_id == company_id, InvoiceHeader.operation_type == "imp", - InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES), + InvoiceHeader.invoice_type.in_(SERIES_INV_TYPES), ) ) for num, ln, rw, sn, md, sm, nid in q_ser.all(): @@ -2660,7 +3086,7 @@ def insert_valid_rows(self, job_id: str, model_target: str): for i, row in enumerate(reader, start=1): if i in error_lines: continue - row_norm = row_from_template(row, "imp_def_series", normalize_header) + row_norm = row_from_template(row, series_row_template_id, normalize_header) err = validate_row_series_impo_def( row_norm, i, @@ -2674,6 +3100,7 @@ def insert_valid_rows(self, job_id: str, model_target: str): existing_series_keys=existing_series_keys, existing_series_data=existing_series_data, warnings=None, + catalog_label="Compras Mexicanas" if is_cmex_series else "Importación Definitiva", ) if err: skipped_invalid += 1 @@ -3089,7 +3516,11 @@ def insert_valid_rows(self, job_id: str, model_target: str): inv_type_value = "MEX" 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 _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}") headers_to_insert = [] 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 ec822a38..8f7981c7 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 @@ -186,6 +186,10 @@ def _resolve_template_columns(template_id: str) -> Optional[List[Dict[str, Any]] return TEMPLATE_COLUMNS.get("imp_temp_header") if template_id in ("imp_def_details", "exp_def_details"): return TEMPLATE_COLUMNS.get("imp_temp_details") + if template_id == "cmex_details": + return TEMPLATE_COLUMNS.get("imp_temp_details") + if template_id == "cmex_series": + return TEMPLATE_COLUMNS.get("imp_def_series") return None 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 index 0c58884d..06f086a8 100644 --- 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 @@ -28,15 +28,16 @@ def _check_factura_existe_def( invoice_number: str, line_num: int, invoice_id_by_number: Dict[str, int], + catalog_label: str = "Importación Definitiva", ) -> Optional[Dict[str, Any]]: - """Misma lógica que _check_factura_existe; mensaje específico Importación Definitiva (Clarion).""" + """Misma lógica que _check_factura_existe; mensaje específico según catalog_label (DEF o Compras Mexicanas).""" 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. " + f"no existe en el catálogo de {catalog_label} y no se pueden hacer las validaciones. " ), } return None @@ -73,6 +74,7 @@ def validate_row_partidas_impo_def( rfc_exception_num_parte: Optional[Set[str]], valid_part_numbers: Optional[Set[str]], warnings: Optional[List[Dict[str, Any]]] = None, + catalog_label: str = "Importación Definitiva", ) -> Optional[Dict[str, Any]]: """ Valida una fila de CSV de Partidas de Importación Definitiva. @@ -87,7 +89,7 @@ def validate_row_partidas_impo_def( 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) + err = _check_factura_existe_def(invoice_number, line_num, invoice_id_by_number, catalog_label) if err: return err @@ -139,13 +141,13 @@ def validate_row_partidas_impo_def( ) if err: return err - # Importación Definitiva: NUM. PARTE es obligatorio en todas las partidas (el insert lo exige). + # Importación Definitiva / Compras Mexicanas: 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).", + "msg": f"NUM. PARTE: Requerido (obligatorio para partidas de {catalog_label}).", } err = _valida_toda_numericos(row, line_num, calcular_costo_en_base_a_total) if err: diff --git a/backend/api/v1/modules/a76/layouts_csv/facturas/validators/series_impo_def.py b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/series_impo_def.py index d1307a01..d00d228f 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/validators/series_impo_def.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/series_impo_def.py @@ -23,19 +23,20 @@ def _check_factura_existe_def( invoice_number: str, line_num: int, invoice_id_by_number: Dict[str, int], + catalog_label: str = "Importación Definitiva", ) -> Optional[Dict[str, Any]]: - """Factura debe existir en BD (Importación Definitiva: DEF/MATDE/EXDEF).""" + """Factura debe existir en BD (Importación Definitiva o Compras Mexicanas según catalog_label).""" 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 SCAII y no se pueden hacer las validaciones. " + f"no existe en SCAII y no se pueden hacer las validaciones. " ), "solution": ( f"Capturar en la Celda A{line_num} un número de Factura existente " - "al cual desee agregar o actualizar series" + f"al cual desee agregar o actualizar series" ), "identifier": "FAC_IMPO_DEF", "fields": invoice_number, @@ -47,14 +48,15 @@ def _check_factura_no_actualizada_def( invoice_number: str, line_num: int, invoice_updated_by_number: Dict[str, bool], + catalog_label: str = "Importación Definitiva", ) -> Optional[Dict[str, Any]]: - """Si factura ya actualizada (Estatus AC) no se pueden hacer cambios. Mensaje DEF.""" + """Si factura ya actualizada (Estatus AC) no se pueden hacer cambios. Mensaje según catalog_label.""" if invoice_updated_by_number.get(invoice_number, False): return { "line": line_num, "col": "NUMERO FACTURA", "msg": ( - f"Error: (Celda A{line_num}) La Factura de Importación: {invoice_number} " + f"Error: (Celda A{line_num}) La Factura de {catalog_label}: {invoice_number} " "ya existe y está Actualizada, no se puede hacer cambios a las facturas actualizadas." ), "solution": ( @@ -224,6 +226,7 @@ def validate_row_series_impo_def( existing_series_keys: Set[Tuple[str, str, str]], existing_series_data: Optional[Dict[Tuple[str, str, str], Dict[str, Any]]], warnings: Optional[List[Dict[str, Any]]] = None, + catalog_label: str = "Importación Definitiva", ) -> Optional[Dict[str, Any]]: """ Punto de entrada: valida una fila de CSV de Series de Importación Definitiva. @@ -243,11 +246,11 @@ def validate_row_series_impo_def( 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) + err = _check_factura_existe_def(invoice_number, line_num, invoice_id_by_number, catalog_label) if err: return err - err = _check_factura_no_actualizada_def(invoice_number, line_num, invoice_updated_by_number) + err = _check_factura_no_actualizada_def(invoice_number, line_num, invoice_updated_by_number, catalog_label) if err: return err diff --git a/frontend/src/lib/config/csv-upload.ts b/frontend/src/lib/config/csv-upload.ts index b19f5af7..076d6279 100644 --- a/frontend/src/lib/config/csv-upload.ts +++ b/frontend/src/lib/config/csv-upload.ts @@ -403,15 +403,17 @@ export const importacionConfig: CsvUploadItem[] = [ icon: Package, group: 'Compras Mex.', modelTarget: 'invoice_details', - disabled: true, + templateId: 'cmex_details', + layoutModule: 'layouts_csv/facturas' }, { id: 'comp_mex_series', title: 'Series', icon: Hash, group: 'Compras Mex.', - modelTarget: 'InvoiceSeries', - disabled: true, + modelTarget: 'invoice_series', + templateId: 'cmex_series', + layoutModule: 'layouts_csv/facturas' }, ];