feat: Enhance CSV cleaning and management functionalities by adding new date normalization rules, improving document type homologation, and implementing robust sanitization for CSV cell values. Update GUI components to support loading indicators and status messages during CSV processing, ensuring a smoother user experience.

application/csv_cleaning.py

@@ -14,8 +14,10 @@ import logging
 import math
 import random
 import re
+import unicodedata
 from collections import Counter
 from dataclasses import dataclass, field
+from datetime import date
 from typing import Any, Callable, Literal, Optional
 
 import numpy as np
@@ -101,6 +103,35 @@ _TIME_TAIL_RE = re.compile(
   r"[\sT]+(\d{1,2})(?::(\d{1,2})(?::(\d{1,2}))?)?\s*([APap]\.?\s*[Mm]\.?)?\s*$"
 )
 
+_MONTH_NAME_TO_NUM = {
+  "enero": 1,
+  "ene": 1,
+  "febrero": 2,
+  "feb": 2,
+  "marzo": 3,
+  "mar": 3,
+  "abril": 4,
+  "abr": 4,
+  "mayo": 5,
+  "may": 5,
+  "junio": 6,
+  "jun": 6,
+  "julio": 7,
+  "jul": 7,
+  "agosto": 8,
+  "ago": 8,
+  "septiembre": 9,
+  "setiembre": 9,
+  "sep": 9,
+  "sept": 9,
+  "octubre": 10,
+  "oct": 10,
+  "noviembre": 11,
+  "nov": 11,
+  "diciembre": 12,
+  "dic": 12,
+}
+
 
 def _split_date_time(value: str) -> tuple[str, Optional[str]]:
   """Separa la parte de fecha de la parte de hora si está presente.
@@ -312,6 +343,118 @@ def _parse_date_with_format(value: str, fmt: str) -> Optional[tuple[int, int, in
   return (d, m, y)
 
 
+def _normalize_date_text(value: Any) -> str:
+  text = unicodedata.normalize("NFKD", str(value or ""))
+  text = "".join(c for c in text if not unicodedata.combining(c))
+  return text.lower()
+
+
+def _valid_date_tuple(d: int, m: int, y: int) -> Optional[tuple[int, int, int]]:
+  if y < 100:
+    y = 2000 + y if y < 50 else 1900 + y
+  if not (1000 <= y <= 9999):
+    return None
+  try:
+    date(y, m, d)
+  except ValueError:
+    return None
+  return (d, m, y)
+
+
+def _split_glued_year_suffix_hhmm(digits: str) -> Optional[tuple[int, int, int]]:
+  """Detecta `YYYY` + `HHMM` en una sola tira de 8 cifras (p. ej. `20241344`).
+
+  Tras quitar separadores puede quedar pegado al mes (p. ej. `15may20241344`).
+  Devuelve (año, hora, minuto) si la parte horaria es válida; si no, None.
+  """
+  if len(digits) != 8 or not digits.isdigit():
+    return None
+  y = int(digits[:4])
+  hh = int(digits[4:6])
+  mm = int(digits[6:8])
+  if not (1000 <= y <= 9999 and 0 <= hh <= 23 and 0 <= mm <= 59):
+    return None
+  return (y, hh, mm)
+
+
+def _parse_text_month_candidate(
+  first: tuple[int, str],
+  month: int,
+  second: tuple[int, str],
+  *,
+  layout: Literal["around", "after", "before"],
+) -> Optional[tuple[int, int, int]]:
+  a, astr = first
+  b, bstr = second
+  if layout == "around":
+    if len(astr) == 4:
+      return _valid_date_tuple(b, month, a)
+    if len(bstr) == 4:
+      return _valid_date_tuple(a, month, b)
+    return _valid_date_tuple(a, month, b)
+  if layout == "after":
+    if len(astr) == 4:
+      return _valid_date_tuple(b, month, a)
+    return _valid_date_tuple(a, month, b)
+  if len(astr) == 4:
+    return _valid_date_tuple(b, month, a)
+  return _valid_date_tuple(a, month, b)
+
+
+def _parse_date_with_text_month(raw: Any) -> Optional[tuple[int, int, int]]:
+  if _is_nullish_scalar(raw):
+    return None
+  date_part, _ = _split_date_time(str(raw))
+  normalized = _normalize_date_text(date_part or raw)
+  # `\d{8}` antes que `\d{1,4}` para no partir año+hora pegados tras strip agresivo.
+  tokens = re.findall(r"[a-z]+|\d{8}|\d{1,4}", normalized)
+  if not tokens:
+    return None
+
+  for i, token in enumerate(tokens):
+    month = _MONTH_NAME_TO_NUM.get(token)
+    if month is None:
+      continue
+    nums_before = [
+      (int(t), t) for t in tokens[:i] if t.isdigit()
+    ]
+    nums_after = [
+      (int(t), t) for t in tokens[i + 1 :] if t.isdigit()
+    ]
+    candidates: list[Optional[tuple[int, int, int]]] = []
+    if nums_before and len(nums_after) == 1:
+      glued = _split_glued_year_suffix_hhmm(nums_after[0][1])
+      if glued is not None:
+        y, _hh, _mm = glued
+        candidates.append(
+          _parse_text_month_candidate(
+            nums_before[-1], month, (y, str(y)), layout="around"
+          )
+        )
+    if nums_before and nums_after:
+      candidates.append(
+        _parse_text_month_candidate(
+          nums_before[-1], month, nums_after[0], layout="around"
+        )
+      )
+    if len(nums_after) >= 2:
+      candidates.append(
+        _parse_text_month_candidate(
+          nums_after[0], month, nums_after[1], layout="after"
+        )
+      )
+    if len(nums_before) >= 2:
+      candidates.append(
+        _parse_text_month_candidate(
+          nums_before[-2], month, nums_before[-1], layout="before"
+        )
+      )
+    for parsed in candidates:
+      if parsed is not None:
+        return parsed
+  return None
+
+
 # ---------------------------------------------------------------------------
 # ANÁLISIS DE COLUMNAS
 # ---------------------------------------------------------------------------
@@ -856,22 +999,25 @@ def _applicable_rules_for_type(inferred_type: str) -> list[str]:
   if inferred_type == TYPE_INTEGER:
     return base
   if inferred_type == TYPE_DECIMAL:
-    return base + ["to_integer", "round_decimals_custom", "doc_id_split_tipo_numero"]
+    return base + [
+      "to_integer",
+      "round_decimals_custom",
+      "doc_id_split_tipo_numero",
+      "doc_type_homologate",
+    ]
   if inferred_type == TYPE_DATE:
     return base + [
       "doc_id_split_tipo_numero",
-      "date_messy_text_to_ddmmyyyy",
-      "date_numeric_to_ddmmyyyy",
-      "date_alpha_to_ddmmyyyy",
+      "doc_type_homologate",
+      "date_to_ddmmyyyy",
     ]
   if inferred_type == TYPE_MIXED:
     return base + [
       "to_integer",
       "round_decimals_custom",
       "doc_id_split_tipo_numero",
-      "date_messy_text_to_ddmmyyyy",
-      "date_numeric_to_ddmmyyyy",
-      "date_alpha_to_ddmmyyyy",
+      "doc_type_homologate",
+      "date_to_ddmmyyyy",
     ]
   # Texto / categórico / boolean / empty → todas las de limpieza de texto + las
   # conversoras (el usuario sabrá si aplica).
@@ -879,9 +1025,8 @@ def _applicable_rules_for_type(inferred_type: str) -> list[str]:
     "to_integer",
     "round_decimals_custom",
     "doc_id_split_tipo_numero",
-    "date_messy_text_to_ddmmyyyy",
-    "date_numeric_to_ddmmyyyy",
-    "date_alpha_to_ddmmyyyy",
+    "doc_type_homologate",
+    "date_to_ddmmyyyy",
   ]
 
 
@@ -962,8 +1107,8 @@ def _rule_strip_special_spaces(series: pd.Series) -> pd.Series:
 
 
 def _rule_strip_special_keep_date_seps(series: pd.Series) -> pd.Series:
-  """Conserva alfanuméricos y separadores de fecha: '/', '-', espacio."""
-  pattern = re.compile(r"[^A-Za-z0-9\/\- ]", flags=re.UNICODE)
+  """Conserva alfanuméricos, separadores de fecha '/', '-' y hora ':', y espacio."""
+  pattern = re.compile(r"[^A-Za-z0-9\/\-: ]", flags=re.UNICODE)
 
   def _fn(x):
     if pd.isna(x):
@@ -1030,6 +1175,103 @@ def _rule_round_decimals_custom(series: pd.Series) -> pd.Series:
   return numeric.map(_fn).astype("Int64")
 
 
+def _rule_doc_type_homologate(series: pd.Series) -> pd.Series:
+  """Homologa una columna que contiene tipos de documento."""
+  def _fn(x: Any):
+    if _is_nullish_scalar(x):
+      return x
+    tipo, _num = split_doc_tipo_y_numero_texto(x, homologate=True)
+    if tipo is not None:
+      return tipo
+    return homologate_doc_type(x)
+
+  return series.map(_fn)
+
+
+def _detect_compact_date_formats(
+  series: pd.Series,
+) -> tuple[Optional[str], Optional[str]]:
+  cser = series.map(_normalize_messy_date_to_compact)
+  c8s = pd.Series(
+    [x for x in cser.tolist() if isinstance(x, str) and len(x) == 8]
+  )
+  c6s = pd.Series(
+    [x for x in cser.tolist() if isinstance(x, str) and len(x) == 6]
+  )
+  fmt8 = detect_date_format(c8s) if not c8s.empty else None
+  fmt6 = detect_date_format(c6s) if not c6s.empty else None
+  return fmt8, fmt6
+
+
+def _parse_any_date_to_tuple(
+  raw: Any,
+  *,
+  sep_fmt: Optional[str],
+  compact8_fmt: Optional[str],
+  compact6_fmt: Optional[str],
+) -> Optional[tuple[int, int, int]]:
+  parsed_text = _parse_date_with_text_month(raw)
+  if parsed_text is not None:
+    return parsed_text
+
+  if _is_nullish_scalar(raw):
+    return None
+
+  raw_text = str(raw)
+  date_part, _ = _split_date_time(raw_text)
+  target = date_part if date_part else raw_text
+  if sep_fmt:
+    parsed = _parse_date_with_format(target, sep_fmt)
+    if parsed is not None:
+      return parsed
+  for fallback_fmt in ("DD/MM/YYYY", "MM/DD/YYYY", "YYYY/MM/DD"):
+    if fallback_fmt == sep_fmt:
+      continue
+    parsed = _parse_date_with_format(target, fallback_fmt)
+    if parsed is not None:
+      return parsed
+
+  compact = _normalize_messy_date_to_compact(raw)
+  if not isinstance(compact, str):
+    return None
+  if len(compact) == 8:
+    for fmt in (compact8_fmt, "DDMMYYYY", "YYYYMMDD"):
+      if not fmt:
+        continue
+      parsed = _parse_date_with_format(compact, fmt)
+      if parsed is not None:
+        return parsed
+  if len(compact) == 6:
+    parsed = _parse_date_with_format(compact, compact6_fmt or "DDMMYY")
+    if parsed is not None:
+      return parsed
+  return None
+
+
+def _rule_date_to_ddmmyyyy(series: pd.Series) -> pd.Series:
+  """Normaliza fechas numéricas, alfanuméricas, con meses en texto y con ruido."""
+  stripped = series.astype(str).str.replace(r"[\s\/\-\.]", "", regex=True)
+  stripped = stripped.where(series.notna(), other=pd.NA)
+  sep_fmt = detect_date_format(series) or detect_date_format(stripped)
+  compact8_fmt, compact6_fmt = _detect_compact_date_formats(series)
+
+  def _fn(x: Any):
+    if pd.isna(x) or str(x).strip() == "":
+      return x
+    parsed = _parse_any_date_to_tuple(
+      x,
+      sep_fmt=sep_fmt,
+      compact8_fmt=compact8_fmt,
+      compact6_fmt=compact6_fmt,
+    )
+    if parsed is None:
+      return x
+    d, m, y = parsed
+    return f"{d:02d}/{m:02d}/{y:04d}"
+
+  return series.map(_fn)
+
+
 def _rule_date_numeric_to_ddmmyyyy(series: pd.Series) -> pd.Series:
   """Entrada: fechas en formato numérico (ej. 20240315 o 15032024).
 
@@ -1166,12 +1408,102 @@ def _rule_date_messy_text_to_ddmmyyyy(series: pd.Series) -> pd.Series:
 # Tipo y número de documento (CC, NIT, CE, MS, …) → 2 columnas
 # ---------------------------------------------------------------------------
 
-def split_doc_tipo_y_numero_texto(raw: Any) -> tuple[Optional[str], Optional[str]]:
-  """Separa cadenas `CC-######`, `NIT900…`, `CE 12-34-56` en (tipo, solo dígitos).
+_DOC_TYPE_ALIASES: dict[str, tuple[str, ...]] = {
+  "CC": ("CC", "Cedula", "CD", "C", "C.C", "C C", "CCC"),
+  "TI": ("TI", "Tarjeta", "Tarjeta Identidad", "T1", "T", "TL"),
+  "CN": ("CN", "Certificado", "Nacido Vivo", "Certificado Nacido Vivo"),
+  "CE": ("CE", "Cedula de Extranjeria", "Extranjera"),
+  "ME": ("ME", "Menor sin Identificacion"),
+  "PS": ("PS", "Pasaporte"),
+  "PT": ("PT", "Permiso por proteccion Temporal"),
+  "RC": ("RC", "Registro Civil", "Registro", "TC"),
+  "SC": ("SC", "Salvo Conducto", "Salvoconducto"),
+  "PE": ("PE", "Permiso especial de permanencia", "Permiso"),
+}
 
-  Tipo: bloque de letras Unicode al inicio. Número: todos los dígitos a partir de
-  la primera cifra (miles: solo dígitos en salida). Celdas solo numéricas o sin
-  letras iniciales devuelven (None, None). Solo letras, sin cifras: (tipo, None).
+
+def _ascii_upper(value: Any) -> str:
+  text = unicodedata.normalize("NFKD", str(value or ""))
+  text = "".join(c for c in text if not unicodedata.combining(c))
+  return text.upper()
+
+
+def _normalize_doc_type_key(value: Any) -> str:
+  """Clave comparable: sin acentos, espacios, puntos ni símbolos."""
+  return re.sub(r"[^A-Z0-9]", "", _ascii_upper(value))
+
+
+def _clean_doc_type_text(value: Any) -> Optional[str]:
+  cleaned = re.sub(r"[^A-Z0-9]+", " ", _ascii_upper(value)).strip()
+  cleaned = re.sub(r"\s+", " ", cleaned)
+  return cleaned or None
+
+
+_DOC_TYPE_ALIAS_BY_KEY: dict[str, str] = {
+  _normalize_doc_type_key(alias): canonical
+  for canonical, aliases in _DOC_TYPE_ALIASES.items()
+  for alias in aliases
+}
+
+_DOC_TYPE_ALIAS_DISPLAY_BY_KEY: dict[str, str] = {}
+for _canonical, _aliases in _DOC_TYPE_ALIASES.items():
+  for _alias in _aliases:
+    _alias_key = _normalize_doc_type_key(_alias)
+    _alias_display = _clean_doc_type_text(_alias)
+    if _alias_key and _alias_display:
+      _DOC_TYPE_ALIAS_DISPLAY_BY_KEY.setdefault(_alias_key, _alias_display)
+
+_DOC_TYPE_ALIAS_KEYS_SORTED: list[str] = sorted(
+  _DOC_TYPE_ALIAS_BY_KEY,
+  key=lambda k: (len(k), k),
+  reverse=True,
+)
+
+
+def homologate_doc_type(value: Any) -> Optional[str]:
+  """Homologa un tipo documental a sigla; si no existe, conserva texto limpio."""
+  if _is_nullish_scalar(value):
+    return None
+  key = _normalize_doc_type_key(value)
+  if not key or key.lower() in ("nan", "none", "nat"):
+    return None
+  return _DOC_TYPE_ALIAS_BY_KEY.get(key) or _clean_doc_type_text(value)
+
+
+def _compact_doc_text(value: Any) -> str:
+  return _normalize_doc_type_key(value)
+
+
+def _looks_like_short_alias_prefix(compact: str, alias_key: str) -> bool:
+  """Evita que alias de 1-2 chars capturen palabras como CODIGO o TOTAL."""
+  if not compact.startswith(alias_key):
+    return False
+  rest = compact[len(alias_key):]
+  if not rest:
+    return True
+  return bool(rest[0].isdigit())
+
+
+def _match_known_doc_type_prefix(compact: str) -> Optional[tuple[str, str]]:
+  for alias_key in _DOC_TYPE_ALIAS_KEYS_SORTED:
+    if len(alias_key) <= 2:
+      if not _looks_like_short_alias_prefix(compact, alias_key):
+        continue
+    elif not compact.startswith(alias_key):
+      continue
+    return _DOC_TYPE_ALIAS_BY_KEY[alias_key], alias_key
+  return None
+
+
+def split_doc_tipo_y_numero_texto(
+  raw: Any,
+  *,
+  homologate: bool = True,
+) -> tuple[Optional[str], Optional[str]]:
+  """Separa tipo documental y número, tolerando nombres largos y símbolos.
+
+  Si `homologate` es True, tipos conocidos se devuelven como sigla (CC, TI,
+  etc.). Los tipos desconocidos se conservan limpios en mayúsculas.
   """
   if _is_nullish_scalar(raw) and not (isinstance(raw, str) and not str(raw).strip()):
     return (None, None)
@@ -1180,23 +1512,33 @@ def split_doc_tipo_y_numero_texto(raw: Any) -> tuple[Optional[str], Optional[str
     return (None, None)
   if t and t[0].isdigit():
     return (None, None)
+
+  compact = _compact_doc_text(t)
+  if not compact:
+    return (None, None)
+
+  match = _match_known_doc_type_prefix(compact)
+  if match is not None:
+    canonical, alias_key = match
+    rest = compact[len(alias_key):]
+    nums = re.sub(r"[^\d]", "", rest)
+    if homologate:
+      return (canonical, nums if nums else None)
+    raw_type = _DOC_TYPE_ALIAS_DISPLAY_BY_KEY.get(alias_key) or canonical
+    return (raw_type, nums if nums else None)
+
   m = re.match(
-    r"^([A-Za-zÑñ\u00C0-\u017F\u00D1\u00F1\u0100-\u02CC]+)"  # letras (1+)
-    r"[\s\-.#:/_,\u00A0]*"  # separación opcional
-    r"([0-9].*)\s*$",  # primera cifra
+    r"^([^\d]+?)\s*([0-9].*)\s*$",
     t,
   )
   if m:
-    to_t = "".join(c for c in m.group(1) if c.isalpha())
-    to_t = to_t.upper() if to_t else None
+    to_t = _clean_doc_type_text(m.group(1))
     nums = re.sub(r"[^\d]", "", m.group(2))
-    return (to_t, nums if nums else None)
-  m2 = re.match(
-    r"^([A-Za-zÑñ\u00C0-\u017F\u00D1\u00F1\u0100-\u02CC]+)\s*$", t,
-  )
-  if m2 and not re.search(r"\d", t):
-    to_t = "".join(c for c in m2.group(1) if c.isalpha())
-    return (to_t.upper() if to_t else None, None)
+    return (homologate_doc_type(to_t) if homologate else to_t, nums if nums else None)
+
+  if not re.search(r"\d", t):
+    to_t = homologate_doc_type(t) if homologate else _clean_doc_type_text(t)
+    return (to_t, None)
   return (None, None)
 
 
@@ -1218,13 +1560,13 @@ def _rule_passthrough_series(s: pd.Series) -> pd.Series:
 
 
 def _apply_doc_id_split_columns(
-  result: pd.DataFrame, col: str, log: list[str]
+  result: pd.DataFrame, col: str, log: list[str], *, homologate_type: bool = True
 ) -> None:
   tcol, ncol = _alloc_doc_split_column_names(str(col), set(result.columns))
   st_tipo: list[Optional[str]] = []
   st_num: list[Any] = []
   for v in result[col].tolist():
-    tt, nn = split_doc_tipo_y_numero_texto(v)
+    tt, nn = split_doc_tipo_y_numero_texto(v, homologate=homologate_type)
     st_tipo.append(tt)
     st_num.append(nn if nn is not None else pd.NA)
   result[tcol] = st_tipo
@@ -1239,13 +1581,21 @@ CLEANING_RULES: list[dict[str, Any]] = [
   {
     "id": "strip_special_spaces",
     "name": "Eliminar caracteres especiales y espacios",
-    "description": "Deja solo caracteres alfanuméricos (A-Z, a-z, 0-9).",
+    "description": (
+      "Deja solo caracteres alfanuméricos (A-Z, a-z, 0-9). No usar antes de "
+      "«Normalizar fecha» en columnas con fecha y hora: pega cifras y puede impedir "
+      "interpretar el valor (aplique primero la normalización de fecha o use la regla "
+      "que conserva separadores)."
+    ),
     "fn": _rule_strip_special_spaces,
   },
   {
     "id": "strip_special_keep_date_seps",
-    "name": "Quitar símbolos; conserva /, - y espacios (fechas o texto con fechas)",
-    "description": "Quita comas, $, comillas, etc. y deja letras, números, /, - y espacios. Adecuado p. ej. en columna 'Fecha Servicio'.",
+    "name": "Quitar símbolos; conserva /, -, : y espacios (fechas o texto con fechas)",
+    "description": (
+      "Quita comas, $, comillas, etc. y deja letras, números, /, -, : (hora) y espacios. "
+      "Adecuado p. ej. en columna 'Fecha Servicio'."
+    ),
     "fn": _rule_strip_special_keep_date_seps,
   },
   {
@@ -1260,6 +1610,15 @@ CLEANING_RULES: list[dict[str, Any]] = [
     "description": "Convierte a enteros con detección de miles/decimal (`,`/`.`) y moneda; redondea decimales.",
     "fn": _rule_to_integer,
   },
+  {
+    "id": "date_to_ddmmyyyy",
+    "name": "Normalizar fecha → DD/MM/AAAA",
+    "description": (
+      "Normaliza fechas numéricas, con separadores, con ruido o con meses en "
+      "texto (ej. 26-abril-2026) al formato DD/MM/AAAA."
+    ),
+    "fn": _rule_date_to_ddmmyyyy,
+  },
   {
     "id": "date_numeric_to_ddmmyyyy",
     "name": "Conversión de formato numérico (Fechas) → DD/MM/AAAA",
@@ -1293,6 +1652,15 @@ CLEANING_RULES: list[dict[str, Any]] = [
     ),
     "fn": _rule_passthrough_series,
   },
+  {
+    "id": "doc_type_homologate",
+    "name": "Homologar tipo de documento",
+    "description": (
+      "Convierte tipos de documento escritos como nombre largo, sigla o texto con "
+      "símbolos a su abreviatura estándar (CC, TI, CE, RC, etc.)."
+    ),
+    "fn": _rule_doc_type_homologate,
+  },
 ]
 
 _RULE_BY_ID: dict[str, Callable[[pd.Series], pd.Series]] = {
@@ -1333,14 +1701,21 @@ def apply_rules(
     if col not in result.columns:
       log.append(f"[WARN] Columna no encontrada: {col}")
       continue
+    split_has_homologation = (
+      "doc_id_split_tipo_numero" in rule_ids and "doc_type_homologate" in rule_ids
+    )
     for rule_id in rule_ids:
       if rule_id == "doc_id_split_tipo_numero":
         try:
-          _apply_doc_id_split_columns(result, col, log)
+          _apply_doc_id_split_columns(
+            result, col, log, homologate_type=split_has_homologation
+          )
         except Exception as exc:  # noqa: BLE001
           logger.exception("Error aplicando regla %s a %s", rule_id, col)
           log.append(f"[ERROR] '{col}' regla '{rule_id}': {exc}")
         continue
+      if rule_id == "doc_type_homologate" and split_has_homologation:
+        continue
       fn = _RULE_BY_ID.get(rule_id)
       if fn is None:
         log.append(f"[WARN] Regla desconocida '{rule_id}' en columna '{col}'")

application/csv_manager.py

@@ -2,25 +2,51 @@
 
 Maneja la escritura incremental segura (con lock) en archivos CSV.
 """
-import os
 import csv
+import os
+import re
 import threading
-from typing import List, Dict, Any, Optional
+from typing import Any, Dict, List
+
+_LINE_BREAK_RE = re.compile(r"[\r\n\v\f\x85\u2028\u2029]+")
+
+
+def sanitize_csv_cell_value(value: Any) -> Any:
+  """Reemplaza saltos de línea en valores de celda por un espacio (una fila física por registro).
+
+  Conserva ``None`` y escalares numéricos/booleanos sin tocar. Cadenas y demás
+  tipos se normalizan solo si la representación en texto contiene saltos.
+  """
+  if value is None:
+    return value
+  if isinstance(value, bool):
+    return value
+  if isinstance(value, (int, float)):
+    return value
+  if isinstance(value, str):
+    if not _LINE_BREAK_RE.search(value):
+      return value
+    return _LINE_BREAK_RE.sub(" ", value).strip()
+  s = str(value)
+  if not _LINE_BREAK_RE.search(s):
+    return value
+  return _LINE_BREAK_RE.sub(" ", s).strip()
+
 
 class CSVManager:
-    def __init__(self):
-        self._lock = threading.Lock()
+  def __init__(self):
+    self._lock = threading.Lock()
 
-    def append_rows(self, csv_path: str, rows: List[Dict[str, Any]], header: List[str]) -> None:
-        """Anexa filas a un CSV, creando encabezado si el archivo no existe."""
-        os.makedirs(os.path.dirname(csv_path) or ".", exist_ok=True)
-        file_exists = os.path.exists(csv_path) and os.path.getsize(csv_path) > 0
-        
-        with self._lock:
-            with open(csv_path, "a", encoding="utf-8", newline="") as f:
-                writer = csv.DictWriter(f, fieldnames=header, extrasaction="ignore", delimiter='|')
-                if not file_exists:
-                    writer.writeheader()
-                for r in rows:
-                    writer.writerow(r)
+  def append_rows(self, csv_path: str, rows: List[Dict[str, Any]], header: List[str]) -> None:
+    """Anexa filas a un CSV, creando encabezado si el archivo no existe."""
+    os.makedirs(os.path.dirname(csv_path) or ".", exist_ok=True)
+    file_exists = os.path.exists(csv_path) and os.path.getsize(csv_path) > 0
 
+    with self._lock:
+      with open(csv_path, "a", encoding="utf-8", newline="") as f:
+        writer = csv.DictWriter(f, fieldnames=header, extrasaction="ignore", delimiter="|")
+        if not file_exists:
+          writer.writeheader()
+        for r in rows:
+          sanitized = {k: sanitize_csv_cell_value(v) for k, v in r.items()}
+          writer.writerow(sanitized)

gui/controller.py

@@ -1201,35 +1201,71 @@ class AppController:
     if not filepath:
       return
 
-    try:
-      df, delimiter = csv_cleaning.load_csv(filepath)
-    except Exception as e:
-      logger.exception("Error leyendo CSV para limpieza")
-      mb.showerror("Error", f"No se pudo leer el CSV:\n{e}")
-      return
-
-    self._cleaning_df = df
     self._cleaning_result_df = None
     self.view.set_cleaning_csv_path(filepath)
+    self.view.enable_cleaning_controls(False)
     self.view.enable_cleaning_save(False)
     self.view.set_cleaning_result(None)
+    self.view.set_cleaning_preview(None)
+    self.view.set_cleaning_status("Cargando CSV...")
+    try:
+      self.view.set_cleaning_loading(True, "Leyendo archivo CSV...")
+    except Exception:
+      pass
+
+    def worker():
+      try:
+        df, delimiter = csv_cleaning.load_csv(filepath)
+        self.view.after(
+          0,
+          lambda d=df, delim=delimiter: self._on_cleaning_csv_loaded(
+            d, delim
+          ),
+        )
+        summaries = csv_cleaning.analyze_dataframe(df)
+        self._cleaning_df = df
+        self.view.after(
+          0,
+          lambda d=df, s=summaries: self._finish_cleaning_csv_load(d, s),
+        )
+      except Exception as e:
+        logger.exception("Error leyendo o analizando CSV para limpieza")
+        err = str(e)
+        self.view.after(0, lambda m=err: self._fail_cleaning_csv_load(m))
+
+    threading.Thread(target=worker, daemon=True).start()
+
+  def _on_cleaning_csv_loaded(self, df: pd.DataFrame, delimiter: str):
     self.view.set_cleaning_status(
-      f"CSV cargado: {len(df):,} filas x {len(df.columns)} columnas."
+      f"CSV leído: {len(df):,} filas x {len(df.columns)} columnas. Analizando columnas..."
     )
     try:
+      self.view.set_cleaning_loading(True, "Analizando columnas...")
       self.view.set_cleaning_delimiter_chip(delimiter)
     except Exception:
       pass
 
-    try:
-      summaries = csv_cleaning.analyze_dataframe(df)
-    except Exception as e:
-      logger.exception("Error analizando CSV para limpieza")
-      mb.showerror("Error", f"No se pudo analizar el CSV:\n{e}")
-      return
-
+  def _finish_cleaning_csv_load(self, df: pd.DataFrame, summaries):
     self.view.populate_cleaning_columns(summaries)
     self.view.set_cleaning_preview(df)
+    self.view.set_cleaning_status(
+      f"CSV cargado: {len(df):,} filas x {len(df.columns)} columnas."
+    )
+    try:
+      self.view.set_cleaning_loading(False)
+    except Exception:
+      pass
+    self.view.enable_cleaning_controls(True)
+
+  def _fail_cleaning_csv_load(self, message: str):
+    self._cleaning_df = None
+    self.view.set_cleaning_status("Error al cargar CSV")
+    try:
+      self.view.set_cleaning_loading(False)
+    except Exception:
+      pass
+    self.view.enable_cleaning_controls(True)
+    mb.showerror("Error", f"No se pudo cargar o analizar el CSV:\n{message}")
 
   def on_cleaning_apply(self):
     """Aplica las reglas seleccionadas en un hilo y actualiza el resultado."""

gui/tabs/cleaning_tab.py

@@ -56,15 +56,17 @@ _TYPE_MENU_VALUES: list[str] = [
 ]
 
 _RULE_SHORT_LABELS: dict[str, str] = {
-  "strip_special_spaces": "Sin especiales/espacios",
-  "strip_special_keep_date_seps": "Símbolos: conserva /, -, espacios",
+  "strip_special_spaces": "Sin especiales/espacios (no antes de fecha+hora)",
+  "strip_special_keep_date_seps": "Símbolos: conserva /, -, :, espacios",
   "strip_spaces": "Sin espacios",
   "to_integer": "A entero",
+  "date_to_ddmmyyyy": "Normalizar fecha → DD/MM/AAAA",
   "date_numeric_to_ddmmyyyy": "Fecha numérica → DD/MM/AAAA",
   "date_alpha_to_ddmmyyyy": "Fecha alfanumérica → DD/MM/AAAA",
   "date_messy_text_to_ddmmyyyy": "Fecha con ruido (texto) → DD/MM/AAAA",
   "round_decimals_custom": "Aprox. decimales (>0.5 sube)",
   "doc_id_split_tipo_numero": "Dividir doc (tipo + cifras)",
+  "doc_type_homologate": "Homologar tipo doc",
 }
 
 
@@ -109,6 +111,7 @@ class CleaningTab:
     self._search_var: Optional[ctk.StringVar] = None
     # Tipo de dato forzado por el usuario (id interno) por columna; vacío = inferido
     self._type_overrides: dict[str, str] = {}
+    self._syncing_rule_vars = False
 
     self._setup_layout()
     self._create_file_section()
@@ -163,6 +166,27 @@ class CleaningTab:
     self.chip_delim = self._make_chip(chips_frame, "Delim: —")
     self.chip_delim.pack(side="left")
 
+    self.loading_frame = ctk.CTkFrame(frame, fg_color="transparent")
+    self.loading_frame.grid(
+      row=2, column=0, columnspan=4, sticky="ew", padx=15, pady=(0, 12)
+    )
+    self.loading_frame.grid_columnconfigure(1, weight=1)
+    self.loading_label = ctk.CTkLabel(
+      self.loading_frame,
+      text="",
+      font=self.fonts["italic_small"],
+      text_color=self.colors["gray_light"],
+      anchor="w",
+    )
+    self.loading_label.grid(row=0, column=0, sticky="w", padx=(0, 10))
+    self.loading_progress = ctk.CTkProgressBar(
+      self.loading_frame,
+      mode="indeterminate",
+      height=8,
+    )
+    self.loading_progress.grid(row=0, column=1, sticky="ew")
+    self.loading_frame.grid_remove()
+
   def _make_chip(self, parent, text: str) -> ctk.CTkLabel:
     return ctk.CTkLabel(
       parent, text=text, font=self.fonts["italic_small"],
@@ -429,6 +453,23 @@ class CleaningTab:
   def set_status(self, text: str):
     self.lbl_status.configure(text=text)
 
+  def set_loading(self, loading: bool, text: str = ""):
+    """Muestra u oculta una barra indeterminada para operaciones de CSV."""
+    if loading:
+      self.loading_label.configure(text=text or "Procesando CSV...")
+      self.loading_frame.grid()
+      try:
+        self.loading_progress.start()
+      except Exception:
+        pass
+    else:
+      try:
+        self.loading_progress.stop()
+      except Exception:
+        pass
+      self.loading_label.configure(text="")
+      self.loading_frame.grid_remove()
+
   def set_preview(self, df: Optional[pd.DataFrame]):
     """Renderiza el DataFrame original en la tabla de vista previa."""
     if df is None or len(df) == 0:
@@ -643,7 +684,10 @@ class CleaningTab:
       if rule is None:
         continue
       var = ctk.BooleanVar(value=False)
-      var.trace_add("write", lambda *_a, c=summary.name: self._on_rule_toggle(c))
+      var.trace_add(
+        "write",
+        lambda *_a, c=summary.name, r=rid: self._on_rule_toggle(c, r),
+      )
       label = _RULE_SHORT_LABELS.get(rid, rule["name"])
       chk = ctk.CTkCheckBox(
         rules_frame, text=label, variable=var,
@@ -850,7 +894,10 @@ class CleaningTab:
       if rule is None:
         continue
       var = ctk.BooleanVar(value=False)
-      var.trace_add("write", lambda *_a, c=col_name: self._on_rule_toggle(c))
+      var.trace_add(
+        "write",
+        lambda *_a, c=col_name, r=rid: self._on_rule_toggle(c, r),
+      )
       label = _RULE_SHORT_LABELS.get(rid, rule["name"])
       chk = ctk.CTkCheckBox(
         ui.rules_frame, text=label, variable=var,
@@ -883,10 +930,30 @@ class CleaningTab:
       if ui.expanded != expanded:
         self._toggle_column(col_name)
 
-  def _on_rule_toggle(self, col_name: str):
+  def _on_rule_toggle(self, col_name: str, changed_rule_id: Optional[str] = None):
     ui = self._column_uis.get(col_name)
     if ui is None:
       return
+    if (
+      changed_rule_id == "doc_id_split_tipo_numero"
+      and not self._syncing_rule_vars
+    ):
+      split_selected = any(
+        rid == "doc_id_split_tipo_numero" and bool(var.get())
+        for rid, _chk, var in ui.rule_vars
+      )
+      if split_selected:
+        for rid, _chk, var in ui.rule_vars:
+          if rid == "doc_type_homologate" and not bool(var.get()):
+            self._syncing_rule_vars = True
+            try:
+              var.set(True)
+            finally:
+              self._syncing_rule_vars = False
+            self.set_status(
+              f"Columna «{col_name}»: se activó homologación de tipo de documento."
+            )
+            break
     selected = sum(1 for _rid, _chk, var in ui.rule_vars if bool(var.get()))
     text = f"{selected} regla{'s' if selected != 1 else ''}"
     color = self.colors["success"] if selected else self.colors["gray_light"]

gui/view.py

@@ -399,5 +399,8 @@ class InvoiceAppView(ctk.CTk):
   def set_cleaning_status(self, text: str):
     self.cleaning_tab.set_status(text)
 
+  def set_cleaning_loading(self, loading: bool, text: str = ""):
+    self.cleaning_tab.set_loading(loading, text)
+
   def set_cleaning_delimiter_chip(self, delimiter: str):
     self.cleaning_tab.set_delimiter_chip(delimiter)

tests/test_csv_cleaning.py

@@ -0,0 +1,233 @@
+from __future__ import annotations
+
+import pandas as pd
+
+from application.csv_cleaning import (
+  apply_rules,
+  homologate_doc_type,
+  split_doc_tipo_y_numero_texto,
+)
+
+
+def test_homologate_doc_type_aliases() -> None:
+  cases = {
+    "Cedula": "CC",
+    "C.C": "CC",
+    "C C": "CC",
+    "Registro Civil": "RC",
+    "Salvoconducto": "SC",
+    "TC": "RC",
+    "C\u00e9dula de Extranjeria": "CE",
+  }
+
+  for raw, expected in cases.items():
+    assert homologate_doc_type(raw) == expected
+
+
+def test_split_doc_type_with_long_names() -> None:
+  cases = {
+    "Cedula 123.456": ("CC", "123456"),
+    "Registro Civil: 001-23": ("RC", "00123"),
+    "Permiso especial de permanencia 999": ("PE", "999"),
+    "Permiso por proteccion Temporal / 555": ("PT", "555"),
+  }
+
+  for raw, expected in cases.items():
+    assert split_doc_tipo_y_numero_texto(raw) == expected
+
+
+def test_split_doc_type_with_abbreviations_and_noise() -> None:
+  cases = {
+    "C.C-123": ("CC", "123"),
+    "TL_987": ("TI", "987"),
+    "CE/1.234.567": ("CE", "1234567"),
+    "T1 # 44": ("TI", "44"),
+  }
+
+  for raw, expected in cases.items():
+    assert split_doc_tipo_y_numero_texto(raw) == expected
+
+
+def test_split_unknown_type_preserves_clean_uppercase() -> None:
+  assert split_doc_tipo_y_numero_texto("ABC Especial: 123-45") == (
+    "ABC ESPECIAL",
+    "12345",
+  )
+
+
+def test_split_without_homologation_preserves_clean_type() -> None:
+  assert split_doc_tipo_y_numero_texto(
+    "Cedula 123.456",
+    homologate=False,
+  ) == ("CEDULA", "123456")
+
+
+def test_apply_rules_homologates_type_column_only() -> None:
+  df = pd.DataFrame(
+    {
+      "tipo": [
+        "Cedula",
+        "C.C",
+        "Registro Civil",
+        "Salvoconducto",
+        "TC",
+        "Extranjera",
+      ]
+    }
+  )
+
+  result, log = apply_rules(df, {"tipo": ["doc_type_homologate"]})
+
+  assert result["tipo"].tolist() == ["CC", "CC", "RC", "SC", "RC", "CE"]
+  assert any("Homologar tipo de documento" in line for line in log)
+
+
+def test_apply_rules_split_uses_homologation_when_selected() -> None:
+  df = pd.DataFrame(
+    {
+      "documento": [
+        "Cedula 123.456",
+        "Registro Civil: 001-23",
+        "ABC Especial: 123-45",
+      ]
+    }
+  )
+
+  result, _log = apply_rules(
+    df,
+    {"documento": ["doc_id_split_tipo_numero", "doc_type_homologate"]},
+  )
+
+  assert result["documento_doc_tipo"].tolist() == ["CC", "RC", "ABC ESPECIAL"]
+  assert result["documento_doc_numeros"].tolist() == ["123456", "00123", "12345"]
+  assert result["documento"].tolist() == df["documento"].tolist()
+
+
+def test_apply_rules_split_can_skip_homologation() -> None:
+  df = pd.DataFrame({"documento": ["Cedula 123.456"]})
+
+  result, _log = apply_rules(df, {"documento": ["doc_id_split_tipo_numero"]})
+
+  assert result["documento_doc_tipo"].tolist() == ["CEDULA"]
+  assert result["documento_doc_numeros"].tolist() == ["123456"]
+
+
+def test_apply_rules_unified_date_rule_numeric_and_separated() -> None:
+  df = pd.DataFrame(
+    {
+      "fecha": [
+        "20260426",
+        "26042026",
+        "260426",
+        "26/04/2026",
+        "04/26/2026",
+        "2026-04-26 10:30",
+      ]
+    }
+  )
+
+  result, log = apply_rules(df, {"fecha": ["date_to_ddmmyyyy"]})
+
+  assert result["fecha"].tolist() == [
+    "26/04/2026",
+    "26/04/2026",
+    "26/04/2026",
+    "26/04/2026",
+    "26/04/2026",
+    "26/04/2026",
+  ]
+  assert any("Normalizar fecha" in line for line in log)
+
+
+def test_apply_rules_unified_date_rule_text_months() -> None:
+  df = pd.DataFrame(
+    {
+      "fecha": [
+        "26-abril-2026",
+        "26 abril 2026",
+        "26/abr/2026",
+        "abril 26 2026",
+        "2026 abril 26",
+        "15-sept-25",
+        "07 setiembre 2024",
+      ]
+    }
+  )
+
+  result, _log = apply_rules(df, {"fecha": ["date_to_ddmmyyyy"]})
+
+  assert result["fecha"].tolist() == [
+    "26/04/2026",
+    "26/04/2026",
+    "26/04/2026",
+    "26/04/2026",
+    "26/04/2026",
+    "15/09/2025",
+    "07/09/2024",
+  ]
+
+
+def test_apply_rules_unified_date_spanish_abbr_month_with_time() -> None:
+  df = pd.DataFrame(
+    {
+      "fecha": [
+        "15/may./2024 13:44",
+        "22/may./2024 11:09",
+        "30/04/2024",
+      ]
+    }
+  )
+
+  result, _log = apply_rules(df, {"fecha": ["date_to_ddmmyyyy"]})
+
+  assert result["fecha"].tolist() == [
+    "15/05/2024",
+    "22/05/2024",
+    "30/04/2024",
+  ]
+
+
+def test_apply_rules_date_after_strip_special_spaces_glued_year_time() -> None:
+  df = pd.DataFrame({"fecha": ["15/may./2024 13:44"]})
+
+  result, _log = apply_rules(
+    df,
+    {"fecha": ["strip_special_spaces", "date_to_ddmmyyyy"]},
+  )
+
+  assert result["fecha"].tolist() == ["15/05/2024"]
+
+
+def test_apply_rules_keep_date_seps_preserves_time_then_normalizes() -> None:
+  df = pd.DataFrame({"fecha": ["30/04/2024 13:44"]})
+
+  result, _log = apply_rules(
+    df,
+    {"fecha": ["strip_special_keep_date_seps", "date_to_ddmmyyyy"]},
+  )
+
+  assert result["fecha"].tolist() == ["30/04/2024"]
+
+
+def test_apply_rules_unified_date_rule_noise_and_unparseable_values() -> None:
+  df = pd.DataFrame(
+    {
+      "fecha": [
+        "**26_abril_2026++",
+        "Fecha: 26-abril-2026",
+        "xx26/04/2026yy",
+        "no es fecha",
+        "31-febrero-2026",
+      ]
+    }
+  )
+
+  result, _log = apply_rules(df, {"fecha": ["date_to_ddmmyyyy"]})
+
+  assert result["fecha"].tolist() == [
+    "26/04/2026",
+    "26/04/2026",
+    "26/04/2026",
+    "no es fecha",
+    "31-febrero-2026",
+  ]

tests/test_csv_manager.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+import tempfile
+from pathlib import Path
+
+from application.csv_manager import CSVManager, sanitize_csv_cell_value
+
+
+def test_sanitize_collapses_newlines() -> None:
+  assert sanitize_csv_cell_value("a\nb") == "a b"
+  assert sanitize_csv_cell_value("a\r\nb") == "a b"
+  assert sanitize_csv_cell_value("  x\v\fy  ") == "x y"
+
+
+def test_sanitize_unicode_line_separators() -> None:
+  assert sanitize_csv_cell_value("a\u2028b\u2029c") == "a b c"
+  assert sanitize_csv_cell_value("a\x85b") == "a b"
+
+
+def test_sanitize_preserves_none_and_scalars() -> None:
+  assert sanitize_csv_cell_value(None) is None
+  assert sanitize_csv_cell_value(42) == 42
+  assert sanitize_csv_cell_value(3.14) == 3.14
+  assert sanitize_csv_cell_value(False) is False
+  assert sanitize_csv_cell_value(True) is True
+
+
+def test_sanitize_plain_string_unchanged_identity() -> None:
+  s = "sin saltos"
+  assert sanitize_csv_cell_value(s) is s
+
+
+def test_append_rows_one_physical_line_per_record() -> None:
+  with tempfile.TemporaryDirectory() as tmp:
+    path = Path(tmp) / "batch.csv"
+    mgr = CSVManager()
+    header = ["radicado", "descripcion"]
+    rows = [
+      {
+        "radicado": "235662283",
+        "descripcion": (
+          '"Las retenciones son a nivel informativo.\n'
+          "Paciente: 1010079549 - HERAZO\n"
+          'Contrato No. URGENCIAS."'
+        ),
+      },
+      {"radicado": "2", "descripcion": "ok"},
+    ]
+    mgr.append_rows(str(path), rows, header)
+    text = path.read_text(encoding="utf-8")
+    physical_lines = text.splitlines()
+    assert len(physical_lines) == 3