"""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", ""): return True return False def _strip_currency(s: str) -> str: t = s.replace("\u00a0", " ").replace("\u202f", " ") t = re.sub(r"[\s]+", "", t) t = t.strip() if not t or t.lower() in ("nan", "none", "nat"): return "" neg = False if t[0:1] == "-": neg, t = not neg, t[1:].lstrip() if t[0:1] == "(" and t.endswith(")"): neg, t = not neg, t[1:-1] t = re.sub( r"^[\$€\u00a3\u0e3f\uffe5\uffe6\uff04\ufe69]+", "", t, count=1, ) t = re.sub(r"[\$€%]+$", "", t) if not t and neg: return "-0" return ("-" if neg else "") + t def _extract_numeric_chunk(s: str) -> str: t = s.strip() t = t.replace("$", "").replace("€", "").replace("\u00a3", "") t = t.replace(" ", "") cands1 = re.findall( r"[+\-]?(?:\d{1,3}(?:[.,]\d{3})*|\d+)(?:[.,]\d+)?", t ) cands2 = re.findall( r"[+\-]?\d[.,\d ]*\d", t ) cands = cands1 + cands2 if not cands: m = re.search(r"[\d.,\-]+", t) if m: c = m.group(0).replace("(", "-").replace(")", "") return c.lstrip("+") return "" best = max(cands, key=len) if cands else "" best = best.replace(" ", "").replace("(", "-").replace(")", "") if not re.search(r"\d", best): return "" return best.lstrip("+") def _as_float_s(s: str) -> Optional[float]: if not s: return None try: f = float(s) if not np.isfinite(f) or abs(f) > 1e100: return None return f except (ValueError, TypeError, OverflowError): return None def _parse_both_separators(body: str) -> Optional[float]: """El separador más a la DERECHA entre `,` y `.` es el decimal; el otro, miles.""" if "," not in body or "." not in body: return None t = body.strip() if re.search(r"[^0-9,.\-]", t): t2 = re.sub(r"[^0-9,.\-]", "", t) t = t2 if not t: return None neg = t[0:1] == "-" b = t[1:].lstrip() if neg else t if not b: return None i_comma, i_dot = b.rfind(","), b.rfind(".") if min(i_comma, i_dot) < 0: return None dpos = max(i_comma, i_dot) th = "," if b[dpos] == "." else "." intp, fr = b[:dpos], b[dpos + 1 :] if not re.fullmatch(r"\d{1,4}$", fr): return None intc = intp.replace(th, "") if not intc.isdigit(): return None v = _as_float_s(f"{intc}.{fr}") if v is None: return None if neg and v and v is not None: return -abs(v) return v def _only_commas(w: str, dec_hint: Optional[Literal[",", "."]]) -> Optional[float]: neg = w[0:1] == "-" wn = w[1:].lstrip() if neg else w if wn and re.search(r"[^0-9,]", wn): wn = re.sub(r"[^0-9,]", "", wn) n = wn.count(",") if n == 0 and wn.isdigit(): v = float(wn) return -v if neg else v if n == 0 or "." in wn: return None if n == 1: i = wn.rfind(",") h, t = wn[:i], wn[i + 1 :] if not t.isdigit() or not h.isdigit(): return None if len(t) in (1, 2): v = _as_float_s(f"{h}.{t}") if v is None: return None return -v if neg else v if len(t) == 3: if dec_hint is None or dec_hint == ",": v = _as_float_s(f"{h}{t}") if v is None: return None return -v if neg else v if dec_hint == ".": v = _as_float_s(wn.replace(",", "")) if v is None: return None return -v if neg else v v = _as_float_s(wn.replace(",", "")) if v is None: return None return -v if neg else v def _only_dots(w: str, dec_hint: Optional[Literal[",", "."]]) -> Optional[float]: neg = w[0:1] == "-" wn = w[1:].lstrip() if neg else w if wn and re.search(r"[^0-9.]", wn): wn = re.sub(r"[^0-9.]", "", wn) n = wn.count(".") if n == 0 and wn.isdigit(): v = float(wn) return -v if neg else v if n == 0 or "," in wn: return None if n == 1: i = wn.rfind(".") h, t = wn[:i], wn[i + 1 :] if not t.isdigit() or not h.isdigit(): return None if len(t) in (1, 2): v = _as_float_s(f"{h}.{t}") if v is None: return None return -v if neg else v if len(t) == 3: if dec_hint is None or dec_hint == ".": v = _as_float_s(f"{h}{t}") if v is None: return None return -v if neg else v if dec_hint == ",": v = _as_float_s(wn.replace(".", "")) if v is None: return None return -v if neg else v v = _as_float_s(wn.replace(".", "")) if v is None: return None return -v if neg else v def _parse_lenient_one( raw: Any, decimal_hint: Optional[Literal[",", "."]] = None, ) -> float: if _is_nullish_scalar(raw): return float("nan") t = _strip_currency(str(raw)) if not t or t == "-": return float("nan") c = _extract_numeric_chunk(t) or t c = c.strip() if not c or c == "-": return float("nan") n_comma, n_dot = c.count(","), c.count(".") if n_comma and n_dot: f = _parse_both_separators(c) elif n_comma and not n_dot: f = _only_commas(c, decimal_hint) elif n_dot and not n_comma: f = _only_dots(c, decimal_hint) else: c2 = re.sub(r"[^0-9.\-]", "", c) f = _as_float_s(c2) if c2 and re.fullmatch(r"-?\d+$", c2) else None if f is not None and np.isfinite(f): return f return float("nan") def _infer_decimal_hint_from_column( series: pd.Series, ) -> Optional[Literal[",", "."]]: """Voto con celdas que traen `,` y `.`: gana el de más a la derecha = decimal.""" votes: list[Literal[",", "."]] = [] for raw in series.dropna(): if _is_nullish_scalar(raw): continue t = _strip_currency(str(raw)) c = _extract_numeric_chunk(t) or t c = c.strip() if "," not in c or "." not in c or _parse_both_separators(c) is None: continue if c.rfind(",") > c.rfind("."): votes.append(",") else: votes.append(".") n = len(votes) if n < _MIN_HINT_VOTES: return None a = sum(1 for v in votes if v == ",") b = sum(1 for v in votes if v == ".") if a + b == 0: return None if a / n >= _HINT_MAJORITY and a > b: return "," if b / n >= _HINT_MAJORITY and b > a: return "." return None def _to_numeric_lenient( series: pd.Series, decimal_hint: Optional[Literal[",", "."]] = None, ) -> pd.Series: """Convierte a flotante; pista de columna por voto (si `decimal_hint` es None).""" hint = decimal_hint if decimal_hint is not None else _infer_decimal_hint_from_column( series ) def _cell(x: Any) -> float: if _is_nullish_scalar(x): return float("nan") return _parse_lenient_one(x, hint) return series.map(_cell) def _column_name_suggests_date_like(name: str) -> bool: """Columnas cuyo nombre pide tratar el contenido como fecha a priori.""" n = re.sub(r"[_\s.]+", " ", (name or "")).lower().strip() if not n: return False if re.search( r"\b(fecha|date|año|mes|dia|día|f[_-]?(alta|baja|proc|corte|pago|ini|fin|" r"desde|hasta|vigenc|vence|venc|caduc|f\s?ec)\b|periodo|anio|ano|expir)", n, ) or n.startswith("fec_") or n.startswith("fecha"): return True return False def _column_name_suggests_monetary_value(name: str) -> bool: """importe, valor, neto, IVA, etc. — no interpretar cifras como fechas. Encabezados de tipo `fecha_…` no se tratan como moneda salvo mezcla con monto/valor. """ n = re.sub(r"[_\s.]+", " ", (name or "")).lower().strip() if not n: return False if _column_name_suggests_date_like(name) and not re.search( r"\b(monto|valor|importe|pago|total|bruto|neto|import|unit|precio|copago)\b", n ): return False for key in ( "monto", "importe", "import", "bruto", "neto", "subtotal", "sub total", "unitario", "unidad", "valor", "precio", "pago", "copago", "sald", "dcto", "franq", "cuota", "iva", "tarif", "costo", "carg", "reten", "retencion", ): if key in n: if key == "iva" and re.search(r"(priva|giva|diva|viva|cliva)\b", n): continue if key == "carg" and re.search(r"descarg|recarg|abord", n): continue return True return False def _column_name_suggests_code_id(name: str) -> bool: """Código, ref., radicado, tipoynúm de documento — no forzar 6-8 cifras como fechas (DDMMYY, etc.).""" n = re.sub(r"[_\s.]+", " ", (name or "")).lower().strip() if not n or _column_name_suggests_date_like(name): return False for key in ( "codigo", "código", "codig", "códig", "cód_", "sku", "ean", "upc", "isbn", "upn", "ref", "referen", "radic", "nro_fac", "nro fac", "nit", "cédul", "cedul", "autoriz", "correlat", "secuenc", "línea de", "linea de", "cupo", "tipy", "tipi", "tipynum", "typnum", "type_n", "n_doc", "n_docu", "núm doc", "n_docu", "doc_id", "docum", "id_doc", ): if key in n: return True if re.search( r"tipo[_\s\\-]*(&|/|u|e|y)[_\s\\-]*num|tipo[_\s\\-]+doc|tipo_y_n", n, re.IGNORECASE, ) or n.startswith("tip_"): return True if n.startswith("cod_") or n.startswith("cód_") or n.startswith("ref_"): return True return bool(re.search(r"\b(nro[._]fac|#?\s*autor|#?\s*ref)\b", n, re.IGNORECASE)) def _column_name_suggests_quantity_or_number_not_bool(name: str) -> bool: """Nombres típicos de cantidad, importe, fechas: no inferir bool por 0/1.""" n = re.sub(r"[_\s.]+", " ", (name or "")).lower().strip() if not n: return False if _column_name_suggests_monetary_value(name) or _column_name_suggests_code_id(name): return True for key in ( "fecha", "monto", "importe", "bruto", "neto", "subtotal", "unitario", "unidad", "valor", "precio", "pago", "copago", "cantid", "quantity", "qty", "nro", "núm", "n°", ): if key in n: return True return False def _raw_strings_suggest_monetary_presentation(non_null: pd.Series) -> bool: """$ €, miles 1,234,12 o .00: presentación monetaria aunque el float sea "entero".""" if non_null is None or len(non_null) == 0: return False n_hit = 0 n_seen = 0 for raw in non_null.head(500): if _is_nullish_scalar(raw): continue t = str(raw) s = t.strip() n_seen += 1 if re.search(r"[$\u00a3\u20ac\uffe5\uffe6\ufe69\uff04]", t): n_hit += 1 continue if re.search(r"\b\d{1,3}(?:[.,\u00a0]\d{3})+\b", s) or re.search( r"(?= 0.10 def _cell_looks_cifra_con_moneda_iso_al_final(s: str) -> bool: """Solo cifra/símbolo (no prefijo de documento) y código de moneda al final. Excluye filas al estilo `CC12…` (típico alfanumérico). """ t = str(s).strip() u = t.upper() if not re.search( r"(?i)(USD|COP|EUR|CLP|ARS|GBP|MXN|PEN|DOP|UYU|VES|CRC|PYG|BRL|BOB|GTQ)\b\s*\Z", u ): return False if re.match( r"^(?i)(CC|CE|NIT|NUIP|TI|PA|MS|RC|DNI|NUIP|CN|SC|AS|PE|PT)\D*[\d0]", u ): return False return re.match( r"^[\$€0-9\s,.\-+%/]", t, ) is not None def _raw_strings_suggest_alnum_mixed_presentation(non_null: pd.Series) -> bool: """Letras y dígitos juntos (p. ej. `CC32…`, `A1-2B`); no es puro 'entero/decimal' lógicamente. Excluido: columnas con apariencia monetaria, y celdas solo cifra + ISO al final (100 USD). """ if _raw_strings_suggest_monetary_presentation(non_null): return False n_m = 0 n_s = 0 for raw in non_null.head(500): if _is_nullish_scalar(raw): continue t = str(raw).strip() n_s += 1 if not re.search(r"\d", t) or not re.search( r"[A-Za-zÑñ\u00c0-\u024f]", t, re.IGNORECASE ): continue if _cell_looks_cifra_con_moneda_iso_al_final(t): continue n_m += 1 if n_s == 0: return False return n_m / n_s >= 0.20 def _cell_looks_like_documento_id_tipo_nit(s: str) -> bool: """Cadenas estilo `CC32324029`, `CE 1.234.567` — no son fechas (evita 8 cifras como DDMM). Al menos una letra al inicio y 5-16 dígitos en la celda (no `Jan`+`2020` = 4 dígitos). """ t = str(s).strip() if not t or len(t) < 3: return False if not re.match(r"^[A-Za-z]{1,3}", t): return False dig = re.sub(r"[^\d]", "", t) if len(dig) < 5 or len(dig) > 16: return False return True def _content_looks_like_doc_id_letter_prefix_column(non_null: pd.Series) -> bool: """Muestras con prefijo documental (CC, NIT, …) y ≥5 dígitos → no columna de fecha.""" n_ok = 0 n_seen = 0 for raw in non_null.head(500): if _is_nullish_scalar(raw): continue s = str(raw).strip() n_seen += 1 if _cell_looks_like_documento_id_tipo_nit(s): n_ok += 1 if n_seen == 0: return False return n_ok / n_seen >= 0.20 def _detect_messy_text_date_format_guess(non_null: pd.Series) -> Optional[str]: """Fechas con ruido: compacta a dígitos; si hay consenso, reutiliza el detector. (Depende de `_normalize_messy_date_to_compact`, definida más abajo en el módulo.) """ # resolución en tiempo de ejecución norm = non_null.map(_normalize_messy_date_to_compact) # type: ignore[name-defined] compacts: list[str] = [] for c in norm: if isinstance(c, str) and len(c) in (6, 8): compacts.append(c) n = max(len(non_null), 1) if not compacts or len(compacts) < max(2, int(0.25 * n)): return None s_compact = pd.Series(compacts, dtype=object) return detect_date_format(s_compact) def _infer_column_type( series: pd.Series, column_name: str = "" ) -> tuple[str, Optional[str]]: """Devuelve (tipo, formato_fecha detectado; solo si aplica a tipo fecha).""" non_null = series.dropna().astype(str).str.strip() non_null = non_null[non_null != ""] if non_null.empty: return TYPE_EMPTY, None 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", ""): 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", ""): 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", ""): 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}")