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 87c9bd13..8892781e 100644 --- a/backend/api/v1/modules/a76/layouts_csv/exportacion/routes.py +++ b/backend/api/v1/modules/a76/layouts_csv/exportacion/routes.py @@ -33,7 +33,7 @@ def _get_redis(): @router.post("/upload/{model_target}", response_model=ImportJobResponse) async def upload_import_file( - model_target: Literal["invoice_header", "invoice_details"], + model_target: Literal["invoice_header", "invoice_details", "invoice_series"], file: UploadFile = File(...), footer_config: Optional[str] = Form(None), template_id: Optional[str] = Form(None), @@ -56,13 +56,18 @@ async def upload_import_file( contents = await file.read() file_key, meta_key, _ = common_storage.storage_keys(JOB_TYPE, job_id) + default_template = ( + "exp_def_header" if model_target == "invoice_header" + else "exp_def_series" if model_target == "invoice_series" + else "exp_def_partidas" + ) meta_data = { "tenant_id": tenant_id, "company_id": company_id, "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_partidas"), + "template_id": template_id or default_template, } try: diff --git a/backend/api/v1/modules/a76/layouts_csv/exportacion/schemas.py b/backend/api/v1/modules/a76/layouts_csv/exportacion/schemas.py index 2a043f30..ca97e233 100644 --- a/backend/api/v1/modules/a76/layouts_csv/exportacion/schemas.py +++ b/backend/api/v1/modules/a76/layouts_csv/exportacion/schemas.py @@ -9,7 +9,7 @@ class ImportJobResponse(BaseModel): class CommitRequest(BaseModel): - model_target: Literal["invoice_header", "invoice_details"] + model_target: Literal["invoice_header", "invoice_details", "invoice_series"] class ImportJobStatus(BaseModel): 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 7a6800e5..a2b61da2 100644 --- a/backend/api/v1/modules/a76/layouts_csv/exportacion/tasks.py +++ b/backend/api/v1/modules/a76/layouts_csv/exportacion/tasks.py @@ -50,6 +50,10 @@ 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_details", config, job_type_override="exp") + if model_target == "invoice_series": + from api.v1.modules.a76.layouts_csv.facturas.tasks import _do_scan_file + return _do_scan_file(job_id, "invoice_series", config, job_type_override="exp") + # Fallback (e.g. unknown model_target) file_path = _ensure_file(job_id) if not file_path: @@ -96,6 +100,10 @@ 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_details", job_type_override="exp") + if model_target == "invoice_series": + from api.v1.modules.a76.layouts_csv.facturas.tasks import _do_insert_valid_rows + return _do_insert_valid_rows(job_id, "invoice_series", job_type_override="exp") + # Fallback: stub sin inserción file_path = _ensure_file(job_id) if not file_path: 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 59247c4a..e7b6124e 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 @@ -87,6 +87,18 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { {"canonical": "NUM. PARTE", "aliases": ["NUMPARTE", "NUMERO PARTE", "NUM PARTE"]}, {"canonical": "ORDEN DE COMPRA", "aliases": ["ORDENCOMPRA", "ORDEN DE VENTA"]}, ], + # Series de exportación definitiva (misma estructura que imp_def_series; Clarion SERIES EXPO) + "exp_def_series": [ + {"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA", "FACTURA EXPO"]}, + {"canonical": "LINEA FACTURA", "aliases": ["LINEA", "PARTIDA"]}, + {"canonical": "LINEA SERIE", "aliases": ["RENGLON", "LINEA SER"]}, + {"canonical": "SERIE", "aliases": ["NUMERO SERIE"]}, + {"canonical": "MODELO"}, + {"canonical": "NUM PARTE", "aliases": ["NUMPARTE", "NUMERO PARTE", "NUM PART"]}, + {"canonical": "SUB MODELO", "aliases": ["SUBMODELO", "SUB MODE"]}, + {"canonical": "NUMERO ID", "aliases": ["NUMEROID"]}, + {"canonical": "COL_EXTRA"}, + ], } 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 8f3990bd..1b338e51 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py @@ -227,6 +227,8 @@ def _do_scan_file(job_id: str, model_target: str, config: Optional[str] = None, template_id = "exp_def_header" if job_type_override == "exp" and model_target == "invoice_details": template_id = "exp_def_partidas" + if job_type_override == "exp" and model_target == "invoice_series": + template_id = "exp_def_series" 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" @@ -409,6 +411,183 @@ def _do_scan_file(job_id: str, model_target: str, config: Optional[str] = None, logger.exception("Series importación definitiva scan failed: %s", e) return {"status": "failed", "error": str(e)} + # --- Series de Exportación Definitiva: flujo exp_def_series (Clarion VALIDA_TODA_SERIES_EXPO / VALIDA_PARCIAL) --- + if model_target == "invoice_series" and template_id == "exp_def_series": + logger.info("Series expo scan: running validation for job %s", job_id) + 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 api.v1.modules.a76.general_catalogs.company.models import Company + from .validators.series_expo import validate_row_series_expo + + _fc = parse_footer_config(meta.get("footer_config")) + autonumerar = meta.get("autonumerar", True) + actualizar = meta.get("actualizar", False) + validar_series_exception = meta.get("validar_series", False) + if _fc: + if "autonumerar" in _fc: + autonumerar = bool(_fc["autonumerar"]) + elif _fc.get("autonumber_series", "true") is not None: + autonumerar = str(_fc.get("autonumber_series", "true")).lower() in ("true", "1", "si", "sí", "yes") + if "actualizar" in _fc: + actualizar = bool(_fc["actualizar"]) + if "validar_series" in _fc: + validar_series_exception = bool(_fc["validar_series"]) + + 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 "" + + 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) + + 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 == "exp", + ) + ) + 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 == "exp", + ) + ) + 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] = {} + + invoice_numbers_from_csv: 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) + for row in reader: + row_norm = row_from_template(row, "exp_def_series", normalize_header) + inv = (row_norm.get("NUMERO FACTURA") or row_norm.get("NUM FACTURA") 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: Set[str] = invoice_numbers_from_csv + else: + rfc_exception_updated = set() + + with open(file_path, "r", encoding="utf-8-sig") as f_in, open(error_path, "w", encoding="utf-8") as f_err: + f_in.seek(0) + try: + dialect = csv.Sniffer().sniff(f_in.read(2048), delimiters=",;\t") + except Exception: + dialect = "excel" + f_in.seek(0) + 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, "exp_def_series", normalize_header) + warnings_list: List[Dict[str, Any]] = [] + err = validate_row_series_expo( + 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, + rfc_exception_updated=rfc_exception_updated, + 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, + ) + 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 row_norm.get("FACTURA EXPO") 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 + + 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, total_rows_in_file=total_rows + ) + except Exception as e: + logger.exception("Series exportació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: @@ -3590,7 +3769,7 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt # 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", "cmex_series") + or meta.get("template_id") in ("imp_temp_series", "imp_def_series", "cmex_series", "exp_def_series") ) _footer_for_series = parse_footer_config(meta.get("footer_config")) or {} @@ -3603,6 +3782,182 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt or _inv_type_series in ("DEF", "MATDE", "EXDEF") ) ) + use_expo_series_commit = use_series_flow and meta.get("template_id") == "exp_def_series" + + # --- Series de Exportación Definitiva: commit (INSERT/UPDATE item_line_series para facturas exp) --- + if use_expo_series_commit: + 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 .validators.series_expo import row_to_series_normalized_expo + + actualizar = meta.get("actualizar", False) + autonumerar = meta.get("autonumerar", True) + _fc = parse_footer_config(meta.get("footer_config")) + if _fc: + if "actualizar" in _fc: + actualizar = bool(_fc["actualizar"]) + 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") + + with CoreSessionLocal() as session: + q_inv = ( + session.query(InvoiceHeader.invoice_number, InvoiceHeader.id) + .filter( + InvoiceHeader.tenant_id == tenant_id, + InvoiceHeader.company_id == company_id, + InvoiceHeader.operation_type == "exp", + ) + ) + invoice_id_by_number: Dict[str, int] = {} + for num, iid in q_inv.all(): + if num: + invoice_id_by_number[str(num).strip()] = iid + + inserted_count = 0 + updated_count = 0 + skipped_invalid = 0 + skipped_details: List[Dict[str, Any]] = [] + + with open(file_path, "r", encoding="utf-8-sig") as f: + sample = f.read(2048) + f.seek(0) + try: + dialect = csv.Sniffer().sniff(sample, delimiters=",;\t") + except Exception: + dialect = "excel" + reader = csv.DictReader(f, dialect=dialect) + + for i, row in enumerate(reader, start=1): + if i in error_lines: + skipped_invalid += 1 + continue + row_norm = row_from_template(row, "exp_def_series", normalize_header) + data = row_to_series_normalized_expo(row_norm) + invoice_number = data["NUMERO FACTURA"] + linea_factura = data["LINEA FACTURA"] + linea_serie = data["LINEA SERIE"] + + if not invoice_number or invoice_number not in invoice_id_by_number: + skipped_invalid += 1 + skipped_details.append({ + "line": i, + "invoice": invoice_number or "(vacío)", + "reason": "Factura de exportación no encontrada.", + }) + continue + invoice_id = invoice_id_by_number[invoice_number] + line_number_val = parse_int(linea_factura) + if line_number_val is None: + skipped_invalid += 1 + skipped_details.append({"line": i, "invoice": invoice_number, "reason": "LINEA FACTURA debe ser numérico."}) + continue + line_item = ( + session.query(LineItem) + .filter( + LineItem.invoice_id == invoice_id, + LineItem.line_number == line_number_val, + LineItem.tenant_id == tenant_id, + LineItem.company_id == company_id, + ) + .first() + ) + if not line_item: + line_numbers = [ + r[0] for r in + session.query(LineItem.line_number) + .filter( + LineItem.invoice_id == invoice_id, + LineItem.tenant_id == tenant_id, + LineItem.company_id == company_id, + ) + .order_by(LineItem.line_number) + .all() + ] + existing_str = ", ".join(str(n) for n in line_numbers) if line_numbers else "ninguna" + skipped_invalid += 1 + skipped_details.append({ + "line": i, + "invoice": invoice_number, + "reason": f"Partida línea {linea_factura} no existe en la factura. Partidas existentes: {existing_str}.", + }) + continue + + if autonumerar: + max_row = ( + session.query(Serie.row) + .filter(Serie.line_item_id == line_item.id) + .order_by(Serie.row.desc()) + .limit(1) + .scalar() + ) + row_num = (max_row or 0) + 1 + else: + row_num = parse_int(linea_serie) + if row_num is None: + skipped_invalid += 1 + skipped_details.append({"line": i, "invoice": invoice_number, "reason": "LINEA SERIE debe ser numérico."}) + continue + + existing_serie = ( + session.query(Serie) + .filter( + Serie.line_item_id == line_item.id, + Serie.row == row_num, + ) + .first() + ) + + if existing_serie: + if actualizar: + existing_serie.serial_numbers = data["SERIE"] or existing_serie.serial_numbers + existing_serie.model = data["MODELO"] or existing_serie.model + existing_serie.sub_model = data["SUB MODELO"] or existing_serie.sub_model + existing_serie.number_id = data["NUMERO ID"] or existing_serie.number_id + session.add(existing_serie) + updated_count += 1 + else: + skipped_invalid += 1 + skipped_details.append({"line": i, "invoice": invoice_number, "reason": "Serie ya existe (use actualizar)."}) + else: + new_serie = Serie( + tenant_id=tenant_id, + company_id=company_id, + line_item_id=line_item.id, + row=row_num, + serial_numbers=data["SERIE"] or None, + model=data["MODELO"] or None, + sub_model=data["SUB MODELO"] or None, + number_id=data["NUMERO ID"] or None, + ) + session.add(new_serie) + inserted_count += 1 + + session.commit() + + common_storage.cleanup_import_job(effective_job_type, job_id, file_path=file_path, error_path=error_path, meta_path=meta_path) + status = "finished" if (inserted_count + updated_count) > 0 else ("warning" if skipped_invalid else "failed") + out = { + "status": status, + "inserted": inserted_count, + "updated": updated_count, + "skipped_invalid": skipped_invalid, + "skipped_missing_fk": 0, + "skipped_duplicate": 0, + "skipped_details": skipped_details, + } + if status == "failed": + out["error"] = "No hay registros válidos en el archivo CSV." + elif status == "warning" and skipped_invalid: + out["message"] = f"No se insertaron registros. {skipped_invalid} fueron rechazados." + return out + except Exception as e: + logger.exception("Series exportación definitiva commit failed: %s", e) + return {"status": "failed", "error": str(e)} # --- Series de Importación Definitiva: commit (INSERT/UPDATE item_line_series para facturas DEF o MEX) --- if use_def_series_commit: @@ -4158,6 +4513,8 @@ def _do_insert_valid_rows(job_id: str, model_target: str, job_type_override: Opt ) if job_type_override == "exp" and model_target == "invoice_details": _template_id_insert = "exp_def_partidas" + if job_type_override == "exp" and model_target == "invoice_series": + _template_id_insert = "exp_def_series" if model_target == "invoice_header" and _template_id_insert == "imp_def_header": inv_type_value = "DEF" if model_target == "invoice_header" and _template_id_insert == "cmex_header": 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 effa0d2c..711f4653 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 @@ -243,6 +243,19 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { {"canonical": "NUMERO ID", "aliases": ["NUMEROID"]}, {"canonical": "COL_EXTRA"}, ], + # --- Series de Exportación Definitiva (Clarion VALIDA_TODA_SERIES_EXPO / VALIDA_PARCIAL_SERIES_EXPO) --- + # Misma estructura: NUMERO FACTURA, LINEA FACTURA, LINEA SERIE, SERIE, MODELO, NUM PARTE, SUB MODELO, NUMERO ID + "exp_def_series": [ + {"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA", "FACTURA EXPO"]}, + {"canonical": "LINEA FACTURA", "aliases": ["LINEA", "PARTIDA"]}, + {"canonical": "LINEA SERIE", "aliases": ["RENGLON", "LINEA SER"]}, + {"canonical": "SERIE", "aliases": ["NUMERO SERIE"]}, + {"canonical": "MODELO"}, + {"canonical": "NUM PARTE", "aliases": ["NUMPARTE", "NUMERO PARTE", "NUM PART"]}, + {"canonical": "SUB MODELO", "aliases": ["SUBMODELO", "SUB MODE"]}, + {"canonical": "NUMERO ID", "aliases": ["NUMEROID"]}, + {"canonical": "COL_EXTRA"}, + ], } @@ -260,6 +273,8 @@ def _resolve_template_columns(template_id: str) -> Optional[List[Dict[str, Any]] return TEMPLATE_COLUMNS.get("imp_temp_details") if template_id == "cmex_series": return TEMPLATE_COLUMNS.get("imp_def_series") + if template_id == "exp_def_series": + return TEMPLATE_COLUMNS.get("exp_def_series") return None 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 b5e920b7..2c363064 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 @@ -17,6 +17,10 @@ from .series_impo_def import ( validate_row_series_impo_def, row_to_series_normalized_def, ) +from .series_expo import ( + validate_row_series_expo, + row_to_series_normalized_expo, +) __all__ = [ "validate_row_encabezados_impo_temp", @@ -27,6 +31,8 @@ __all__ = [ "validate_row_partidas_impo_def", "validate_row_series_impo_def", "row_to_series_normalized_def", + "validate_row_series_expo", + "row_to_series_normalized_expo", "row_to_transport_type_clarion", "parse_pedimento_col_a", "parse_pedimento_col_a_impo_def", diff --git a/backend/api/v1/modules/a76/layouts_csv/facturas/validators/series_expo.py b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/series_expo.py new file mode 100644 index 00000000..9d143635 --- /dev/null +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/series_expo.py @@ -0,0 +1,296 @@ +""" +Validaciones CSV para Series de Exportación Definitiva. +Paridad Clarion: VALIDA_TODA_SERIES_EXPO, VALIDA_PARCIAL_SERIES_EXPO, LLENA_SERIES_EXPO. +Estructura: NUMERO FACTURA, LINEA FACTURA, LINEA SERIE, SERIE, MODELO, NUM PARTE, SUB MODELO, NUMERO ID. +Reutiliza helpers de series_impo_temp y series_impo_def; factura = exportación (FAC_EXPO), rfc_exception_updated. +""" +from typing import Dict, Any, Optional, Set, Tuple, List + +from .series_impo_temp import ( + _clip, + normalize_sacarcomasenters, + _check_desfase, + _check_linea_factura_vacia, + _check_linea_serie_si_no_autonumerar, + _warn_apostrofes, + _check_max_length, + MAX_LEN, +) +from .series_impo_def import ( + _check_partida_existe_en_factura, + _check_cantidad_series_vs_partida, + _get_val_def, +) + + +def _check_factura_vacia_expo(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]: + """NUMERO FACTURA (A) vacío → error. Mensaje Factura de Exportación.""" + val = _clip(row.get("NUMERO FACTURA") or row.get("NUM FACTURA") or row.get("FACTURA") or row.get("FACTURA EXPO")) + if not val: + return { + "line": line_num, + "col": "NUMERO FACTURA", + "msg": ( + f"Error: (Celda A{line_num}) La Factura de Exportación está vacía y no se pueden hacer las validaciones. " + ), + "solution": ( + f"Capturar en la Celda A{line_num} un número de Factura existente " + f"al cual desee agregar o actualizar series" + ), + } + return None + + +def _check_factura_existe_expo( + invoice_number: str, + line_num: int, + invoice_id_by_number: Dict[str, int], +) -> Optional[Dict[str, Any]]: + """Factura debe existir en BD (catálogo exportación).""" + 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 Exportación {invoice_number} " + 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 " + f"al cual desee agregar o actualizar series" + ), + "identifier": "FAC_EXPO", + "fields": invoice_number, + } + return None + + +def _check_factura_no_actualizada_expo( + invoice_number: str, + line_num: int, + invoice_updated_by_number: Dict[str, bool], + rfc_exception_updated: Set[str], +) -> Optional[Dict[str, Any]]: + """Si factura ya actualizada (Estatus AC) no se pueden hacer cambios; excepción por RFC (ej. EGM0303257J1).""" + 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{line_num}) La Factura de Exportación: {invoice_number} " + "ya existe y está Actualizada, no se puede hacer cambios a las facturas actualizadas." + ), + "solution": "Capturar otro número de Factura de Exportación o Desactualizar la factura.", + "identifier": "FAC_EXPO", + } + return None + + +def valida_toda_series_expo( + row: Dict[str, Any], + line_num: int, + validar_series_exception: bool, + warnings: Optional[List[Dict[str, Any]]], +) -> Optional[Dict[str, Any]]: + """ + VALIDA_TODA_SERIES_EXPO: cuando no existe la serie o autonumerar=SI. + Si D+E+F están todos vacíos y no aplica excepción ValidarSeries → error obligatorios. + Valida longitudes máximas. + """ + d = _get_val_def(row, "SERIE") + e = _get_val_def(row, "MODELO") + f = _get_val_def(row, "NUM PARTE") + campos = d + e + f + + if not campos and not validar_series_exception: + obligatorios = [] + if not d: + obligatorios.append("(Col.D) Serie") + if not e: + obligatorios.append("(Col.E) Modelo") + if not f: + obligatorios.append("(Col.F) Num. Parte") + if obligatorios: + return { + "line": line_num, + "col": "SERIE", + "msg": ( + f"Existen campos vacíos que son obligatorios al no tener ningun campo, " + f"es la {', '.join(obligatorios)}." + ), + "solution": "Revisar la línea del archivo y capturar los campos con la información correcta.", + "identifier": "ARCHIVO CSV", + } + + for col, key, max_len in [ + ("SERIE", "SERIE", MAX_LEN["serial_numbers"]), + ("MODELO", "MODELO", MAX_LEN["model"]), + ("NUM PARTE", "NUM PARTE", 50), + ("SUB MODELO", "SUB MODELO", MAX_LEN["sub_model"]), + ("NUMERO ID", "NUMERO ID", MAX_LEN["number_id"]), + ]: + val = _get_val_def(row, key) + if val: + err = _check_max_length(col, val, line_num, max_len) + if err: + return err + return None + + +def valida_parcial_series_expo( + row: Dict[str, Any], + line_num: int, + existing_series_data: Dict[str, Any], + warnings: Optional[List[Dict[str, Any]]], +) -> Optional[Dict[str, Any]]: + """ + VALIDA_PARCIAL_SERIES_EXPO: actualizar serie existente; campos vacíos se rellenan con existente. + Solo validar longitudes en campos no vacíos. + """ + for col, key, max_len in [ + ("SERIE", "SERIE", MAX_LEN["serial_numbers"]), + ("MODELO", "MODELO", MAX_LEN["model"]), + ("NUM PARTE", "NUM PARTE", 50), + ("SUB MODELO", "SUB MODELO", MAX_LEN["sub_model"]), + ("NUMERO ID", "NUMERO ID", MAX_LEN["number_id"]), + ]: + val = _get_val_def(row, key) + if val: + err = _check_max_length(col, val, line_num, max_len) + if err: + return err + return None + + +def _series_key(invoice_number: str, linea_factura: str, linea_serie: str) -> Tuple[str, str, str]: + return (invoice_number.strip(), _clip(linea_factura), _clip(linea_serie)) + + +def validate_row_series_expo( + row: Dict[str, Any], + line_num: int, + actualizar: bool, + autonumerar: bool, + validar_series_exception: bool, + invoice_id_by_number: Dict[str, int], + invoice_updated_by_number: Dict[str, bool], + rfc_exception_updated: Set[str], + partida_max_series: Dict[Tuple[str, str], int], + csv_series_count_so_far: Dict[Tuple[str, str], int], + 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, +) -> Optional[Dict[str, Any]]: + """ + Punto de entrada: valida una fila de CSV de Series de Exportación Definitiva. + Clarion: decisión VALIDA_TODA vs VALIDA_PARCIAL según autonumerar, actualizar y si la serie existe. + rfc_exception_updated: set de números de factura que se consideran no actualizadas (ej. EGM0303257J1). + """ + desfase = _check_desfase(row, line_num) + if desfase and warnings is not None: + warnings.append(desfase) + + err = _check_factura_vacia_expo(row, line_num) + if err: + return err + + invoice_number = _clip( + row.get("NUMERO FACTURA") or row.get("NUM FACTURA") or row.get("FACTURA") or row.get("FACTURA EXPO") + ) + if not invoice_number: + return {"line": line_num, "col": "NUMERO FACTURA", "msg": "Requerido"} + + err = _check_factura_existe_expo(invoice_number, line_num, invoice_id_by_number) + if err: + return err + + err = _check_factura_no_actualizada_expo( + invoice_number, line_num, invoice_updated_by_number, rfc_exception_updated + ) + if err: + return err + + err = _check_linea_factura_vacia(row, line_num) + if err: + return err + + linea_factura = _clip(row.get("LINEA FACTURA") or row.get("LINEA") or row.get("PARTIDA")) + err = _check_partida_existe_en_factura( + invoice_number, + linea_factura, + line_num, + partida_max_series, + ) + if err: + return err + + err = _check_cantidad_series_vs_partida( + invoice_number, + linea_factura, + line_num, + partida_max_series, + csv_series_count_so_far, + ) + if err: + return err + + err = _check_linea_serie_si_no_autonumerar(row, line_num, autonumerar) + if err: + return err + + _warn_apostrofes(row, line_num, warnings) + + linea_serie = _clip(row.get("LINEA SERIE") or row.get("RENGLON")) + key = _series_key(invoice_number, linea_factura, linea_serie) + + use_partial = ( + actualizar + and not autonumerar + and bool(linea_serie) + and key in (existing_series_keys or set()) + ) + + if use_partial and existing_series_data and key in existing_series_data: + return valida_parcial_series_expo( + row, line_num, existing_series_data[key], warnings + ) + return valida_toda_series_expo( + row, line_num, validar_series_exception, warnings + ) + + +def row_to_series_normalized_expo(row: Dict[str, Any]) -> Dict[str, Any]: + """ + Normaliza fila para guardar: SACARCOMASENTERS en D, E, F, G, H. + Clarion LLENA_SERIES_EXPO: asigna QueCSV a SerExpo. + NUM PARTE se valida pero el modelo Serie no tiene campo parte. + """ + def clip(col: str, alt: Optional[List[str]] = None) -> str: + v = row.get(col) + if alt: + for k in alt: + if v is None or (isinstance(v, str) and not v.strip()): + v = row.get(k) + return _clip(v) if v is not None else "" + + def norm(col: str, alt: Optional[List[str]] = None, max_len: int = 50) -> str: + v = row.get(col) + if alt: + for k in alt: + if v is None or (isinstance(v, str) and not v.strip()): + v = row.get(k) + s = normalize_sacarcomasenters(v) if v is not None else "" + return s[:max_len] if s else "" + + return { + "NUMERO FACTURA": clip("NUMERO FACTURA", ["NUM FACTURA", "FACTURA", "FACTURA EXPO"]), + "LINEA FACTURA": clip("LINEA FACTURA", ["LINEA", "PARTIDA"]), + "LINEA SERIE": clip("LINEA SERIE", ["RENGLON"]), + "SERIE": norm("SERIE", max_len=MAX_LEN["serial_numbers"]), + "MODELO": norm("MODELO", max_len=MAX_LEN["model"]), + "NUM PARTE": norm("NUM PARTE", ["NUMPARTE", "NUMERO PARTE"]), + "SUB MODELO": norm("SUB MODELO", ["SUBMODELO"], MAX_LEN["sub_model"]), + "NUMERO ID": norm("NUMERO ID", ["NUMEROID"], MAX_LEN["number_id"]), + } diff --git a/frontend/src/lib/config/csv-upload.ts b/frontend/src/lib/config/csv-upload.ts index 076d6279..3cc39412 100644 --- a/frontend/src/lib/config/csv-upload.ts +++ b/frontend/src/lib/config/csv-upload.ts @@ -444,8 +444,9 @@ export const exportacionConfig: CsvUploadItem[] = [ title: 'Series', icon: Hash, group: 'Expo. Def./Cam. Reg.', - modelTarget: 'InvoiceSeries', - disabled: true, + modelTarget: 'invoice_series', + templateId: 'exp_def_series', + layoutModule: 'layouts_csv/exportacion' }, { id: 'exp_def_nodes',