feat: Enhance search functionality in the database manager to support relaxed criteria for result retrieval, improving user experience by allowing broader searches when no results are found. Update GUI components to reflect these changes, including new filter notes and options for relaxing search criteria.

This commit is contained in:
erik
2026-04-26 19:49:20 -05:00
parent 6b3eccdc19
commit 37d32a5a96
6 changed files with 1048 additions and 123 deletions
+749 -61
View File
@@ -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", "<na>"):
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"(?<!\d)\d+[.,][0-9]{1,2}\b", s
):
n_hit += 1
if n_seen == 0:
return False
return n_hit / n_seen >= 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", "<na>"):
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", "<na>"):
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:
+63 -13
View File
@@ -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)
+16 -9
View File
@@ -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")
+155 -25
View File
@@ -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:
+47 -1
View File
@@ -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():
+18 -14
View File
@@ -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)