diff --git a/backend/api/v1/modules/a76/layouts_csv/exportacion/routes.py b/backend/api/v1/modules/a76/layouts_csv/exportacion/routes.py index a357f1f0..87c9bd13 100644 --- a/backend/api/v1/modules/a76/layouts_csv/exportacion/routes.py +++ b/backend/api/v1/modules/a76/layouts_csv/exportacion/routes.py @@ -62,7 +62,7 @@ async def upload_import_file( "user_id": current_user.get("id"), "footer_config": footer_config, "operation_type": operation_type or "exp", - "template_id": template_id or ("exp_def_header" if model_target == "invoice_header" else "exp_def_details"), + "template_id": template_id or ("exp_def_header" if model_target == "invoice_header" else "exp_def_partidas"), } try: diff --git a/backend/api/v1/modules/a76/layouts_csv/exportacion/tasks.py b/backend/api/v1/modules/a76/layouts_csv/exportacion/tasks.py index 96c41870..7a6800e5 100644 --- a/backend/api/v1/modules/a76/layouts_csv/exportacion/tasks.py +++ b/backend/api/v1/modules/a76/layouts_csv/exportacion/tasks.py @@ -38,7 +38,7 @@ def _norm_row(row: Dict[str, Any], template_id: str) -> Dict[str, Any]: def scan_file(self, job_id: str, model_target: str, config: str = None): """ Para invoice_header: delega en facturas._do_scan_file con storage "exp" (validaciones FK, encabezados_expo). - Para invoice_details: scan sin validaciones (stub). + Para invoice_details: delega en facturas._do_scan_file con template exp_def_partidas (validaciones partidas expo). """ logger.info("Exportación import: starting scan for job %s target %s", job_id, model_target) @@ -46,45 +46,37 @@ def scan_file(self, job_id: str, model_target: str, config: str = None): from api.v1.modules.a76.layouts_csv.facturas.tasks import _do_scan_file return _do_scan_file(job_id, "invoice_header", config, job_type_override="exp") - # invoice_details: stub sin validaciones + if model_target == "invoice_details": + from api.v1.modules.a76.layouts_csv.facturas.tasks import _do_scan_file + return _do_scan_file(job_id, "invoice_details", config, job_type_override="exp") + + # Fallback (e.g. unknown model_target) file_path = _ensure_file(job_id) if not file_path: return {"status": "failed", "error": "Archivo no encontrado (expirado o no subido). Sube de nuevo."} if os.path.getsize(file_path) == 0: return {"status": "failed", "error": "El archivo está vacío."} _ensure_meta(job_id, file_path) - try: common_meta.require_tenant_context(file_path) except ValueError as e: return {"status": "failed", "error": str(e)} - meta = common_meta.load_meta(file_path) or {} - template_id = meta.get("template_id") or ( - "exp_def_header" if model_target == "invoice_header" else "exp_def_details" - ) - - total_rows = 0 - processed_rows = 0 - + template_id = meta.get("template_id") or "exp_def_details" try: total_rows = common_csv_reader.count_csv_rows(file_path, has_header=True) except Exception as e: return {"status": "failed", "error": str(e)} - - def on_progress(current: int, total: int) -> None: - self.update_state(state="PROGRESS", meta={"current": current, "total": total}) - + processed_rows = 0 try: for i, row in common_csv_reader.iter_csv_rows(file_path, fieldnames=None): if i % 500 == 0: - on_progress(i, total_rows) + self.update_state(state="PROGRESS", meta={"current": i, "total": total_rows}) _norm_row(row, template_id) processed_rows += 1 except Exception as e: logger.error("Exportación import scan failed: %s", e) return {"status": "failed", "error": str(e)} - return common_responses.scan_result(job_id, processed_rows, 0, []) @@ -92,7 +84,7 @@ def scan_file(self, job_id: str, model_target: str, config: str = None): def insert_valid_rows(self, job_id: str, model_target: str): """ Commit: para invoice_header delega en facturas (inserción real en BD con storage "exp"). - Para invoice_details mantiene stub (sin inserción). + Para invoice_details delega en facturas (inserción partidas expo cuando esté implementada; mientras tanto mismo flujo). """ logger.info("Exportación import: starting commit for job %s target %s", job_id, model_target) @@ -100,7 +92,11 @@ def insert_valid_rows(self, job_id: str, model_target: str): from api.v1.modules.a76.layouts_csv.facturas.tasks import _do_insert_valid_rows return _do_insert_valid_rows(job_id, "invoice_header", job_type_override="exp") - # invoice_details: stub (sin inserción en BD) + if model_target == "invoice_details": + from api.v1.modules.a76.layouts_csv.facturas.tasks import _do_insert_valid_rows + return _do_insert_valid_rows(job_id, "invoice_details", job_type_override="exp") + + # Fallback: stub sin inserción file_path = _ensure_file(job_id) if not file_path: alt_path = common_storage.file_path_for_job(JOB_TYPE, job_id) @@ -121,7 +117,7 @@ def insert_valid_rows(self, job_id: str, model_target: str): error_lines = common_storage.get_error_lines(JOB_TYPE, job_id, error_path) template_id = meta.get("template_id") or ( - "exp_def_header" if model_target == "invoice_header" else "exp_def_details" + "exp_def_header" if model_target == "invoice_header" else "exp_def_partidas" ) try: diff --git a/backend/api/v1/modules/a76/layouts_csv/exportacion/template_config.py b/backend/api/v1/modules/a76/layouts_csv/exportacion/template_config.py index 60ce5cb5..59247c4a 100644 --- a/backend/api/v1/modules/a76/layouts_csv/exportacion/template_config.py +++ b/backend/api/v1/modules/a76/layouts_csv/exportacion/template_config.py @@ -57,6 +57,36 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { {"canonical": "FRACCION"}, {"canonical": "ORDEN DE COMPRA", "aliases": ["ORDENCOMPRA"]}, ], + # Partidas exportación definitiva (Clarion EstructuraParExpoCamReg: A–V) + "exp_def_partidas": [ + {"canonical": "NUMERO FACTURA EXPO", "aliases": ["NUMERO FACTURA EXPO.", "NUM FACTURA EXPO", "FACTURA EXPO"]}, + {"canonical": "LINEA EXPO", "aliases": ["LINEA EXPO.", "RENGLON EXPO"]}, + {"canonical": "TIPO DE IMPO", "aliases": ["TIPO DE IMPO.", "TIPO IMPO", "PROCEDENCIA"]}, + {"canonical": "FACTURA IMPO", "aliases": ["FACTURA IMPO.", "FACTURA IMPORTACION"]}, + {"canonical": "LINEA IMPO", "aliases": ["LINEA IMPO.", "LINEA IMPORTACION"]}, + {"canonical": "GENERA DESCARGA", "aliases": ["GENERA DESCARGA?", "DESCARGA"]}, + {"canonical": "CANTIDAD EXPORTADA/DESCARGAR", "aliases": ["CANTIDAD EXPORTADA", "CANTIDAD EXPORTADA/DESCARGAR", "CANTIDAD"]}, + {"canonical": "UNIDAD DE MEDIDA", "aliases": ["U.M.", "UNIDAD MEDIDA"]}, + {"canonical": "COSTO UNITARIO", "aliases": ["COSTOUNITARIO"]}, + {"canonical": "PESO NETO", "aliases": ["PESONETO"]}, + {"canonical": "PESO BRUTO", "aliases": ["PESOBRUTO"]}, + {"canonical": "SE PAGO IMPUESTO", "aliases": ["SE PAGO IMPUESTO? (SI o NO)", "SEPAGOIMPUESTO"]}, + {"canonical": "FORMA DE PAGO", "aliases": ["FORMADEPAGO", "FORMA PAGO"]}, + {"canonical": "DESCRIPCION EXTRA", "aliases": ["DESCRIPCION EXTRA", "DESCRIPCIONEXTRA"]}, + {"canonical": "INFORMACION ADICIONAL", "aliases": ["INFORMACION ADICIONAL", "INFORMACIONADICIONAL"]}, + {"canonical": "AGREGAR(A)/SUSTITUIR(S)", "aliases": ["AGREGAR(A)/SUSTITUIR(S)", "AGREGAR/SUSTITUIR", "SUSTITUIR"]}, + {"canonical": "LOTE"}, + {"canonical": "NUMERO ENTRADA", "aliases": ["NUMERO ENTRADA", "NUM ENTRADA"]}, + {"canonical": "ES PARTIDA/SUBPARTIDA", "aliases": ["ES PARTIDA/SUBPARTIDA", "ESSUBPARTIDA", "ES PARTIDA O SUBPARTIDA"]}, + {"canonical": "LINEA PRINCIPAL", "aliases": ["LINEAPRINCIPAL", "PARTIDA PRINCIPAL"]}, + {"canonical": "FRACCION AMERICANA", "aliases": ["FRACCION AMERICANA", "FRACCIONAMERICANA"]}, + {"canonical": "FRACCION ARANCELARIA", "aliases": ["FRACCION ARANCELARIA", "FRACCIONARANCELARIA"]}, + {"canonical": "CANTIDAD BULTOS", "aliases": ["CANTIDADBULTOS"]}, + {"canonical": "CLAVE BULTOS", "aliases": ["CLAVEBULTOS"]}, + {"canonical": "PAIS ORIGEN", "aliases": ["PAISORIGEN", "PAIS"]}, + {"canonical": "NUM. PARTE", "aliases": ["NUMPARTE", "NUMERO PARTE", "NUM PARTE"]}, + {"canonical": "ORDEN DE COMPRA", "aliases": ["ORDENCOMPRA", "ORDEN DE VENTA"]}, + ], } 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 429a310b..b2c0aea3 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py @@ -222,9 +222,11 @@ def _do_scan_file(job_id: str, model_target: str, config: Optional[str] = None, "imp_temp_header" if model_target == "invoice_header" else "imp_temp_details" if model_target == "invoice_details" else "imp_temp_series" ) - # Cuando el scan viene de Exportación (job_type_override "exp"), forzar exp_def_header para que corran las validaciones FK en el escaneo + # Cuando el scan viene de Exportación (job_type_override "exp"), forzar exp_def_header o exp_def_partidas if job_type_override == "exp" and model_target == "invoice_header": template_id = "exp_def_header" + if job_type_override == "exp" and model_target == "invoice_details": + template_id = "exp_def_partidas" inv_type_value = normalize_public_code(footer_config.get("invoice_type") or meta.get("invoice_type") or "TEM") if not inv_type_value: inv_type_value = "TEM" @@ -1202,6 +1204,269 @@ def _do_scan_file(job_id: str, model_target: str, config: Optional[str] = None, logger.exception("Partidas importación definitiva scan failed: %s", e) return {"status": "failed", "error": str(e)} + # --- Partidas Exportación Definitiva: Clarion VALIDA_TODA_PAR_EXPO / VALIDA_PARCIAL_PAR_EXPO / VALIDACIONES_PAR_EXPO --- + if model_target == "invoice_details" and template_id == "exp_def_partidas": + try: + from api.v1.modules.a76.invoices.models import InvoiceHeader + from api.v1.modules.a76.items.models import LineItem + 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.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_expo import validate_row_partidas_expo + + _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) + 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 "validar_decimales_pza" in _fc: + validar_decimales_pza = bool(_fc["validar_decimales_pza"]) + + RFC_EXCEPTION_UPDATED = set() + RFC_EXCEPTION_EGM = {"EGM0303257J1"} + + with CoreSessionLocal() as session: + company = session.query(Company).filter(Company.id == company_id).first() + company_rfc = (company.rfc or "").strip().upper() if company else "" + if company_rfc in RFC_EXCEPTION_EGM: + rfc_exception_egm = True + else: + rfc_exception_egm = False + + q_inv_expo = ( + session.query(InvoiceHeader.invoice_number, InvoiceHeader.id, InvoiceHeader.is_updated) + .filter( + InvoiceHeader.tenant_id == tenant_id, + InvoiceHeader.company_id == company_id, + InvoiceHeader.operation_type == "exp", + ) + ) + invoice_id_by_number: Dict[str, int] = {} + invoice_updated_by_number: Dict[str, bool] = {} + for num, iid, is_upd in q_inv_expo.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_expo = ( + 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 == "exp", + ) + ) + for num, ln in q_li_expo.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 == "exp", + 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 + + factura_impo_tem_by_number: Dict[str, int] = {} + q_tem = session.query(InvoiceHeader.invoice_number, InvoiceHeader.id).filter( + InvoiceHeader.tenant_id == tenant_id, + InvoiceHeader.company_id == company_id, + InvoiceHeader.operation_type == "imp", + InvoiceHeader.invoice_type == "TEM", + ) + for num, iid in q_tem.all(): + if num: + factura_impo_tem_by_number[str(num).strip()] = iid + + factura_impo_def_by_number: Dict[str, int] = {} + DEF_INVOICE_TYPES = ("DEF", "MATDE", "EXDEF") + q_def = session.query(InvoiceHeader.invoice_number, 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, iid in q_def.all(): + if num: + factura_impo_def_by_number[str(num).strip()] = iid + + line_exists_tem: Set[Tuple[int, str]] = set() + q_li_tem = ( + session.query(LineItem.invoice_id, LineItem.line_number) + .join(InvoiceHeader, InvoiceHeader.id == LineItem.invoice_id) + .filter( + InvoiceHeader.tenant_id == tenant_id, + InvoiceHeader.company_id == company_id, + InvoiceHeader.operation_type == "imp", + InvoiceHeader.invoice_type == "TEM", + ) + ) + for inv_id, ln in q_li_tem.all(): + if inv_id is not None and ln is not None: + line_exists_tem.add((inv_id, str(ln).strip())) + + line_exists_def: Set[Tuple[int, str]] = set() + q_li_def = ( + session.query(LineItem.invoice_id, LineItem.line_number) + .join(InvoiceHeader, InvoiceHeader.id == LineItem.invoice_id) + .filter( + InvoiceHeader.tenant_id == tenant_id, + InvoiceHeader.company_id == company_id, + InvoiceHeader.operation_type == "imp", + InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES), + ) + ) + for inv_id, ln in q_li_def.all(): + if inv_id is not None and ln is not None: + line_exists_def.add((inv_id, str(ln).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_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_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()) + + 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()) + + invoice_numbers_from_csv = 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) + + for row in rows_list: + row_norm = row_from_template(row, "exp_def_partidas", normalize_header) + inv = (row_norm.get("NUMERO FACTURA EXPO") or row_norm.get("NUMERO FACTURA EXPO.") or row_norm.get("FACTURA EXPO") or "").strip() + if inv: + invoice_numbers_from_csv.add(inv) + if company_rfc in RFC_EXCEPTION_EGM: + rfc_exception_updated = invoice_numbers_from_csv + else: + rfc_exception_updated = set() + + line_counts_csv: Dict[Tuple[str, str], int] = {} + partidas_principales_csv: Set[Tuple[str, str]] = set() + + def _get_row_expo(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, "exp_def_partidas", normalize_header) + inv = _get_row_expo(row_norm, "NUMERO FACTURA EXPO", "NUMERO FACTURA EXPO.", "FACTURA EXPO") + linea = _get_row_expo(row_norm, "LINEA EXPO", "LINEA EXPO.", "RENGLON EXPO") + if inv and linea: + key = (inv, linea) + line_counts_csv[key] = line_counts_csv.get(key, 0) + 1 + u = _get_row_expo(row_norm, "ES PARTIDA/SUBPARTIDA", "ESSUBPARTIDA", "ES PARTIDA O SUBPARTIDA").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, "exp_def_partidas", normalize_header) + err = validate_row_partidas_expo( + row_norm, + i, + autonumerar=autonumerar, + actualizar=actualizar, + levantar_subpartidas=levantar_subpartidas, + 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_uom_codes=valid_uom_codes, + valid_bulks_codes=valid_bulks_codes, + valid_payment_methods=valid_payment_methods, + valid_fraction_ame=valid_fraction_ame, + valid_part_numbers=valid_part_numbers, + factura_impo_tem_by_number=factura_impo_tem_by_number, + factura_impo_def_by_number=factura_impo_def_by_number, + line_exists_tem=line_exists_tem, + line_exists_def=line_exists_def, + rfc_exception_egm=rfc_exception_egm, + validar_decimales_pza=validar_decimales_pza, + ) + 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 + + common_storage.store_error_lines(effective_job_type, job_id, error_lines_list) + return common_responses.scan_result(job_id, processed_rows, error_count, errors_detail) + except Exception as e: + logger.exception("Partidas exportació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: @@ -3308,6 +3573,24 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt error_path = common_storage.error_path_for_job(effective_job_type, job_id) error_lines = common_storage.get_error_lines(effective_job_type, job_id, error_path) + # Partidas exportación definitiva: inserción pendiente (Phase 2); solo cleanup y respuesta + if model_target == "invoice_details" and meta.get("template_id") == "exp_def_partidas": + common_storage.cleanup_import_job( + effective_job_type, job_id, + file_path=file_path, + error_path=error_path, + meta_path=meta_path, + ) + return { + "status": "finished", + "inserted": 0, + "skipped_invalid": len(error_lines), + "skipped_missing_fk": 0, + "skipped_duplicate": 0, + "skipped_details": [], + "message": "Validación de partidas de exportación completada. Inserción en BD pendiente de implementación (Phase 2).", + } + # 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" 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 e4c01912..effa0d2c 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 @@ -188,6 +188,36 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { # --- Partidas: Impo Def y Expo - misma estructura --- "imp_def_details": None, "exp_def_details": None, + # --- Partidas exportación definitiva (Clarion EstructuraParExpoCamReg A–V) --- + "exp_def_partidas": [ + {"canonical": "NUMERO FACTURA EXPO", "aliases": ["NUMERO FACTURA EXPO.", "NUM FACTURA EXPO", "FACTURA EXPO"]}, + {"canonical": "LINEA EXPO", "aliases": ["LINEA EXPO.", "RENGLON EXPO"]}, + {"canonical": "TIPO DE IMPO", "aliases": ["TIPO DE IMPO.", "TIPO IMPO", "PROCEDENCIA"]}, + {"canonical": "FACTURA IMPO", "aliases": ["FACTURA IMPO.", "FACTURA IMPORTACION"]}, + {"canonical": "LINEA IMPO", "aliases": ["LINEA IMPO.", "LINEA IMPORTACION"]}, + {"canonical": "GENERA DESCARGA", "aliases": ["GENERA DESCARGA?", "DESCARGA"]}, + {"canonical": "CANTIDAD EXPORTADA/DESCARGAR", "aliases": ["CANTIDAD EXPORTADA", "CANTIDAD EXPORTADA/DESCARGAR", "CANTIDAD"]}, + {"canonical": "UNIDAD DE MEDIDA", "aliases": ["U.M.", "UNIDAD MEDIDA"]}, + {"canonical": "COSTO UNITARIO", "aliases": ["COSTOUNITARIO"]}, + {"canonical": "PESO NETO", "aliases": ["PESONETO"]}, + {"canonical": "PESO BRUTO", "aliases": ["PESOBRUTO"]}, + {"canonical": "SE PAGO IMPUESTO", "aliases": ["SE PAGO IMPUESTO? (SI o NO)", "SEPAGOIMPUESTO"]}, + {"canonical": "FORMA DE PAGO", "aliases": ["FORMADEPAGO", "FORMA PAGO"]}, + {"canonical": "DESCRIPCION EXTRA", "aliases": ["DESCRIPCION EXTRA", "DESCRIPCIONEXTRA"]}, + {"canonical": "INFORMACION ADICIONAL", "aliases": ["INFORMACION ADICIONAL", "INFORMACIONADICIONAL"]}, + {"canonical": "AGREGAR(A)/SUSTITUIR(S)", "aliases": ["AGREGAR(A)/SUSTITUIR(S)", "AGREGAR/SUSTITUIR", "SUSTITUIR"]}, + {"canonical": "LOTE"}, + {"canonical": "NUMERO ENTRADA", "aliases": ["NUMERO ENTRADA", "NUM ENTRADA"]}, + {"canonical": "ES PARTIDA/SUBPARTIDA", "aliases": ["ES PARTIDA/SUBPARTIDA", "ESSUBPARTIDA", "ES PARTIDA O SUBPARTIDA"]}, + {"canonical": "LINEA PRINCIPAL", "aliases": ["LINEAPRINCIPAL", "PARTIDA PRINCIPAL"]}, + {"canonical": "FRACCION AMERICANA", "aliases": ["FRACCION AMERICANA", "FRACCIONAMERICANA"]}, + {"canonical": "FRACCION ARANCELARIA", "aliases": ["FRACCION ARANCELARIA", "FRACCIONARANCELARIA"]}, + {"canonical": "CANTIDAD BULTOS", "aliases": ["CANTIDADBULTOS"]}, + {"canonical": "CLAVE BULTOS", "aliases": ["CLAVEBULTOS"]}, + {"canonical": "PAIS ORIGEN", "aliases": ["PAISORIGEN", "PAIS"]}, + {"canonical": "NUM. PARTE", "aliases": ["NUMPARTE", "NUMERO PARTE", "NUM PARTE"]}, + {"canonical": "ORDEN DE COMPRA", "aliases": ["ORDENCOMPRA", "ORDEN DE VENTA"]}, + ], # --- Series de Importación Temporal (EstructuraSeriesFacImpoTemp.xls) --- # Clarion: NUMERO FACTURA, LINEA FACTURA, LINEA SERIE, SERIE, MODELO, NUM PARTE, SUB MODELO, NUMERO ID "imp_temp_series": [ @@ -224,6 +254,8 @@ 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 == "exp_def_partidas": + return TEMPLATE_COLUMNS.get("exp_def_partidas") if template_id == "cmex_details": return TEMPLATE_COLUMNS.get("imp_temp_details") if template_id == "cmex_series": 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 d244c132..b5e920b7 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 @@ -11,6 +11,7 @@ from .encabezados_impo_def import ( ) from .encabezados_cmex import validate_row_encabezados_cmex from .encabezados_expo import validate_row_encabezados_expo +from .partidas_expo import validate_row_partidas_expo from .partidas_impo_def import validate_row_partidas_impo_def from .series_impo_def import ( validate_row_series_impo_def, @@ -22,6 +23,7 @@ __all__ = [ "validate_row_encabezados_impo_def", "validate_row_encabezados_cmex", "validate_row_encabezados_expo", + "validate_row_partidas_expo", "validate_row_partidas_impo_def", "validate_row_series_impo_def", "row_to_series_normalized_def", diff --git a/backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_expo.py b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_expo.py new file mode 100644 index 00000000..e198d087 --- /dev/null +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/partidas_expo.py @@ -0,0 +1,558 @@ +""" +Validaciones CSV para Partidas de Exportación Definitiva (y Cambio de Régimen). +Paridad Clarion: VALIDA_TODA_PAR_EXPO, VALIDA_PARCIAL_PAR_EXPO, VALIDACIONES_PAR_EXPO. +Estructura: NUMERO FACTURA EXPO, LINEA EXPO, TIPO DE IMPO, FACTURA IMPO, LINEA IMPO, GENERA DESCARGA, ... +Variante RFC EGM0303257J1: columnas ES PARTIDA/SUBPARTIDA y LINEA PRINCIPAL en T y U. +""" +from decimal import Decimal, InvalidOperation +from typing import Dict, Any, Optional, Set, Tuple, List + +from .partidas_impo_temp import _clip, _get + +# Longitudes Clarion partidas expo +MAX_LEN_FACTURA_EXPO = 15 +MAX_LEN_LINEA_EXPO = 5 +MAX_LEN_TIPO_IMPO = 3 +MAX_LEN_ORDEN_COMPRA = 20 +MAX_LEN_NUM_PARTE = 30 + +TIPO_IMPO_VALIDOS = frozenset({"TEM", "DEF"}) +GENERA_DESCARGA_VALIDOS = frozenset({"SI", "NO"}) +SE_PAGO_IMPUESTO_VALIDOS = frozenset({"SI", "NO"}) + + +def _err( + line_num: int, + col: str, + msg: str, + identifier: str = "ARCHIVO CSV", +) -> Dict[str, Any]: + return {"line": line_num, "col": col, "msg": msg} + + +def _parse_decimal(val: Any) -> Optional[Decimal]: + if val is None: + return None + s = _clip(val) + if not s: + return None + s = str(s).replace(",", "") + try: + return Decimal(s) + except (InvalidOperation, ValueError): + return None + + +def _check_factura_expo_vacia(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]: + val = _get(row, "NUMERO FACTURA EXPO", "NUMERO FACTURA EXPO.", "NUM FACTURA EXPO", "FACTURA EXPO") + if not val: + return _err( + line_num, + "NUMERO FACTURA EXPO", + f"Error: (Celda A{line_num}) La Factura de Exportación está vacía y no se pueden hacer las validaciones. " + "Capturar en la Celda A un número de Factura existente al cual desee agregar o actualizar partidas.", + ) + return None + + +def _check_factura_expo_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 _err( + line_num, + "NUMERO FACTURA EXPO", + f"Error: (Celda A{line_num}) La Factura de Exportación {invoice_number} no existe en SCAII y no se pueden hacer las validaciones. " + "Capturar en la Celda A un número de Factura existente al cual desee agregar o actualizar partidas.", + ) + return None + + +def _check_factura_expo_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 _err( + line_num, + "NUMERO FACTURA EXPO", + f"Error: (Celda A{line_num}) La Factura de Exportación: {invoice_number} ya existe y está Actualizada, no se puede hacer cambios a las facturas actualizadas. " + "Capturar otro número de Factura de Exportación o Desactualizar la factura.", + ) + return None + + +def _check_linea_expo_si_no_autonumerar( + row: Dict[str, Any], line_num: int, autonumerar: bool +) -> Optional[Dict[str, Any]]: + if autonumerar: + return None + val = _get(row, "LINEA EXPO", "LINEA EXPO.", "RENGLON EXPO") + if not val: + return _err( + line_num, + "LINEA EXPO", + f"Error: (Celda B{line_num}) 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. " + "Capturar en la Celda B la línea de la partida al cual desee agregar o actualizar información.", + ) + return None + + +def _check_levantar_subpartidas_expo( + row: Dict[str, Any], + line_num: int, + levantar_subpartidas: bool, + rfc_exception_egm: bool, +) -> Optional[Dict[str, Any]]: + """Clarion: si LevantarSubpartidas=S, obligatorios ES PARTIDA/SUBPARTIDA y LINEA PRINCIPAL (col S/T o T/U para EGM0303257J1).""" + if not levantar_subpartidas: + return None + es_sub = _get(row, "ES PARTIDA/SUBPARTIDA", "ESSUBPARTIDA", "ES PARTIDA O SUBPARTIDA") + if not es_sub: + col = "T" if rfc_exception_egm else "S" + return _err( + line_num, + "ES PARTIDA/SUBPARTIDA", + f"Error: (Celda {col}{line_num}) El campo del tipo de la partida (partida o subpartida) está vacío. " + "Capturar el tipo de la partida. [P] = Partida o [S] = Subpartida.", + ) + linea_principal = _get(row, "LINEA PRINCIPAL", "LINEAPRINCIPAL", "PARTIDA PRINCIPAL") + if not linea_principal: + col = "U" if rfc_exception_egm else "T" + return _err( + line_num, + "LINEA PRINCIPAL", + f"Error: (Celda {col}{line_num}) El campo de la partida principal está vacío. " + f"Capturar en la Celda {col} la partida principal.", + ) + return None + + +def _valida_toda_obligatorios_expo( + row: Dict[str, Any], + line_num: int, + levantar_subpartidas: bool, + rfc_exception_egm: bool, +) -> Optional[Dict[str, Any]]: + """VALIDA_TODA_PAR_EXPO obligatorios: C (tipo impo), F (descarga), G (cantidad); si descarga=SI: D, E; si subpartidas: S/T o T/U.""" + obligatorios: List[str] = [] + tipo_impo = _get(row, "TIPO DE IMPO", "TIPO DE IMPO.", "TIPO IMPO", "PROCEDENCIA") + if not tipo_impo: + obligatorios.append("(Col.C) Procedencia de la Importación.") + descarga = _get(row, "GENERA DESCARGA", "GENERA DESCARGA?", "DESCARGA") + if descarga == "SI": + if not _get(row, "FACTURA IMPO", "FACTURA IMPO.", "FACTURA IMPORTACION"): + obligatorios.append("(Col.D) Factura de Impo.") + if not _get(row, "LINEA IMPO", "LINEA IMPO.", "LINEA IMPORTACION"): + obligatorios.append("(Col.E) Línea de Impo.") + if not descarga: + obligatorios.append("(Col.F) Descarga? SI o NO") + if not _get(row, "CANTIDAD EXPORTADA/DESCARGAR", "CANTIDAD EXPORTADA", "CANTIDAD"): + obligatorios.append("(Col.G) Cantidad Expo.") + if levantar_subpartidas: + if rfc_exception_egm: + if not _get(row, "ES PARTIDA/SUBPARTIDA", "ESSUBPARTIDA", "ES PARTIDA O SUBPARTIDA"): + obligatorios.append("(Col.T) EsSubpartida?.") + if not _get(row, "LINEA PRINCIPAL", "LINEAPRINCIPAL", "PARTIDA PRINCIPAL"): + obligatorios.append("(Col.U) Partida Principal.") + else: + if not _get(row, "ES PARTIDA/SUBPARTIDA", "ESSUBPARTIDA", "ES PARTIDA O SUBPARTIDA"): + obligatorios.append("(Col.S) EsSubpartida?.") + if not _get(row, "LINEA PRINCIPAL", "LINEAPRINCIPAL", "PARTIDA PRINCIPAL"): + obligatorios.append("(Col.T) Partida Principal.") + if obligatorios: + return _err( + line_num, + "TIPO DE IMPO", + f"Existen campos vacíos que son obligatorios, es la {', '.join(obligatorios)}. " + "Revisar la línea del archivo y capturar los campos con la información correcta.", + ) + return None + + +def _valida_subpartidas_duplicados_expo( + factura_expo: str, + linea_expo: str, + line_num: int, + line_counts_csv: Dict[Tuple[str, str], int], +) -> Optional[Dict[str, Any]]: + key = (factura_expo.strip(), _clip(linea_expo)) + if line_counts_csv.get(key, 0) > 1: + return _err( + line_num, + "LINEA EXPO", + f"Error: (Celda B{line_num}) El campo de la partida está duplicado entre las partidas. " + f"Capturar en la Celda B{line_num} otro número de partida.", + ) + return None + + +def _valida_subpartida_tiene_principal_expo( + row: Dict[str, Any], + line_num: int, + factura_expo: str, + partidas_principales_csv: Set[Tuple[str, str]], + partidas_principales_bd: Set[Tuple[str, str]], + rfc_exception_egm: bool, +) -> Optional[Dict[str, Any]]: + es_sub = _get(row, "ES PARTIDA/SUBPARTIDA", "ESSUBPARTIDA", "ES PARTIDA O SUBPARTIDA").upper() + linea_principal = _get(row, "LINEA PRINCIPAL", "LINEAPRINCIPAL", "PARTIDA PRINCIPAL") + if es_sub != "S" or not linea_principal or linea_principal == "0": + return None + key_principal = (factura_expo.strip(), _clip(linea_principal)) + if key_principal in partidas_principales_csv or key_principal in partidas_principales_bd: + return None + col = "U" if rfc_exception_egm else "T" + return _err( + line_num, + "LINEA PRINCIPAL", + f"Error: (Celda {col}{line_num}) La partida principal {linea_principal} no existe. " + f"Capturar en la Celda {col} la partida principal y/o verificar que si permita contener subpartidas.", + ) + + +def _valida_subpartida_linea_principal_no_cero_expo( + row: Dict[str, Any], line_num: int, rfc_exception_egm: bool +) -> Optional[Dict[str, Any]]: + es_sub = _get(row, "ES PARTIDA/SUBPARTIDA", "ESSUBPARTIDA", "ES PARTIDA O SUBPARTIDA").upper() + linea_principal = _get(row, "LINEA PRINCIPAL", "LINEAPRINCIPAL", "PARTIDA PRINCIPAL") + if es_sub == "S" and linea_principal == "0": + col = "U" if rfc_exception_egm else "T" + return _err( + line_num, + "LINEA PRINCIPAL", + f"Error: (Celda {col}{line_num}) La SubPartida no tiene asignada una partida principal. " + f"Capturar en la Celda {col} una partida principal.", + ) + return None + + +def _validaciones_par_expo( + row: Dict[str, Any], + line_num: int, + factura_expo: str, + valid_uom_codes: Set[str], + valid_bulks_codes: Set[str], + valid_payment_methods: Set[str], + valid_fraction_ame: Set[str], + valid_part_numbers: Optional[Set[str]], + factura_impo_tem_by_number: Dict[str, int], + factura_impo_def_by_number: Dict[str, int], + line_exists_tem: Set[Tuple[int, str]], + line_exists_def: Set[Tuple[int, str]], + validar_decimales_pza: bool = False, +) -> Optional[Dict[str, Any]]: + """VALIDACIONES_PAR_EXPO: longitudes, TEM/DEF, FK factura impo + línea, descarga SI/NO, cantidad, U.M., bultos, forma pago, impuesto, fracción ame, num parte, decimales PZA.""" + tipo_impo = _get(row, "TIPO DE IMPO", "TIPO DE IMPO.", "TIPO IMPO", "PROCEDENCIA").strip().upper() + if not tipo_impo: + tipo_impo = "TEM" + factura_impo = _get(row, "FACTURA IMPO", "FACTURA IMPO.", "FACTURA IMPORTACION") + linea_impo = _get(row, "LINEA IMPO", "LINEA IMPO.", "LINEA IMPORTACION") + descarga = _get(row, "GENERA DESCARGA", "GENERA DESCARGA?", "DESCARGA").upper() + + # Longitud A + if factura_expo and len(factura_expo) > MAX_LEN_FACTURA_EXPO: + return _err( + line_num, + "NUMERO FACTURA EXPO", + f"Error: (Celda A{line_num}) La Factura de Exportación: {factura_expo} supera la longitud de caracteres. " + "Capturar en la Celda A el campo Factura de Exportación con formato ###############.", + ) + # Longitud B + linea_expo = _get(row, "LINEA EXPO", "LINEA EXPO.", "RENGLON EXPO") + if linea_expo and len(linea_expo) > MAX_LEN_LINEA_EXPO: + return _err( + line_num, + "LINEA EXPO", + f"Error: (Celda B{line_num}) La Línea de Exportación: {linea_expo} supera la longitud de caracteres. " + "Capturar en la Celda B una Línea de Exportación con formato #####.", + ) + # Longitud C + if tipo_impo and len(tipo_impo) > MAX_LEN_TIPO_IMPO: + return _err( + line_num, + "TIPO DE IMPO", + f"Error: (Celda C{line_num}) La Procedencia debe ser especificada como TEM o DEF. " + "Capturar en la Celda C una procedencia no mayor de 3 caracteres.", + ) + # C: TEM o DEF + if tipo_impo and tipo_impo not in TIPO_IMPO_VALIDOS: + return _err( + line_num, + "TIPO DE IMPO", + f"Error: (Celda C{line_num}) El Tipo de Descargo: {tipo_impo} no es valido. Capturar uno valido como TEM o DEF.", + ) + # D, E: factura impo + línea existen en catálogo TEM o DEF + if factura_impo and tipo_impo: + consec_tem = factura_impo_tem_by_number.get(factura_impo.strip()) + consec_def = factura_impo_def_by_number.get(factura_impo.strip()) + if tipo_impo == "TEM": + if consec_tem is None: + return _err( + line_num, + "FACTURA IMPO", + f"Error: (Celda D{line_num}) La Factura: {factura_impo} de Importación Temporal no existe. " + "Capturar un Número de Factura que exista en el Catálogo de Importaciones Temporales.", + ) + key_line = (consec_tem, _clip(linea_impo)) + if linea_impo and key_line not in line_exists_tem: + return _err( + line_num, + "LINEA IMPO", + f"Error: (Celda D{line_num}, E{line_num}) La Factura: {factura_impo} con línea: {linea_impo} de Importación Temporal no existe. " + "Capturar un Número de Factura con diferente línea que este en el Catálogo de Importaciones Temporales.", + ) + elif tipo_impo == "DEF": + if consec_def is None: + return _err( + line_num, + "FACTURA IMPO", + f"Error: (Celda D{line_num}) La Factura: {factura_impo} de Importación Definitiva no existe. " + "Capturar un Número de Factura que exista en el Catálogo de Importaciones Definitivas.", + ) + key_line = (consec_def, _clip(linea_impo)) + if linea_impo and key_line not in line_exists_def: + return _err( + line_num, + "LINEA IMPO", + f"Error: (Celda D{line_num}, E{line_num}) La Factura: {factura_impo} con línea: {linea_impo} de Importación Definitiva no existe. " + "Capturar un Número de Factura con diferente línea que este en el Catálogo de Importaciones Definitivas.", + ) + # F: SI o NO + if descarga and descarga not in GENERA_DESCARGA_VALIDOS: + return _err( + line_num, + "GENERA DESCARGA", + f"Error: (Celda F{line_num}) La captura: {descarga} no es valido para la opción de que la partida genere descarga. " + "Capturar un valor valido como SI o NO o dejar vacio y lo tomará como un SI.", + ) + # G: cantidad no cero + cant_str = _get(row, "CANTIDAD EXPORTADA/DESCARGAR", "CANTIDAD EXPORTADA", "CANTIDAD") + if cant_str: + cant = _parse_decimal(cant_str) + if cant is not None and cant == 0: + return _err( + line_num, + "CANTIDAD EXPORTADA/DESCARGAR", + f"Error: (Celda G{line_num}) La Cantidad a Exportar: {cant_str} no puede ser cero. Capturar una cantidad a exportar valida.", + ) + # H: U.M. en catálogo (si no se toma de impo) + um = _get(row, "UNIDAD DE MEDIDA", "U.M.", "UNIDAD MEDIDA") + if um and valid_uom_codes and um.upper() not in valid_uom_codes: + return _err( + line_num, + "UNIDAD DE MEDIDA", + f"Error: (Celda H{line_num}) La U.M.: {um} no existe en catálogo de Unidades de Medida. " + "Capturar una Clave de Unidad de Medida que exista en el Catálogo.", + ) + # Bultos: cantidad + clave (Clarion Col L, M) + clave_bultos = _get(row, "CLAVE BULTOS", "CLAVEBULTOS") + cant_bultos = row.get("CANTIDAD BULTOS") or row.get("CANTIDADBULTOS") + if clave_bultos: + if valid_bulks_codes and clave_bultos not in valid_bulks_codes: + return _err( + line_num, + "CLAVE BULTOS", + f"Error: (Celda M{line_num}) La Clave de Bulto: {clave_bultos} no existe en el Catálogo de Claves de Bultos. " + "Darlo de alta en el Catálogo de Claves de Bultos o capturar uno ya existente.", + ) + cant_bultos_val = _parse_decimal(cant_bultos) + if cant_bultos_val is None: + return _err( + line_num, + "CANTIDAD BULTOS", + f"Error: (Celda L{line_num}) La Cantidad de Bultos está vacía y en la Celda M{line_num} se tienen la Clave de Bulto: {clave_bultos}.", + ) + if cant_bultos_val == 0: + return _err( + line_num, + "CANTIDAD BULTOS", + f"Error: (Celda L{line_num}) La Cantidad de Bultos es cero y en la Celda M{line_num} se tienen la Clave de Bulto: {clave_bultos}.", + ) + else: + cant_bultos_val = _parse_decimal(cant_bultos) + if cant_bultos_val is not None and cant_bultos_val > 0: + return _err( + line_num, + "CANTIDAD BULTOS", + f"Error: (Celda L{line_num}) La Cantidad de Bultos es {cant_bultos} y en la Celda M{line_num} no se tienen la Clave de Bulto.", + ) + # L: Se pagó impuesto SI/NO + se_pago = _get(row, "SE PAGO IMPUESTO", "SE PAGO IMPUESTO? (SI o NO)", "SEPAGOIMPUESTO") + if se_pago and se_pago.upper() not in SE_PAGO_IMPUESTO_VALIDOS: + return _err( + line_num, + "SE PAGO IMPUESTO", + f"Error: (Celda L{line_num}) El Valor Capturado para Se Pago Impuesto no es Válido. Capturar en la Celda L{line_num} SI o NO.", + ) + # M: Forma de pago en catálogo + forma_pago = _get(row, "FORMA DE PAGO", "FORMADEPAGO", "FORMA PAGO") + if forma_pago and valid_payment_methods and forma_pago not in valid_payment_methods: + return _err( + line_num, + "FORMA DE PAGO", + f"Error: (Celda M{line_num}) La Forma de Pago Capturado no es Válido. " + "Capturar en la Celda M una Forma de Pago dentro del Catálogo General de Formas de Pago.", + ) + # Fracción americana (Clarion Col R) + frac_ame = _get(row, "FRACCION AMERICANA", "FRACCIONAMERICANA") + if frac_ame and valid_fraction_ame and frac_ame not in valid_fraction_ame: + return _err( + line_num, + "FRACCION AMERICANA", + f"Error: (Celda R{line_num}) La Fracción Americana: {frac_ame} no está en el Catálogo de Fracciones Americanas. " + "Capturar en la Celda R una fracción que se encuentre en el catálogo o dar la de alta.", + ) + # Orden de compra / orden venta (Clarion Col S) máx 20 + orden = _get(row, "ORDEN DE COMPRA", "ORDENCOMPRA", "ORDEN DE VENTA") + if orden and len(orden) > MAX_LEN_ORDEN_COMPRA: + return _err( + line_num, + "ORDEN DE COMPRA", + f"Error: (Celda S{line_num}) La Orden de Venta: {orden} supera la cantidad de caracteres permitidos. " + "Capturar en la Celda S una orden de compra no mayor de 20 caracteres.", + ) + # Número de parte (Clarion Col T): longitud y catálogo + num_parte = _get(row, "NUM. PARTE", "NUMPARTE", "NUMERO PARTE", "NUM PARTE") + if num_parte: + if len(num_parte) > MAX_LEN_NUM_PARTE: + return _err( + line_num, + "NUM. PARTE", + f"Error: (Celda T{line_num}) El Número de Parte: {num_parte} supera la longitud de caracteres. " + "Capturar en la Celda T el Número de Parte no mayor de 30 caracteres.", + ) + if valid_part_numbers is not None and num_parte.upper() not in valid_part_numbers: + return _err( + line_num, + "NUM. PARTE", + f"Error: (Celda T{line_num}) El Número de Parte: {num_parte} no existe en el Catálogo de Partes. Darlo de alta en el Catálogo de Partes.", + ) + # Decimales PZA + if validar_decimales_pza and um and um.upper() == "PZA" and cant_str: + d = _parse_decimal(cant_str) + if d is not None and d != int(d): + return _err( + line_num, + "CANTIDAD EXPORTADA/DESCARGAR", + f"Error: (Celda G{line_num}) La Unidad de Medida es PZA, Por lo Tanto no es Válida la Captura de Decimales. Asignar una Cantidad sin decimales.", + ) + return None + + +def validate_row_partidas_expo( + row: Dict[str, Any], + line_num: int, + autonumerar: bool, + actualizar: bool, + levantar_subpartidas: 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_uom_codes: Set[str], + valid_bulks_codes: Set[str], + valid_payment_methods: Set[str], + valid_fraction_ame: Set[str], + valid_part_numbers: Optional[Set[str]], + factura_impo_tem_by_number: Dict[str, int], + factura_impo_def_by_number: Dict[str, int], + line_exists_tem: Set[Tuple[int, str]], + line_exists_def: Set[Tuple[int, str]], + rfc_exception_egm: bool = False, + validar_decimales_pza: bool = False, +) -> Optional[Dict[str, Any]]: + """ + Valida una fila de CSV de Partidas de Exportación Definitiva (o Cambio de Régimen). + Clarion: VALIDA_TODA_PAR_EXPO vs VALIDA_PARCIAL_PAR_EXPO según autonumerar, actualizar y si la partida existe. + """ + err = _check_factura_expo_vacia(row, line_num) + if err: + return err + + factura_expo = _get(row, "NUMERO FACTURA EXPO", "NUMERO FACTURA EXPO.", "NUM FACTURA EXPO", "FACTURA EXPO") + if not factura_expo: + return _err(line_num, "NUMERO FACTURA EXPO", "Requerido") + + err = _check_factura_expo_existe(factura_expo, line_num, invoice_id_by_number) + if err: + return err + + err = _check_factura_expo_no_actualizada( + factura_expo, line_num, invoice_updated_by_number, rfc_exception_updated + ) + if err: + return err + + err = _check_linea_expo_si_no_autonumerar(row, line_num, autonumerar) + if err: + return err + + err = _check_levantar_subpartidas_expo(row, line_num, levantar_subpartidas, rfc_exception_egm) + if err: + return err + + linea_expo = _get(row, "LINEA EXPO", "LINEA EXPO.", "RENGLON EXPO") + existing_lines = existing_line_keys_by_invoice.get(factura_expo.strip(), set()) + partida_existe = bool(linea_expo and linea_expo in existing_lines) + use_partial = actualizar and not autonumerar and partida_existe + + if use_partial: + return _validaciones_par_expo( + row, + line_num, + factura_expo, + valid_uom_codes=valid_uom_codes, + valid_bulks_codes=valid_bulks_codes, + valid_payment_methods=valid_payment_methods, + valid_fraction_ame=valid_fraction_ame, + valid_part_numbers=valid_part_numbers, + factura_impo_tem_by_number=factura_impo_tem_by_number, + factura_impo_def_by_number=factura_impo_def_by_number, + line_exists_tem=line_exists_tem, + line_exists_def=line_exists_def, + validar_decimales_pza=validar_decimales_pza, + ) + else: + err = _valida_toda_obligatorios_expo(row, line_num, levantar_subpartidas, rfc_exception_egm) + if err: + return err + if levantar_subpartidas: + err = _valida_subpartidas_duplicados_expo( + factura_expo, linea_expo or "", line_num, line_counts_csv + ) + if err: + return err + err = _valida_subpartida_tiene_principal_expo( + row, line_num, factura_expo, + partidas_principales_csv, partidas_principales_bd, rfc_exception_egm + ) + if err: + return err + err = _valida_subpartida_linea_principal_no_cero_expo(row, line_num, rfc_exception_egm) + if err: + return err + return _validaciones_par_expo( + row, + line_num, + factura_expo, + valid_uom_codes=valid_uom_codes, + valid_bulks_codes=valid_bulks_codes, + valid_payment_methods=valid_payment_methods, + valid_fraction_ame=valid_fraction_ame, + valid_part_numbers=valid_part_numbers, + factura_impo_tem_by_number=factura_impo_tem_by_number, + factura_impo_def_by_number=factura_impo_def_by_number, + line_exists_tem=line_exists_tem, + line_exists_def=line_exists_def, + validar_decimales_pza=validar_decimales_pza, + )