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 b4efce49..2646d960 100644 --- a/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/tasks.py @@ -217,10 +217,181 @@ def scan_file(self, job_id: str, model_target: str, config: str = None): "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" + if model_target == "invoice_series" and inv_type_value in ("DEF", "MATDE", "EXDEF"): + template_id = "imp_def_series" + + # --- Series de Importación Definitiva: flujo específico (Clarion VALIDA_TODA_SERIES_IMPO_DEF / VALIDA_PARCIAL) --- + DEF_SERIES_TEMPLATE_OR_TYPE = ( + model_target == "invoice_series" + and ( + template_id == "imp_def_series" + or (inv_type_value in ("DEF", "MATDE", "EXDEF")) + ) + ) + if DEF_SERIES_TEMPLATE_OR_TYPE: + try: + from api.v1.modules.a76.invoices.models import InvoiceHeader + from api.v1.modules.a76.items.models import LineItem + from api.v1.modules.a76.items.series.models import Serie + from api.v1.modules.a76.items.line_quantities.models import LineQuantity + from .validators.series_impo_def import ( + validate_row_series_impo_def, + ) + + DEF_INVOICE_TYPES = ("DEF", "MATDE", "EXDEF") + 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.in_(DEF_INVOICE_TYPES), + ) + ) + rows_inv = q.all() + invoice_id_by_number: Dict[str, int] = {} + invoice_updated_by_number: Dict[str, bool] = {} + for num, iid, is_upd in rows_inv: + if num: + invoice_id_by_number[str(num).strip()] = iid + invoice_updated_by_number[str(num).strip()] = bool(is_upd) + + partida_max_series: Dict[Tuple[str, str], int] = {} + q_qty = ( + session.query( + InvoiceHeader.invoice_number, + LineItem.line_number, + LineQuantity.quantity, + ) + .join(LineItem, LineItem.invoice_id == InvoiceHeader.id) + .outerjoin(LineQuantity, LineQuantity.item_line_id == LineItem.id) + .filter( + InvoiceHeader.tenant_id == tenant_id, + InvoiceHeader.company_id == company_id, + InvoiceHeader.operation_type == "imp", + InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES), + ) + ) + for num, ln, qty in q_qty.all(): + if num is not None and ln is not None: + key = (str(num).strip(), str(ln).strip()) + if qty is not None: + partida_max_series[key] = int(qty) if qty else 0 + else: + partida_max_series[key] = 0 + + existing_series_keys: Set[Tuple[str, str, str]] = set() + existing_series_data: Dict[Tuple[str, str, str], Dict[str, Any]] = {} + if actualizar and not autonumerar: + q_ser = ( + session.query( + InvoiceHeader.invoice_number, + LineItem.line_number, + Serie.row, + Serie.serial_numbers, + Serie.model, + Serie.sub_model, + Serie.number_id, + ) + .join(LineItem, LineItem.invoice_id == InvoiceHeader.id) + .join(Serie, Serie.line_item_id == LineItem.id) + .filter( + InvoiceHeader.tenant_id == tenant_id, + InvoiceHeader.company_id == company_id, + InvoiceHeader.operation_type == "imp", + InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES), + ) + ) + for num, ln, rw, sn, md, sm, nid in q_ser.all(): + if num is not None: + k = (str(num).strip(), str(ln).strip(), str(rw).strip()) + existing_series_keys.add(k) + existing_series_data.setdefault(k, { + "serial_numbers": sn or "", + "model": md or "", + "sub_model": sm or "", + "number_id": nid or "", + }) + + csv_series_count_so_far: Dict[Tuple[str, str], int] = {} + + with open(file_path, "r", encoding="utf-8-sig") as f_in, open(error_path, "w", encoding="utf-8") as f_err: + sample = f_in.read(2048) + f_in.seek(0) + try: + dialect = csv.Sniffer().sniff(sample, delimiters=",;\t") + except Exception: + dialect = "excel" + reader = csv.DictReader(f_in, dialect=dialect) + errors_detail = [] + error_lines_list: List[int] = [] + for i, row in enumerate(reader, start=1): + if i % 1000 == 0: + self.update_state(state="PROGRESS", meta={"current": i, "total": total_rows, "errors": len(error_lines_list)}) + row_norm = row_from_template(row, "imp_def_series", normalize_header) + warnings_list: List[Dict[str, Any]] = [] + err = validate_row_series_impo_def( + row_norm, + i, + actualizar=actualizar, + autonumerar=autonumerar, + validar_series_exception=validar_series_exception, + invoice_id_by_number=invoice_id_by_number, + invoice_updated_by_number=invoice_updated_by_number, + partida_max_series=partida_max_series, + csv_series_count_so_far=csv_series_count_so_far, + existing_series_keys=existing_series_keys, + existing_series_data=existing_series_data, + warnings=warnings_list, + ) + if err and not err.get("warning"): + error_count += 1 + error_lines_list.append(err["line"]) + f_err.write(json.dumps(err) + "\n") + if len(errors_detail) < 500: + errors_detail.append({"line": err["line"], "col": err.get("col", ""), "msg": err.get("msg", "")}) + else: + inv_num = (row_norm.get("NUMERO FACTURA") or row_norm.get("NUM FACTURA") or "").strip() + line_fac = (row_norm.get("LINEA FACTURA") or row_norm.get("LINEA") or row_norm.get("PARTIDA") or "").strip() + if inv_num and line_fac: + key_csv = (inv_num, line_fac) + csv_series_count_so_far[key_csv] = csv_series_count_so_far.get(key_csv, 0) + 1 + for w in warnings_list: + if len(errors_detail) < 500: + errors_detail.append({"line": w["line"], "col": w.get("col", ""), "msg": w.get("msg", ""), "warning": True}) + processed_rows += 1 + + if error_lines_list: + common_storage.store_error_lines(JOB_TYPE, job_id, error_lines_list) + return common_responses.scan_result( + job_id, processed_rows, error_count, errors_detail, total_rows_in_file=total_rows + ) + except Exception as e: + logger.exception("Series importación definitiva scan failed: %s", e) + return {"status": "failed", "error": str(e)} # --- Series de Importación Temporal: flujo específico (Clarion VALIDA_TODA / VALIDA_PARCIAL) --- if model_target == "invoice_series": @@ -2171,12 +2342,289 @@ 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 + # 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") == "imp_temp_series" + or meta.get("template_id") in ("imp_temp_series", "imp_def_series") ) + _footer_for_series = parse_footer_config(meta.get("footer_config")) or {} + _inv_type_series = normalize_public_code(_footer_for_series.get("invoice_type") or meta.get("invoice_type") or "") + use_def_series_commit = ( + use_series_flow + and ( + meta.get("template_id") == "imp_def_series" + or _inv_type_series in ("DEF", "MATDE", "EXDEF") + ) + ) + + # --- Series de Importación Definitiva: commit (INSERT/UPDATE item_line_series para facturas DEF) --- + if use_def_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 api.v1.modules.a76.items.line_quantities.models import LineQuantity + from .validators.series_impo_def import ( + validate_row_series_impo_def, + row_to_series_normalized_def, + ) + + DEF_INVOICE_TYPES = ("DEF", "MATDE", "EXDEF") + 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.in_(DEF_INVOICE_TYPES), + ) + ) + rows_inv = q.all() + invoice_id_by_number: Dict[str, int] = {} + invoice_updated_by_number: Dict[str, bool] = {} + for num, iid, is_upd in rows_inv: + if num: + invoice_id_by_number[str(num).strip()] = iid + invoice_updated_by_number[str(num).strip()] = bool(is_upd) + + partida_max_series = {} + q_qty = ( + session.query( + InvoiceHeader.invoice_number, + LineItem.line_number, + LineQuantity.quantity, + ) + .join(LineItem, LineItem.invoice_id == InvoiceHeader.id) + .outerjoin(LineQuantity, LineQuantity.item_line_id == LineItem.id) + .filter( + InvoiceHeader.tenant_id == tenant_id, + InvoiceHeader.company_id == company_id, + InvoiceHeader.operation_type == "imp", + InvoiceHeader.invoice_type.in_(DEF_INVOICE_TYPES), + ) + ) + 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()) + partida_max_series[key] = int(qty) if qty else 0 + + existing_series_keys = 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.in_(DEF_INVOICE_TYPES), + ) + ) + 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] = {} + 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_def_series", normalize_header) + err = validate_row_series_impo_def( + row_norm, + i, + actualizar=actualizar, + autonumerar=autonumerar, + validar_series_exception=validar_series_exception, + invoice_id_by_number=invoice_id_by_number, + invoice_updated_by_number=invoice_updated_by_number, + partida_max_series=partida_max_series, + csv_series_count_so_far=csv_series_count_so_far, + existing_series_keys=existing_series_keys, + existing_series_data=existing_series_data, + warnings=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_def(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: + 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 en la factura: {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 + + key_csv = (invoice_number.strip(), (linea_factura or "").strip()) + if key_csv[0] and key_csv[1]: + csv_series_count_so_far[key_csv] = csv_series_count_so_far.get(key_csv, 0) + 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 importación definitiva commit failed: %s", e) + return {"status": "failed", "error": str(e)} + # --- Series de Importación Temporal: commit (INSERT/UPDATE item_line_series) --- if use_series_flow: try: diff --git a/backend/api/v1/modules/a76/layouts_csv/facturas/template_config.py b/backend/api/v1/modules/a76/layouts_csv/facturas/template_config.py index 5b7999fc..a5e371ff 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 @@ -136,6 +136,18 @@ TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { {"canonical": "NUMERO ID", "aliases": ["NUMEROID"]}, {"canonical": "COL_EXTRA"}, # Optional; if has value → desfase warning (Clarion) ], + # --- Series de Importación Definitiva (misma estructura que TEM; cabeceras imagen: NUMERO/LINEA FAC, LINEA SER, etc.) --- + "imp_def_series": [ + {"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA", "NUMERO / LINEA FAC"]}, + {"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/validators/__init__.py b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/__init__.py index 89fb2a9f..650f053a 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 @@ -10,11 +10,17 @@ from .encabezados_impo_def import ( parse_pedimento_col_a_impo_def, ) from .partidas_impo_def import validate_row_partidas_impo_def +from .series_impo_def import ( + validate_row_series_impo_def, + row_to_series_normalized_def, +) __all__ = [ "validate_row_encabezados_impo_temp", "validate_row_encabezados_impo_def", "validate_row_partidas_impo_def", + "validate_row_series_impo_def", + "row_to_series_normalized_def", "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_impo_def.py b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/series_impo_def.py new file mode 100644 index 00000000..d1307a01 --- /dev/null +++ b/backend/api/v1/modules/a76/layouts_csv/facturas/validators/series_impo_def.py @@ -0,0 +1,335 @@ +""" +Validaciones CSV para Series de Importación Definitiva. +Paridad Clarion: VALIDA_TODA_SERIES_IMPO_DEF, VALIDA_PARCIAL_SERIES_IMPO_DEF, LLENA_SERIES_IMPO_DEF. +Estructura: NUMERO FACTURA, LINEA FACTURA, LINEA SERIE, SERIE, MODELO, NUM PARTE, SUB MODELO, NUMERO ID. +Reutiliza helpers de series_impo_temp; añade reglas DEF: factura en catálogo DEF, cantidad series vs partida. +""" +from typing import Dict, Any, Optional, Set, Tuple, List + +from .series_impo_temp import ( + _clip, + normalize_sacarcomasenters, + _check_desfase, + _check_factura_vacia, + _check_linea_factura_vacia, + _check_linea_serie_si_no_autonumerar, + _warn_apostrofes, + _check_max_length, + MAX_LEN, +) + + +def _check_factura_existe_def( + invoice_number: str, + line_num: int, + invoice_id_by_number: Dict[str, int], +) -> Optional[Dict[str, Any]]: + """Factura debe existir en BD (Importación Definitiva: DEF/MATDE/EXDEF).""" + if invoice_number not in invoice_id_by_number: + return { + "line": line_num, + "col": "NUMERO FACTURA", + "msg": ( + f"Error: (Celda A{line_num}) La Factura de Importación {invoice_number} " + "no existe en SCAII y no se pueden hacer las validaciones. " + ), + "solution": ( + f"Capturar en la Celda A{line_num} un número de Factura existente " + "al cual desee agregar o actualizar series" + ), + "identifier": "FAC_IMPO_DEF", + "fields": invoice_number, + } + return None + + +def _check_factura_no_actualizada_def( + 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. Mensaje DEF.""" + if invoice_updated_by_number.get(invoice_number, False): + return { + "line": line_num, + "col": "NUMERO FACTURA", + "msg": ( + f"Error: (Celda A{line_num}) La Factura de Importación: {invoice_number} " + "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_DEF", + } + return None + + +def _check_partida_existe_en_factura( + invoice_number: str, + linea_factura: str, + line_num: int, + partida_max_series: Dict[Tuple[str, str], int], +) -> Optional[Dict[str, Any]]: + """ + Clarion: la partida (línea de factura) debe existir en la factura. + partida_max_series tiene como claves (invoice_number, line_number) de las partidas existentes. + """ + if not linea_factura: + return None + key = (invoice_number.strip(), _clip(linea_factura)) + if key in partida_max_series: + return None + return { + "line": line_num, + "col": "LINEA FACTURA", + "msg": f"Partida línea {linea_factura} no existe en la factura.", + "solution": "Usar un número de línea de partida que exista en la factura (capturar partidas antes de cargar series).", + "identifier": "ARCHIVO CSV", + } + + +def _check_cantidad_series_vs_partida( + invoice_number: str, + linea_factura: str, + line_num: int, + partida_max_series: Dict[Tuple[str, str], int], + csv_series_count_so_far: Dict[Tuple[str, str], int], +) -> Optional[Dict[str, Any]]: + """ + Clarion: si cantidad de series en CSV para (factura, partida) excede CantImpoDef → error. + partida_max_series[(inv, line)] = máximo permitido (desde LineQuantity.quantity). + csv_series_count_so_far = conteo actual de filas válidas ya procesadas por (inv, line). + """ + key = (invoice_number.strip(), _clip(linea_factura)) + max_allowed = partida_max_series.get(key) + if max_allowed is None or max_allowed <= 0: + return None + current = csv_series_count_so_far.get(key, 0) + if current + 1 > max_allowed: + return { + "line": line_num, + "col": "LINEA FACTURA", + "msg": ( + f"Error: La cantidad de Series de la partida {_clip(linea_factura)} en la factura " + f"{invoice_number} es menor al número de Series en el archivo. " + ), + "solution": "Nivelar la cantidad de la Partida o el número de Series.", + "identifier": "ARCHIVO CSV", + } + return None + + +def _get_val_def(row: Dict[str, Any], k: str) -> str: + """Obtener valor normalizado por clave (aliases DEF/TEM).""" + 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)) + + +def valida_toda_series_impo_def( + 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_DEF: 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_impo_def( + 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_DEF: 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_impo_def( + 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], + 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 Importación Definitiva. + Clarion: decisión VALIDA_TODA vs VALIDA_PARCIAL según autonumerar, actualizar y si la serie existe. + partida_max_series: máximo de series por (invoice_number, linea_factura); 0 o ausente = no validar. + csv_series_count_so_far: conteo de filas ya aceptadas por (invoice_number, linea_factura); el caller debe incrementar al aceptar. + """ + desfase = _check_desfase(row, line_num) + if desfase and warnings is not None: + warnings.append(desfase) + + 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"} + + err = _check_factura_existe_def(invoice_number, line_num, invoice_id_by_number) + if err: + return err + + err = _check_factura_no_actualizada_def(invoice_number, line_num, invoice_updated_by_number) + 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_impo_def( + row, line_num, existing_series_data[key], warnings + ) + return valida_toda_series_impo_def( + row, line_num, validar_series_exception, warnings + ) + + +def row_to_series_normalized_def(row: Dict[str, Any]) -> Dict[str, Any]: + """ + Normaliza fila para guardar: SACARCOMASENTERS en D, E, F, G, H. + Clarion LLENA_SERIES_IMPO_DEF: asigna QueCSV a SerDef. + Incluye NUM PARTE para validación; el modelo Serie actual no tiene campo parte (documentar si se añade). + """ + 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 2a990325..3f67eb5b 100644 --- a/frontend/src/lib/config/csv-upload.ts +++ b/frontend/src/lib/config/csv-upload.ts @@ -383,8 +383,9 @@ export const importacionConfig: CsvUploadItem[] = [ title: 'Series', icon: Hash, group: 'Impo. Def.', - modelTarget: 'InvoiceSeries', - disabled: true, + modelTarget: 'invoice_series', + templateId: 'imp_def_series', + layoutModule: 'layouts_csv/facturas' }, // Compras Mex {