From 4dc7fbd8ed6153e733f051cebd5e88c5e68f3a4b Mon Sep 17 00:00:00 2001 From: hreyes Date: Mon, 9 Mar 2026 07:28:05 -0600 Subject: [PATCH] feature/series-invoices-clarion-validations-csv --- .../a76/layouts_csv/facturas/routes.py | 2 +- .../a76/layouts_csv/facturas/schemas.py | 2 +- .../modules/a76/layouts_csv/facturas/tasks.py | 365 +++++++++++++++++- .../layouts_csv/facturas/template_config.py | 13 + .../facturas/validators/__init__.py | 1 + .../facturas/validators/series_impo_temp.py | 362 +++++++++++++++++ frontend/src/lib/config/csv-upload.ts | 5 +- .../routes/dashboard/csv-upload/+page.svelte | 2 +- 8 files changed, 745 insertions(+), 7 deletions(-) create mode 100644 backend/api/v1/modules/a76/layouts_csv/facturas/validators/__init__.py create mode 100644 backend/api/v1/modules/a76/layouts_csv/facturas/validators/series_impo_temp.py diff --git a/backend/api/v1/modules/a76/layouts_csv/facturas/routes.py b/backend/api/v1/modules/a76/layouts_csv/facturas/routes.py index c7e9c570..cda7106d 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/routes.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/routes.py @@ -35,7 +35,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), # JSON string with settings template_id: Optional[str] = Form(None), # id de la plantilla (ej. imp_temp_header) para respetar columnas diff --git a/backend/api/v1/modules/a76/layouts_csv/facturas/schemas.py b/backend/api/v1/modules/a76/layouts_csv/facturas/schemas.py index 63c0202a..4cf4db4d 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/schemas.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/schemas.py @@ -7,7 +7,7 @@ class ImportJobResponse(BaseModel): message: str class CommitRequest(BaseModel): - model_target: Literal["invoice_header", "invoice_details"] + model_target: Literal["invoice_header", "invoice_details", "invoice_series"] class ImportJobStatus(BaseModel): status: str 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 a3fe37be..c3cf8d72 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py @@ -6,7 +6,7 @@ import json import logging import re import unicodedata -from typing import Dict, Any, Optional, List +from typing import Dict, Any, Optional, List, Set, Tuple from core.celery_app import celery_app from core.database import CoreSessionLocal @@ -165,12 +165,142 @@ def scan_file(self, job_id: str, model_target: str, config: str = None): meta = common_meta.load_meta(file_path) template_id = meta.get("template_id") or ( - "imp_temp_header" if model_target == "invoice_header" else "imp_temp_details" + "imp_temp_header" if model_target == "invoice_header" else + "imp_temp_details" if model_target == "invoice_details" else "imp_temp_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" + + # --- Series de Importación Temporal: flujo específico (Clarion VALIDA_TODA / VALIDA_PARCIAL) --- + if model_target == "invoice_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 .validators.series_impo_temp import ( + validate_row_series_impo_temp, + ) + + 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: + # Invoice lookup: imp + TEM + 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 == "TEM", + ) + ) + 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) + + # Existing series keys: (invoice_number, linea_factura, linea_serie) + 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 == "TEM", + ) + ) + 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 "", + }) + + 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: List[Dict[str, Any]] = [] + 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, "imp_temp_series", normalize_header) + warnings_list: List[Dict[str, Any]] = [] + err = validate_row_series_impo_temp( + 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, + 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", "")}) + 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 import scan failed: %s", e) + return {"status": "failed", "error": str(e)} try: from api.v1.modules.a76.invoices.models import InvoiceHeader @@ -711,6 +841,237 @@ def insert_valid_rows(self, job_id: str, model_target: str): error_path = common_storage.error_path_for_job(JOB_TYPE, job_id) error_lines = common_storage.get_error_lines(JOB_TYPE, job_id, error_path) + # Si el upload fue de series (template_id imp_temp_series), usar flujo series aunque model_target venga mal + use_series_flow = ( + model_target == "invoice_series" + or meta.get("template_id") == "imp_temp_series" + ) + + # --- Series de Importación Temporal: commit (INSERT/UPDATE item_line_series) --- + if use_series_flow: + 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_impo_temp import ( + validate_row_series_impo_temp, + row_to_series_normalized, + ) + + 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 == "TEM", + ) + ) + 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) + + 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 == "TEM", + ) + ) + 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 "", + }) + + 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: + continue + row_norm = row_from_template(row, "imp_temp_series", normalize_header) + err = validate_row_series_impo_temp( + 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, + existing_series_keys=existing_series_keys, + existing_series_data=existing_series_data, + warnings=None, + ) + if err: + skipped_invalid += 1 + skipped_details.append({ + "line": i, + "invoice": (row_norm.get("NUMERO FACTURA") or row_norm.get("NUM FACTURA") or "").strip(), + "reason": err.get("msg", ""), + }) + continue + + data = row_to_series_normalized(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 + 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: + skipped_invalid += 1 + skipped_details.append({ + "line": i, + "invoice": invoice_number, + "reason": f"Partida línea {linea_factura} no existe en la factura.", + }) + 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(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 import commit failed: %s", e) + return {"status": "failed", "error": str(e)} + try: from api.v1.modules.a76.invoices.models import ( InvoiceHeader, 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 1a3fc274..2d6122a9 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 @@ -69,6 +69,19 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { # --- Partidas: Impo Def y Expo - misma estructura --- "imp_def_details": None, "exp_def_details": None, + # --- 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": [ + {"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA"]}, + {"canonical": "LINEA FACTURA", "aliases": ["LINEA", "PARTIDA"]}, + {"canonical": "LINEA SERIE", "aliases": ["RENGLON"]}, + {"canonical": "SERIE", "aliases": ["NUMERO SERIE"]}, + {"canonical": "MODELO"}, + {"canonical": "NUM PARTE", "aliases": ["NUMPARTE", "NUMERO PARTE"]}, + {"canonical": "SUB MODELO", "aliases": ["SUBMODELO"]}, + {"canonical": "NUMERO ID", "aliases": ["NUMEROID"]}, + {"canonical": "COL_EXTRA"}, # Optional; if has value → desfase warning (Clarion) + ], } 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 new file mode 100644 index 00000000..83516424 --- /dev/null +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/__init__.py @@ -0,0 +1 @@ +# Validators for invoice CSV imports (header, details, series). diff --git a/backend/api/v1/modules/a76/layouts_csv/facturas/validators/series_impo_temp.py b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/series_impo_temp.py new file mode 100644 index 00000000..44668900 --- /dev/null +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/series_impo_temp.py @@ -0,0 +1,362 @@ +""" +Validaciones CSV para Series de Importación Temporal. +Paridad Clarion: VALIDA_TODA_SERIES_IMPO_TEM, VALIDA_PARCIAL_SERIES_IMPO_TEM, LLENA_SERIES_IMPO_TEM. +Estructura: NUMERO FACTURA, LINEA FACTURA, LINEA SERIE, SERIE, MODELO, NUM PARTE, SUB MODELO, NUMERO ID. +""" +from typing import Dict, Any, Optional, Set, Tuple, List + +# Longitudes máximas según modelo Serie (item_line_series) +MAX_LEN = { + "serial_numbers": 50, + "model": 50, + "sub_model": 50, + "number_id": 25, +} + +APOSTROFE = "'" + + +def _clip(val: Any) -> str: + if val is None: + return "" + return str(val).strip() + + +def normalize_sacarcomasenters(val: Any) -> str: + """Clarion SACARCOMASENTERS: quitar comas y saltos de línea.""" + s = _clip(val) + s = s.replace(",", " ").replace("\n", " ").replace("\r", " ") + return " ".join(s.split()) + + +def _check_desfase(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]: + """Si COL_EXTRA tiene valor → advertencia desfase (no bloqueante).""" + val = _clip(row.get("COL_EXTRA")) + if not val: + return None + return { + "line": line_num, + "col": "COL_EXTRA", + "msg": "Advertencia: Podría existir un desfase en esta línea.", + "solution": "Revisar esta línea del archivo CSV y verificar cada campo esté en la posición correcta.", + "warning": True, + } + + +def _check_factura_vacia(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]: + """NUMERO FACTURA (A) vacío → error bloqueante.""" + val = _clip(row.get("NUMERO FACTURA") or row.get("NUM FACTURA") or row.get("FACTURA")) + if not val: + return { + "line": line_num, + "col": "NUMERO FACTURA", + "msg": "Error: (Celda A) La Factura de Importación está vacía y no se pueden hacer las validaciones.", + "solution": "Capturar en la Celda A un número de Factura existente al cual desee agregar o actualizar series.", + } + return None + + +def _check_factura_existe( + invoice_number: str, + line_num: int, + invoice_id_by_number: Dict[str, int], +) -> Optional[Dict[str, Any]]: + """Factura debe existir en BD (imp + TEM).""" + if invoice_number not in invoice_id_by_number: + return { + "line": line_num, + "col": "NUMERO FACTURA", + "msg": f"Error: (Celda A) La Factura de Importación {invoice_number} no existe en SCAII y no se pueden hacer las validaciones.", + "solution": "Capturar en la Celda A un número de Factura existente al cual desee agregar o actualizar partidas.", + "identifier": "FAC_IMPO_TEM", + "fields": invoice_number, + } + return None + + +def _check_factura_no_actualizada( + invoice_number: str, + line_num: int, + invoice_updated_by_number: Dict[str, bool], +) -> Optional[Dict[str, Any]]: + """Si factura ya actualizada (Estatus AC) no se pueden hacer cambios.""" + if invoice_updated_by_number.get(invoice_number, False): + return { + "line": line_num, + "col": "NUMERO FACTURA", + "msg": f"Error: (Celda A) La Factura de Importación: {invoice_number} ya existe y está Actualizada, no se puede hacer cambios a las facturas actualizadas.", + "solution": "Capturar otro número de Factura de Importación Temporal o Desactualizar la factura.", + "identifier": "FAC_IMPO_TEM", + } + return None + + +def _check_linea_factura_vacia(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]: + """LINEA FACTURA (B) vacía → error.""" + val = _clip(row.get("LINEA FACTURA") or row.get("LINEA") or row.get("PARTIDA")) + if not val: + return { + "line": line_num, + "col": "LINEA FACTURA", + "msg": "Error: (Celda B) El campo de la línea de la partida está vacío y no se pueden hacer las validaciones.", + "solution": "Capturar en la Celda B la línea de la partida al cual desee agregar o actualizar información.", + } + return None + + +def _check_linea_serie_si_no_autonumerar( + row: Dict[str, Any], + line_num: int, + autonumerar: bool, +) -> Optional[Dict[str, Any]]: + """Si Autonumerar = NO, LINEA SERIE (C) es obligatoria.""" + if autonumerar: + return None + val = _clip(row.get("LINEA SERIE") or row.get("RENGLON")) + if not val: + return { + "line": line_num, + "col": "LINEA SERIE", + "msg": "Error: (Celda C) El campo de Renglón está vacío y no se pueden hacer las validaciones, ya que se tiene la opción de Autonumerar como NO.", + "solution": "Capturar en la Celda C el renglón de la serie la cual desee agregar o actualizar información.", + } + return None + + +def _warn_apostrofes( + row: Dict[str, Any], + line_num: int, + warnings: Optional[List[Dict[str, Any]]], +) -> None: + """Advertencias si SERIE, MODELO o NUM PARTE contienen apostrofe (no bloqueante).""" + if warnings is None: + return + checks = [ + ("SERIE", "Número de Serie", row.get("SERIE")), + ("MODELO", "Número de Modelo", row.get("MODELO")), + ("NUM PARTE", "Número de Parte", row.get("NUM PARTE") or row.get("NUMPARTE") or row.get("NUMERO PARTE")), + ] + for col, label, val in checks: + v = _clip(val) if val is not None else "" + if v and APOSTROFE in v: + warnings.append({ + "line": line_num, + "col": col, + "msg": f"Advertencia: El {label}: {v} Contiene Apostrofes.", + "solution": "Se Omitirá el Apostrofe para Subir.", + "warning": True, + }) + + +def _check_max_length(col: str, val: str, line_num: int, max_len: int) -> Optional[Dict[str, Any]]: + if not val or len(val) <= max_len: + return None + return { + "line": line_num, + "col": col, + "msg": f"Máximo {max_len} caracteres", + } + + +def valida_toda_series_impo_tem( + row: Dict[str, Any], + line_num: int, + validar_series_exception: bool, + warnings: Optional[List[Dict[str, Any]]], +) -> Optional[Dict[str, Any]]: + """ + VALIDA_TODA_SERIES_IMPO_TEM: cuando no existe la partida/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 = _clip(row.get("SERIE")) + e = _clip(row.get("MODELO")) + f = _clip(row.get("NUM PARTE") or row.get("NUMPARTE") or row.get("NUMERO 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: {', '.join(obligatorios)}.", + "solution": "Revisar la línea del archivo y capturar los campos con la información correcta.", + } + + # Longitudes (solo si hay valor) + def get_val(k: str) -> str: + if k == "NUM PARTE": + return _clip(row.get("NUM PARTE") or row.get("NUMPARTE") or row.get("NUMERO PARTE")) + if k == "SUB MODELO": + return _clip(row.get("SUB MODELO") or row.get("SUBMODELO")) + if k == "NUMERO ID": + return _clip(row.get("NUMERO ID") or row.get("NUMEROID")) + return _clip(row.get(k)) + + 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(key) + if val: + err = _check_max_length(col, val, line_num, max_len) + if err: + return err + + return None + + +def valida_parcial_series_impo_tem( + 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_IMPO_TEM: actualizar serie existente; campos vacíos se rellenan con existente. + Solo validar longitudes en campos no vacíos. + """ + def get_val(k: str) -> str: + if k == "NUM PARTE": + return _clip(row.get("NUM PARTE") or row.get("NUMPARTE") or row.get("NUMERO PARTE")) + if k == "SUB MODELO": + return _clip(row.get("SUB MODELO") or row.get("SUBMODELO")) + if k == "NUMERO ID": + return _clip(row.get("NUMERO ID") or row.get("NUMEROID")) + return _clip(row.get(k)) + + 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(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_impo_temp( + 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], + 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 Importación Temporal. + Clarion: decisión VALIDA_TODA vs VALIDA_PARCIAL según autonumerar, actualizar y si la serie existe. + """ + # Desfase (solo advertencia) + desfase = _check_desfase(row, line_num) + if desfase and warnings is not None: + warnings.append(desfase) + + # Factura vacía + err = _check_factura_vacia(row, line_num) + if err: + return err + + invoice_number = _clip(row.get("NUMERO FACTURA") or row.get("NUM FACTURA") or row.get("FACTURA")) + if not invoice_number: + return {"line": line_num, "col": "NUMERO FACTURA", "msg": "Requerido"} + + # Factura existe + err = _check_factura_existe(invoice_number, line_num, invoice_id_by_number) + if err: + return err + + # Factura no actualizada + err = _check_factura_no_actualizada(invoice_number, line_num, invoice_updated_by_number) + if err: + return err + + # LINEA FACTURA vacía + err = _check_linea_factura_vacia(row, line_num) + if err: + return err + + # Autonumerar NO y LINEA SERIE vacía + err = _check_linea_serie_si_no_autonumerar(row, line_num, autonumerar) + if err: + return err + + # Advertencias apostrofes (no bloqueante) + _warn_apostrofes(row, line_num, warnings) + + # Decisión VALIDA_TODA vs VALIDA_PARCIAL + linea_factura = _clip(row.get("LINEA FACTURA") or row.get("LINEA") or row.get("PARTIDA")) + 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_impo_tem( + row, line_num, existing_series_data[key], warnings + ) + return valida_toda_series_impo_tem( + row, line_num, validar_series_exception, warnings + ) + + +def row_to_series_normalized(row: Dict[str, Any]) -> Dict[str, Any]: + """ + Normaliza fila para guardar: SACARCOMASENTERS en D, E, F, G, H. + Clarion LLENA_SERIES_IMPO_TEM: asigna QueCSV a SerImp. + """ + 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"]), + "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 3af99c66..2a990325 100644 --- a/frontend/src/lib/config/csv-upload.ts +++ b/frontend/src/lib/config/csv-upload.ts @@ -355,8 +355,9 @@ export const importacionConfig: CsvUploadItem[] = [ title: 'Series', icon: Hash, group: 'Impo. Temp.', - modelTarget: 'InvoiceSeries', - disabled: true, + modelTarget: 'invoice_series', + templateId: 'imp_temp_series', + layoutModule: 'layouts_csv/facturas' }, // Impo Def { diff --git a/frontend/src/routes/dashboard/csv-upload/+page.svelte b/frontend/src/routes/dashboard/csv-upload/+page.svelte index 5f79624d..ea90ee62 100644 --- a/frontend/src/routes/dashboard/csv-upload/+page.svelte +++ b/frontend/src/routes/dashboard/csv-upload/+page.svelte @@ -431,7 +431,7 @@ footerConfig, companyId, opType, - config.id + config.templateId || config.id ); if (res.data?.job_id) { currentJobId = res.data.job_id;