diff --git a/application/csv_cleaning.py b/application/csv_cleaning.py index 7dd04c4..b9f105d 100644 --- a/application/csv_cleaning.py +++ b/application/csv_cleaning.py @@ -11,10 +11,12 @@ from __future__ import annotations import csv as _csv import logging +import math import random import re +from collections import Counter from dataclasses import dataclass, field -from typing import Any, Callable, Optional +from typing import Any, Callable, Literal, Optional import numpy as np import pandas as pd @@ -317,73 +319,557 @@ def _parse_date_with_format(value: str, fmt: str) -> Optional[tuple[int, int, in _BOOLEAN_VALUES = {"true", "false", "si", "sí", "no", "1", "0", "yes"} -def _infer_column_type(series: pd.Series) -> tuple[str, Optional[str]]: - """Devuelve (tipo, formato_fecha_detectado).""" +# PARSING numérico tolerante (miles/decimal) sin fijar un solo locale +_HINT_MAJORITY = 0.6 +_MIN_HINT_VOTES = 1 + + +def _is_nullish_scalar(x: Any) -> bool: + if x is None or x is pd.NA: + return True + if isinstance(x, (bool, np.bool_)): + return False + if isinstance(x, (int, float, np.integer, np.floating)) and not isinstance( + x, (bool, np.bool_) + ): + if isinstance(x, float) and (np.isnan(x) or pd.isna(x)): + return True + return False + t = str(x).strip().lower() + if not t or t in ("nan", "none", "nat", "null", ""): + return True + return False + + +def _strip_currency(s: str) -> str: + t = s.replace("\u00a0", " ").replace("\u202f", " ") + t = re.sub(r"[\s]+", "", t) + t = t.strip() + if not t or t.lower() in ("nan", "none", "nat"): + return "" + neg = False + if t[0:1] == "-": + neg, t = not neg, t[1:].lstrip() + if t[0:1] == "(" and t.endswith(")"): + neg, t = not neg, t[1:-1] + t = re.sub( + r"^[\$€\u00a3\u0e3f\uffe5\uffe6\uff04\ufe69]+", + "", + t, + count=1, + ) + t = re.sub(r"[\$€%]+$", "", t) + if not t and neg: + return "-0" + return ("-" if neg else "") + t + + +def _extract_numeric_chunk(s: str) -> str: + t = s.strip() + t = t.replace("$", "").replace("€", "").replace("\u00a3", "") + t = t.replace(" ", "") + cands1 = re.findall( + r"[+\-]?(?:\d{1,3}(?:[.,]\d{3})*|\d+)(?:[.,]\d+)?", t + ) + cands2 = re.findall( + r"[+\-]?\d[.,\d ]*\d", t + ) + cands = cands1 + cands2 + if not cands: + m = re.search(r"[\d.,\-]+", t) + if m: + c = m.group(0).replace("(", "-").replace(")", "") + return c.lstrip("+") + return "" + best = max(cands, key=len) if cands else "" + best = best.replace(" ", "").replace("(", "-").replace(")", "") + if not re.search(r"\d", best): + return "" + return best.lstrip("+") + + +def _as_float_s(s: str) -> Optional[float]: + if not s: + return None + try: + f = float(s) + if not np.isfinite(f) or abs(f) > 1e100: + return None + return f + except (ValueError, TypeError, OverflowError): + return None + + +def _parse_both_separators(body: str) -> Optional[float]: + """El separador más a la DERECHA entre `,` y `.` es el decimal; el otro, miles.""" + if "," not in body or "." not in body: + return None + t = body.strip() + if re.search(r"[^0-9,.\-]", t): + t2 = re.sub(r"[^0-9,.\-]", "", t) + t = t2 + if not t: + return None + neg = t[0:1] == "-" + b = t[1:].lstrip() if neg else t + if not b: + return None + i_comma, i_dot = b.rfind(","), b.rfind(".") + if min(i_comma, i_dot) < 0: + return None + dpos = max(i_comma, i_dot) + th = "," if b[dpos] == "." else "." + intp, fr = b[:dpos], b[dpos + 1 :] + if not re.fullmatch(r"\d{1,4}$", fr): + return None + intc = intp.replace(th, "") + if not intc.isdigit(): + return None + v = _as_float_s(f"{intc}.{fr}") + if v is None: + return None + if neg and v and v is not None: + return -abs(v) + return v + + +def _only_commas(w: str, dec_hint: Optional[Literal[",", "."]]) -> Optional[float]: + neg = w[0:1] == "-" + wn = w[1:].lstrip() if neg else w + if wn and re.search(r"[^0-9,]", wn): + wn = re.sub(r"[^0-9,]", "", wn) + n = wn.count(",") + if n == 0 and wn.isdigit(): + v = float(wn) + return -v if neg else v + if n == 0 or "." in wn: + return None + if n == 1: + i = wn.rfind(",") + h, t = wn[:i], wn[i + 1 :] + if not t.isdigit() or not h.isdigit(): + return None + if len(t) in (1, 2): + v = _as_float_s(f"{h}.{t}") + if v is None: + return None + return -v if neg else v + if len(t) == 3: + if dec_hint is None or dec_hint == ",": + v = _as_float_s(f"{h}{t}") + if v is None: + return None + return -v if neg else v + if dec_hint == ".": + v = _as_float_s(wn.replace(",", "")) + if v is None: + return None + return -v if neg else v + v = _as_float_s(wn.replace(",", "")) + if v is None: + return None + return -v if neg else v + + +def _only_dots(w: str, dec_hint: Optional[Literal[",", "."]]) -> Optional[float]: + neg = w[0:1] == "-" + wn = w[1:].lstrip() if neg else w + if wn and re.search(r"[^0-9.]", wn): + wn = re.sub(r"[^0-9.]", "", wn) + n = wn.count(".") + if n == 0 and wn.isdigit(): + v = float(wn) + return -v if neg else v + if n == 0 or "," in wn: + return None + if n == 1: + i = wn.rfind(".") + h, t = wn[:i], wn[i + 1 :] + if not t.isdigit() or not h.isdigit(): + return None + if len(t) in (1, 2): + v = _as_float_s(f"{h}.{t}") + if v is None: + return None + return -v if neg else v + if len(t) == 3: + if dec_hint is None or dec_hint == ".": + v = _as_float_s(f"{h}{t}") + if v is None: + return None + return -v if neg else v + if dec_hint == ",": + v = _as_float_s(wn.replace(".", "")) + if v is None: + return None + return -v if neg else v + v = _as_float_s(wn.replace(".", "")) + if v is None: + return None + return -v if neg else v + + +def _parse_lenient_one( + raw: Any, + decimal_hint: Optional[Literal[",", "."]] = None, +) -> float: + if _is_nullish_scalar(raw): + return float("nan") + t = _strip_currency(str(raw)) + if not t or t == "-": + return float("nan") + c = _extract_numeric_chunk(t) or t + c = c.strip() + if not c or c == "-": + return float("nan") + n_comma, n_dot = c.count(","), c.count(".") + if n_comma and n_dot: + f = _parse_both_separators(c) + elif n_comma and not n_dot: + f = _only_commas(c, decimal_hint) + elif n_dot and not n_comma: + f = _only_dots(c, decimal_hint) + else: + c2 = re.sub(r"[^0-9.\-]", "", c) + f = _as_float_s(c2) if c2 and re.fullmatch(r"-?\d+$", c2) else None + if f is not None and np.isfinite(f): + return f + return float("nan") + + +def _infer_decimal_hint_from_column( + series: pd.Series, +) -> Optional[Literal[",", "."]]: + """Voto con celdas que traen `,` y `.`: gana el de más a la derecha = decimal.""" + votes: list[Literal[",", "."]] = [] + for raw in series.dropna(): + if _is_nullish_scalar(raw): + continue + t = _strip_currency(str(raw)) + c = _extract_numeric_chunk(t) or t + c = c.strip() + if "," not in c or "." not in c or _parse_both_separators(c) is None: + continue + if c.rfind(",") > c.rfind("."): + votes.append(",") + else: + votes.append(".") + n = len(votes) + if n < _MIN_HINT_VOTES: + return None + a = sum(1 for v in votes if v == ",") + b = sum(1 for v in votes if v == ".") + if a + b == 0: + return None + if a / n >= _HINT_MAJORITY and a > b: + return "," + if b / n >= _HINT_MAJORITY and b > a: + return "." + return None + + +def _to_numeric_lenient( + series: pd.Series, + decimal_hint: Optional[Literal[",", "."]] = None, +) -> pd.Series: + """Convierte a flotante; pista de columna por voto (si `decimal_hint` es None).""" + hint = decimal_hint if decimal_hint is not None else _infer_decimal_hint_from_column( + series + ) + + def _cell(x: Any) -> float: + if _is_nullish_scalar(x): + return float("nan") + return _parse_lenient_one(x, hint) + + return series.map(_cell) + + +def _column_name_suggests_date_like(name: str) -> bool: + """Columnas cuyo nombre pide tratar el contenido como fecha a priori.""" + n = re.sub(r"[_\s.]+", " ", (name or "")).lower().strip() + if not n: + return False + if re.search( + r"\b(fecha|date|año|mes|dia|día|f[_-]?(alta|baja|proc|corte|pago|ini|fin|" + r"desde|hasta|vigenc|vence|venc|caduc|f\s?ec)\b|periodo|anio|ano|expir)", + n, + ) or n.startswith("fec_") or n.startswith("fecha"): + return True + return False + + +def _column_name_suggests_monetary_value(name: str) -> bool: + """importe, valor, neto, IVA, etc. — no interpretar cifras como fechas. + + Encabezados de tipo `fecha_…` no se tratan como moneda salvo mezcla con monto/valor. + """ + n = re.sub(r"[_\s.]+", " ", (name or "")).lower().strip() + if not n: + return False + if _column_name_suggests_date_like(name) and not re.search( + r"\b(monto|valor|importe|pago|total|bruto|neto|import|unit|precio|copago)\b", n + ): + return False + for key in ( + "monto", "importe", "import", "bruto", "neto", "subtotal", "sub total", + "unitario", "unidad", "valor", "precio", "pago", "copago", "sald", "dcto", + "franq", "cuota", "iva", "tarif", "costo", "carg", "reten", "retencion", + ): + if key in n: + if key == "iva" and re.search(r"(priva|giva|diva|viva|cliva)\b", n): + continue + if key == "carg" and re.search(r"descarg|recarg|abord", n): + continue + return True + return False + + +def _column_name_suggests_code_id(name: str) -> bool: + """Código, ref., radicado, tipoynúm de documento — no forzar 6-8 cifras como fechas (DDMMYY, etc.).""" + n = re.sub(r"[_\s.]+", " ", (name or "")).lower().strip() + if not n or _column_name_suggests_date_like(name): + return False + for key in ( + "codigo", "código", "codig", "códig", "cód_", + "sku", "ean", "upc", "isbn", "upn", + "ref", "referen", "radic", "nro_fac", "nro fac", "nit", "cédul", "cedul", + "autoriz", "correlat", "secuenc", "línea de", "linea de", "cupo", + "tipy", "tipi", "tipynum", "typnum", "type_n", "n_doc", "n_docu", + "núm doc", "n_docu", "doc_id", "docum", "id_doc", + ): + if key in n: + return True + if re.search( + r"tipo[_\s\\-]*(&|/|u|e|y)[_\s\\-]*num|tipo[_\s\\-]+doc|tipo_y_n", + n, + re.IGNORECASE, + ) or n.startswith("tip_"): + return True + if n.startswith("cod_") or n.startswith("cód_") or n.startswith("ref_"): + return True + return bool(re.search(r"\b(nro[._]fac|#?\s*autor|#?\s*ref)\b", n, re.IGNORECASE)) + + +def _column_name_suggests_quantity_or_number_not_bool(name: str) -> bool: + """Nombres típicos de cantidad, importe, fechas: no inferir bool por 0/1.""" + n = re.sub(r"[_\s.]+", " ", (name or "")).lower().strip() + if not n: + return False + if _column_name_suggests_monetary_value(name) or _column_name_suggests_code_id(name): + return True + for key in ( + "fecha", + "monto", + "importe", + "bruto", + "neto", + "subtotal", + "unitario", + "unidad", + "valor", + "precio", + "pago", + "copago", + "cantid", + "quantity", + "qty", + "nro", + "núm", + "n°", + ): + if key in n: + return True + return False + + +def _raw_strings_suggest_monetary_presentation(non_null: pd.Series) -> bool: + """$ €, miles 1,234,12 o .00: presentación monetaria aunque el float sea "entero".""" + if non_null is None or len(non_null) == 0: + return False + n_hit = 0 + n_seen = 0 + for raw in non_null.head(500): + if _is_nullish_scalar(raw): + continue + t = str(raw) + s = t.strip() + n_seen += 1 + if re.search(r"[$\u00a3\u20ac\uffe5\uffe6\ufe69\uff04]", t): + n_hit += 1 + continue + if re.search(r"\b\d{1,3}(?:[.,\u00a0]\d{3})+\b", s) or re.search( + r"(?= 0.10 + + +def _cell_looks_cifra_con_moneda_iso_al_final(s: str) -> bool: + """Solo cifra/símbolo (no prefijo de documento) y código de moneda al final. + + Excluye filas al estilo `CC12…` (típico alfanumérico). + """ + t = str(s).strip() + u = t.upper() + if not re.search( + r"(?i)(USD|COP|EUR|CLP|ARS|GBP|MXN|PEN|DOP|UYU|VES|CRC|PYG|BRL|BOB|GTQ)\b\s*\Z", u + ): + return False + if re.match( + r"^(?i)(CC|CE|NIT|NUIP|TI|PA|MS|RC|DNI|NUIP|CN|SC|AS|PE|PT)\D*[\d0]", u + ): + return False + return re.match( + r"^[\$€0-9\s,.\-+%/]", t, + ) is not None + + +def _raw_strings_suggest_alnum_mixed_presentation(non_null: pd.Series) -> bool: + """Letras y dígitos juntos (p. ej. `CC32…`, `A1-2B`); no es puro 'entero/decimal' lógicamente. + + Excluido: columnas con apariencia monetaria, y celdas solo cifra + ISO al final (100 USD). + """ + if _raw_strings_suggest_monetary_presentation(non_null): + return False + n_m = 0 + n_s = 0 + for raw in non_null.head(500): + if _is_nullish_scalar(raw): + continue + t = str(raw).strip() + n_s += 1 + if not re.search(r"\d", t) or not re.search( + r"[A-Za-zÑñ\u00c0-\u024f]", t, re.IGNORECASE + ): + continue + if _cell_looks_cifra_con_moneda_iso_al_final(t): + continue + n_m += 1 + if n_s == 0: + return False + return n_m / n_s >= 0.20 + + +def _cell_looks_like_documento_id_tipo_nit(s: str) -> bool: + """Cadenas estilo `CC32324029`, `CE 1.234.567` — no son fechas (evita 8 cifras como DDMM). + + Al menos una letra al inicio y 5-16 dígitos en la celda (no `Jan`+`2020` = 4 dígitos). + """ + t = str(s).strip() + if not t or len(t) < 3: + return False + if not re.match(r"^[A-Za-z]{1,3}", t): + return False + dig = re.sub(r"[^\d]", "", t) + if len(dig) < 5 or len(dig) > 16: + return False + return True + + +def _content_looks_like_doc_id_letter_prefix_column(non_null: pd.Series) -> bool: + """Muestras con prefijo documental (CC, NIT, …) y ≥5 dígitos → no columna de fecha.""" + n_ok = 0 + n_seen = 0 + for raw in non_null.head(500): + if _is_nullish_scalar(raw): + continue + s = str(raw).strip() + n_seen += 1 + if _cell_looks_like_documento_id_tipo_nit(s): + n_ok += 1 + if n_seen == 0: + return False + return n_ok / n_seen >= 0.20 + + +def _detect_messy_text_date_format_guess(non_null: pd.Series) -> Optional[str]: + """Fechas con ruido: compacta a dígitos; si hay consenso, reutiliza el detector. + (Depende de `_normalize_messy_date_to_compact`, definida más abajo en el módulo.) + """ + # resolución en tiempo de ejecución + norm = non_null.map(_normalize_messy_date_to_compact) # type: ignore[name-defined] + compacts: list[str] = [] + for c in norm: + if isinstance(c, str) and len(c) in (6, 8): + compacts.append(c) + n = max(len(non_null), 1) + if not compacts or len(compacts) < max(2, int(0.25 * n)): + return None + s_compact = pd.Series(compacts, dtype=object) + return detect_date_format(s_compact) + + +def _infer_column_type( + series: pd.Series, column_name: str = "" +) -> tuple[str, Optional[str]]: + """Devuelve (tipo, formato_fecha detectado; solo si aplica a tipo fecha).""" non_null = series.dropna().astype(str).str.strip() non_null = non_null[non_null != ""] if non_null.empty: return TYPE_EMPTY, None - # Booleano - lowered = non_null.str.lower() - if lowered.isin(_BOOLEAN_VALUES).all() and lowered.nunique() <= 3: - return TYPE_BOOLEAN, None + cname = str(column_name or "") - # Numérico - # Aceptamos separador de miles con "." o "," y decimal con "," o ".". - # Estrategia: intentar to_numeric tras normalizar. - numeric_candidate = _to_numeric_lenient(non_null) + # Boolean: no aplicar 0/1/ sí-no a columnas cuyo nombre pide importe, cantid., fecha… + if not _column_name_suggests_quantity_or_number_not_bool(cname): + lowered = non_null.str.lower() + if lowered.isin(_BOOLEAN_VALUES).all() and lowered.nunique() <= 3: + return TYPE_BOOLEAN, None + + is_money = _column_name_suggests_monetary_value(cname) + is_code = _column_name_suggests_code_id(cname) + is_doc_pattern = _content_looks_like_doc_id_letter_prefix_column(non_null) + + # Fecha: no forzar en importe/código, ni con datos tipo `CC`+cifra (6–8 dígitos sueltos) + if not is_money and not is_code and not is_doc_pattern: + date_fmt = detect_date_format(non_null) + if date_fmt: + return (TYPE_DATE, date_fmt) + messy_date_fmt = _detect_messy_text_date_format_guess(non_null) + if messy_date_fmt: + return (TYPE_DATE, messy_date_fmt) + + numeric_candidate = _to_numeric_lenient(non_null, decimal_hint=None) numeric_ratio = numeric_candidate.notna().mean() if numeric_ratio >= 0.9: + if _raw_strings_suggest_alnum_mixed_presentation(non_null): + return (TYPE_MIXED, None) is_all_integer = (numeric_candidate.dropna() % 1 == 0).all() - return (TYPE_INTEGER if is_all_integer else TYPE_DECIMAL), None + if is_all_integer and not _raw_strings_suggest_monetary_presentation(non_null): + return (TYPE_INTEGER, None) + return (TYPE_DECIMAL, None) - # Fecha - date_fmt = detect_date_format(non_null) - if date_fmt: - return TYPE_DATE, date_fmt - - # Categórico vs texto unique_ratio = non_null.nunique() / len(non_null) if non_null.nunique() <= 15 and unique_ratio < 0.3: return TYPE_CATEGORICAL, None - - # Mixto si mezcla números con texto y nada es dominante if 0.1 < numeric_ratio < 0.9: return TYPE_MIXED, None - return TYPE_TEXT, None -def _to_numeric_lenient(series: pd.Series) -> pd.Series: - """Intenta convertir a numérico aceptando separadores comunes.""" - s = series.astype(str).str.strip() - s = s.str.replace(r"\s+", "", regex=True) - - # Si hay ambos "." y ",", asumir "." como miles y "," como decimal - def _norm(x: str) -> str: - if "." in x and "," in x: - return x.replace(".", "").replace(",", ".") - if "," in x and "." not in x: - # coma decimal - return x.replace(",", ".") - return x - - s = s.map(_norm) - return pd.to_numeric(s, errors="coerce") - - def _applicable_rules_for_type(inferred_type: str) -> list[str]: - """Reglas sugeridas según tipo inferido.""" + """Identificadores de reglas a mostrar según el tipo inferido (sin redundancias).""" base = ["strip_special_spaces", "strip_special_keep_date_seps", "strip_spaces"] - if inferred_type in (TYPE_INTEGER, TYPE_DECIMAL): - return base + ["to_integer", "round_decimals_custom"] + if inferred_type == TYPE_INTEGER: + return base + if inferred_type == TYPE_DECIMAL: + return base + ["to_integer", "round_decimals_custom", "doc_id_split_tipo_numero"] if inferred_type == TYPE_DATE: - return base + ["date_numeric_to_ddmmyyyy", "date_alpha_to_ddmmyyyy"] + return base + [ + "doc_id_split_tipo_numero", + "date_messy_text_to_ddmmyyyy", + "date_numeric_to_ddmmyyyy", + "date_alpha_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", ] @@ -392,11 +878,16 @@ def _applicable_rules_for_type(inferred_type: str) -> list[str]: 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", ] +applicable_rule_ids_for_type = _applicable_rules_for_type + + def analyze_dataframe(df: pd.DataFrame) -> list[ColumnSummary]: """Genera un resumen por columna del DataFrame.""" summaries: list[ColumnSummary] = [] @@ -422,7 +913,7 @@ def analyze_dataframe(df: pd.DataFrame) -> list[ColumnSummary]: unique_count = int(non_null.nunique()) unique_pct = (unique_count / total_rows * 100) if total_rows else 0.0 - inferred_type, date_fmt = _infer_column_type(series) + inferred_type, date_fmt = _infer_column_type(series, str(col)) summary = ColumnSummary( name=str(col), @@ -498,23 +989,43 @@ def _rule_to_integer(series: pd.Series) -> pd.Series: return numeric.round().astype("Int64") -def _rule_round_decimals_custom(series: pd.Series) -> pd.Series: - """Aproximación custom: fracción > 0.5 → entero superior, resto → inferior. +def _round_to_int_half_tie_le(x: float) -> int: + """Fracc. > 0,5 hacia arriba; 0,5 o menos hacia abajo. Banda 1e-9 en 0,5 (float). - Funciona con positivos y negativos (usa parte entera hacia 0 y suma o - resta 1 según el signo y la fracción). + Misma noción que en la descripción de la regla. Trabaja sobre |x| y + recompone el signo. """ + sign = -1 if x < 0 else 1 + if x == 0.0 or x == -0.0: + return 0 + ax = abs(float(x)) + ip = math.trunc(ax) + frac = ax - ip + if frac < 0.5 - 1e-9: + out = ip + elif frac > 0.5 + 1e-9: + out = ip + 1 + else: + out = ip + return -out if sign < 0 else out + + +def _rule_round_decimals_custom(series: pd.Series) -> pd.Series: + """Mismo parser tolerante que convertir a entero; aprox. con fracc. >0,5 arriba.""" numeric = _to_numeric_lenient(series.astype(str)) def _fn(x): - if pd.isna(x): + if _is_nullish_scalar(x) or (isinstance(x, (float, np.floating)) and (not np.isfinite(x) or np.isnan(x))): return pd.NA - sign = -1 if x < 0 else 1 - ax = abs(float(x)) - integer_part = int(ax) - frac = ax - integer_part - rounded = integer_part + (1 if frac > 0.5 else 0) - return sign * rounded + try: + f = float(x) + except (TypeError, ValueError, OverflowError): + return pd.NA + if not np.isfinite(f): + return pd.NA + if f == 0.0 or f == -0.0: + return 0 + return _round_to_int_half_tie_le(f) return numeric.map(_fn).astype("Int64") @@ -567,6 +1078,162 @@ def _format_dates_to_ddmmyyyy(series: pd.Series, fmt: Optional[str]) -> pd.Serie return series.map(_fn) +# --------------------------------------------------------------------------- +# Fechas con ruido (texto sucio) → compacto numérico + DD/MM/AAAA +# --------------------------------------------------------------------------- + + +def _normalize_messy_date_to_compact(raw: Any) -> Optional[str]: + """Quita ruido (* + _ / espacios, etc.); deja 6 o 8 dígitos o None. + + Soporta separadores entremezclados; hora al final vía `_split_date_time`. + Con 9+ dígitos, heurística: tomar 8 iniciales (p. ej. 9.º = 0 colado) y + validar luego con `_parse_date_with_format`; con más de 8 sobrantes, + 8 iniciales. + """ + if _is_nullish_scalar(raw): + return None + s = str(raw).strip() + if s.lower() in ("nan", "none", "nat", ""): + return None + date_part, _ = _split_date_time(s) + body = date_part if date_part else s + d = re.sub(r"[^\d]", "", body) + if not d: + return None + n = len(d) + if n in (6, 8): + return d + if n < 6 or n == 7: + return None + # 9+ dígitos: 8 iniciales (p. ej. 9.º 0 sobrante, o ruido al final) + return d[:8] + + +def _rule_date_messy_text_to_ddmmyyyy(series: pd.Series) -> pd.Series: + """Fechas con símbolos y basura entre cifras → DD/MM/AAAA. + + Normaliza a solo 6/8 dígitos, vota el formato con `detect_date_format` por + separado en compact8 y compact6; si el detector no devuelve nada, usa + `DDMMYYYY` / `DDMMYY` (alineado con la preferencia hispana en fechas + numéricas). + """ + 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] + ) + if c8s.empty and c6s.empty: + return series.copy() + + fmt8: Optional[str] = None + fmt6: Optional[str] = None + if not c8s.empty: + fmt8 = detect_date_format(c8s) or "DDMMYYYY" + if not c6s.empty: + fmt6 = detect_date_format(c6s) or "DDMMYY" + + def _one(x: Any, c: Any) -> Any: + if _is_nullish_scalar(x): + return x + if not isinstance(c, str) or len(c) not in (6, 8): + return x + f = (fmt8 or "DDMMYYYY") if len(c) == 8 else (fmt6 or "DDMMYY") + t = _parse_date_with_format(c, f) + if t is None and len(c) == 8: + t = _parse_date_with_format(c, "DDMMYYYY") or _parse_date_with_format( + c, "YYYYMMDD" + ) + if t is None and len(c) == 6: + t = _parse_date_with_format(c, "DDMMYY") + if t is None: + return x + d, m, y = t + return f"{d:02d}/{m:02d}/{y:04d}" + + return pd.Series( + [ + _one(v, cser.iloc[i] if i < len(cser) else None) for i, v in enumerate(series) + ], + index=series.index, + dtype=object, + ) + + +# --------------------------------------------------------------------------- +# 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). + + 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). + """ + if _is_nullish_scalar(raw) and not (isinstance(raw, str) and not str(raw).strip()): + return (None, None) + t = str(raw).strip() + if t.lower() in ("nan", "none", "nat", ""): + return (None, None) + if t and t[0].isdigit(): + return (None, 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 + 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 + 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 (None, None) + + +def _alloc_doc_split_column_names( + col_name: str, existing: set[str] +) -> tuple[str, str]: + """Nombres únicos para las dos columnas derivadas.""" + base = re.sub(r"[^\w]+", "_", str(col_name).strip(), flags=re.ASCII) or "col" + for i in range(0, 200): + suf = f"_{i}" if i else "" + t, n = f"{base}{suf}_doc_tipo", f"{base}{suf}_doc_numeros" + if t not in existing and n not in existing: + return t, n + raise ValueError("No se pudieron asignar nombres de columnas únicos.") + + +def _rule_passthrough_series(s: pd.Series) -> pd.Series: + return s.copy() + + +def _apply_doc_id_split_columns( + result: pd.DataFrame, col: str, log: list[str] +) -> 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) + st_tipo.append(tt) + st_num.append(nn if nn is not None else pd.NA) + result[tcol] = st_tipo + result[ncol] = st_num + log.append( + f"[OK] '{str(col)}' -> columnas '{tcol}' y '{ncol}' <- Dividir tipo y número (documento)" + ) + + # Registro de reglas CLEANING_RULES: list[dict[str, Any]] = [ { @@ -577,8 +1244,8 @@ CLEANING_RULES: list[dict[str, Any]] = [ }, { "id": "strip_special_keep_date_seps", - "name": "Eliminar caracteres especiales y espacios (excepto / - y espacio, p. ej. Fecha Servicio)", - "description": "Conserva alfanuméricos y los separadores '/', '-' y espacios.", + "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'.", "fn": _rule_strip_special_keep_date_seps, }, { @@ -590,7 +1257,7 @@ CLEANING_RULES: list[dict[str, Any]] = [ { "id": "to_integer", "name": "Convertir a Entero", - "description": "Convierte la columna a enteros (los decimales se redondean).", + "description": "Convierte a enteros con detección de miles/decimal (`,`/`.`) y moneda; redondea decimales.", "fn": _rule_to_integer, }, { @@ -605,12 +1272,27 @@ CLEANING_RULES: list[dict[str, Any]] = [ "description": "Detecta el formato con separadores predominante (DD/MM/YYYY, MM/DD/YYYY, YYYY/MM/DD, etc.) y normaliza a DD/MM/AAAA.", "fn": _rule_date_alpha_to_ddmmyyyy, }, + { + "id": "date_messy_text_to_ddmmyyyy", + "name": "Texto a fecha (ruido en fechas) → DD/MM/AAAA", + "description": "Quita ruido entre cifras (*, +, _, /, etc.), detecta el orden en la columna (DDMMYYYY, YYYYMMDD, DDMMYY) y unifica a DD/MM/AAAA.", + "fn": _rule_date_messy_text_to_ddmmyyyy, + }, { "id": "round_decimals_custom", - "name": "Aproximación Decimales (>0.5 sube, ≤0.5 baja)", - "description": "Si la fracción decimal es mayor a 0.5 se aproxima al entero superior; de lo contrario, al inferior.", + "name": "Aproximación Decimales (>0,5 sube, ≤0,5 baja)", + "description": "Usa el mismo análisis de miles, decimales y moneda que «Entero». Tras leer el número, si la parte fraccionaria de |valor| es mayor a 0,5 se aproxima al entero hacia lejos de cero; en caso contrario, hacia cero. El 0,5 exacto se baja.", "fn": _rule_round_decimals_custom, }, + { + "id": "doc_id_split_tipo_numero", + "name": "Dividir tipo y número (documento)", + "description": ( + "Añade dos columnas (nombres únicos p. ej. TipoID_doc_tipo, TipoID_doc_numeros): " + "prefijo con letras (NIT, CC, CE, …) y cifras sin separadores. No modifica la columna de origen." + ), + "fn": _rule_passthrough_series, + }, ] _RULE_BY_ID: dict[str, Callable[[pd.Series], pd.Series]] = { @@ -652,6 +1334,13 @@ def apply_rules( log.append(f"[WARN] Columna no encontrada: {col}") continue for rule_id in rule_ids: + if rule_id == "doc_id_split_tipo_numero": + try: + _apply_doc_id_split_columns(result, col, log) + 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 fn = _RULE_BY_ID.get(rule_id) if fn is None: log.append(f"[WARN] Regla desconocida '{rule_id}' en columna '{col}'") @@ -713,7 +1402,6 @@ def detect_delimiter(filepath: str) -> str: if max(counts) == 0: continue # Modo (valor más frecuente) - from collections import Counter common = Counter(counts).most_common(1)[0] mode_count, mode_freq = common if mode_count == 0: diff --git a/application/database.py b/application/database.py index f54c7e1..be87fc8 100644 --- a/application/database.py +++ b/application/database.py @@ -159,41 +159,91 @@ class DatabaseManager: conn.close() return [dict(row) for row in rows] - def search_results(self, run_id: Optional[int] = None, status: Optional[str] = None, text: Optional[str] = None) -> List[Dict[str, Any]]: - """Busca resultados con filtros.""" + def _search_results_query( + self, run_id: Optional[int] = None, status: Optional[str] = None, text: Optional[str] = None + ) -> List[Dict[str, Any]]: + """Una búsqueda; todos los criterios activos se combinan con AND. + + Orden de importancia usado al relajar (en `search_results`): + prioridad 1: run, 2: estado, 3: texto. No se suelta el criterio de + `run` al relajar — solo texto y, si aplica, estado. + """ + tx = (text or "").strip() or None conn = sqlite3.connect(self.db_path) conn.row_factory = sqlite3.Row c = conn.cursor() - query = ''' SELECT r.id, r.run_id, r.radicado, r.file_path, r.status, r.raw_response, runs.timestamp FROM results r JOIN runs ON r.run_id = runs.id WHERE 1=1 ''' - params = [] - - if run_id: + params: list = [] + if run_id is not None: query += " AND r.run_id = ?" params.append(run_id) - if status and status != "Todos": query += " AND r.status = ?" params.append(status) - - if text: - # Busqueda simple en radicado o path + if tx: query += " AND (r.radicado LIKE ? OR r.file_path LIKE ?)" - wild = f"%{text}%" + wild = f"%{tx}%" params.extend([wild, wild]) - query += " ORDER BY r.id DESC LIMIT 1000" - c.execute(query, params) rows = c.fetchall() conn.close() return [dict(row) for row in rows] + def search_results( + self, + run_id: Optional[int] = None, + status: Optional[str] = None, + text: Optional[str] = None, + relax_if_empty: bool = False, + ) -> tuple[List[Dict[str, Any]], Optional[str]]: + """Busca resultados; opcionalmente relaja criterios si no hay filas. + + Prioridad: **run (si aplica) > estado > criterio de búsqueda (texto)**. + Con `relax_if_empty=True`, se omite el texto, luego el estado; el + filtro de run fijado no se omite. + + Devuelve (filas, nota) donde `nota` informa criterios omitidos al + relajar, o `None` si se usó la consulta estricta o no aplica. + """ + had_text = bool((text or "").strip()) + had_status = bool(status and status != "Todos") + + rows = self._search_results_query(run_id, status, text) + if rows or not relax_if_empty: + return (rows, None) + if not had_text and not had_status: + return ([], None) + if had_text: + looser = self._search_results_query(run_id, status, None) + if looser: + return ( + looser, + "Búsqueda ampliada: se omitió el criterio de búsqueda (texto) " + "(el de menor importancia: run > estado > texto; el run no se descarta al relajar).", + ) + if had_status: + looser2 = self._search_results_query(run_id, None, None) + if looser2: + if had_text: + nota = ( + "Búsqueda ampliada: se omitieron el criterio de búsqueda (texto) y el filtro de estado, " + "en ese orden, por prioridad (run > estado > texto; el run no se descarta al relajar)." + ) + else: + nota = ( + "Búsqueda ampliada: se omitió el filtro de estado, " + "criterio más débil al combinar con un run. Prioridad: run > estado > búsqueda; " + "el run no se descarta al relajar." + ) + return (looser2, nota) + return ([], None) + def update_result_status(self, result_id: int, status: str) -> None: """Actualiza el status de un resultado existente.""" conn = sqlite3.connect(self.db_path) diff --git a/gui/controller.py b/gui/controller.py index 2cdbda7..6641b1b 100644 --- a/gui/controller.py +++ b/gui/controller.py @@ -310,7 +310,9 @@ class AppController: # Mostrar resumen rápido # Usar tabla results (1 fila = 1 documento), no el join con matches - docs = self.db_manager.search_results(run_id=run_id, status="Todos", text=None) + docs, _ = self.db_manager.search_results( + run_id=run_id, status="Todos", text=None, relax_if_empty=False + ) match_count = sum(1 for r in docs if r.get("status") == "MATCH") self.view.append_preclass_output_safe(f"Total procesados: {len(docs)}") self.view.append_preclass_output_safe(f"Coincidencias (MATCH): {match_count}") @@ -427,20 +429,24 @@ class AppController: def on_res_search(self): """Busca resultados con los filtros actuales.""" + self.view.clear_res_filter_note() filters = self.view.get_res_filters() run_id = None if filters["run_id"] != "Todos": try: run_id = int(filters["run_id"]) except: pass - + try: - rows = self.db_manager.search_results( + rows, relax_note = self.db_manager.search_results( run_id=run_id, status=filters["status"], - text=filters["text"] + text=filters["text"], + relax_if_empty=bool(filters.get("relax_on_empty", False)), ) self.view.update_res_table(rows) + if relax_note: + self.view.set_res_filter_note(relax_note) except Exception as e: logger.error(f"Error buscando resultados: {e}") mb.showerror("Error", f"Error en búsqueda: {e}") @@ -462,13 +468,14 @@ class AppController: if filters["run_id"] != "Todos": try: run_id = int(filters["run_id"]) except: pass - - rows = self.db_manager.search_results( + + rows, _ = self.db_manager.search_results( run_id=run_id, status=filters["status"], - text=filters["text"] + text=filters["text"], + relax_if_empty=bool(filters.get("relax_on_empty", False)), ) - + if not rows: mb.showwarning("Exportar", "No hay datos para exportar.") return @@ -1292,7 +1299,7 @@ class AppController: if path.lower().endswith(".xlsx"): self._cleaning_result_df.to_excel(path, index=False) else: - self._cleaning_result_df.to_csv(path, index=False) + self._cleaning_result_df.to_csv(path, index=False, sep="|") mb.showinfo("Guardado", f"Archivo guardado correctamente:\n{path}") except Exception as e: logger.exception("Error guardando CSV limpio") diff --git a/gui/tabs/cleaning_tab.py b/gui/tabs/cleaning_tab.py index a5b608c..884a097 100644 --- a/gui/tabs/cleaning_tab.py +++ b/gui/tabs/cleaning_tab.py @@ -12,7 +12,11 @@ from typing import Optional import customtkinter as ctk import pandas as pd -from application.csv_cleaning import CLEANING_RULES, ColumnSummary +from application.csv_cleaning import ( + CLEANING_RULES, + ColumnSummary, + applicable_rule_ids_for_type, +) from gui.styles import AppStyles @@ -43,14 +47,24 @@ _TYPE_COLORS: dict[str, tuple[str, str]] = { "empty": ("#6B7280", "#6B7280"), } +# Orden fijo en el desplegable de anulación manual +_USER_EDITABLE_TYPE_IDS: tuple[str, ...] = ( + "integer", "decimal", "date", "boolean", "categorical", "text", "mixed", "empty", +) +_TYPE_MENU_VALUES: list[str] = [ + _TYPE_LABELS.get(t, t) for t in _USER_EDITABLE_TYPE_IDS +] + _RULE_SHORT_LABELS: dict[str, str] = { "strip_special_spaces": "Sin especiales/espacios", - "strip_special_keep_date_seps": "Sin especiales (mantiene / - espacio)", + "strip_special_keep_date_seps": "Símbolos: conserva /, -, espacios", "strip_spaces": "Sin espacios", "to_integer": "A entero", "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)", } @@ -73,7 +87,10 @@ class _ColumnUI: card: ctk.CTkFrame body: ctk.CTkFrame toggle_btn: ctk.CTkButton + type_menu: ctk.CTkOptionMenu count_label: ctk.CTkLabel + stats_frame: ctk.CTkFrame + rules_frame: ctk.CTkFrame rule_vars: list[tuple[str, ctk.CTkCheckBox, ctk.BooleanVar]] summary: ColumnSummary expanded: bool = True @@ -90,6 +107,8 @@ class CleaningTab: self._column_uis: dict[str, _ColumnUI] = {} 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._setup_layout() self._create_file_section() @@ -275,13 +294,14 @@ class CleaningTab: ) self.bulk_type_menu.pack(side="left", padx=(0, 6)) - rule_options = [r["name"] for r in CLEANING_RULES] - self._bulk_rule_var = ctk.StringVar(value=rule_options[0]) + self._bulk_rule_var = ctk.StringVar(value=CLEANING_RULES[0]["name"]) self.bulk_rule_menu = ctk.CTkOptionMenu( - bulk_row, values=rule_options, variable=self._bulk_rule_var, + bulk_row, values=[r["name"] for r in CLEANING_RULES], variable=self._bulk_rule_var, width=380, font=self.fonts["ui"], ) self.bulk_rule_menu.pack(side="left", padx=(0, 6)) + self._sync_bulk_rule_menu() + self._bulk_type_var.trace_add("write", lambda *_: self._sync_bulk_rule_menu()) self.btn_bulk_apply = ctk.CTkButton( bulk_row, text="Aplicar a todas", width=140, height=28, @@ -314,8 +334,9 @@ class CleaningTab: ) self.lbl_global_count.pack(side="right") - self.columns_container = ctk.CTkScrollableFrame( - frame, fg_color="transparent", height=420, + # Frame plano: el scroll único es el de `self.scroll` (evita doble rueda del ratón) + self.columns_container = ctk.CTkFrame( + frame, fg_color="transparent", ) self.columns_container.grid(row=1, column=0, sticky="ew", padx=10, pady=(0, 10)) self.columns_container.grid_columnconfigure(0, weight=1) @@ -460,11 +481,33 @@ class CleaningTab: self.btn_expand_all.configure(state=state) self.btn_collapse_all.configure(state=state) + def _sync_bulk_rule_menu(self) -> None: + """Acota las reglas del desplegable según el tipo (p. ej. entero: sin a entero/approx).""" + if not hasattr(self, "bulk_rule_menu") or not hasattr( + self, "_bulk_type_var", + ): + return + type_label = self._bulk_type_var.get() + type_id = next( + (k for k, v in _TYPE_LABELS.items() if v == type_label), "text" + ) + want = set(applicable_rule_ids_for_type(type_id)) + names = [r["name"] for r in CLEANING_RULES if r["id"] in want] + if not names: + return + cur = self._bulk_rule_var.get() + self.bulk_rule_menu.configure(values=names) + if cur in names: + self._bulk_rule_var.set(cur) + else: + self._bulk_rule_var.set(names[0]) + def populate_columns(self, summaries: list[ColumnSummary]): """Renderiza la lista de columnas con sus reglas aplicables.""" for child in list(self.columns_container.winfo_children()): child.destroy() self._column_uis.clear() + self._type_overrides.clear() if self._search_var is not None: try: self._search_var.set("") @@ -523,15 +566,28 @@ class CleaningTab: ) toggle_btn.grid(row=0, column=0, padx=(0, 6), sticky="w") - type_color = _TYPE_COLORS.get(summary.inferred_type, ("gray60", "gray40")) - type_label = _TYPE_LABELS.get(summary.inferred_type, summary.inferred_type) - badge = ctk.CTkLabel( - header, text=type_label.upper(), - font=ctk.CTkFont(size=10, weight="bold"), - fg_color=type_color, text_color="white", - corner_radius=10, padx=10, pady=2, + eff0 = self._type_overrides.get(summary.name) or summary.inferred_type + if eff0 not in _TYPE_LABELS: + eff0 = "text" + tcol = _TYPE_COLORS.get(eff0, ("#64748B", "#64748B")) + tmenu = ctk.CTkOptionMenu( + header, + values=_TYPE_MENU_VALUES, + width=150, + height=28, + font=self.fonts["ui"], + corner_radius=10, + button_color=tcol[0] if tcol[0] else tcol[1], + button_hover_color=tcol[0], + fg_color=tcol[0] if tcol[0] else tcol[1], + ) + tmenu.set(_TYPE_LABELS.get(eff0, _TYPE_LABELS["text"])) + tmenu.grid(row=0, column=1, padx=(0, 6), sticky="w") + tmenu.configure( + command=lambda v, c=summary.name, m=tmenu: self._on_user_column_type_change( + c, m, v + ) ) - badge.grid(row=0, column=1, padx=(0, 10), sticky="w") name_lbl = ctk.CTkLabel( header, text=summary.name, font=self.fonts["bold_large"], anchor="w", @@ -559,7 +615,8 @@ class CleaningTab: ) lbl_example.grid(row=0, column=0, sticky="ew", padx=4, pady=(0, 4)) - self._build_stats_chips(body, summary).grid( + stats_frame = self._build_stats_chips(body, summary, eff0) + stats_frame.grid( row=1, column=0, sticky="ew", padx=4, pady=(0, 4), ) @@ -572,8 +629,11 @@ class CleaningTab: rules_frame.grid(row=3, column=0, sticky="ew", padx=0, pady=(2, 0)) rules_frame.grid_columnconfigure((0, 1), weight=1, uniform="rules") - applicable_ids = summary.applicable_rules or [r["id"] for r in CLEANING_RULES] - ordered_ids = [r["id"] for r in CLEANING_RULES if r["id"] in applicable_ids] + eff_rules = self._type_overrides.get(summary.name) or summary.inferred_type + if eff_rules not in _TYPE_LABELS: + eff_rules = "text" + applic = set(applicable_rule_ids_for_type(eff_rules)) + ordered_ids = [r["id"] for r in CLEANING_RULES if r["id"] in applic] vars_list: list[tuple[str, ctk.CTkCheckBox, ctk.BooleanVar]] = [] col_in_row = 0 @@ -597,8 +657,15 @@ class CleaningTab: row_in_rules += 1 ui = _ColumnUI( - card=card, body=body, toggle_btn=toggle_btn, - count_label=count_lbl, rule_vars=vars_list, summary=summary, + card=card, + body=body, + toggle_btn=toggle_btn, + type_menu=tmenu, + count_label=count_lbl, + stats_frame=stats_frame, + rules_frame=rules_frame, + rule_vars=vars_list, + summary=summary, ) toggle_btn.configure(command=lambda c=summary.name: self._toggle_column(c)) return ui @@ -607,9 +674,14 @@ class CleaningTab: # Stats visuales por columna # =========================================================================== - def _build_stats_chips(self, parent, summary: ColumnSummary) -> ctk.CTkFrame: + def _build_stats_chips( + self, parent, summary: ColumnSummary, display_type: Optional[str] = None + ) -> ctk.CTkFrame: """Construye una fila de chips con la información clave de la columna.""" row = ctk.CTkFrame(parent, fg_color="transparent") + dtype = (display_type or summary.inferred_type) or "text" + if dtype not in _TYPE_LABELS: + dtype = "text" # Sin datos nulls_color = ( @@ -638,7 +710,7 @@ class CleaningTab: ).pack(side="left", padx=(0, 6), pady=2) # Stats por tipo - if summary.inferred_type in ("integer", "decimal"): + if dtype in ("integer", "decimal"): if summary.min is not None: self._make_stat_chip( row, label="Mín", value=_fmt_number(summary.min), @@ -654,7 +726,7 @@ class CleaningTab: row, label="Promedio", value=_fmt_number(summary.mean), bg=("#DCFCE7", "#14532D"), text_color=("#166534", "#BBF7D0"), ).pack(side="left", padx=(0, 6), pady=2) - elif summary.inferred_type == "date": + elif dtype == "date": if summary.detected_date_format: self._make_stat_chip( row, label="Formato detectado", @@ -662,7 +734,7 @@ class CleaningTab: bg=("#FFEDD5", "#7C2D12"), text_color=("#9A3412", "#FED7AA"), ).pack(side="left", padx=(0, 6), pady=2) - elif summary.inferred_type in ("text", "categorical"): + elif dtype in ("text", "categorical"): if summary.avg_length is not None: self._make_stat_chip( row, label="Longitud promedio", @@ -736,6 +808,64 @@ class CleaningTab: # Lógica interna # =========================================================================== + def _effective_inferred_type(self, ui: _ColumnUI) -> str: + """Tipo mostrado para la columna: anulación del usuario o inferido.""" + t = self._type_overrides.get(ui.summary.name) or ui.summary.inferred_type + return t if t in _TYPE_LABELS else "text" + + def _on_user_column_type_change( + self, col_name: str, type_menu: ctk.CTkOptionMenu, spanish_label: str + ) -> None: + type_id = next( + (k for k, v in _TYPE_LABELS.items() if v == spanish_label), "text" + ) + ui = self._column_uis.get(col_name) + if ui is None: + return + s = ui.summary + if type_id == s.inferred_type: + self._type_overrides.pop(col_name, None) + else: + self._type_overrides[col_name] = type_id + eff = self._effective_inferred_type(ui) + tcol = _TYPE_COLORS.get(eff, ("#64748B", "#64748B")) + type_menu.configure( + button_color=tcol[0] if tcol[0] else tcol[1], + button_hover_color=tcol[0], + fg_color=tcol[0] if tcol[0] else tcol[1], + ) + ui.stats_frame.destroy() + new_stats = self._build_stats_chips(ui.body, s, eff) + new_stats.grid(row=1, column=0, sticky="ew", padx=4, pady=(0, 4)) + ui.stats_frame = new_stats + for w in list(ui.rules_frame.winfo_children()): + w.destroy() + applic = set(applicable_rule_ids_for_type(eff)) + new_vars: list[tuple[str, ctk.CTkCheckBox, ctk.BooleanVar]] = [] + col_in_row = 0 + row_in_rules = 0 + rmap = {r["id"]: r for r in CLEANING_RULES} + for rid in (r["id"] for r in CLEANING_RULES if r["id"] in applic): + rule = rmap.get(rid) + if rule is None: + continue + var = ctk.BooleanVar(value=False) + var.trace_add("write", lambda *_a, c=col_name: self._on_rule_toggle(c)) + label = _RULE_SHORT_LABELS.get(rid, rule["name"]) + chk = ctk.CTkCheckBox( + ui.rules_frame, text=label, variable=var, + font=self.fonts["ui"], checkbox_width=18, checkbox_height=18, + ) + chk.grid(row=row_in_rules, column=col_in_row, sticky="w", padx=10, pady=4) + new_vars.append((rid, chk, var)) + col_in_row += 1 + if col_in_row >= 2: + col_in_row = 0 + row_in_rules += 1 + ui.rule_vars = new_vars + self.set_status(f"Columna «{col_name}»: tipo «{_TYPE_LABELS.get(eff, eff)}».") + self._on_rule_toggle(col_name) + def _toggle_column(self, col_name: str): ui = self._column_uis.get(col_name) if ui is None: @@ -804,7 +934,7 @@ class CleaningTab: matched = 0 for ui in self._column_uis.values(): - if ui.summary.inferred_type != type_id: + if self._effective_inferred_type(ui) != type_id: continue for rid, _chk, var in ui.rule_vars: if rid == rule_id: diff --git a/gui/tabs/results_tab.py b/gui/tabs/results_tab.py index 4b7c82b..f1b1206 100644 --- a/gui/tabs/results_tab.py +++ b/gui/tabs/results_tab.py @@ -1,5 +1,6 @@ """Pestaña de Resultados de la aplicación GUI.""" import os +import tkinter as tk import customtkinter as ctk import tkinter.ttk as ttk from gui.styles import AppStyles @@ -88,6 +89,43 @@ class ResultsTab: self.btn_res_search = ctk.CTkButton(self.frame_res_filters, text="Buscar", width=100) self.btn_res_search.grid(row=0, column=5, padx=10, pady=5) + + self.lbl_res_priority = ctk.CTkLabel( + self.frame_res_filters, + text=( + "Filtros estrictos: se combinan (AND) con importancia: " + "Run > estado (MATCH, etc.) > criterio de búsqueda. " + "Puede ampliar la búsqueda (sin tocar el run) abajo." + ), + text_color=("#5C6370", "#9CA3AF"), + font=ctk.CTkFont(size=12), + anchor="w", + wraplength=920, + ) + self.lbl_res_priority.grid( + row=1, column=0, columnspan=6, padx=5, pady=(4, 0), sticky="w" + ) + self._res_relax_var = tk.BooleanVar(value=False) + self.chk_res_relax = ctk.CTkCheckBox( + self.frame_res_filters, + text="Si 0 resultados, ampliar: quitar criterio de búsqueda, luego estado (el run fijado no se descarta).", + font=ctk.CTkFont(size=12), + variable=self._res_relax_var, + ) + self.chk_res_relax.grid( + row=2, column=0, columnspan=6, padx=5, pady=4, sticky="w" + ) + self.lbl_res_filter_note = ctk.CTkLabel( + self.frame_res_filters, + text="", + text_color=("#1D4ED8", "#60A5FA"), + font=ctk.CTkFont(size=12), + anchor="w", + wraplength=920, + ) + self.lbl_res_filter_note.grid( + row=3, column=0, columnspan=6, padx=5, pady=(0, 4), sticky="w" + ) def _create_tables_section(self): """Crea la sección con las tablas (principal y de detalle).""" @@ -238,9 +276,17 @@ class ResultsTab: return { "run_id": self.combo_res_run.get(), "status": self.combo_res_status.get(), - "text": self.entry_res_search.get() + "text": self.entry_res_search.get(), + "relax_on_empty": bool(self._res_relax_var.get()), } + def set_filter_note(self, text: str) -> None: + """Mensaje bajo ampliar (p. ej. qué criterio se relajó). Texto vacío la limpia.""" + self.lbl_res_filter_note.configure(text=(text or "").strip()) + + def clear_filter_note(self) -> None: + self.set_filter_note("") + def update_results_table(self, rows): """Actualiza la tabla de resultados con los datos proporcionados.""" for item in self.tree_results.get_children(): diff --git a/gui/view.py b/gui/view.py index e0901a9..043c15c 100644 --- a/gui/view.py +++ b/gui/view.py @@ -39,12 +39,12 @@ class InvoiceAppView(ctk.CTk): self.tabs.grid(row=0, column=0, sticky="nsew", padx=20, pady=20) self.tab_main = self.tabs.add("Procesar Documentos") + self.tab_cleaning = self.tabs.add("Proceso de Limpieza") self.tab_preclass = self.tabs.add("Preclasificación") self.tab_results = self.tabs.add("Resultados") self.tab_settings = self.tabs.add("Configuración") self.tab_usage = self.tabs.add("Dashboard & Métricas") self.tab_postprocess = self.tabs.add("Post-Procesamiento") - self.tab_cleaning = self.tabs.add("Proceso de Limpieza") # ========================================================================= # --- TAB: PROCESAR DOCUMENTOS --- @@ -52,21 +52,24 @@ class InvoiceAppView(ctk.CTk): self.main_tab = MainProcessingTab(self.tab_main) # ========================================================================= - # --- TAB: RESULTADOS --- + # --- TAB: PROCESO DE LIMPIEZA --- # ========================================================================= - self.results_tab = ResultsTab(self.tab_results) - - - # ========================================================================= - # --- TAB: CONFIGURACIÓN --- - # ========================================================================= - self.settings_tab = SettingsTab(self.tab_settings) + self.cleaning_tab = CleaningTab(self.tab_cleaning) # ========================================================================= # --- TAB: PRECLASIFICACIÓN --- # ========================================================================= self.preclass_tab = PreclassificationTab(self.tab_preclass) + # ========================================================================= + # --- TAB: RESULTADOS --- + # ========================================================================= + self.results_tab = ResultsTab(self.tab_results) + + # ========================================================================= + # --- TAB: CONFIGURACIÓN --- + # ========================================================================= + self.settings_tab = SettingsTab(self.tab_settings) # ========================================================================= # --- TAB: DASHBOARD & MÉTRICAS --- @@ -78,11 +81,6 @@ class InvoiceAppView(ctk.CTk): # ========================================================================= self.postprocess_tab = PostProcessTab(self.tab_postprocess) - # ========================================================================= - # --- TAB: PROCESO DE LIMPIEZA --- - # ========================================================================= - self.cleaning_tab = CleaningTab(self.tab_cleaning) - # --------------------------------------------------------------------------- # MÉTODOS PÚBLICOS (API COMPATIBLE) @@ -161,6 +159,12 @@ class InvoiceAppView(ctk.CTk): def get_res_filters(self): return self.results_tab.get_filters() + def set_res_filter_note(self, text: str) -> None: + self.results_tab.set_filter_note(text) + + def clear_res_filter_note(self) -> None: + self.results_tab.clear_filter_note() + def update_res_table(self, rows): self.results_tab.update_results_table(rows)