diff --git a/application/csv_cleaning.py b/application/csv_cleaning.py index b9f105d..2bf4e8e 100644 --- a/application/csv_cleaning.py +++ b/application/csv_cleaning.py @@ -14,8 +14,10 @@ import logging import math import random import re +import unicodedata from collections import Counter from dataclasses import dataclass, field +from datetime import date from typing import Any, Callable, Literal, Optional import numpy as np @@ -101,6 +103,35 @@ _TIME_TAIL_RE = re.compile( r"[\sT]+(\d{1,2})(?::(\d{1,2})(?::(\d{1,2}))?)?\s*([APap]\.?\s*[Mm]\.?)?\s*$" ) +_MONTH_NAME_TO_NUM = { + "enero": 1, + "ene": 1, + "febrero": 2, + "feb": 2, + "marzo": 3, + "mar": 3, + "abril": 4, + "abr": 4, + "mayo": 5, + "may": 5, + "junio": 6, + "jun": 6, + "julio": 7, + "jul": 7, + "agosto": 8, + "ago": 8, + "septiembre": 9, + "setiembre": 9, + "sep": 9, + "sept": 9, + "octubre": 10, + "oct": 10, + "noviembre": 11, + "nov": 11, + "diciembre": 12, + "dic": 12, +} + def _split_date_time(value: str) -> tuple[str, Optional[str]]: """Separa la parte de fecha de la parte de hora si está presente. @@ -312,6 +343,118 @@ def _parse_date_with_format(value: str, fmt: str) -> Optional[tuple[int, int, in return (d, m, y) +def _normalize_date_text(value: Any) -> str: + text = unicodedata.normalize("NFKD", str(value or "")) + text = "".join(c for c in text if not unicodedata.combining(c)) + return text.lower() + + +def _valid_date_tuple(d: int, m: int, y: int) -> Optional[tuple[int, int, int]]: + if y < 100: + y = 2000 + y if y < 50 else 1900 + y + if not (1000 <= y <= 9999): + return None + try: + date(y, m, d) + except ValueError: + return None + return (d, m, y) + + +def _split_glued_year_suffix_hhmm(digits: str) -> Optional[tuple[int, int, int]]: + """Detecta `YYYY` + `HHMM` en una sola tira de 8 cifras (p. ej. `20241344`). + + Tras quitar separadores puede quedar pegado al mes (p. ej. `15may20241344`). + Devuelve (año, hora, minuto) si la parte horaria es válida; si no, None. + """ + if len(digits) != 8 or not digits.isdigit(): + return None + y = int(digits[:4]) + hh = int(digits[4:6]) + mm = int(digits[6:8]) + if not (1000 <= y <= 9999 and 0 <= hh <= 23 and 0 <= mm <= 59): + return None + return (y, hh, mm) + + +def _parse_text_month_candidate( + first: tuple[int, str], + month: int, + second: tuple[int, str], + *, + layout: Literal["around", "after", "before"], +) -> Optional[tuple[int, int, int]]: + a, astr = first + b, bstr = second + if layout == "around": + if len(astr) == 4: + return _valid_date_tuple(b, month, a) + if len(bstr) == 4: + return _valid_date_tuple(a, month, b) + return _valid_date_tuple(a, month, b) + if layout == "after": + if len(astr) == 4: + return _valid_date_tuple(b, month, a) + return _valid_date_tuple(a, month, b) + if len(astr) == 4: + return _valid_date_tuple(b, month, a) + return _valid_date_tuple(a, month, b) + + +def _parse_date_with_text_month(raw: Any) -> Optional[tuple[int, int, int]]: + if _is_nullish_scalar(raw): + return None + date_part, _ = _split_date_time(str(raw)) + normalized = _normalize_date_text(date_part or raw) + # `\d{8}` antes que `\d{1,4}` para no partir año+hora pegados tras strip agresivo. + tokens = re.findall(r"[a-z]+|\d{8}|\d{1,4}", normalized) + if not tokens: + return None + + for i, token in enumerate(tokens): + month = _MONTH_NAME_TO_NUM.get(token) + if month is None: + continue + nums_before = [ + (int(t), t) for t in tokens[:i] if t.isdigit() + ] + nums_after = [ + (int(t), t) for t in tokens[i + 1 :] if t.isdigit() + ] + candidates: list[Optional[tuple[int, int, int]]] = [] + if nums_before and len(nums_after) == 1: + glued = _split_glued_year_suffix_hhmm(nums_after[0][1]) + if glued is not None: + y, _hh, _mm = glued + candidates.append( + _parse_text_month_candidate( + nums_before[-1], month, (y, str(y)), layout="around" + ) + ) + if nums_before and nums_after: + candidates.append( + _parse_text_month_candidate( + nums_before[-1], month, nums_after[0], layout="around" + ) + ) + if len(nums_after) >= 2: + candidates.append( + _parse_text_month_candidate( + nums_after[0], month, nums_after[1], layout="after" + ) + ) + if len(nums_before) >= 2: + candidates.append( + _parse_text_month_candidate( + nums_before[-2], month, nums_before[-1], layout="before" + ) + ) + for parsed in candidates: + if parsed is not None: + return parsed + return None + + # --------------------------------------------------------------------------- # ANÁLISIS DE COLUMNAS # --------------------------------------------------------------------------- @@ -856,22 +999,25 @@ def _applicable_rules_for_type(inferred_type: str) -> list[str]: if inferred_type == TYPE_INTEGER: return base if inferred_type == TYPE_DECIMAL: - return base + ["to_integer", "round_decimals_custom", "doc_id_split_tipo_numero"] + return base + [ + "to_integer", + "round_decimals_custom", + "doc_id_split_tipo_numero", + "doc_type_homologate", + ] if inferred_type == TYPE_DATE: return base + [ "doc_id_split_tipo_numero", - "date_messy_text_to_ddmmyyyy", - "date_numeric_to_ddmmyyyy", - "date_alpha_to_ddmmyyyy", + "doc_type_homologate", + "date_to_ddmmyyyy", ] if inferred_type == TYPE_MIXED: return base + [ "to_integer", "round_decimals_custom", "doc_id_split_tipo_numero", - "date_messy_text_to_ddmmyyyy", - "date_numeric_to_ddmmyyyy", - "date_alpha_to_ddmmyyyy", + "doc_type_homologate", + "date_to_ddmmyyyy", ] # Texto / categórico / boolean / empty → todas las de limpieza de texto + las # conversoras (el usuario sabrá si aplica). @@ -879,9 +1025,8 @@ def _applicable_rules_for_type(inferred_type: str) -> list[str]: "to_integer", "round_decimals_custom", "doc_id_split_tipo_numero", - "date_messy_text_to_ddmmyyyy", - "date_numeric_to_ddmmyyyy", - "date_alpha_to_ddmmyyyy", + "doc_type_homologate", + "date_to_ddmmyyyy", ] @@ -962,8 +1107,8 @@ def _rule_strip_special_spaces(series: pd.Series) -> pd.Series: def _rule_strip_special_keep_date_seps(series: pd.Series) -> pd.Series: - """Conserva alfanuméricos y separadores de fecha: '/', '-', espacio.""" - pattern = re.compile(r"[^A-Za-z0-9\/\- ]", flags=re.UNICODE) + """Conserva alfanuméricos, separadores de fecha '/', '-' y hora ':', y espacio.""" + pattern = re.compile(r"[^A-Za-z0-9\/\-: ]", flags=re.UNICODE) def _fn(x): if pd.isna(x): @@ -1030,6 +1175,103 @@ def _rule_round_decimals_custom(series: pd.Series) -> pd.Series: return numeric.map(_fn).astype("Int64") +def _rule_doc_type_homologate(series: pd.Series) -> pd.Series: + """Homologa una columna que contiene tipos de documento.""" + def _fn(x: Any): + if _is_nullish_scalar(x): + return x + tipo, _num = split_doc_tipo_y_numero_texto(x, homologate=True) + if tipo is not None: + return tipo + return homologate_doc_type(x) + + return series.map(_fn) + + +def _detect_compact_date_formats( + series: pd.Series, +) -> tuple[Optional[str], Optional[str]]: + cser = series.map(_normalize_messy_date_to_compact) + c8s = pd.Series( + [x for x in cser.tolist() if isinstance(x, str) and len(x) == 8] + ) + c6s = pd.Series( + [x for x in cser.tolist() if isinstance(x, str) and len(x) == 6] + ) + fmt8 = detect_date_format(c8s) if not c8s.empty else None + fmt6 = detect_date_format(c6s) if not c6s.empty else None + return fmt8, fmt6 + + +def _parse_any_date_to_tuple( + raw: Any, + *, + sep_fmt: Optional[str], + compact8_fmt: Optional[str], + compact6_fmt: Optional[str], +) -> Optional[tuple[int, int, int]]: + parsed_text = _parse_date_with_text_month(raw) + if parsed_text is not None: + return parsed_text + + if _is_nullish_scalar(raw): + return None + + raw_text = str(raw) + date_part, _ = _split_date_time(raw_text) + target = date_part if date_part else raw_text + if sep_fmt: + parsed = _parse_date_with_format(target, sep_fmt) + if parsed is not None: + return parsed + for fallback_fmt in ("DD/MM/YYYY", "MM/DD/YYYY", "YYYY/MM/DD"): + if fallback_fmt == sep_fmt: + continue + parsed = _parse_date_with_format(target, fallback_fmt) + if parsed is not None: + return parsed + + compact = _normalize_messy_date_to_compact(raw) + if not isinstance(compact, str): + return None + if len(compact) == 8: + for fmt in (compact8_fmt, "DDMMYYYY", "YYYYMMDD"): + if not fmt: + continue + parsed = _parse_date_with_format(compact, fmt) + if parsed is not None: + return parsed + if len(compact) == 6: + parsed = _parse_date_with_format(compact, compact6_fmt or "DDMMYY") + if parsed is not None: + return parsed + return None + + +def _rule_date_to_ddmmyyyy(series: pd.Series) -> pd.Series: + """Normaliza fechas numéricas, alfanuméricas, con meses en texto y con ruido.""" + stripped = series.astype(str).str.replace(r"[\s\/\-\.]", "", regex=True) + stripped = stripped.where(series.notna(), other=pd.NA) + sep_fmt = detect_date_format(series) or detect_date_format(stripped) + compact8_fmt, compact6_fmt = _detect_compact_date_formats(series) + + def _fn(x: Any): + if pd.isna(x) or str(x).strip() == "": + return x + parsed = _parse_any_date_to_tuple( + x, + sep_fmt=sep_fmt, + compact8_fmt=compact8_fmt, + compact6_fmt=compact6_fmt, + ) + if parsed is None: + return x + d, m, y = parsed + return f"{d:02d}/{m:02d}/{y:04d}" + + return series.map(_fn) + + def _rule_date_numeric_to_ddmmyyyy(series: pd.Series) -> pd.Series: """Entrada: fechas en formato numérico (ej. 20240315 o 15032024). @@ -1166,12 +1408,102 @@ def _rule_date_messy_text_to_ddmmyyyy(series: pd.Series) -> pd.Series: # Tipo y número de documento (CC, NIT, CE, MS, …) → 2 columnas # --------------------------------------------------------------------------- -def split_doc_tipo_y_numero_texto(raw: Any) -> tuple[Optional[str], Optional[str]]: - """Separa cadenas `CC-######`, `NIT900…`, `CE 12-34-56` en (tipo, solo dígitos). +_DOC_TYPE_ALIASES: dict[str, tuple[str, ...]] = { + "CC": ("CC", "Cedula", "CD", "C", "C.C", "C C", "CCC"), + "TI": ("TI", "Tarjeta", "Tarjeta Identidad", "T1", "T", "TL"), + "CN": ("CN", "Certificado", "Nacido Vivo", "Certificado Nacido Vivo"), + "CE": ("CE", "Cedula de Extranjeria", "Extranjera"), + "ME": ("ME", "Menor sin Identificacion"), + "PS": ("PS", "Pasaporte"), + "PT": ("PT", "Permiso por proteccion Temporal"), + "RC": ("RC", "Registro Civil", "Registro", "TC"), + "SC": ("SC", "Salvo Conducto", "Salvoconducto"), + "PE": ("PE", "Permiso especial de permanencia", "Permiso"), +} - Tipo: bloque de letras Unicode al inicio. Número: todos los dígitos a partir de - la primera cifra (miles: solo dígitos en salida). Celdas solo numéricas o sin - letras iniciales devuelven (None, None). Solo letras, sin cifras: (tipo, None). + +def _ascii_upper(value: Any) -> str: + text = unicodedata.normalize("NFKD", str(value or "")) + text = "".join(c for c in text if not unicodedata.combining(c)) + return text.upper() + + +def _normalize_doc_type_key(value: Any) -> str: + """Clave comparable: sin acentos, espacios, puntos ni símbolos.""" + return re.sub(r"[^A-Z0-9]", "", _ascii_upper(value)) + + +def _clean_doc_type_text(value: Any) -> Optional[str]: + cleaned = re.sub(r"[^A-Z0-9]+", " ", _ascii_upper(value)).strip() + cleaned = re.sub(r"\s+", " ", cleaned) + return cleaned or None + + +_DOC_TYPE_ALIAS_BY_KEY: dict[str, str] = { + _normalize_doc_type_key(alias): canonical + for canonical, aliases in _DOC_TYPE_ALIASES.items() + for alias in aliases +} + +_DOC_TYPE_ALIAS_DISPLAY_BY_KEY: dict[str, str] = {} +for _canonical, _aliases in _DOC_TYPE_ALIASES.items(): + for _alias in _aliases: + _alias_key = _normalize_doc_type_key(_alias) + _alias_display = _clean_doc_type_text(_alias) + if _alias_key and _alias_display: + _DOC_TYPE_ALIAS_DISPLAY_BY_KEY.setdefault(_alias_key, _alias_display) + +_DOC_TYPE_ALIAS_KEYS_SORTED: list[str] = sorted( + _DOC_TYPE_ALIAS_BY_KEY, + key=lambda k: (len(k), k), + reverse=True, +) + + +def homologate_doc_type(value: Any) -> Optional[str]: + """Homologa un tipo documental a sigla; si no existe, conserva texto limpio.""" + if _is_nullish_scalar(value): + return None + key = _normalize_doc_type_key(value) + if not key or key.lower() in ("nan", "none", "nat"): + return None + return _DOC_TYPE_ALIAS_BY_KEY.get(key) or _clean_doc_type_text(value) + + +def _compact_doc_text(value: Any) -> str: + return _normalize_doc_type_key(value) + + +def _looks_like_short_alias_prefix(compact: str, alias_key: str) -> bool: + """Evita que alias de 1-2 chars capturen palabras como CODIGO o TOTAL.""" + if not compact.startswith(alias_key): + return False + rest = compact[len(alias_key):] + if not rest: + return True + return bool(rest[0].isdigit()) + + +def _match_known_doc_type_prefix(compact: str) -> Optional[tuple[str, str]]: + for alias_key in _DOC_TYPE_ALIAS_KEYS_SORTED: + if len(alias_key) <= 2: + if not _looks_like_short_alias_prefix(compact, alias_key): + continue + elif not compact.startswith(alias_key): + continue + return _DOC_TYPE_ALIAS_BY_KEY[alias_key], alias_key + return None + + +def split_doc_tipo_y_numero_texto( + raw: Any, + *, + homologate: bool = True, +) -> tuple[Optional[str], Optional[str]]: + """Separa tipo documental y número, tolerando nombres largos y símbolos. + + Si `homologate` es True, tipos conocidos se devuelven como sigla (CC, TI, + etc.). Los tipos desconocidos se conservan limpios en mayúsculas. """ if _is_nullish_scalar(raw) and not (isinstance(raw, str) and not str(raw).strip()): return (None, None) @@ -1180,23 +1512,33 @@ def split_doc_tipo_y_numero_texto(raw: Any) -> tuple[Optional[str], Optional[str return (None, None) if t and t[0].isdigit(): return (None, None) + + compact = _compact_doc_text(t) + if not compact: + return (None, None) + + match = _match_known_doc_type_prefix(compact) + if match is not None: + canonical, alias_key = match + rest = compact[len(alias_key):] + nums = re.sub(r"[^\d]", "", rest) + if homologate: + return (canonical, nums if nums else None) + raw_type = _DOC_TYPE_ALIAS_DISPLAY_BY_KEY.get(alias_key) or canonical + return (raw_type, nums if nums else None) + m = re.match( - r"^([A-Za-zÑñ\u00C0-\u017F\u00D1\u00F1\u0100-\u02CC]+)" # letras (1+) - r"[\s\-.#:/_,\u00A0]*" # separación opcional - r"([0-9].*)\s*$", # primera cifra + r"^([^\d]+?)\s*([0-9].*)\s*$", t, ) if m: - to_t = "".join(c for c in m.group(1) if c.isalpha()) - to_t = to_t.upper() if to_t else None + to_t = _clean_doc_type_text(m.group(1)) nums = re.sub(r"[^\d]", "", m.group(2)) - return (to_t, nums if nums else None) - m2 = re.match( - r"^([A-Za-zÑñ\u00C0-\u017F\u00D1\u00F1\u0100-\u02CC]+)\s*$", t, - ) - if m2 and not re.search(r"\d", t): - to_t = "".join(c for c in m2.group(1) if c.isalpha()) - return (to_t.upper() if to_t else None, None) + return (homologate_doc_type(to_t) if homologate else to_t, nums if nums else None) + + if not re.search(r"\d", t): + to_t = homologate_doc_type(t) if homologate else _clean_doc_type_text(t) + return (to_t, None) return (None, None) @@ -1218,13 +1560,13 @@ def _rule_passthrough_series(s: pd.Series) -> pd.Series: def _apply_doc_id_split_columns( - result: pd.DataFrame, col: str, log: list[str] + result: pd.DataFrame, col: str, log: list[str], *, homologate_type: bool = True ) -> None: tcol, ncol = _alloc_doc_split_column_names(str(col), set(result.columns)) st_tipo: list[Optional[str]] = [] st_num: list[Any] = [] for v in result[col].tolist(): - tt, nn = split_doc_tipo_y_numero_texto(v) + tt, nn = split_doc_tipo_y_numero_texto(v, homologate=homologate_type) st_tipo.append(tt) st_num.append(nn if nn is not None else pd.NA) result[tcol] = st_tipo @@ -1239,13 +1581,21 @@ CLEANING_RULES: list[dict[str, Any]] = [ { "id": "strip_special_spaces", "name": "Eliminar caracteres especiales y espacios", - "description": "Deja solo caracteres alfanuméricos (A-Z, a-z, 0-9).", + "description": ( + "Deja solo caracteres alfanuméricos (A-Z, a-z, 0-9). No usar antes de " + "«Normalizar fecha» en columnas con fecha y hora: pega cifras y puede impedir " + "interpretar el valor (aplique primero la normalización de fecha o use la regla " + "que conserva separadores)." + ), "fn": _rule_strip_special_spaces, }, { "id": "strip_special_keep_date_seps", - "name": "Quitar símbolos; conserva /, - y espacios (fechas o texto con fechas)", - "description": "Quita comas, $, comillas, etc. y deja letras, números, /, - y espacios. Adecuado p. ej. en columna 'Fecha Servicio'.", + "name": "Quitar símbolos; conserva /, -, : y espacios (fechas o texto con fechas)", + "description": ( + "Quita comas, $, comillas, etc. y deja letras, números, /, -, : (hora) y espacios. " + "Adecuado p. ej. en columna 'Fecha Servicio'." + ), "fn": _rule_strip_special_keep_date_seps, }, { @@ -1260,6 +1610,15 @@ CLEANING_RULES: list[dict[str, Any]] = [ "description": "Convierte a enteros con detección de miles/decimal (`,`/`.`) y moneda; redondea decimales.", "fn": _rule_to_integer, }, + { + "id": "date_to_ddmmyyyy", + "name": "Normalizar fecha → DD/MM/AAAA", + "description": ( + "Normaliza fechas numéricas, con separadores, con ruido o con meses en " + "texto (ej. 26-abril-2026) al formato DD/MM/AAAA." + ), + "fn": _rule_date_to_ddmmyyyy, + }, { "id": "date_numeric_to_ddmmyyyy", "name": "Conversión de formato numérico (Fechas) → DD/MM/AAAA", @@ -1293,6 +1652,15 @@ CLEANING_RULES: list[dict[str, Any]] = [ ), "fn": _rule_passthrough_series, }, + { + "id": "doc_type_homologate", + "name": "Homologar tipo de documento", + "description": ( + "Convierte tipos de documento escritos como nombre largo, sigla o texto con " + "símbolos a su abreviatura estándar (CC, TI, CE, RC, etc.)." + ), + "fn": _rule_doc_type_homologate, + }, ] _RULE_BY_ID: dict[str, Callable[[pd.Series], pd.Series]] = { @@ -1333,14 +1701,21 @@ def apply_rules( if col not in result.columns: log.append(f"[WARN] Columna no encontrada: {col}") continue + split_has_homologation = ( + "doc_id_split_tipo_numero" in rule_ids and "doc_type_homologate" in rule_ids + ) for rule_id in rule_ids: if rule_id == "doc_id_split_tipo_numero": try: - _apply_doc_id_split_columns(result, col, log) + _apply_doc_id_split_columns( + result, col, log, homologate_type=split_has_homologation + ) except Exception as exc: # noqa: BLE001 logger.exception("Error aplicando regla %s a %s", rule_id, col) log.append(f"[ERROR] '{col}' regla '{rule_id}': {exc}") continue + if rule_id == "doc_type_homologate" and split_has_homologation: + continue fn = _RULE_BY_ID.get(rule_id) if fn is None: log.append(f"[WARN] Regla desconocida '{rule_id}' en columna '{col}'") diff --git a/application/csv_manager.py b/application/csv_manager.py index a288517..553dcb8 100644 --- a/application/csv_manager.py +++ b/application/csv_manager.py @@ -2,25 +2,51 @@ Maneja la escritura incremental segura (con lock) en archivos CSV. """ -import os import csv +import os +import re import threading -from typing import List, Dict, Any, Optional +from typing import Any, Dict, List + +_LINE_BREAK_RE = re.compile(r"[\r\n\v\f\x85\u2028\u2029]+") + + +def sanitize_csv_cell_value(value: Any) -> Any: + """Reemplaza saltos de línea en valores de celda por un espacio (una fila física por registro). + + Conserva ``None`` y escalares numéricos/booleanos sin tocar. Cadenas y demás + tipos se normalizan solo si la representación en texto contiene saltos. + """ + if value is None: + return value + if isinstance(value, bool): + return value + if isinstance(value, (int, float)): + return value + if isinstance(value, str): + if not _LINE_BREAK_RE.search(value): + return value + return _LINE_BREAK_RE.sub(" ", value).strip() + s = str(value) + if not _LINE_BREAK_RE.search(s): + return value + return _LINE_BREAK_RE.sub(" ", s).strip() + class CSVManager: - def __init__(self): - self._lock = threading.Lock() + def __init__(self): + self._lock = threading.Lock() - def append_rows(self, csv_path: str, rows: List[Dict[str, Any]], header: List[str]) -> None: - """Anexa filas a un CSV, creando encabezado si el archivo no existe.""" - os.makedirs(os.path.dirname(csv_path) or ".", exist_ok=True) - file_exists = os.path.exists(csv_path) and os.path.getsize(csv_path) > 0 - - with self._lock: - with open(csv_path, "a", encoding="utf-8", newline="") as f: - writer = csv.DictWriter(f, fieldnames=header, extrasaction="ignore", delimiter='|') - if not file_exists: - writer.writeheader() - for r in rows: - writer.writerow(r) + def append_rows(self, csv_path: str, rows: List[Dict[str, Any]], header: List[str]) -> None: + """Anexa filas a un CSV, creando encabezado si el archivo no existe.""" + os.makedirs(os.path.dirname(csv_path) or ".", exist_ok=True) + file_exists = os.path.exists(csv_path) and os.path.getsize(csv_path) > 0 + with self._lock: + with open(csv_path, "a", encoding="utf-8", newline="") as f: + writer = csv.DictWriter(f, fieldnames=header, extrasaction="ignore", delimiter="|") + if not file_exists: + writer.writeheader() + for r in rows: + sanitized = {k: sanitize_csv_cell_value(v) for k, v in r.items()} + writer.writerow(sanitized) diff --git a/gui/controller.py b/gui/controller.py index 6641b1b..76741ec 100644 --- a/gui/controller.py +++ b/gui/controller.py @@ -1201,35 +1201,71 @@ class AppController: if not filepath: return - try: - df, delimiter = csv_cleaning.load_csv(filepath) - except Exception as e: - logger.exception("Error leyendo CSV para limpieza") - mb.showerror("Error", f"No se pudo leer el CSV:\n{e}") - return - - self._cleaning_df = df self._cleaning_result_df = None self.view.set_cleaning_csv_path(filepath) + self.view.enable_cleaning_controls(False) self.view.enable_cleaning_save(False) self.view.set_cleaning_result(None) + self.view.set_cleaning_preview(None) + self.view.set_cleaning_status("Cargando CSV...") + try: + self.view.set_cleaning_loading(True, "Leyendo archivo CSV...") + except Exception: + pass + + def worker(): + try: + df, delimiter = csv_cleaning.load_csv(filepath) + self.view.after( + 0, + lambda d=df, delim=delimiter: self._on_cleaning_csv_loaded( + d, delim + ), + ) + summaries = csv_cleaning.analyze_dataframe(df) + self._cleaning_df = df + self.view.after( + 0, + lambda d=df, s=summaries: self._finish_cleaning_csv_load(d, s), + ) + except Exception as e: + logger.exception("Error leyendo o analizando CSV para limpieza") + err = str(e) + self.view.after(0, lambda m=err: self._fail_cleaning_csv_load(m)) + + threading.Thread(target=worker, daemon=True).start() + + def _on_cleaning_csv_loaded(self, df: pd.DataFrame, delimiter: str): self.view.set_cleaning_status( - f"CSV cargado: {len(df):,} filas x {len(df.columns)} columnas." + f"CSV leído: {len(df):,} filas x {len(df.columns)} columnas. Analizando columnas..." ) try: + self.view.set_cleaning_loading(True, "Analizando columnas...") self.view.set_cleaning_delimiter_chip(delimiter) except Exception: pass - try: - summaries = csv_cleaning.analyze_dataframe(df) - except Exception as e: - logger.exception("Error analizando CSV para limpieza") - mb.showerror("Error", f"No se pudo analizar el CSV:\n{e}") - return - + def _finish_cleaning_csv_load(self, df: pd.DataFrame, summaries): self.view.populate_cleaning_columns(summaries) self.view.set_cleaning_preview(df) + self.view.set_cleaning_status( + f"CSV cargado: {len(df):,} filas x {len(df.columns)} columnas." + ) + try: + self.view.set_cleaning_loading(False) + except Exception: + pass + self.view.enable_cleaning_controls(True) + + def _fail_cleaning_csv_load(self, message: str): + self._cleaning_df = None + self.view.set_cleaning_status("Error al cargar CSV") + try: + self.view.set_cleaning_loading(False) + except Exception: + pass + self.view.enable_cleaning_controls(True) + mb.showerror("Error", f"No se pudo cargar o analizar el CSV:\n{message}") def on_cleaning_apply(self): """Aplica las reglas seleccionadas en un hilo y actualiza el resultado.""" diff --git a/gui/tabs/cleaning_tab.py b/gui/tabs/cleaning_tab.py index 884a097..30cfcf6 100644 --- a/gui/tabs/cleaning_tab.py +++ b/gui/tabs/cleaning_tab.py @@ -56,15 +56,17 @@ _TYPE_MENU_VALUES: list[str] = [ ] _RULE_SHORT_LABELS: dict[str, str] = { - "strip_special_spaces": "Sin especiales/espacios", - "strip_special_keep_date_seps": "Símbolos: conserva /, -, espacios", + "strip_special_spaces": "Sin especiales/espacios (no antes de fecha+hora)", + "strip_special_keep_date_seps": "Símbolos: conserva /, -, :, espacios", "strip_spaces": "Sin espacios", "to_integer": "A entero", + "date_to_ddmmyyyy": "Normalizar fecha → DD/MM/AAAA", "date_numeric_to_ddmmyyyy": "Fecha numérica → DD/MM/AAAA", "date_alpha_to_ddmmyyyy": "Fecha alfanumérica → DD/MM/AAAA", "date_messy_text_to_ddmmyyyy": "Fecha con ruido (texto) → DD/MM/AAAA", "round_decimals_custom": "Aprox. decimales (>0.5 sube)", "doc_id_split_tipo_numero": "Dividir doc (tipo + cifras)", + "doc_type_homologate": "Homologar tipo doc", } @@ -109,6 +111,7 @@ class CleaningTab: self._search_var: Optional[ctk.StringVar] = None # Tipo de dato forzado por el usuario (id interno) por columna; vacío = inferido self._type_overrides: dict[str, str] = {} + self._syncing_rule_vars = False self._setup_layout() self._create_file_section() @@ -163,6 +166,27 @@ class CleaningTab: self.chip_delim = self._make_chip(chips_frame, "Delim: —") self.chip_delim.pack(side="left") + self.loading_frame = ctk.CTkFrame(frame, fg_color="transparent") + self.loading_frame.grid( + row=2, column=0, columnspan=4, sticky="ew", padx=15, pady=(0, 12) + ) + self.loading_frame.grid_columnconfigure(1, weight=1) + self.loading_label = ctk.CTkLabel( + self.loading_frame, + text="", + font=self.fonts["italic_small"], + text_color=self.colors["gray_light"], + anchor="w", + ) + self.loading_label.grid(row=0, column=0, sticky="w", padx=(0, 10)) + self.loading_progress = ctk.CTkProgressBar( + self.loading_frame, + mode="indeterminate", + height=8, + ) + self.loading_progress.grid(row=0, column=1, sticky="ew") + self.loading_frame.grid_remove() + def _make_chip(self, parent, text: str) -> ctk.CTkLabel: return ctk.CTkLabel( parent, text=text, font=self.fonts["italic_small"], @@ -429,6 +453,23 @@ class CleaningTab: def set_status(self, text: str): self.lbl_status.configure(text=text) + def set_loading(self, loading: bool, text: str = ""): + """Muestra u oculta una barra indeterminada para operaciones de CSV.""" + if loading: + self.loading_label.configure(text=text or "Procesando CSV...") + self.loading_frame.grid() + try: + self.loading_progress.start() + except Exception: + pass + else: + try: + self.loading_progress.stop() + except Exception: + pass + self.loading_label.configure(text="") + self.loading_frame.grid_remove() + def set_preview(self, df: Optional[pd.DataFrame]): """Renderiza el DataFrame original en la tabla de vista previa.""" if df is None or len(df) == 0: @@ -643,7 +684,10 @@ class CleaningTab: if rule is None: continue var = ctk.BooleanVar(value=False) - var.trace_add("write", lambda *_a, c=summary.name: self._on_rule_toggle(c)) + var.trace_add( + "write", + lambda *_a, c=summary.name, r=rid: self._on_rule_toggle(c, r), + ) label = _RULE_SHORT_LABELS.get(rid, rule["name"]) chk = ctk.CTkCheckBox( rules_frame, text=label, variable=var, @@ -850,7 +894,10 @@ class CleaningTab: if rule is None: continue var = ctk.BooleanVar(value=False) - var.trace_add("write", lambda *_a, c=col_name: self._on_rule_toggle(c)) + var.trace_add( + "write", + lambda *_a, c=col_name, r=rid: self._on_rule_toggle(c, r), + ) label = _RULE_SHORT_LABELS.get(rid, rule["name"]) chk = ctk.CTkCheckBox( ui.rules_frame, text=label, variable=var, @@ -883,10 +930,30 @@ class CleaningTab: if ui.expanded != expanded: self._toggle_column(col_name) - def _on_rule_toggle(self, col_name: str): + def _on_rule_toggle(self, col_name: str, changed_rule_id: Optional[str] = None): ui = self._column_uis.get(col_name) if ui is None: return + if ( + changed_rule_id == "doc_id_split_tipo_numero" + and not self._syncing_rule_vars + ): + split_selected = any( + rid == "doc_id_split_tipo_numero" and bool(var.get()) + for rid, _chk, var in ui.rule_vars + ) + if split_selected: + for rid, _chk, var in ui.rule_vars: + if rid == "doc_type_homologate" and not bool(var.get()): + self._syncing_rule_vars = True + try: + var.set(True) + finally: + self._syncing_rule_vars = False + self.set_status( + f"Columna «{col_name}»: se activó homologación de tipo de documento." + ) + break selected = sum(1 for _rid, _chk, var in ui.rule_vars if bool(var.get())) text = f"{selected} regla{'s' if selected != 1 else ''}" color = self.colors["success"] if selected else self.colors["gray_light"] diff --git a/gui/view.py b/gui/view.py index 043c15c..a0c246f 100644 --- a/gui/view.py +++ b/gui/view.py @@ -399,5 +399,8 @@ class InvoiceAppView(ctk.CTk): def set_cleaning_status(self, text: str): self.cleaning_tab.set_status(text) + def set_cleaning_loading(self, loading: bool, text: str = ""): + self.cleaning_tab.set_loading(loading, text) + def set_cleaning_delimiter_chip(self, delimiter: str): self.cleaning_tab.set_delimiter_chip(delimiter) diff --git a/tests/test_csv_cleaning.py b/tests/test_csv_cleaning.py new file mode 100644 index 0000000..530dca5 --- /dev/null +++ b/tests/test_csv_cleaning.py @@ -0,0 +1,233 @@ +from __future__ import annotations + +import pandas as pd + +from application.csv_cleaning import ( + apply_rules, + homologate_doc_type, + split_doc_tipo_y_numero_texto, +) + + +def test_homologate_doc_type_aliases() -> None: + cases = { + "Cedula": "CC", + "C.C": "CC", + "C C": "CC", + "Registro Civil": "RC", + "Salvoconducto": "SC", + "TC": "RC", + "C\u00e9dula de Extranjeria": "CE", + } + + for raw, expected in cases.items(): + assert homologate_doc_type(raw) == expected + + +def test_split_doc_type_with_long_names() -> None: + cases = { + "Cedula 123.456": ("CC", "123456"), + "Registro Civil: 001-23": ("RC", "00123"), + "Permiso especial de permanencia 999": ("PE", "999"), + "Permiso por proteccion Temporal / 555": ("PT", "555"), + } + + for raw, expected in cases.items(): + assert split_doc_tipo_y_numero_texto(raw) == expected + + +def test_split_doc_type_with_abbreviations_and_noise() -> None: + cases = { + "C.C-123": ("CC", "123"), + "TL_987": ("TI", "987"), + "CE/1.234.567": ("CE", "1234567"), + "T1 # 44": ("TI", "44"), + } + + for raw, expected in cases.items(): + assert split_doc_tipo_y_numero_texto(raw) == expected + + +def test_split_unknown_type_preserves_clean_uppercase() -> None: + assert split_doc_tipo_y_numero_texto("ABC Especial: 123-45") == ( + "ABC ESPECIAL", + "12345", + ) + + +def test_split_without_homologation_preserves_clean_type() -> None: + assert split_doc_tipo_y_numero_texto( + "Cedula 123.456", + homologate=False, + ) == ("CEDULA", "123456") + + +def test_apply_rules_homologates_type_column_only() -> None: + df = pd.DataFrame( + { + "tipo": [ + "Cedula", + "C.C", + "Registro Civil", + "Salvoconducto", + "TC", + "Extranjera", + ] + } + ) + + result, log = apply_rules(df, {"tipo": ["doc_type_homologate"]}) + + assert result["tipo"].tolist() == ["CC", "CC", "RC", "SC", "RC", "CE"] + assert any("Homologar tipo de documento" in line for line in log) + + +def test_apply_rules_split_uses_homologation_when_selected() -> None: + df = pd.DataFrame( + { + "documento": [ + "Cedula 123.456", + "Registro Civil: 001-23", + "ABC Especial: 123-45", + ] + } + ) + + result, _log = apply_rules( + df, + {"documento": ["doc_id_split_tipo_numero", "doc_type_homologate"]}, + ) + + assert result["documento_doc_tipo"].tolist() == ["CC", "RC", "ABC ESPECIAL"] + assert result["documento_doc_numeros"].tolist() == ["123456", "00123", "12345"] + assert result["documento"].tolist() == df["documento"].tolist() + + +def test_apply_rules_split_can_skip_homologation() -> None: + df = pd.DataFrame({"documento": ["Cedula 123.456"]}) + + result, _log = apply_rules(df, {"documento": ["doc_id_split_tipo_numero"]}) + + assert result["documento_doc_tipo"].tolist() == ["CEDULA"] + assert result["documento_doc_numeros"].tolist() == ["123456"] + + +def test_apply_rules_unified_date_rule_numeric_and_separated() -> None: + df = pd.DataFrame( + { + "fecha": [ + "20260426", + "26042026", + "260426", + "26/04/2026", + "04/26/2026", + "2026-04-26 10:30", + ] + } + ) + + result, log = apply_rules(df, {"fecha": ["date_to_ddmmyyyy"]}) + + assert result["fecha"].tolist() == [ + "26/04/2026", + "26/04/2026", + "26/04/2026", + "26/04/2026", + "26/04/2026", + "26/04/2026", + ] + assert any("Normalizar fecha" in line for line in log) + + +def test_apply_rules_unified_date_rule_text_months() -> None: + df = pd.DataFrame( + { + "fecha": [ + "26-abril-2026", + "26 abril 2026", + "26/abr/2026", + "abril 26 2026", + "2026 abril 26", + "15-sept-25", + "07 setiembre 2024", + ] + } + ) + + result, _log = apply_rules(df, {"fecha": ["date_to_ddmmyyyy"]}) + + assert result["fecha"].tolist() == [ + "26/04/2026", + "26/04/2026", + "26/04/2026", + "26/04/2026", + "26/04/2026", + "15/09/2025", + "07/09/2024", + ] + + +def test_apply_rules_unified_date_spanish_abbr_month_with_time() -> None: + df = pd.DataFrame( + { + "fecha": [ + "15/may./2024 13:44", + "22/may./2024 11:09", + "30/04/2024", + ] + } + ) + + result, _log = apply_rules(df, {"fecha": ["date_to_ddmmyyyy"]}) + + assert result["fecha"].tolist() == [ + "15/05/2024", + "22/05/2024", + "30/04/2024", + ] + + +def test_apply_rules_date_after_strip_special_spaces_glued_year_time() -> None: + df = pd.DataFrame({"fecha": ["15/may./2024 13:44"]}) + + result, _log = apply_rules( + df, + {"fecha": ["strip_special_spaces", "date_to_ddmmyyyy"]}, + ) + + assert result["fecha"].tolist() == ["15/05/2024"] + + +def test_apply_rules_keep_date_seps_preserves_time_then_normalizes() -> None: + df = pd.DataFrame({"fecha": ["30/04/2024 13:44"]}) + + result, _log = apply_rules( + df, + {"fecha": ["strip_special_keep_date_seps", "date_to_ddmmyyyy"]}, + ) + + assert result["fecha"].tolist() == ["30/04/2024"] + + +def test_apply_rules_unified_date_rule_noise_and_unparseable_values() -> None: + df = pd.DataFrame( + { + "fecha": [ + "**26_abril_2026++", + "Fecha: 26-abril-2026", + "xx26/04/2026yy", + "no es fecha", + "31-febrero-2026", + ] + } + ) + + result, _log = apply_rules(df, {"fecha": ["date_to_ddmmyyyy"]}) + + assert result["fecha"].tolist() == [ + "26/04/2026", + "26/04/2026", + "26/04/2026", + "no es fecha", + "31-febrero-2026", + ] diff --git a/tests/test_csv_manager.py b/tests/test_csv_manager.py new file mode 100644 index 0000000..489db75 --- /dev/null +++ b/tests/test_csv_manager.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +import tempfile +from pathlib import Path + +from application.csv_manager import CSVManager, sanitize_csv_cell_value + + +def test_sanitize_collapses_newlines() -> None: + assert sanitize_csv_cell_value("a\nb") == "a b" + assert sanitize_csv_cell_value("a\r\nb") == "a b" + assert sanitize_csv_cell_value(" x\v\fy ") == "x y" + + +def test_sanitize_unicode_line_separators() -> None: + assert sanitize_csv_cell_value("a\u2028b\u2029c") == "a b c" + assert sanitize_csv_cell_value("a\x85b") == "a b" + + +def test_sanitize_preserves_none_and_scalars() -> None: + assert sanitize_csv_cell_value(None) is None + assert sanitize_csv_cell_value(42) == 42 + assert sanitize_csv_cell_value(3.14) == 3.14 + assert sanitize_csv_cell_value(False) is False + assert sanitize_csv_cell_value(True) is True + + +def test_sanitize_plain_string_unchanged_identity() -> None: + s = "sin saltos" + assert sanitize_csv_cell_value(s) is s + + +def test_append_rows_one_physical_line_per_record() -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "batch.csv" + mgr = CSVManager() + header = ["radicado", "descripcion"] + rows = [ + { + "radicado": "235662283", + "descripcion": ( + '"Las retenciones son a nivel informativo.\n' + "Paciente: 1010079549 - HERAZO\n" + 'Contrato No. URGENCIAS."' + ), + }, + {"radicado": "2", "descripcion": "ok"}, + ] + mgr.append_rows(str(path), rows, header) + text = path.read_text(encoding="utf-8") + physical_lines = text.splitlines() + assert len(physical_lines) == 3