2131 lines
65 KiB
Python
2131 lines
65 KiB
Python
"""Limpieza de CSV con análisis inteligente local y reglas configurables.
|
||
|
||
Este módulo es pura lógica (sin GUI, sin LLM). Provee:
|
||
- `analyze_dataframe` para obtener un resumen por columna (tipo inferido,
|
||
ejemplo aleatorio, nulos, únicos, estadísticas específicas por tipo).
|
||
- Un registro `CLEANING_RULES` con reglas aplicables a columnas.
|
||
- `apply_rules` que aplica las reglas seleccionadas por columna y devuelve
|
||
el DataFrame resultante junto con un log de operaciones/advertencias.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import csv as _csv
|
||
import logging
|
||
import math
|
||
import random
|
||
import re
|
||
import unicodedata
|
||
from collections import Counter
|
||
from dataclasses import dataclass, field
|
||
from datetime import date
|
||
from typing import Any, Callable, Literal, Optional
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# CUOTA MODERADORA (COP/MOD)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
DEFAULT_MODERADORA_BY_YEAR: dict[int, set[int]] = {
|
||
2023: {4100, 16400, 43000},
|
||
2024: {4500, 18200, 47700},
|
||
2025: {4700, 19200, 50300},
|
||
2026: {5000, 20100, 52800},
|
||
}
|
||
|
||
|
||
def normalize_moderadora_by_year_dict(raw: Any) -> dict[int, set[int]]:
|
||
"""Normaliza un dict JSON año → lista de enteros (cuotas moderadora)."""
|
||
if not isinstance(raw, dict):
|
||
return {}
|
||
out: dict[int, set[int]] = {}
|
||
for k, v in raw.items():
|
||
try:
|
||
yi = int(k)
|
||
except (TypeError, ValueError):
|
||
continue
|
||
if not isinstance(v, (list, tuple)):
|
||
continue
|
||
nums: set[int] = set()
|
||
for x in v:
|
||
try:
|
||
nums.add(int(round(float(x))))
|
||
except (TypeError, ValueError):
|
||
continue
|
||
if nums:
|
||
out[yi] = nums
|
||
return out
|
||
|
||
|
||
def merge_moderadora_config(
|
||
cop_mod_config: Optional[dict[str, Any]],
|
||
) -> dict[int, set[int]]:
|
||
"""Fusiona defaults con `moderadora_by_year` opcional del usuario."""
|
||
merged = {y: set(vals) for y, vals in DEFAULT_MODERADORA_BY_YEAR.items()}
|
||
if cop_mod_config and isinstance(cop_mod_config.get("moderadora_by_year"), dict):
|
||
custom = normalize_moderadora_by_year_dict(
|
||
cop_mod_config["moderadora_by_year"]
|
||
)
|
||
for y, vals in custom.items():
|
||
merged[y] = set(vals)
|
||
return merged
|
||
|
||
|
||
def _alloc_cop_mod_column_names(
|
||
col_name: str, existing: set[str]
|
||
) -> tuple[str, str]:
|
||
"""Nombres únicos para copago y moderadora."""
|
||
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 ""
|
||
ccol, mcol = f"{base}{suf}_copago", f"{base}{suf}_moderadora"
|
||
if ccol not in existing and mcol not in existing:
|
||
return ccol, mcol
|
||
raise ValueError("No se pudieron asignar nombres de columnas únicos.")
|
||
|
||
|
||
def _amount_cell_to_optional_int(
|
||
val: Any,
|
||
decimal_hint: Optional[Literal[",", "."]],
|
||
) -> Optional[int]:
|
||
try:
|
||
f = _parse_lenient_one(val, decimal_hint)
|
||
except Exception:
|
||
return None
|
||
if f is None or (isinstance(f, float) and not np.isfinite(f)) or (
|
||
isinstance(f, (float, np.floating)) and np.isnan(f)
|
||
):
|
||
return None
|
||
try:
|
||
return int(round(float(f)))
|
||
except (TypeError, ValueError, OverflowError):
|
||
return None
|
||
|
||
|
||
def _apply_cop_mod_split(
|
||
result: pd.DataFrame,
|
||
amount_col: str,
|
||
fecha_column: str,
|
||
log: list[str],
|
||
*,
|
||
moderadora_by_year: dict[int, set[int]],
|
||
) -> None:
|
||
"""Añade columnas copago y moderadora según cuotas por año de la fecha."""
|
||
if fecha_column not in result.columns:
|
||
log.append(
|
||
f"[WARN] COP/MOD: columna de fecha '{fecha_column}' no existe; se omite."
|
||
)
|
||
return
|
||
if amount_col not in result.columns:
|
||
log.append(f"[WARN] COP/MOD: columna de importe '{amount_col}' no existe; se omite.")
|
||
return
|
||
|
||
fec_ser = result[fecha_column]
|
||
amt_ser = result[amount_col]
|
||
stripped = fec_ser.astype(str).str.replace(r"[\s\/\-\.]", "", regex=True)
|
||
stripped = stripped.where(fec_ser.notna(), other=pd.NA)
|
||
sep_fmt = detect_date_format(fec_ser) or detect_date_format(stripped)
|
||
compact8_fmt, compact6_fmt = _detect_compact_date_formats(fec_ser)
|
||
amt_hint = _infer_decimal_hint_from_column(amt_ser.astype(str))
|
||
|
||
ccol, mcol = _alloc_cop_mod_column_names(str(amount_col), set(result.columns))
|
||
cop_vals: list[Any] = []
|
||
mod_vals: list[Any] = []
|
||
|
||
for i in range(len(result)):
|
||
amt_raw = amt_ser.iloc[i]
|
||
fec_raw = fec_ser.iloc[i]
|
||
amt_i = _amount_cell_to_optional_int(amt_raw, amt_hint)
|
||
if amt_i is None:
|
||
cop_vals.append(0)
|
||
mod_vals.append(0)
|
||
continue
|
||
|
||
parsed_d = _parse_any_date_to_tuple(
|
||
fec_raw,
|
||
sep_fmt=sep_fmt,
|
||
compact8_fmt=compact8_fmt,
|
||
compact6_fmt=compact6_fmt,
|
||
)
|
||
if parsed_d is None:
|
||
cop_vals.append(amt_i)
|
||
mod_vals.append(0)
|
||
continue
|
||
_d, _m, year = parsed_d
|
||
tiers = moderadora_by_year.get(year)
|
||
if tiers is None:
|
||
cop_vals.append(amt_i)
|
||
mod_vals.append(0)
|
||
continue
|
||
|
||
if amt_i in tiers:
|
||
cop_vals.append(0)
|
||
mod_vals.append(amt_i)
|
||
else:
|
||
cop_vals.append(amt_i)
|
||
mod_vals.append(0)
|
||
|
||
result[ccol] = cop_vals
|
||
result[mcol] = mod_vals
|
||
log.append(
|
||
f"[OK] '{amount_col}' -> '{ccol}', '{mcol}' <- Separa COP/MOD "
|
||
f"(fecha: '{fecha_column}')"
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# MODELO DE DATOS
|
||
# ---------------------------------------------------------------------------
|
||
|
||
TYPE_INTEGER = "integer"
|
||
TYPE_DECIMAL = "decimal"
|
||
TYPE_DATE = "date"
|
||
TYPE_BOOLEAN = "boolean"
|
||
TYPE_CATEGORICAL = "categorical"
|
||
TYPE_TEXT = "text"
|
||
TYPE_MIXED = "mixed"
|
||
TYPE_EMPTY = "empty"
|
||
|
||
|
||
@dataclass
|
||
class ColumnSummary:
|
||
"""Resumen inteligente para una columna del DataFrame."""
|
||
name: str
|
||
inferred_type: str
|
||
example: str = ""
|
||
null_count: int = 0
|
||
null_pct: float = 0.0
|
||
unique_count: int = 0
|
||
unique_pct: float = 0.0
|
||
min: Optional[float] = None
|
||
max: Optional[float] = None
|
||
mean: Optional[float] = None
|
||
avg_length: Optional[float] = None
|
||
top_values: list[tuple[str, int]] = field(default_factory=list)
|
||
detected_date_format: Optional[str] = None
|
||
applicable_rules: list[str] = field(default_factory=list)
|
||
|
||
def info_text(self) -> str:
|
||
"""Texto humano con información adicional (para subtítulo de la UI)."""
|
||
parts: list[str] = []
|
||
parts.append(f"nulos: {self.null_count} ({self.null_pct:.0f}%)")
|
||
parts.append(f"únicos: {self.unique_count}")
|
||
if self.inferred_type in (TYPE_INTEGER, TYPE_DECIMAL):
|
||
if self.min is not None:
|
||
parts.append(f"min: {_fmt_num(self.min)}")
|
||
if self.max is not None:
|
||
parts.append(f"max: {_fmt_num(self.max)}")
|
||
if self.mean is not None:
|
||
parts.append(f"media: {_fmt_num(self.mean)}")
|
||
elif self.inferred_type == TYPE_DATE:
|
||
if self.detected_date_format:
|
||
parts.append(f"formato: {self.detected_date_format}")
|
||
elif self.inferred_type in (TYPE_TEXT, TYPE_CATEGORICAL):
|
||
if self.avg_length is not None:
|
||
parts.append(f"long. prom: {self.avg_length:.1f}")
|
||
if self.top_values:
|
||
tops = ", ".join(f'"{v}" ({c})' for v, c in self.top_values[:3])
|
||
parts.append(f"top: {tops}")
|
||
return " · ".join(parts)
|
||
|
||
|
||
def _fmt_num(x: float) -> str:
|
||
if x is None:
|
||
return ""
|
||
if isinstance(x, float) and x.is_integer():
|
||
return str(int(x))
|
||
if isinstance(x, float):
|
||
return f"{x:.2f}"
|
||
return str(x)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# DETECCIÓN DE FECHAS
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_DATE_NUM_RE = re.compile(r"^\s*(\d{1,4})[\s\/\-\.](\d{1,2})[\s\/\-\.](\d{1,4})\s*$")
|
||
_DATE_COMPACT_RE = re.compile(r"^\s*(\d{6}|\d{8})\s*$")
|
||
# Hora opcional: HH[:MM[:SS]][ AM|PM] al final de la cadena
|
||
_TIME_TAIL_RE = re.compile(
|
||
r"[\sT]+(\d{1,2})(?::(\d{1,2})(?::(\d{1,2}))?)?\s*([APap]\.?\s*[Mm]\.?)?\s*$"
|
||
)
|
||
|
||
_MONTH_NAME_TO_NUM = {
|
||
"enero": 1,
|
||
"ene": 1,
|
||
"febrero": 2,
|
||
"feb": 2,
|
||
"marzo": 3,
|
||
"mar": 3,
|
||
"abril": 4,
|
||
"abr": 4,
|
||
"mayo": 5,
|
||
"may": 5,
|
||
"junio": 6,
|
||
"jun": 6,
|
||
"julio": 7,
|
||
"jul": 7,
|
||
"agosto": 8,
|
||
"ago": 8,
|
||
"septiembre": 9,
|
||
"setiembre": 9,
|
||
"sep": 9,
|
||
"sept": 9,
|
||
"octubre": 10,
|
||
"oct": 10,
|
||
"noviembre": 11,
|
||
"nov": 11,
|
||
"diciembre": 12,
|
||
"dic": 12,
|
||
}
|
||
|
||
|
||
def _split_date_time(value: str) -> tuple[str, Optional[str]]:
|
||
"""Separa la parte de fecha de la parte de hora si está presente.
|
||
|
||
Devuelve `(date_part, time_part_normalized)`. `time_part_normalized` es
|
||
None si no hay hora. La hora se normaliza a `HH:MM[:SS][ AM|PM]`.
|
||
"""
|
||
if value is None:
|
||
return "", None
|
||
s = str(value).strip()
|
||
if not s:
|
||
return "", None
|
||
|
||
m = _TIME_TAIL_RE.search(s)
|
||
if not m:
|
||
return s, None
|
||
|
||
hh = int(m.group(1))
|
||
mm = int(m.group(2)) if m.group(2) else 0
|
||
ss = m.group(3)
|
||
ampm = m.group(4)
|
||
if not (0 <= hh <= 23 and 0 <= mm <= 59):
|
||
return s, None
|
||
if ss is not None:
|
||
try:
|
||
ssi = int(ss)
|
||
if not (0 <= ssi <= 59):
|
||
return s, None
|
||
except ValueError:
|
||
return s, None
|
||
# Validar contra AM/PM (con AM/PM, hora 1-12)
|
||
if ampm and not (1 <= hh <= 12):
|
||
return s, None
|
||
|
||
# Quitar separadores habituales entre fecha y hora (p. ej. "12/27/24, 12:00 AM").
|
||
date_part = s[: m.start()].rstrip(" \t,;T")
|
||
if not date_part:
|
||
return s, None
|
||
|
||
parts: list[str] = [f"{hh:02d}:{mm:02d}"]
|
||
if ss is not None:
|
||
parts[0] += f":{int(ss):02d}"
|
||
if ampm:
|
||
norm_ampm = ampm.replace(".", "").replace(" ", "").upper()
|
||
if norm_ampm in ("AM", "PM"):
|
||
parts.append(norm_ampm)
|
||
time_norm = " ".join(parts)
|
||
return date_part, time_norm
|
||
|
||
|
||
def _extract_date_parts(value: str) -> Optional[tuple[int, int, int, str]]:
|
||
"""Intenta extraer (a, b, c, layout) de una fecha en cualquier formato común.
|
||
|
||
`layout` es uno de: 'sep' (con separadores), 'compact8' (YYYYMMDD o
|
||
DDMMYYYY), 'compact6' (DDMMYY o YYMMDD). No interpreta todavía qué
|
||
componente es día/mes/año; eso lo decide `detect_date_format` viendo la
|
||
columna completa. Acepta valores con hora al final (ignorada para los
|
||
componentes de fecha).
|
||
"""
|
||
if value is None:
|
||
return None
|
||
s = str(value).strip()
|
||
if not s or s.lower() in ("nan", "none", "nat"):
|
||
return None
|
||
|
||
date_part, _ = _split_date_time(s)
|
||
if date_part:
|
||
s = date_part
|
||
|
||
m = _DATE_NUM_RE.match(s)
|
||
if m:
|
||
try:
|
||
a, b, c = int(m.group(1)), int(m.group(2)), int(m.group(3))
|
||
return (a, b, c, "sep")
|
||
except ValueError:
|
||
return None
|
||
|
||
m2 = _DATE_COMPACT_RE.match(s)
|
||
if m2:
|
||
digits = m2.group(1)
|
||
if len(digits) == 8:
|
||
return (int(digits[0:2]), int(digits[2:4]), int(digits[4:8]), "compact8")
|
||
if len(digits) == 6:
|
||
return (int(digits[0:2]), int(digits[2:4]), int(digits[4:6]), "compact6")
|
||
|
||
return None
|
||
|
||
|
||
def detect_date_format(series: pd.Series, sample_size: int = 500) -> Optional[str]:
|
||
"""Analiza la serie y determina el formato de fecha dominante.
|
||
|
||
Devuelve un string con el layout detectado (p. ej. "DD/MM/YYYY",
|
||
"MM/DD/YYYY", "YYYYMMDD", "DDMMYYYY") o None si no parece fechas.
|
||
"""
|
||
if series is None or len(series) == 0:
|
||
return None
|
||
|
||
non_null = series.dropna()
|
||
if non_null.empty:
|
||
return None
|
||
|
||
if len(non_null) > sample_size:
|
||
non_null = non_null.sample(sample_size, random_state=42)
|
||
|
||
layouts = {"sep": 0, "compact8": 0, "compact6": 0}
|
||
sep_first_gt12 = 0 # primer componente > 12 → probable día
|
||
sep_second_gt12 = 0 # segundo componente > 12 → probable mes solo posible si MM/DD
|
||
sep_third_4digits = 0 # tercer componente de 4 dígitos → año al final
|
||
sep_first_4digits = 0 # primer componente de 4 dígitos → año al inicio
|
||
compact8_year_first = 0 # primeros 4 dígitos > 1900 → YYYYMMDD
|
||
compact8_year_last = 0 # últimos 4 dígitos > 1900 → DDMMYYYY
|
||
total_parsed = 0
|
||
|
||
for val in non_null:
|
||
parsed = _extract_date_parts(val)
|
||
if parsed is None:
|
||
continue
|
||
total_parsed += 1
|
||
a, b, c, layout = parsed
|
||
layouts[layout] = layouts.get(layout, 0) + 1
|
||
|
||
if layout == "sep":
|
||
# año puede estar al inicio o al final según cantidad de dígitos
|
||
if _digit_count(a) == 4:
|
||
sep_first_4digits += 1
|
||
if _digit_count(c) == 4:
|
||
sep_third_4digits += 1
|
||
if a > 12:
|
||
sep_first_gt12 += 1
|
||
if b > 12:
|
||
sep_second_gt12 += 1
|
||
elif layout == "compact8":
|
||
# a son los primeros 2 dígitos, c son los últimos 4
|
||
if c >= 1900:
|
||
compact8_year_last += 1
|
||
# a concatenado con b son los primeros 4 dígitos
|
||
first4 = a * 100 + b
|
||
if first4 >= 1900:
|
||
compact8_year_first += 1
|
||
|
||
if total_parsed == 0:
|
||
return None
|
||
|
||
# Requerimos que >=70% de los no nulos parseen como fecha para considerarla
|
||
# columna de fechas.
|
||
if total_parsed / len(non_null) < 0.7:
|
||
return None
|
||
|
||
dominant = max(layouts, key=lambda k: layouts[k])
|
||
|
||
if dominant == "compact8":
|
||
if compact8_year_first > compact8_year_last:
|
||
return "YYYYMMDD"
|
||
return "DDMMYYYY"
|
||
if dominant == "compact6":
|
||
return "DDMMYY"
|
||
|
||
# Caso con separadores
|
||
if sep_first_4digits > sep_third_4digits:
|
||
return "YYYY/MM/DD"
|
||
# año al final
|
||
if sep_first_gt12 > 0 and sep_second_gt12 == 0:
|
||
return "DD/MM/YYYY"
|
||
if sep_second_gt12 > 0 and sep_first_gt12 == 0:
|
||
return "MM/DD/YYYY"
|
||
# ambiguo → preferencia hispana
|
||
return "DD/MM/YYYY"
|
||
|
||
|
||
def _digit_count(n: int) -> int:
|
||
return len(str(abs(n)))
|
||
|
||
|
||
def _parse_date_with_format(value: str, fmt: str) -> Optional[tuple[int, int, int]]:
|
||
"""Convierte un valor de fecha a (día, mes, año) según el layout detectado."""
|
||
parsed = _extract_date_parts(value)
|
||
if parsed is None:
|
||
return None
|
||
a, b, c, layout = parsed
|
||
|
||
if fmt == "DD/MM/YYYY":
|
||
d, m, y = a, b, c
|
||
elif fmt == "MM/DD/YYYY":
|
||
m, d, y = a, b, c
|
||
elif fmt == "YYYY/MM/DD":
|
||
y, m, d = a, b, c
|
||
elif fmt == "YYYYMMDD" and layout == "compact8":
|
||
# reconstruir: primer4 = a*100 + b, luego mes y día
|
||
y = a * 100 + b
|
||
m = c // 100
|
||
d = c % 100
|
||
elif fmt == "DDMMYYYY" and layout == "compact8":
|
||
d = a
|
||
m = b
|
||
y = c
|
||
elif fmt == "DDMMYY" and layout == "compact6":
|
||
d = a
|
||
m = b
|
||
y = 2000 + c if c < 50 else 1900 + c
|
||
else:
|
||
# intentar como sep genérico
|
||
d, m, y = a, b, c
|
||
|
||
if not (1 <= m <= 12 and 1 <= d <= 31 and 1000 <= y <= 9999):
|
||
# año en 2 dígitos
|
||
if y < 100:
|
||
y = 2000 + y if y < 50 else 1900 + y
|
||
if not (1 <= m <= 12 and 1 <= d <= 31 and 1000 <= y <= 9999):
|
||
return None
|
||
return (d, m, y)
|
||
|
||
|
||
def _normalize_date_text(value: Any) -> str:
|
||
text = unicodedata.normalize("NFKD", str(value or ""))
|
||
text = "".join(c for c in text if not unicodedata.combining(c))
|
||
return text.lower()
|
||
|
||
|
||
def _valid_date_tuple(d: int, m: int, y: int) -> Optional[tuple[int, int, int]]:
|
||
if y < 100:
|
||
y = 2000 + y if y < 50 else 1900 + y
|
||
if not (1000 <= y <= 9999):
|
||
return None
|
||
try:
|
||
date(y, m, d)
|
||
except ValueError:
|
||
return None
|
||
return (d, m, y)
|
||
|
||
|
||
def _split_glued_year_suffix_hhmm(digits: str) -> Optional[tuple[int, int, int]]:
|
||
"""Detecta `YYYY` + `HHMM` en una sola tira de 8 cifras (p. ej. `20241344`).
|
||
|
||
Tras quitar separadores puede quedar pegado al mes (p. ej. `15may20241344`).
|
||
Devuelve (año, hora, minuto) si la parte horaria es válida; si no, None.
|
||
"""
|
||
if len(digits) != 8 or not digits.isdigit():
|
||
return None
|
||
y = int(digits[:4])
|
||
hh = int(digits[4:6])
|
||
mm = int(digits[6:8])
|
||
if not (1000 <= y <= 9999 and 0 <= hh <= 23 and 0 <= mm <= 59):
|
||
return None
|
||
return (y, hh, mm)
|
||
|
||
|
||
def _parse_text_month_candidate(
|
||
first: tuple[int, str],
|
||
month: int,
|
||
second: tuple[int, str],
|
||
*,
|
||
layout: Literal["around", "after", "before"],
|
||
) -> Optional[tuple[int, int, int]]:
|
||
a, astr = first
|
||
b, bstr = second
|
||
if layout == "around":
|
||
if len(astr) == 4:
|
||
return _valid_date_tuple(b, month, a)
|
||
if len(bstr) == 4:
|
||
return _valid_date_tuple(a, month, b)
|
||
return _valid_date_tuple(a, month, b)
|
||
if layout == "after":
|
||
if len(astr) == 4:
|
||
return _valid_date_tuple(b, month, a)
|
||
return _valid_date_tuple(a, month, b)
|
||
if len(astr) == 4:
|
||
return _valid_date_tuple(b, month, a)
|
||
return _valid_date_tuple(a, month, b)
|
||
|
||
|
||
def _parse_date_with_text_month(raw: Any) -> Optional[tuple[int, int, int]]:
|
||
if _is_nullish_scalar(raw):
|
||
return None
|
||
date_part, _ = _split_date_time(str(raw))
|
||
normalized = _normalize_date_text(date_part or raw)
|
||
# `\d{8}` antes que `\d{1,4}` para no partir año+hora pegados tras strip agresivo.
|
||
tokens = re.findall(r"[a-z]+|\d{8}|\d{1,4}", normalized)
|
||
if not tokens:
|
||
return None
|
||
|
||
for i, token in enumerate(tokens):
|
||
month = _MONTH_NAME_TO_NUM.get(token)
|
||
if month is None:
|
||
continue
|
||
nums_before = [
|
||
(int(t), t) for t in tokens[:i] if t.isdigit()
|
||
]
|
||
nums_after = [
|
||
(int(t), t) for t in tokens[i + 1 :] if t.isdigit()
|
||
]
|
||
candidates: list[Optional[tuple[int, int, int]]] = []
|
||
if nums_before and len(nums_after) == 1:
|
||
glued = _split_glued_year_suffix_hhmm(nums_after[0][1])
|
||
if glued is not None:
|
||
y, _hh, _mm = glued
|
||
candidates.append(
|
||
_parse_text_month_candidate(
|
||
nums_before[-1], month, (y, str(y)), layout="around"
|
||
)
|
||
)
|
||
if nums_before and nums_after:
|
||
candidates.append(
|
||
_parse_text_month_candidate(
|
||
nums_before[-1], month, nums_after[0], layout="around"
|
||
)
|
||
)
|
||
if len(nums_after) >= 2:
|
||
candidates.append(
|
||
_parse_text_month_candidate(
|
||
nums_after[0], month, nums_after[1], layout="after"
|
||
)
|
||
)
|
||
if len(nums_before) >= 2:
|
||
candidates.append(
|
||
_parse_text_month_candidate(
|
||
nums_before[-2], month, nums_before[-1], layout="before"
|
||
)
|
||
)
|
||
for parsed in candidates:
|
||
if parsed is not None:
|
||
return parsed
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# ANÁLISIS DE COLUMNAS
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_BOOLEAN_VALUES = {"true", "false", "si", "sí", "no", "1", "0", "yes"}
|
||
|
||
|
||
# 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
|
||
|
||
cname = str(column_name or "")
|
||
|
||
# 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()
|
||
if is_all_integer and not _raw_strings_suggest_monetary_presentation(non_null):
|
||
return (TYPE_INTEGER, None)
|
||
return (TYPE_DECIMAL, None)
|
||
|
||
unique_ratio = non_null.nunique() / len(non_null)
|
||
if non_null.nunique() <= 15 and unique_ratio < 0.3:
|
||
return TYPE_CATEGORICAL, None
|
||
if 0.1 < numeric_ratio < 0.9:
|
||
return TYPE_MIXED, None
|
||
return TYPE_TEXT, None
|
||
|
||
|
||
def _applicable_rules_for_type(inferred_type: str) -> list[str]:
|
||
"""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 == TYPE_INTEGER:
|
||
return base + ["cop_mod_split"]
|
||
if inferred_type == TYPE_DECIMAL:
|
||
return base + [
|
||
"to_integer",
|
||
"round_decimals_custom",
|
||
"doc_id_split_tipo_numero",
|
||
"doc_type_homologate",
|
||
"cop_mod_split",
|
||
]
|
||
if inferred_type == TYPE_DATE:
|
||
return base + [
|
||
"doc_id_split_tipo_numero",
|
||
"doc_type_homologate",
|
||
"date_to_ddmmyyyy",
|
||
"nota_paciente_autorizacion",
|
||
]
|
||
if inferred_type == TYPE_MIXED:
|
||
return base + [
|
||
"to_integer",
|
||
"round_decimals_custom",
|
||
"doc_id_split_tipo_numero",
|
||
"doc_type_homologate",
|
||
"date_to_ddmmyyyy",
|
||
"cop_mod_split",
|
||
"nota_paciente_autorizacion",
|
||
]
|
||
# Texto / categórico / boolean / empty → todas las de limpieza de texto + las
|
||
# conversoras (el usuario sabrá si aplica).
|
||
return base + [
|
||
"to_integer",
|
||
"round_decimals_custom",
|
||
"doc_id_split_tipo_numero",
|
||
"doc_type_homologate",
|
||
"date_to_ddmmyyyy",
|
||
"cop_mod_split",
|
||
"nota_paciente_autorizacion",
|
||
]
|
||
|
||
|
||
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] = []
|
||
total_rows = len(df)
|
||
rng = random.Random(42)
|
||
|
||
for col in df.columns:
|
||
series = df[col]
|
||
null_count = int(series.isna().sum() + (series.astype(str).str.strip() == "").sum())
|
||
# Evitar doble conteo: considerar vacío solo si no-NaN
|
||
null_count = int((series.isna() | (series.astype(str).str.strip() == "")).sum())
|
||
null_pct = (null_count / total_rows * 100) if total_rows else 0.0
|
||
|
||
non_null = series.dropna().astype(str)
|
||
non_null = non_null[non_null.str.strip() != ""]
|
||
|
||
example = ""
|
||
if not non_null.empty:
|
||
example = str(rng.choice(non_null.tolist()))
|
||
if len(example) > 80:
|
||
example = example[:77] + "..."
|
||
|
||
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, str(col))
|
||
|
||
summary = ColumnSummary(
|
||
name=str(col),
|
||
inferred_type=inferred_type,
|
||
example=example,
|
||
null_count=null_count,
|
||
null_pct=null_pct,
|
||
unique_count=unique_count,
|
||
unique_pct=unique_pct,
|
||
detected_date_format=date_fmt,
|
||
)
|
||
|
||
if inferred_type in (TYPE_INTEGER, TYPE_DECIMAL):
|
||
numeric = _to_numeric_lenient(non_null)
|
||
if numeric.notna().any():
|
||
summary.min = float(numeric.min())
|
||
summary.max = float(numeric.max())
|
||
summary.mean = float(numeric.mean())
|
||
elif inferred_type in (TYPE_TEXT, TYPE_CATEGORICAL):
|
||
if not non_null.empty:
|
||
summary.avg_length = float(non_null.str.len().mean())
|
||
counts = non_null.value_counts().head(3)
|
||
summary.top_values = [(str(v), int(c)) for v, c in counts.items()]
|
||
|
||
summary.applicable_rules = _applicable_rules_for_type(inferred_type)
|
||
summaries.append(summary)
|
||
|
||
return summaries
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# REGLAS DE LIMPIEZA
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _rule_strip_special_spaces(series: pd.Series) -> pd.Series:
|
||
"""Elimina todo lo que no sea alfanumérico (incluye espacios)."""
|
||
pattern = re.compile(r"[^A-Za-z0-9]", flags=re.UNICODE)
|
||
|
||
def _fn(x):
|
||
if pd.isna(x):
|
||
return x
|
||
return pattern.sub("", str(x))
|
||
|
||
return series.map(_fn)
|
||
|
||
|
||
def _rule_strip_special_keep_date_seps(series: pd.Series) -> pd.Series:
|
||
"""Conserva alfanuméricos, separadores de fecha '/', '-' y hora ':', y espacio."""
|
||
pattern = re.compile(r"[^A-Za-z0-9\/\-: ]", flags=re.UNICODE)
|
||
|
||
def _fn(x):
|
||
if pd.isna(x):
|
||
return x
|
||
return pattern.sub("", str(x))
|
||
|
||
return series.map(_fn)
|
||
|
||
|
||
def _rule_strip_spaces(series: pd.Series) -> pd.Series:
|
||
"""Elimina únicamente espacios (incluidos tabs y saltos de línea)."""
|
||
def _fn(x):
|
||
if pd.isna(x):
|
||
return x
|
||
return re.sub(r"\s+", "", str(x))
|
||
|
||
return series.map(_fn)
|
||
|
||
|
||
def _rule_to_integer(series: pd.Series) -> pd.Series:
|
||
"""Convierte a entero (Int64 nullable) con parser tolerante."""
|
||
numeric = _to_numeric_lenient(series.astype(str))
|
||
return numeric.round().astype("Int64")
|
||
|
||
|
||
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).
|
||
|
||
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 _is_nullish_scalar(x) or (isinstance(x, (float, np.floating)) and (not np.isfinite(x) or np.isnan(x))):
|
||
return pd.NA
|
||
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")
|
||
|
||
|
||
def _rule_doc_type_homologate(series: pd.Series) -> pd.Series:
|
||
"""Homologa una columna que contiene tipos de documento."""
|
||
def _fn(x: Any):
|
||
if _is_nullish_scalar(x):
|
||
return x
|
||
tipo, _num = split_doc_tipo_y_numero_texto(x, homologate=True)
|
||
if tipo is not None:
|
||
return tipo
|
||
return homologate_doc_type(x)
|
||
|
||
return series.map(_fn)
|
||
|
||
|
||
def _detect_compact_date_formats(
|
||
series: pd.Series,
|
||
) -> tuple[Optional[str], Optional[str]]:
|
||
cser = series.map(_normalize_messy_date_to_compact)
|
||
c8s = pd.Series(
|
||
[x for x in cser.tolist() if isinstance(x, str) and len(x) == 8]
|
||
)
|
||
c6s = pd.Series(
|
||
[x for x in cser.tolist() if isinstance(x, str) and len(x) == 6]
|
||
)
|
||
fmt8 = detect_date_format(c8s) if not c8s.empty else None
|
||
fmt6 = detect_date_format(c6s) if not c6s.empty else None
|
||
return fmt8, fmt6
|
||
|
||
|
||
def _parse_any_date_to_tuple(
|
||
raw: Any,
|
||
*,
|
||
sep_fmt: Optional[str],
|
||
compact8_fmt: Optional[str],
|
||
compact6_fmt: Optional[str],
|
||
) -> Optional[tuple[int, int, int]]:
|
||
parsed_text = _parse_date_with_text_month(raw)
|
||
if parsed_text is not None:
|
||
return parsed_text
|
||
|
||
if _is_nullish_scalar(raw):
|
||
return None
|
||
|
||
raw_text = str(raw)
|
||
date_part, _ = _split_date_time(raw_text)
|
||
target = date_part if date_part else raw_text
|
||
if sep_fmt:
|
||
parsed = _parse_date_with_format(target, sep_fmt)
|
||
if parsed is not None:
|
||
return parsed
|
||
for fallback_fmt in ("DD/MM/YYYY", "MM/DD/YYYY", "YYYY/MM/DD"):
|
||
if fallback_fmt == sep_fmt:
|
||
continue
|
||
parsed = _parse_date_with_format(target, fallback_fmt)
|
||
if parsed is not None:
|
||
return parsed
|
||
|
||
compact = _normalize_messy_date_to_compact(raw)
|
||
if not isinstance(compact, str):
|
||
return None
|
||
if len(compact) == 8:
|
||
for fmt in (compact8_fmt, "DDMMYYYY", "YYYYMMDD"):
|
||
if not fmt:
|
||
continue
|
||
parsed = _parse_date_with_format(compact, fmt)
|
||
if parsed is not None:
|
||
return parsed
|
||
if len(compact) == 6:
|
||
parsed = _parse_date_with_format(compact, compact6_fmt or "DDMMYY")
|
||
if parsed is not None:
|
||
return parsed
|
||
return None
|
||
|
||
|
||
def _rule_date_to_ddmmyyyy(series: pd.Series) -> pd.Series:
|
||
"""Normaliza fechas numéricas, alfanuméricas, con meses en texto y con ruido."""
|
||
stripped = series.astype(str).str.replace(r"[\s\/\-\.]", "", regex=True)
|
||
stripped = stripped.where(series.notna(), other=pd.NA)
|
||
sep_fmt = detect_date_format(series) or detect_date_format(stripped)
|
||
compact8_fmt, compact6_fmt = _detect_compact_date_formats(series)
|
||
|
||
def _fn(x: Any):
|
||
if pd.isna(x) or str(x).strip() == "":
|
||
return x
|
||
parsed = _parse_any_date_to_tuple(
|
||
x,
|
||
sep_fmt=sep_fmt,
|
||
compact8_fmt=compact8_fmt,
|
||
compact6_fmt=compact6_fmt,
|
||
)
|
||
if parsed is None:
|
||
return x
|
||
d, m, y = parsed
|
||
return f"{d:02d}/{m:02d}/{y:04d}"
|
||
|
||
return series.map(_fn)
|
||
|
||
|
||
def _rule_date_numeric_to_ddmmyyyy(series: pd.Series) -> pd.Series:
|
||
"""Entrada: fechas en formato numérico (ej. 20240315 o 15032024).
|
||
|
||
Salida: "DD/MM/AAAA" tras detectar el formato predominante.
|
||
"""
|
||
# Para esta regla consideramos los valores "numéricos" (quitamos separadores
|
||
# comunes si los hay para detectar un YYYYMMDD/DDMMYYYY confiable).
|
||
stripped = series.astype(str).str.replace(r"[\s\/\-\.]", "", regex=True)
|
||
stripped = stripped.where(series.notna(), other=pd.NA)
|
||
fmt = detect_date_format(stripped)
|
||
if fmt is None:
|
||
# fallback: intentar con detector normal
|
||
fmt = detect_date_format(series)
|
||
return _format_dates_to_ddmmyyyy(series, fmt)
|
||
|
||
|
||
def _rule_date_alpha_to_ddmmyyyy(series: pd.Series) -> pd.Series:
|
||
"""Entrada: fechas en formato alfanumérico con separadores.
|
||
|
||
Salida: "DD/MM/AAAA".
|
||
"""
|
||
fmt = detect_date_format(series)
|
||
return _format_dates_to_ddmmyyyy(series, fmt)
|
||
|
||
|
||
def _format_dates_to_ddmmyyyy(series: pd.Series, fmt: Optional[str]) -> pd.Series:
|
||
"""Convierte cada valor al formato DD/MM/AAAA usando `fmt` como layout de
|
||
entrada detectado. La parte de hora (si existe) se descarta. Mantiene el
|
||
valor original si no puede parsear la fecha.
|
||
"""
|
||
if fmt is None:
|
||
return series.copy()
|
||
|
||
def _fn(x):
|
||
if pd.isna(x) or str(x).strip() == "":
|
||
return x
|
||
raw = str(x)
|
||
date_part, _time_part = _split_date_time(raw)
|
||
target = date_part if date_part else raw
|
||
parsed = _parse_date_with_format(target, fmt)
|
||
if parsed is None:
|
||
return x
|
||
d, m, y = parsed
|
||
return f"{d:02d}/{m:02d}/{y:04d}"
|
||
|
||
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
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_DOC_TYPE_ALIASES: dict[str, tuple[str, ...]] = {
|
||
"CC": ("CC", "Cedula", "CD", "C", "C.C", "C C", "CCC"),
|
||
"TI": ("TI", "Tarjeta", "Tarjeta Identidad", "T1", "T", "TL"),
|
||
"CN": ("CN", "Certificado", "Nacido Vivo", "Certificado Nacido Vivo"),
|
||
"CE": ("CE", "Cedula de Extranjeria", "Extranjera"),
|
||
"ME": ("ME", "Menor sin Identificacion"),
|
||
"PS": ("PS", "Pasaporte"),
|
||
"PT": ("PT", "Permiso por proteccion Temporal"),
|
||
"RC": ("RC", "Registro Civil", "Registro", "TC"),
|
||
"SC": ("SC", "Salvo Conducto", "Salvoconducto"),
|
||
"PE": ("PE", "Permiso especial de permanencia", "Permiso"),
|
||
}
|
||
|
||
|
||
def _ascii_upper(value: Any) -> str:
|
||
text = unicodedata.normalize("NFKD", str(value or ""))
|
||
text = "".join(c for c in text if not unicodedata.combining(c))
|
||
return text.upper()
|
||
|
||
|
||
def _normalize_doc_type_key(value: Any) -> str:
|
||
"""Clave comparable: sin acentos, espacios, puntos ni símbolos."""
|
||
return re.sub(r"[^A-Z0-9]", "", _ascii_upper(value))
|
||
|
||
|
||
def _clean_doc_type_text(value: Any) -> Optional[str]:
|
||
cleaned = re.sub(r"[^A-Z0-9]+", " ", _ascii_upper(value)).strip()
|
||
cleaned = re.sub(r"\s+", " ", cleaned)
|
||
return cleaned or None
|
||
|
||
|
||
_DOC_TYPE_ALIAS_BY_KEY: dict[str, str] = {
|
||
_normalize_doc_type_key(alias): canonical
|
||
for canonical, aliases in _DOC_TYPE_ALIASES.items()
|
||
for alias in aliases
|
||
}
|
||
|
||
_DOC_TYPE_ALIAS_DISPLAY_BY_KEY: dict[str, str] = {}
|
||
for _canonical, _aliases in _DOC_TYPE_ALIASES.items():
|
||
for _alias in _aliases:
|
||
_alias_key = _normalize_doc_type_key(_alias)
|
||
_alias_display = _clean_doc_type_text(_alias)
|
||
if _alias_key and _alias_display:
|
||
_DOC_TYPE_ALIAS_DISPLAY_BY_KEY.setdefault(_alias_key, _alias_display)
|
||
|
||
_DOC_TYPE_ALIAS_KEYS_SORTED: list[str] = sorted(
|
||
_DOC_TYPE_ALIAS_BY_KEY,
|
||
key=lambda k: (len(k), k),
|
||
reverse=True,
|
||
)
|
||
|
||
|
||
def homologate_doc_type(value: Any) -> Optional[str]:
|
||
"""Homologa un tipo documental a sigla; si no existe, conserva texto limpio."""
|
||
if _is_nullish_scalar(value):
|
||
return None
|
||
key = _normalize_doc_type_key(value)
|
||
if not key or key.lower() in ("nan", "none", "nat"):
|
||
return None
|
||
return _DOC_TYPE_ALIAS_BY_KEY.get(key) or _clean_doc_type_text(value)
|
||
|
||
|
||
def _compact_doc_text(value: Any) -> str:
|
||
return _normalize_doc_type_key(value)
|
||
|
||
|
||
def _looks_like_short_alias_prefix(compact: str, alias_key: str) -> bool:
|
||
"""Evita que alias de 1-2 chars capturen palabras como CODIGO o TOTAL."""
|
||
if not compact.startswith(alias_key):
|
||
return False
|
||
rest = compact[len(alias_key):]
|
||
if not rest:
|
||
return True
|
||
return bool(rest[0].isdigit())
|
||
|
||
|
||
def _match_known_doc_type_prefix(compact: str) -> Optional[tuple[str, str]]:
|
||
for alias_key in _DOC_TYPE_ALIAS_KEYS_SORTED:
|
||
if len(alias_key) <= 2:
|
||
if not _looks_like_short_alias_prefix(compact, alias_key):
|
||
continue
|
||
elif not compact.startswith(alias_key):
|
||
continue
|
||
return _DOC_TYPE_ALIAS_BY_KEY[alias_key], alias_key
|
||
return None
|
||
|
||
|
||
def split_doc_tipo_y_numero_texto(
|
||
raw: Any,
|
||
*,
|
||
homologate: bool = True,
|
||
) -> tuple[Optional[str], Optional[str]]:
|
||
"""Separa tipo documental y número, tolerando nombres largos y símbolos.
|
||
|
||
Si `homologate` es True, tipos conocidos se devuelven como sigla (CC, TI,
|
||
etc.). Los tipos desconocidos se conservan limpios en mayúsculas.
|
||
"""
|
||
if _is_nullish_scalar(raw) and not (isinstance(raw, str) and not str(raw).strip()):
|
||
return (None, None)
|
||
t = str(raw).strip()
|
||
if t.lower() in ("nan", "none", "nat", "<na>"):
|
||
return (None, None)
|
||
if t and t[0].isdigit():
|
||
return (None, None)
|
||
|
||
compact = _compact_doc_text(t)
|
||
if not compact:
|
||
return (None, None)
|
||
|
||
match = _match_known_doc_type_prefix(compact)
|
||
if match is not None:
|
||
canonical, alias_key = match
|
||
rest = compact[len(alias_key):]
|
||
nums = re.sub(r"[^\d]", "", rest)
|
||
if homologate:
|
||
return (canonical, nums if nums else None)
|
||
raw_type = _DOC_TYPE_ALIAS_DISPLAY_BY_KEY.get(alias_key) or canonical
|
||
return (raw_type, nums if nums else None)
|
||
|
||
m = re.match(
|
||
r"^([^\d]+?)\s*([0-9].*)\s*$",
|
||
t,
|
||
)
|
||
if m:
|
||
to_t = _clean_doc_type_text(m.group(1))
|
||
nums = re.sub(r"[^\d]", "", m.group(2))
|
||
return (homologate_doc_type(to_t) if homologate else to_t, nums if nums else None)
|
||
|
||
if not re.search(r"\d", t):
|
||
to_t = homologate_doc_type(t) if homologate else _clean_doc_type_text(t)
|
||
return (to_t, None)
|
||
return (None, None)
|
||
|
||
|
||
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], *, homologate_type: bool = True
|
||
) -> None:
|
||
tcol, ncol = _alloc_doc_split_column_names(str(col), set(result.columns))
|
||
st_tipo: list[Optional[str]] = []
|
||
st_num: list[Any] = []
|
||
for v in result[col].tolist():
|
||
tt, nn = split_doc_tipo_y_numero_texto(v, homologate=homologate_type)
|
||
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)"
|
||
)
|
||
|
||
|
||
# --- Texto tipo nota (Paciente / Autorización) --------------------------------
|
||
|
||
_PACIENTE_DOC_NOTA_RE = re.compile(r"(?i)Paciente:\s*(\d+)\s*-\s*")
|
||
_AUTORIZACION_NOTA_RE = re.compile(r"(?i)Autorizaci[oó]n\s*No\.?\s*(\d+)")
|
||
|
||
|
||
def _normalize_nota_text(raw: Any) -> str:
|
||
if _is_nullish_scalar(raw):
|
||
return ""
|
||
try:
|
||
if pd.isna(raw):
|
||
return ""
|
||
except TypeError:
|
||
pass
|
||
s = str(raw).strip()
|
||
if not s or s.lower() in ("nan", "none", "nat", "<na>"):
|
||
return ""
|
||
return re.sub(r"\s+", " ", s)
|
||
|
||
|
||
def extract_paciente_documento_nota(raw: Any) -> Optional[str]:
|
||
"""Número de documento tras ``Paciente:`` y antes del guión del nombre."""
|
||
t = _normalize_nota_text(raw)
|
||
if not t:
|
||
return None
|
||
m = _PACIENTE_DOC_NOTA_RE.search(t)
|
||
return m.group(1) if m else None
|
||
|
||
|
||
def extract_autorizacion_nota(raw: Any) -> Optional[str]:
|
||
"""Número tras ``Autorización No.`` (opcional por fila)."""
|
||
t = _normalize_nota_text(raw)
|
||
if not t:
|
||
return None
|
||
m = _AUTORIZACION_NOTA_RE.search(t)
|
||
return m.group(1) if m else None
|
||
|
||
|
||
def _alloc_nota_extract_column_names(
|
||
col_name: str, existing: set[str]
|
||
) -> tuple[str, str]:
|
||
"""Nombres únicos para documento paciente y autorización."""
|
||
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 ""
|
||
dcol, acol = f"{base}{suf}_nota_paciente_doc", f"{base}{suf}_nota_autorizacion"
|
||
if dcol not in existing and acol not in existing:
|
||
return dcol, acol
|
||
raise ValueError("No se pudieron asignar nombres de columnas únicos.")
|
||
|
||
|
||
def _apply_nota_extract_columns(result: pd.DataFrame, col: str, log: list[str]) -> None:
|
||
"""Añade columnas con documento (Paciente:) y número de autorización si existe."""
|
||
dcol, acol = _alloc_nota_extract_column_names(str(col), set(result.columns))
|
||
docs: list[Any] = []
|
||
auths: list[Any] = []
|
||
for v in result[col].tolist():
|
||
dd = extract_paciente_documento_nota(v)
|
||
aa = extract_autorizacion_nota(v)
|
||
docs.append(dd if dd is not None else pd.NA)
|
||
auths.append(aa if aa is not None else pd.NA)
|
||
result[dcol] = docs
|
||
result[acol] = auths
|
||
log.append(
|
||
f"[OK] '{str(col)}' -> columnas '{dcol}' y '{acol}' <- Extraer datos de notas"
|
||
)
|
||
|
||
|
||
# Registro de reglas
|
||
CLEANING_RULES: list[dict[str, Any]] = [
|
||
{
|
||
"id": "strip_special_spaces",
|
||
"name": "Eliminar caracteres especiales y espacios",
|
||
"description": (
|
||
"Deja solo caracteres alfanuméricos (A-Z, a-z, 0-9). No usar antes de "
|
||
"«Normalizar fecha» en columnas con fecha y hora: pega cifras y puede impedir "
|
||
"interpretar el valor (aplique primero la normalización de fecha o use la regla "
|
||
"que conserva separadores)."
|
||
),
|
||
"fn": _rule_strip_special_spaces,
|
||
},
|
||
{
|
||
"id": "strip_special_keep_date_seps",
|
||
"name": "Quitar símbolos; conserva /, -, : y espacios (fechas o texto con fechas)",
|
||
"description": (
|
||
"Quita comas, $, comillas, etc. y deja letras, números, /, -, : (hora) y espacios. "
|
||
"Adecuado p. ej. en columna 'Fecha Servicio'."
|
||
),
|
||
"fn": _rule_strip_special_keep_date_seps,
|
||
},
|
||
{
|
||
"id": "strip_spaces",
|
||
"name": "Eliminar únicamente espacios",
|
||
"description": "Elimina todos los caracteres de espacio en blanco.",
|
||
"fn": _rule_strip_spaces,
|
||
},
|
||
{
|
||
"id": "to_integer",
|
||
"name": "Convertir a Entero",
|
||
"description": "Convierte a enteros con detección de miles/decimal (`,`/`.`) y moneda; redondea decimales.",
|
||
"fn": _rule_to_integer,
|
||
},
|
||
{
|
||
"id": "date_to_ddmmyyyy",
|
||
"name": "Normalizar fecha → DD/MM/AAAA",
|
||
"description": (
|
||
"Normaliza fechas numéricas, con separadores, con ruido o con meses en "
|
||
"texto (ej. 26-abril-2026) al formato DD/MM/AAAA."
|
||
),
|
||
"fn": _rule_date_to_ddmmyyyy,
|
||
},
|
||
{
|
||
"id": "date_numeric_to_ddmmyyyy",
|
||
"name": "Conversión de formato numérico (Fechas) → DD/MM/AAAA",
|
||
"description": "Detecta el formato numérico predominante (YYYYMMDD, DDMMYYYY, etc.) y normaliza a DD/MM/AAAA.",
|
||
"fn": _rule_date_numeric_to_ddmmyyyy,
|
||
},
|
||
{
|
||
"id": "date_alpha_to_ddmmyyyy",
|
||
"name": "Conversión de formato alfanumérico (Fechas) → DD/MM/AAAA",
|
||
"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": "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,
|
||
},
|
||
{
|
||
"id": "doc_type_homologate",
|
||
"name": "Homologar tipo de documento",
|
||
"description": (
|
||
"Convierte tipos de documento escritos como nombre largo, sigla o texto con "
|
||
"símbolos a su abreviatura estándar (CC, TI, CE, RC, etc.)."
|
||
),
|
||
"fn": _rule_doc_type_homologate,
|
||
},
|
||
{
|
||
"id": "cop_mod_split",
|
||
"name": "Separa COP/MOD (copago / moderadora)",
|
||
"description": (
|
||
"Según el año en la columna Fecha Servicio que elija en la tarjeta, clasifica "
|
||
"el importe en cuota moderadora (valores por año en Configuración) o copago. "
|
||
"Se ejecuta al final del proceso de limpieza para usar fechas ya normalizadas."
|
||
),
|
||
"fn": _rule_passthrough_series,
|
||
},
|
||
{
|
||
"id": "nota_paciente_autorizacion",
|
||
"name": "Extraer datos de notas (documento y autorización)",
|
||
"description": (
|
||
"Desde texto tipo nota hospitalaria, extrae el número tras «Paciente:» (antes del "
|
||
"guión del nombre) y, si existe, el de «Autorización No.». Añade dos columnas; "
|
||
"no modifica el texto original."
|
||
),
|
||
"fn": _rule_passthrough_series,
|
||
},
|
||
]
|
||
|
||
_RULE_BY_ID: dict[str, Callable[[pd.Series], pd.Series]] = {
|
||
r["id"]: r["fn"] for r in CLEANING_RULES
|
||
}
|
||
|
||
_RULE_NAME_BY_ID: dict[str, str] = {r["id"]: r["name"] for r in CLEANING_RULES}
|
||
|
||
|
||
def get_rule(rule_id: str) -> Optional[dict[str, Any]]:
|
||
"""Devuelve el registro de una regla por id."""
|
||
for r in CLEANING_RULES:
|
||
if r["id"] == rule_id:
|
||
return r
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# APLICACIÓN DE REGLAS
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def apply_rules(
|
||
df: pd.DataFrame,
|
||
rules_per_column: dict[str, list[str]],
|
||
*,
|
||
cop_mod_fecha_column: Optional[dict[str, str]] = None,
|
||
cop_mod_config: Optional[dict[str, Any]] = None,
|
||
) -> tuple[pd.DataFrame, list[str]]:
|
||
"""Aplica las reglas seleccionadas a cada columna.
|
||
|
||
`rules_per_column` es un dict de `columna -> lista de IDs de reglas en el
|
||
orden a aplicar`. La regla `cop_mod_split` se difiere a una segunda pasada
|
||
tras el resto, para que la columna de fecha pueda normalizarse antes.
|
||
|
||
`cop_mod_fecha_column`: mapa columna de importe -> nombre de columna Fecha Servicio.
|
||
`cop_mod_config`: opcional; puede incluir ``moderadora_by_year`` (dict año → lista).
|
||
"""
|
||
result = df.copy()
|
||
log: list[str] = []
|
||
|
||
moderadora_years = merge_moderadora_config(cop_mod_config)
|
||
|
||
def _run_rules_for_column(col: str, rule_ids: list[str]) -> None:
|
||
if not rule_ids:
|
||
return
|
||
if col not in result.columns:
|
||
log.append(f"[WARN] Columna no encontrada: {col}")
|
||
return
|
||
split_has_homologation = (
|
||
"doc_id_split_tipo_numero" in rule_ids and "doc_type_homologate" in rule_ids
|
||
)
|
||
for rule_id in rule_ids:
|
||
if rule_id == "doc_id_split_tipo_numero":
|
||
try:
|
||
_apply_doc_id_split_columns(
|
||
result, col, log, homologate_type=split_has_homologation
|
||
)
|
||
except Exception as exc: # noqa: BLE001
|
||
logger.exception("Error aplicando regla %s a %s", rule_id, col)
|
||
log.append(f"[ERROR] '{col}' regla '{rule_id}': {exc}")
|
||
continue
|
||
if rule_id == "nota_paciente_autorizacion":
|
||
try:
|
||
_apply_nota_extract_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
|
||
if rule_id == "doc_type_homologate" and split_has_homologation:
|
||
continue
|
||
fn = _RULE_BY_ID.get(rule_id)
|
||
if fn is None:
|
||
log.append(f"[WARN] Regla desconocida '{rule_id}' en columna '{col}'")
|
||
continue
|
||
try:
|
||
result[col] = fn(result[col])
|
||
log.append(f"[OK] '{col}' <- {_RULE_NAME_BY_ID.get(rule_id, rule_id)}")
|
||
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}")
|
||
|
||
# Fase 1: todas las reglas excepto cop_mod_split
|
||
for col, rule_ids in rules_per_column.items():
|
||
filtered = [r for r in rule_ids if r != "cop_mod_split"]
|
||
_run_rules_for_column(col, filtered)
|
||
|
||
# Fase 2: Separa COP/MOD sobre el DataFrame ya transformado
|
||
fecha_map = cop_mod_fecha_column or {}
|
||
for col, rule_ids in rules_per_column.items():
|
||
if "cop_mod_split" not in rule_ids:
|
||
continue
|
||
fecha_col = fecha_map.get(col)
|
||
if not fecha_col:
|
||
log.append(
|
||
f"[WARN] COP/MOD en '{col}': indique la columna «Fecha Servicio» en la tarjeta; se omite."
|
||
)
|
||
continue
|
||
try:
|
||
_apply_cop_mod_split(
|
||
result,
|
||
col,
|
||
fecha_col,
|
||
log,
|
||
moderadora_by_year=moderadora_years,
|
||
)
|
||
except Exception as exc: # noqa: BLE001
|
||
logger.exception("Error aplicando COP/MOD a %s", col)
|
||
log.append(f"[ERROR] '{col}' regla 'cop_mod_split': {exc}")
|
||
|
||
return result, log
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# CARGA ROBUSTA DE CSV
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_CANDIDATE_DELIMITERS = ["|", ";", "\t", ",", ":"]
|
||
|
||
|
||
def detect_delimiter(filepath: str) -> str:
|
||
"""Detecta el delimitador del CSV de forma robusta.
|
||
|
||
Estrategia:
|
||
1. Intentar `csv.Sniffer` con un sample grande.
|
||
2. Si falla o el resultado parece incorrecto, contar ocurrencias por línea
|
||
de cada delimitador candidato y elegir el que produce el mayor número
|
||
de columnas estable entre líneas.
|
||
"""
|
||
encodings = ("utf-8", "utf-8-sig", "latin-1", "cp1252")
|
||
sample = ""
|
||
for enc in encodings:
|
||
try:
|
||
with open(filepath, "r", encoding=enc, errors="replace") as f:
|
||
sample = f.read(65536)
|
||
break
|
||
except Exception:
|
||
continue
|
||
|
||
if not sample:
|
||
return ","
|
||
|
||
try:
|
||
dialect = _csv.Sniffer().sniff(sample, delimiters="".join(_CANDIDATE_DELIMITERS))
|
||
sniffed = dialect.delimiter
|
||
except _csv.Error:
|
||
sniffed = None
|
||
|
||
lines = [ln for ln in sample.splitlines() if ln.strip()]
|
||
if not lines:
|
||
return sniffed or ","
|
||
|
||
# Para cada candidato, calcular el número promedio de columnas y la
|
||
# consistencia (cuántas líneas tienen el mismo conteo dominante).
|
||
scores: dict[str, tuple[int, int]] = {}
|
||
for delim in _CANDIDATE_DELIMITERS:
|
||
counts = [ln.count(delim) for ln in lines[:50]]
|
||
if max(counts) == 0:
|
||
continue
|
||
# Modo (valor más frecuente)
|
||
common = Counter(counts).most_common(1)[0]
|
||
mode_count, mode_freq = common
|
||
if mode_count == 0:
|
||
continue
|
||
scores[delim] = (mode_count, mode_freq)
|
||
|
||
if scores:
|
||
# Elegir el delimitador con mayor número de columnas (mode_count) y,
|
||
# como desempate, mayor consistencia (mode_freq).
|
||
best = max(scores.items(), key=lambda kv: (kv[1][0], kv[1][1]))
|
||
best_delim = best[0]
|
||
# Si el sniffer dio algo razonable y coincide, prevalece; si difiere y
|
||
# nuestro candidato es claramente mejor, usamos el nuestro.
|
||
if sniffed and sniffed in scores:
|
||
sniffed_score = scores[sniffed]
|
||
if sniffed_score[0] >= best[1][0] - 1:
|
||
return sniffed
|
||
return best_delim
|
||
|
||
return sniffed or ","
|
||
|
||
|
||
def load_csv(filepath: str) -> tuple[pd.DataFrame, str]:
|
||
"""Carga un CSV con detección de delimitador y tolerancia a filas malas.
|
||
|
||
Devuelve `(df, delimiter)`. Lanza excepción si todos los intentos fallan.
|
||
"""
|
||
delimiter = detect_delimiter(filepath)
|
||
|
||
encodings_to_try = ["utf-8", "utf-8-sig", "latin-1", "cp1252"]
|
||
last_error: Optional[Exception] = None
|
||
|
||
for enc in encodings_to_try:
|
||
try:
|
||
df = pd.read_csv(
|
||
filepath,
|
||
delimiter=delimiter,
|
||
dtype=str,
|
||
encoding=enc,
|
||
engine="python",
|
||
on_bad_lines="skip",
|
||
)
|
||
return df, delimiter
|
||
except Exception as exc: # noqa: BLE001
|
||
last_error = exc
|
||
continue
|
||
|
||
# Último intento con quoting permisivo y sin separador detectado
|
||
try:
|
||
df = pd.read_csv(
|
||
filepath,
|
||
sep=None,
|
||
dtype=str,
|
||
engine="python",
|
||
on_bad_lines="skip",
|
||
)
|
||
return df, "(auto)"
|
||
except Exception as exc: # noqa: BLE001
|
||
last_error = exc
|
||
|
||
raise RuntimeError(f"No se pudo leer el CSV: {last_error}")
|