# -*- coding: utf-8 -*- from email.mime import base import shutil import os import re import csv import threading import traceback from pathlib import Path from datetime import datetime import tkinter as tk from tkinter import filedialog, messagebox, ttk import pandas as pd # ============================================================ # CONFIGURACIÓN GENERAL # ============================================================ COLUMNAS_FINAL_IT = [ "radicado", "total_cargos", "autorizacion", "ingreso", "tipo_documento", "documento", "tipo_servicio", "copago", "cuota_moderadora", ] CSV_FINAL_IT_HEADERS = [ "RADICADO", "BRUTO", "AUTORIZACION", "FECHA", "TIPO_ID", "NUMERO", "SERVICIO", "COPAGO", "MODERADORA", ] CSV_FINAL_IT_RENAME = { "radicado": "RADICADO", "total_cargos": "BRUTO", "autorizacion": "AUTORIZACION", "ingreso": "FECHA", "tipo_documento": "TIPO_ID", "documento": "NUMERO", "tipo_servicio": "SERVICIO", "copago": "COPAGO", "cuota_moderadora": "MODERADORA", } # Tabla de cuotas moderadoras por año. TABLA_MODERADORA = { 2023: {4100, 16400, 43000}, 2024: {4500, 18200, 47700}, 2025: {4700, 19200, 50300}, 2026: {5000, 20100, 52800}, } # Longitudes según tabla entregada por el usuario. # Los tipos marcados como alfanuméricos permiten letras y números. REGLAS_DOCUMENTO = { "CC": {"min": 3, "max": 10, "tipo": "num"}, "CE": {"min": 3, "max": 7, "tipo": "num"}, "TI": {"min": 10, "max": 10, "tipo": "num"}, "RC": {"min": 10, "max": 10, "tipo": "num"}, "CN": {"min": 9, "max": 14, "tipo": "num"}, "PT": {"min": 7, "max": 8, "tipo": "num"}, "PE": {"min": 15, "max": 15, "tipo": "num"}, "PA": {"min": 3, "max": 16, "tipo": "alnum"}, "PS": {"min": 3, "max": 16, "tipo": "alnum"}, "AS": {"min": 3, "max": 16, "tipo": "alnum"}, "MS": {"min": 3, "max": 16, "tipo": "alnum"}, } HOMOLOGACION_TIPO_DOC = { "CC": "CC", "C.C": "CC", "C.C.": "CC", "CEDULA": "CC", "CEDULA CIUDADANIA": "CC", "CÉDULA": "CC", "CÉDULA CIUDADANÍA": "CC", "CEDULA DE CIUDADANIA": "CC", "Cédula Ciudadanía": "CC", "Cédula_Ciudadanía": "CC", "TI": "TI", "T.I": "TI", "T.I.": "TI", "TARJETA IDENTIDAD": "TI", "TARJETA DE IDENTIDAD": "TI", "tarjeta identidad": "TI", "tarjeta_identidad": "TI", "RC": "RC", "R.C": "RC", "R.C.": "RC", "REGISTRO CIVIL": "RC", "REGISTRO_CIVIL": "RC", "registro civil": "RC", "registro_civil": "RC", "CE": "CE", "C.E": "CE", "C.E.": "CE", "CEDULA EXTRANJERIA": "CE", "CÉDULA EXTRANJERÍA": "CE", "CEDULA DE EXTRANJERIA": "CE", "Cédula Extranjería": "CE", "Cédula_Extranjería": "CE", "PA": "PA", "PASAPORTE": "PA", "PS": "PS", "PERMISO ESPECIAL": "PS", "PT": "PT", "PERMISO PROTECCION TEMPORAL": "PT", "PERMISO POR PROTECCION TEMPORAL": "PT", "PERMISO PROTECCION TEMPORAL": "PT", "Permiso Protección Temporal": "PT", "Permiso_Protección_Temporal": "PT", "PE": "PE", "PERMISO ESPECIAL": "PE", "PERMISO ESPECIAL DE PERMANENCIA": "PE", "Permiso Especial de Permanencia": "PE", "Permiso_Especial_de_Permanencia": "PE", "CN": "CN", "CERTIFICADO NACIDO VIVO": "CN", "NACIDO VIVO": "CN", "CERTIFICADO NACIDO": "CN", "Certificado Nacido Vivo": "CN", "Certificado_Nacido_Vivo": "CN", "AS": "AS", "ADULTO SIN IDENTIFICACION": "AS", "ADULTO_SIN_IDENTIFICACION": "AS", "MS": "MS", "MENOR SIN IDENTIFICACION": "MS", "MENOR_SIN_IDENTIFICACION": "MS", "MENOR SIN IDENTIFICACIÓN": "MS", "Menor Sin Identificación": "MS", "Menor_Sin_Identificación": "MS", } # Diccionario inicial basado en los ejemplos que compartiste. # Se puede ampliar fácilmente. DICCIONARIO_COLUMNAS = { "radicado": [ "radicado", "radicacion", "radicación", "numero radicado", "numero_radicado", "numero radicacion", "numero_radicacion", "numeroradicacion", "numero de radicacion", "no radicado", "nro radicado", "num radicado", "id radicado" ], "autorizacion": [ "autorizacion", "autorización", "autorrizacion", "autoriz", "autoriza", "numero autorizacion", "num autorizacion", "no autorizacion", "nro autorizacion", "aut", "numaut", "no_aut", "autoritiponumero" ], "ingreso": [ "ingreso", "fecha ingreso", "fec ingreso", "fec.ingreso", "fecha_ingreso", "fecha de ingreso", "fecha servicio", "fec servicio", "fecha_servicio", "fecha", "ingeso" ], "tipo_documento": [ "tipo_documento", "tipo documento", "tipo de documento", "tipo doc", "tipodoc", "tipo", "td", "doc tipo", "documenoto tipo" ], "documento": [ "documento", "docuemnto", "numero documento", "número documento", "no documento", "no. documento", "num documento", "identificacion", "identificación", "cedula", "cédula", "id paciente", "paciente id", "id" ], "tipo_documento_documento": [ "tipo numero", "tiponumero", "tiponum", "tipnumero", "tipnum", "tipnumebr", "tipnumingraut", "tipo / documento", "tipo/documento", "tipo muner", "documento tipo", "autoritiponumero", "tipo y documento", "tipo-documento", "tipo doc documento", "tipo identificacion", "tipo identificación", "tipo id", "identificacion paciente", "identificación paciente", "identipo", "tipo num doc", "tipo nro doc", "tipo no doc" ], "total_cargos": [ "total_cargos", "total cargos", "bruto", "brutos", "brruto", "valor bruto", "valorbruto", "valor factura", "valor_factura", "total factura", "totalfactura", "valor facturado", "valor_facturado" ], "neto": [ "neto", "valor neto", "valorneto", "netos", "valor pagado", "valor_pagado" ], "copago": [ "copago", "copagos", "cop", "valor copago", "valor_copago" ], "cuota_moderadora": [ "moderadora", "cuota moderadora", "cuota_moderadora", "cm", "valor moderadora", "valor_moderadora", "moderradora" ], "cop_mod": [ "cop/mod", "copago/moderadora", "copago cm", "copago/cm", "copago cuota moderadora", "copago/CM", "copago cm", "valor usuario", "valor_usuario", "abono paciente", "cuota usuario", "pago usuario", "pagousuario", "copago/CM" ], "tipo_servicio": [ "tipo_servicio", "tipo servicio", "servicio", "tipo de servicio", "origen_servicio", "origen servicio", "ambito", "ámbito" ], } # ============================================================ # UTILIDADES DE LECTURA # ============================================================ def normalizar_nombre_columna(nombre): if nombre is None: return "" s = str(nombre).strip().lower() s = s.replace("\ufeff", "") s = s.replace("_", " ") s = re.sub(r"\s+", " ", s) # quitar tildes básicas sin depender de librerías externas trans = str.maketrans("áéíóúüñ", "aeiouun") s = s.translate(trans) return s def detectar_delimitador(path): try: with open(path, "r", encoding="utf-8-sig", errors="ignore") as f: muestra = f.read(4096) delimitadores = ["|", ";", ",", "\t"] conteos = {d: muestra.count(d) for d in delimitadores} mejor = max(conteos, key=conteos.get) return mejor if conteos[mejor] > 0 else "," except Exception: return "," def leer_csv_flexible(path): sep = detectar_delimitador(path) intentos = [ {"sep": sep, "encoding": "utf-8-sig"}, {"sep": sep, "encoding": "latin1"}, {"sep": ";", "encoding": "utf-8-sig"}, {"sep": ",", "encoding": "utf-8-sig"}, {"sep": "|", "encoding": "utf-8-sig"}, ] ultimo_error = None for opt in intentos: try: df = pd.read_csv( path, sep=opt["sep"], encoding=opt["encoding"], dtype=str, engine="python", keep_default_na=False, on_bad_lines="skip", ) df.columns = [str(c).strip() for c in df.columns] return df except Exception as e: ultimo_error = e raise RuntimeError(f"No se pudo leer CSV {path.name}: {ultimo_error}") def encontrar_columna(df, campo): cols_norm = {normalizar_nombre_columna(c): c for c in df.columns} equivalencias = DICCIONARIO_COLUMNAS.get(campo, []) for eq in equivalencias: eq_norm = normalizar_nombre_columna(eq) if eq_norm in cols_norm: return cols_norm[eq_norm] # búsqueda por contiene, útil para encabezados largos for eq in equivalencias: eq_norm = normalizar_nombre_columna(eq) for c_norm, c_original in cols_norm.items(): if eq_norm and (eq_norm in c_norm or c_norm in eq_norm): return c_original return None # ============================================================ # LIMPIEZA Y NORMALIZACIÓN # ============================================================ def limpiar_texto(valor): if valor is None: return "" s = str(valor).strip() if s.lower() in {"nan", "none", "null"}: return "" return s def limpiar_numero_entero(valor, default=""): """ Limpia valores monetarios sin inflar decimales. Ejemplos: - $ 150.000,00 -> 150000 - 150000,00 -> 150000 - 150000.00 -> 150000 - 1.250.000,00 -> 1250000 - 1,250,000.00 -> 1250000 - 150.000 -> 150000 - 150,000 -> 150000 """ s = limpiar_texto(valor) if not s: return default s = s.replace("$", "").replace("COP", "").replace("cop", "").strip() s = re.sub(r"[^\d,.\-]", "", s) if not s: return default # Para este proceso esperamos valores positivos. s = s.replace("-", "") # Caso con coma y punto: el último separador define el decimal. if "," in s and "." in s: last_comma = s.rfind(",") last_dot = s.rfind(".") if last_comma > last_dot: # Formato colombiano: 1.250.000,00 entero = s[:last_comma] decimal = s[last_comma + 1:] if len(decimal) <= 2: s = entero.replace(".", "").replace(",", "") else: s = s.replace(".", "").replace(",", "") else: # Formato inglés: 1,250,000.00 entero = s[:last_dot] decimal = s[last_dot + 1:] if len(decimal) <= 2: s = entero.replace(",", "").replace(".", "") else: s = s.replace(",", "").replace(".", "") elif "," in s: partes = s.split(",") ultima = partes[-1] # 150000,00 -> decimal. Se elimina la parte decimal. if len(ultima) in (1, 2): s = "".join(partes[:-1]) else: # 150,000 -> miles s = "".join(partes) elif "." in s: partes = s.split(".") ultima = partes[-1] # 150000.00 -> decimal. # 150.000 -> miles porque la última parte tiene 3 dígitos. if len(ultima) in (1, 2): s = "".join(partes[:-1]) else: s = "".join(partes) s = re.sub(r"\D", "", s) return s if s else default def to_int(valor, default=0): s = limpiar_numero_entero(valor, default="") if not s: return default try: return int(s) except Exception: return default def homologar_tipo_documento(valor): """ Homologa tipo de documento aunque venga con texto adicional. Ejemplos: - C.C. -> CC - CÉDULA CIUDADANIA -> CC - CEDULA CIUDAD 1021676637 -> CC - CC 1012333072 -> CC - RC 1188975098 -> RC """ s = limpiar_texto(valor).upper() if not s: return "" trans = str.maketrans("ÁÉÍÓÚÜÑ", "AEIOUUN") s = s.translate(trans) s = s.replace(".", "").replace("-", " ").replace("_", " ") s = re.sub(r"\s+", " ", s).strip() # Casos textuales de cédula if "CEDULA" in s and ("CIUDAD" in s or "CIUDADANIA" in s): return "CC" # Si empieza por tipo válido y luego trae documento/nombre m = re.match(r"^(CC|TI|RC|CE|PA|PS|PT|PE|CN|AS|MS)\b", s) if m: return m.group(1) return HOMOLOGACION_TIPO_DOC.get(s, s) def limpiar_documento(valor, tipo_doc): s = limpiar_texto(valor).upper() if tipo_doc in {"PA", "PS", "AS", "MS"}: s = re.sub(r"[^A-Z0-9]", "", s) else: s = re.sub(r"\D", "", s) return s def validar_documento(tipo_doc, documento): tipo_doc = homologar_tipo_documento(tipo_doc) documento = limpiar_documento(documento, tipo_doc) # Si el tipo está vacío, pero el documento numérico es usable, pasa porque el bot procesa el número. if not tipo_doc: if documento and documento.isdigit() and 3 <= len(documento) <= 16: return True, "" return False, "Tipo de documento vacío y documento inválido" if tipo_doc not in REGLAS_DOCUMENTO: return False, f"Tipo de documento no parametrizado: {tipo_doc}" regla = REGLAS_DOCUMENTO[tipo_doc] largo = len(documento) if largo < regla["min"] or largo > regla["max"]: if regla["min"] == regla["max"]: return False, f"Longitud inválida para {tipo_doc}. Debe tener {regla['min']} dígitos." return False, f"Longitud inválida para {tipo_doc}. Debe tener entre {regla['min']} y {regla['max']} dígitos." if regla["tipo"] == "num" and not documento.isdigit(): return False, f"Documento inválido para {tipo_doc}. Debe ser numérico." if regla["tipo"] == "alnum" and not re.match(r"^[A-Z0-9]+$", documento): return False, f"Documento inválido para {tipo_doc}. Debe ser alfanumérico." return True, "" def normalizar_fecha(valor): """ Normaliza ingreso a DD/MM/YYYY. Soporta: - 2026-05-13 00:00 - 13/05/2026 - 13-05-2026 - 29 abr. 2026 06:52 p. m. - ATENCION 07/05/2026. 8/05/2026 -> toma la primera fecha válida """ s = limpiar_texto(valor) if not s: return "", "fecha ingreso vacia" s_original = s s = s.strip() s_upper = s.upper() if any(marker in s_upper for marker in ["SIN_FECHA", "SIN FECHA", "NO FECHA", "SINFECHA"]): return "", "sin fecha" trans = str.maketrans("ÁÉÍÓÚÜÑáéíóúüñ", "AEIOUUNaeiouun") s_norm = s.translate(trans) s_norm = re.sub(r"\s+", " ", s_norm).strip() s_norm = s_norm.upper() s_norm = re.sub(r"[^A-Z0-9/\-.:\s]", " ", s_norm) meses = { "ENE": "01", "ENERO": "01", "FEB": "02", "FEBRERO": "02", "MAR": "03", "MARZO": "03", "ABR": "04", "ABRIL": "04", "MAY": "05", "MAYO": "05", "JUN": "06", "JUNIO": "06", "JUL": "07", "JULIO": "07", "AGO": "08", "AGOSTO": "08", "SEP": "09", "SEPT": "09", "SEPTIEMBRE": "09", "OCT": "10", "OCTUBRE": "10", "NOV": "11", "NOVIEMBRE": "11", "DIC": "12", "DICIEMBRE": "12", } def reemplazar_mes(match): token = re.sub(r"[^A-Z]", "", match.group(0).upper()) if token in meses: return meses[token] if token.endswith("BR"): return meses["ABR"] return match.group(0) s_mes = re.sub(r"\b[\w\.]{3,12}\b", reemplazar_mes, s_norm) s_mes = re.sub(r"\bP\s*\.\s*M\b", "PM", s_mes) s_mes = re.sub(r"\bA\s*\.\s*M\b", "AM", s_mes) s_mes = re.sub(r"\bP\s*M\b", "PM", s_mes) s_mes = re.sub(r"\bA\s*M\b", "AM", s_mes) s_mes = re.sub(r"\s+", " ", s_mes).strip() # Buscar primero fechas numéricas dentro del texto y tomar la primera válida. fechas_en_texto = re.findall(r"\b\d{1,2}[/-]\d{1,2}[/-]\d{2,4}\b", s_mes) for f in fechas_en_texto: try: fecha = pd.to_datetime(f, dayfirst=True, errors="raise") if 1900 <= fecha.year <= 2100: return fecha.strftime("%d/%m/%Y"), "" except Exception: pass # Intentar parsear después de reemplazar meses en español. for dayfirst in [True, False]: try: fecha = pd.to_datetime(s_mes, dayfirst=dayfirst, errors="raise") if pd.isna(fecha): continue if 1900 <= fecha.year <= 2100: return fecha.strftime("%d/%m/%Y"), "" except Exception: pass # Serial de Excel aproximado if re.fullmatch(r"\d{5}", s_mes): try: fecha = pd.to_datetime(float(s_mes), unit="D", origin="1899-12-30") return fecha.strftime("%d/%m/%Y"), "" except Exception: pass # Formato compacto YYYYMMDD o DDMMYYYY solo_digitos = re.sub(r"\D", "", s_mes) if len(solo_digitos) == 8: candidatos = [ solo_digitos[:4] + "-" + solo_digitos[4:6] + "-" + solo_digitos[6:], solo_digitos[:2] + "/" + solo_digitos[2:4] + "/" + solo_digitos[4:], ] for c in candidatos: try: fecha = pd.to_datetime(c, dayfirst=("/" in c), errors="raise") if 1900 <= fecha.year <= 2100: return fecha.strftime("%d/%m/%Y"), "" except Exception: pass # ISO claro if re.match(r"^\d{4}-\d{1,2}-\d{1,2}(?:[ T]\d{1,2}:\d{2}(?::\d{2})?(?:[-+]\d{2}:?\d{2})?)?$", s_mes): try: fecha = pd.to_datetime(s_mes, dayfirst=False, errors="raise") if 1900 <= fecha.year <= 2100: return fecha.strftime("%d/%m/%Y"), "" except Exception: pass return "", f"Fecha ingreso inválida: {s_original}" def obtener_anio_fecha(fecha_ddmmyyyy): try: return int(fecha_ddmmyyyy[-4:]) except Exception: return None def normalizar_autorizacion(valor): """ Normaliza autorización. Reglas: - Si viene vacía o con marcador de ausencia -> observación. - Si viene con guion -> se conserva. - Si viene pegada sin guion -> se agrega guion después de los primeros 5 caracteres. Ejemplo: 128242624857057 -> 12824-2624857057 - Después de normalizar debe quedar con largo 16 incluyendo el guion. """ s_raw = limpiar_texto(valor) s = s_raw.upper() if not s: return "", "autorizacion vacia" if any(marker in s for marker in ["SIN_AUTORIZACION", "SIN AUTORIZACION", "NO AUTORIZACION", "NO AUTORIZACIÓN", "NO AUTORIZACION"]): return "", "sin autorizacion" s = s.replace("AUTORIZACION", "").replace("AUTORIZACIÓN", "") s = s.replace(":", " ").strip() s = re.sub(r"[^A-Z0-9\-]", "", s) if not s: return "", "autorizacion vacia" if s.isdigit() and len(s) == 10: return s, "autorizacion incompleta" # Si viene sin guion y con caracteres suficientes, agregar guion después del 5.º carácter. if "-" not in s and len(s) > 5: s = s[:5] + "-" + s[5:] if len(s) == 16: return s, "" return s, "autorizacion invalida" def separar_copago_moderadora(valor, anio): """ Si el valor coincide con tabla moderadora del año => cuota_moderadora. Si no coincide => copago. """ monto = to_int(valor, default=0) if monto == 0: return 0, 0 if anio in TABLA_MODERADORA and monto in TABLA_MODERADORA[anio]: return 0, monto return monto, 0 def normalizar_tipo_servicio(valor): s = limpiar_texto(valor).upper() s = re.sub(r"\s+", " ", s).strip() # Se respeta porque el usuario indicó que ya lo manejan bien. return s # ============================================================ # DETECCIÓN Y EXTRACCIÓN ESPECIAL # ============================================================ def es_archivo_especial(df): """ Heurística: - Alguna columna tiene textos largos. - Y contiene palabras clave típicas del archivo especial: AUTORIZACION, PACIENTE, INGRESO, ABONO. """ if df.empty: return False palabras = ["AUTORIZACION", "AUTORIZACIÓN", "PACIENTE", "INGRESO", "ABONO"] for col in df.columns: serie = df[col].astype(str).head(50) textos = " ".join(serie.tolist()).upper() largo_promedio = serie.map(lambda x: len(str(x))).mean() hits = sum(1 for p in palabras if p in textos) if largo_promedio >= 45 and hits >= 2: return True return False def extraer_tipo_doc_y_documento_desde_texto(texto): """ Extrae tipo y número de documento cuando vienen mezclados. Soporta: - CC 1065138068 - CC-1065138068 - C.C. 1065138068 - CÉDULA CIUDAD 1021676637 NOMBRE PACIENTE - CEDULA CIUDADANIA 1021676637 NOMBRE PACIENTE - PACIENTE: RC-1253158282-NOMBRE - RC 1.067.645.729 - TI 1.066.894.756 - solo número: 1023876746 => tipo vacío, documento 1023876746 """ t = limpiar_texto(texto).upper() if not t: return "", "" trans = str.maketrans("ÁÉÍÓÚÜÑ", "AEIOUUN") t_norm = t.translate(trans) t_norm = t_norm.replace("_", " ") t_norm = re.sub(r"\bC\s*\.?\s*C\.?\b", "CC", t_norm) t_norm = re.sub(r"\bT\s*\.?\s*I\.?\b", "TI", t_norm) t_norm = re.sub(r"\bR\s*\.?\s*C\.?\b", "RC", t_norm) t_norm = re.sub(r"\bC\s*\.?\s*E\.?\b", "CE", t_norm) def limpiar_doc_texto(valor): return re.sub(r"[^A-Z0-9]", "", valor.upper()) m = re.search(r"\bCEDULA(?:\s+DE)?\s+CIUDAD(?:ANIA)?\s+([0-9]{3,16})\b", t_norm) if m: return "CC", m.group(1) m = re.search(r"\bCEDULA\s+([0-9]{3,16})\b", t_norm) if m: return "CC", m.group(1) m = re.match(r"^([A-Z ]+?)\s+([0-9\.\s]+)$", t_norm) if m: tipo_candidato = homologar_tipo_documento(m.group(1)) if tipo_candidato in REGLAS_DOCUMENTO: return tipo_candidato, limpiar_doc_texto(m.group(2)) m = re.search(r"PACIENTE\s*:\s*(CC|TI|RC|CE|PA|PS|PT|PE|CN|AS|MS)\s*[-:/ ]+\s*([0-9\.\s]+)", t_norm) if m: return homologar_tipo_documento(m.group(1)), limpiar_doc_texto(m.group(2)) m = re.search( r"(?:TIPO\s*/?\s*DOCUMENTO|TIPO\s*/?\s*NUMERO|TIPO\s+NUMERO|DOCUMENTO)\s*[:\- ]+\s*" r"(CC|TI|RC|CE|PA|PS|PT|PE|CN|AS|MS)\s*[-:/ ]+\s*([0-9\.\s]+)", t_norm ) if m: return homologar_tipo_documento(m.group(1)), limpiar_doc_texto(m.group(2)) m = re.search(r"\b(CC|TI|RC|CE|PA|PS|PT|PE|CN|AS|MS)\s*[-:/ ]+\s*([0-9\.\s]+)\b", t_norm) if m: return homologar_tipo_documento(m.group(1)), limpiar_doc_texto(m.group(2)) m = re.search(r"\b([0-9\.\s]+)\s*[-:/ ]+\s*(CC|TI|RC|CE|PA|PS|PT|PE|CN|AS|MS)\b", t_norm) if m: return homologar_tipo_documento(m.group(2)), limpiar_doc_texto(m.group(1)) # Solo número: se permite tipo vacío si el documento es usable para el bot. # Evita extraer números desde textos largos con nombres/observaciones. if re.fullmatch(r"[0-9\s.,;:/\-]+", t_norm): cleaned = limpiar_doc_texto(t_norm) if 6 <= len(cleaned) <= 16: return "", cleaned numeros = re.findall(r"\d{5,20}", cleaned) if numeros: candidatos = [n for n in numeros if 6 <= len(n) <= 10] if candidatos: candidatos = sorted(candidatos, key=len) return "", candidatos[0] return "", numeros[-1] # Si el texto contiene solo tipo y número con puntos, limpiar y reintentar. m = re.search(r"\b(CC|TI|RC|CE|PA|PS|PT|PE|CN|AS|MS)\s+([0-9\.\s]+)\b", t_norm) if m: return homologar_tipo_documento(m.group(1)), limpiar_doc_texto(m.group(2)) return "", "" def extraer_autorizacion_desde_texto(texto): t = limpiar_texto(texto).upper() m = re.search(r"AUTORIZACI[ÓO]N\s*:\s*([A-Z0-9\-]+)", t) if m: return m.group(1) return "" def extraer_ingreso_desde_texto(texto): t = limpiar_texto(texto).upper() # Caso especial mencionado: INGRESO: 9 m = re.search(r"INGRESO\s*:\s*([0-9/\-.]+)", t) if m: return m.group(1) return "" def extraer_abono_desde_texto(texto): t = limpiar_texto(texto).upper() m = re.search(r"ABONO\s+PACIENTE\s*:\s*([0-9.,]+)", t) if m: return m.group(1) return "" def obtener_serie_columna_especial(df): """ Devuelve la columna que parece contener el texto mezclado. """ mejor_col = None mejor_score = -1 palabras = ["AUTORIZACION", "AUTORIZACIÓN", "PACIENTE", "INGRESO", "ABONO"] for col in df.columns: serie = df[col].astype(str) textos = " ".join(serie.head(50).tolist()).upper() score = sum(textos.count(p) for p in palabras) + int(serie.map(lambda x: len(str(x))).mean()) if score > mejor_score: mejor_score = score mejor_col = col return mejor_col def construir_base_normal(df, archivo_nombre): """ Construye dataframe base con columnas estándar a partir de archivo normal. """ base = pd.DataFrame() for campo in COLUMNAS_FINAL_IT: base[campo] = "" # columnas origen col_radicado = encontrar_columna(df, "radicado") col_tipo_doc = encontrar_columna(df, "tipo_documento") col_doc = encontrar_columna(df, "documento") col_tipo_doc_doc = encontrar_columna(df, "tipo_documento_documento") col_ingreso = encontrar_columna(df, "ingreso") col_total = encontrar_columna(df, "total_cargos") col_aut = encontrar_columna(df, "autorizacion") col_serv = encontrar_columna(df, "tipo_servicio") col_copago = encontrar_columna(df, "copago") col_mod = encontrar_columna(df, "cuota_moderadora") col_cop_mod = encontrar_columna(df, "cop_mod") col_neto = encontrar_columna(df, "neto") n = len(df) base = pd.DataFrame(index=df.index) base["archivo_origen"] = archivo_nombre base["tipo_archivo"] = "NORMAL" base["radicado"] = df[col_radicado] if col_radicado else "" base["tipo_documento"] = df[col_tipo_doc] if col_tipo_doc else "" base["documento"] = df[col_doc] if col_doc else "" # Si viene tipo y documento mezclado en una misma columna, intentar partir. if col_tipo_doc_doc: columnas_para_extraer = [col_tipo_doc_doc] else: # Fallback: revisar columnas probables sin incluir radicado ni montos. columnas_para_extraer = [ c for c in df.columns if c not in { col_radicado, col_ingreso, col_total, col_aut, col_serv, col_copago, col_mod, col_cop_mod, col_neto } ] for col_mix in columnas_para_extraer: for i, val in df[col_mix].items(): tipo, doc = extraer_tipo_doc_y_documento_desde_texto(val) current_tipo = limpiar_texto(base.at[i, "tipo_documento"]) current_doc = limpiar_texto(base.at[i, "documento"]) raw_val = limpiar_texto(val) if col_mix == col_tipo_doc_doc and (current_tipo == raw_val or current_doc == raw_val): if tipo: base.at[i, "tipo_documento"] = tipo if doc: base.at[i, "documento"] = doc continue if col_mix != col_tipo_doc_doc and current_tipo and current_doc: continue if tipo and not current_tipo: base.at[i, "tipo_documento"] = tipo if doc and not current_doc: base.at[i, "documento"] = doc # Si no se detectó columna radicado explícita, intentar inferir radicado desde cualquier columna # buscando secuencias largas de dígitos (heurística). No sobrescribe documento ya detectado. for col in df.columns: for i, val in df[col].items(): if limpiar_texto(base.at[i, "radicado"]): continue s = limpiar_texto(val) if not s: continue # Buscar secuencia de al menos 7 dígitos hasta 13 m = re.search(r"(\d{7,13})", str(s)) if not m: continue candidate = m.group(1) # Evitar tomar si coincide con documento ya detectado if limpiar_texto(base.at[i, "documento"]) == candidate: continue # Evitar sobrescribir si ya hay tipo_documento (probablemente documento legítimo) if limpiar_texto(base.at[i, "tipo_documento"]): continue base.at[i, "radicado"] = candidate base["ingreso"] = df[col_ingreso] if col_ingreso else "" base["total_cargos"] = df[col_total] if col_total else "" base["autorizacion"] = df[col_aut] if col_aut else "" base["tipo_servicio"] = df[col_serv] if col_serv else "" base["copago_origen"] = df[col_copago] if col_copago else "" base["moderadora_origen"] = df[col_mod] if col_mod else "" base["cop_mod_origen"] = df[col_cop_mod] if col_cop_mod else "" base["neto_origen"] = df[col_neto] if col_neto else "" return base def construir_base_especial(df, archivo_nombre): """ Reconstruye columnas estándar desde texto mezclado. """ col_texto = obtener_serie_columna_especial(df) col_radicado = encontrar_columna(df, "radicado") col_total = encontrar_columna(df, "total_cargos") col_serv = encontrar_columna(df, "tipo_servicio") base = pd.DataFrame(index=df.index) base["archivo_origen"] = archivo_nombre base["tipo_archivo"] = "ESPECIAL" base["radicado"] = df[col_radicado] if col_radicado else "" base["tipo_documento"] = "" base["documento"] = "" base["ingreso"] = "" base["total_cargos"] = df[col_total] if col_total else "" base["autorizacion"] = "" base["tipo_servicio"] = df[col_serv] if col_serv else "" base["copago_origen"] = "" base["moderadora_origen"] = "" base["cop_mod_origen"] = "" base["neto_origen"] = "" if col_texto: for i, texto in df[col_texto].items(): tipo, doc = extraer_tipo_doc_y_documento_desde_texto(texto) aut = extraer_autorizacion_desde_texto(texto) ingreso = extraer_ingreso_desde_texto(texto) abono = extraer_abono_desde_texto(texto) base.at[i, "tipo_documento"] = tipo base.at[i, "documento"] = doc base.at[i, "autorizacion"] = aut base.at[i, "ingreso"] = ingreso base.at[i, "cop_mod_origen"] = abono return base # ============================================================ # NORMALIZACIÓN FINAL Y VALIDACIONES # ============================================================ def resolver_tipo_y_documento(tipo_doc_raw, documento_raw): """ Resuelve casos donde el tipo y documento vienen mezclados en cualquiera de las dos columnas. Regla: - Si tipo_documento trae "CC 1012333072", separa. - Si tipo_documento trae "CÉDULA CIUDADANIA" y documento viene aparte, homologa a CC. - Si documento trae "RC 1188975098", separa. - Si solo viene número, permite tipo vacío y documento numérico. """ tipo_raw = limpiar_texto(tipo_doc_raw) doc_raw = limpiar_texto(documento_raw) # 1. Intentar extraer desde la columna tipo_documento si viene mezclada. tipo_mix, doc_mix = extraer_tipo_doc_y_documento_desde_texto(tipo_raw) if tipo_mix or doc_mix: tipo_final = homologar_tipo_documento(tipo_mix) doc_final = limpiar_documento(doc_mix, tipo_final) # Si solo encontró tipo, usar documento de la otra columna. if tipo_final and not doc_final and doc_raw: doc_final = limpiar_documento(doc_raw, tipo_final) if tipo_final or doc_final: return tipo_final, doc_final # 2. Intentar extraer desde la columna documento si viene mezclada. tipo_mix, doc_mix = extraer_tipo_doc_y_documento_desde_texto(doc_raw) if tipo_mix or doc_mix: tipo_final = homologar_tipo_documento(tipo_mix) doc_final = limpiar_documento(doc_mix, tipo_final) # Si el tipo no salió en documento, usar tipo_raw homologado. if not tipo_final: tipo_final = homologar_tipo_documento(tipo_raw) if tipo_final or doc_final: return tipo_final, doc_final # 3. Caso normal. tipo_final = homologar_tipo_documento(tipo_raw) doc_final = limpiar_documento(doc_raw, tipo_final) return tipo_final, doc_final def normalizar_registros(base): registros = [] errores = [] for idx, row in base.iterrows(): obs = [] archivo = limpiar_texto(row.get("archivo_origen", "")) tipo_archivo = limpiar_texto(row.get("tipo_archivo", "")) radicado = limpiar_texto(row.get("radicado", "")) # IMPORTANTE: # Primero separar tipo/documento si vienen mezclados, luego homologar y validar. tipo_doc, documento = resolver_tipo_y_documento( row.get("tipo_documento", ""), row.get("documento", "") ) ingreso, err_fecha = normalizar_fecha(row.get("ingreso", "")) # No detener el proceso si la fecha no se reconoce. if err_fecha: ingreso = "" obs.append(err_fecha) anio = obtener_anio_fecha(ingreso) total_cargos = limpiar_numero_entero(row.get("total_cargos", ""), default="") neto_origen = limpiar_numero_entero(row.get("neto_origen", ""), default="") if not total_cargos and neto_origen: total_cargos = neto_origen if not total_cargos: obs.append("Total cargos / bruto vacío o inválido") autorizacion, err_aut = normalizar_autorizacion(row.get("autorizacion", "")) # No detener el proceso si la autorización no se reconoce. if err_aut: autorizacion = autorizacion or "" obs.append(err_aut) tipo_servicio = normalizar_tipo_servicio(row.get("tipo_servicio", "")) ok_doc, err_doc = validar_documento(tipo_doc, documento) if not ok_doc: obs.append(err_doc) # Copago / moderadora: copago = to_int(row.get("copago_origen", ""), default=0) moderadora = to_int(row.get("moderadora_origen", ""), default=0) cop_mod = to_int(row.get("cop_mod_origen", ""), default=0) # Si hay cop_mod mezclado, clasificar por tabla. if cop_mod > 0: c, m = separar_copago_moderadora(cop_mod, anio) copago += c moderadora += m # Si hay neto y bruto, usar diferencia para inferir cop/mod. neto = to_int(row.get("neto_origen", ""), default=0) bruto = to_int(total_cargos, default=0) if bruto > 0 and neto > 0 and bruto >= neto: diferencia = bruto - neto if diferencia > 0 and copago == 0 and moderadora == 0: c, m = separar_copago_moderadora(diferencia, anio) copago = c moderadora = m final = { "archivo_origen": archivo, "tipo_archivo": tipo_archivo, "radicado": radicado, "tipo_documento": tipo_doc, "documento": documento, "ingreso": ingreso, "total_cargos": total_cargos, "autorizacion": autorizacion, "tipo_servicio": tipo_servicio, "copago": str(copago), "cuota_moderadora": str(moderadora), "observacion": " | ".join(obs), } if obs: errores.append(final) else: registros.append(final) return pd.DataFrame(registros), pd.DataFrame(errores) def separar_duplicados(df_ok): """ Lógica tipo CONTAR.SI del usuario: Si un radicado aparece 2 o más veces: → TODOS esos registros van a DUPLICADOS.xlsx → NINGUNO pasa al CSV_FINAL_IT """ if df_ok.empty: return df_ok, pd.DataFrame() # Limpiar radicado temporalmente temp = df_ok.copy() temp["_radicado_clean"] = temp["radicado"].astype(str).str.strip() # Conteo tipo CONTAR.SI conteo = temp.groupby("_radicado_clean")["_radicado_clean"].transform("count") # Duplicado = aparece 2 o más veces dup_mask = (conteo >= 2) & temp["_radicado_clean"].ne("") duplicados = temp[dup_mask].copy() unicos = temp[~dup_mask].copy() if not duplicados.empty: duplicados["cantidad_duplicados"] = conteo[dup_mask] duplicados["observacion"] = ( "Radicado duplicado detectado por CONTAR.SI. " "Cantidad: " + duplicados["cantidad_duplicados"].astype(str) ) # Quitar columnas temporales for df_ in [duplicados, unicos]: if "_radicado_clean" in df_.columns: df_.drop(columns=["_radicado_clean"], inplace=True, errors="ignore") return unicos, duplicados def clasificar_errores_por_hoja(df_errores): """ Separa los errores en hojas por categoría. Un mismo registro puede aparecer en varias hojas si tiene varios errores. """ if df_errores.empty or "observacion" not in df_errores.columns: return {} categorias = { "Autorizaciones": [ "autorización", "autorizacion" ], "Fechas": [ "fecha ingreso", "fecha" ], "Documentos": [ "documento inválido", "documento invalido", "longitud inválida", "longitud invalida", "tipo de documento" ], "Total_Bruto": [ "total cargos", "bruto" ], "Tipo_Documento": [ "tipo de documento vacío", "tipo de documento vacio", "tipo de documento no parametrizado" ], "Otros": [] } hojas = {} obs_norm = df_errores["observacion"].astype(str).str.lower() usados = pd.Series(False, index=df_errores.index) for hoja, palabras in categorias.items(): if hoja == "Otros": continue mask = pd.Series(False, index=df_errores.index) for palabra in palabras: mask = mask | obs_norm.str.contains(palabra, na=False, regex=False) if mask.any(): hojas[hoja] = df_errores[mask].copy() usados = usados | mask otros = df_errores[~usados].copy() if not otros.empty: hojas["Otros"] = otros return hojas def escribir_errores_por_hojas(writer, df_err_total): """ Escribe ERRORES_NORMALIZACION.xlsx con: - Hoja Todos - Hojas por categoría: Autorizaciones, Fechas, Documentos, Total_Bruto, Tipo_Documento, Otros """ columnas_base = [ "archivo_origen", "tipo_archivo", *COLUMNAS_FINAL_IT, "observacion" ] if df_err_total.empty: pd.DataFrame(columns=columnas_base).to_excel(writer, index=False, sheet_name="Todos") return df_err_total.to_excel(writer, index=False, sheet_name="Todos") hojas = clasificar_errores_por_hoja(df_err_total) for nombre_hoja, df_hoja in hojas.items(): # Excel permite máximo 31 caracteres en nombre de hoja nombre_seguro = str(nombre_hoja)[:31] df_hoja.to_excel(writer, index=False, sheet_name=nombre_seguro) def leer_archivo_flexible(path): """Lee CSV o Excel manteniendo todo como texto.""" path = Path(path) ext = path.suffix.lower() if ext == ".csv": return leer_csv_flexible(path) if ext in {".xlsx", ".xlsm", ".xls"}: return pd.read_excel(path, dtype=str, keep_default_na=False) raise RuntimeError(f"Formato no soportado: {path.name}. Usa CSV o Excel.") def listar_archivos_entrada(carpeta): carpeta = Path(carpeta) archivos = [] for patron in ("*.csv", "*.xlsx", "*.xlsm", "*.xls"): archivos.extend(carpeta.glob(patron)) # Evitar leer salidas anteriores o temporales. archivos_limpios = [] for a in sorted(archivos): nombre = a.name.upper() if a.parent.name.upper() == "SALIDA_NORMALIZADOR_IT": continue if nombre.startswith("CONSOLIDADO_BOT_IT"): continue if "ORIGEN_SERVICIO" in nombre: continue if nombre.startswith("~$"): continue archivos_limpios.append(a) return archivos_limpios def normalizar_clave_radicado(valor): """ Normaliza el radicado para poder cruzarlo contra Origen Servicio. Evita fallos por espacios, guiones, .0 de Excel o separadores. """ s = limpiar_texto(valor).upper() if not s: return "" # Cuando Excel deja 100000069634188.0 s = re.sub(r"\.0$", "", s) # Quitar espacios y separadores comunes s = re.sub(r"[\s\-_.:,;/\\]+", "", s) # Si quedó algo tipo RAD100000..., conservar letras y números s = re.sub(r"[^A-Z0-9]", "", s) return s def cargar_origen_servicio(ruta_archivo): """ Carga el Excel/CSV seleccionado por el usuario y arma un diccionario: radicado -> tipo_servicio. Soporta encabezados como: - radicado / NumeroRadicacion / numero radicacion - tipo_servicio / servicio / Origen_Servicio """ if not ruta_archivo: return {} archivo_servicio = Path(ruta_archivo) if not archivo_servicio.exists(): return {} df_serv = leer_archivo_flexible(archivo_servicio) df_serv.columns = [str(c).strip() for c in df_serv.columns] col_rad = encontrar_columna(df_serv, "radicado") col_serv = encontrar_columna(df_serv, "tipo_servicio") # Refuerzo por si el encabezado viene pegado tipo NumeroRadicacion if not col_rad: for c in df_serv.columns: c_norm = normalizar_nombre_columna(c).replace(" ", "") if "radic" in c_norm: col_rad = c break # Refuerzo por si el encabezado viene pegado tipo OrigenServicio if not col_serv: for c in df_serv.columns: c_norm = normalizar_nombre_columna(c).replace(" ", "") if "servicio" in c_norm or "origenservicio" in c_norm: col_serv = c break if not col_rad or not col_serv: return {} mapa = {} for _, row in df_serv.iterrows(): rad_original = limpiar_texto(row.get(col_rad, "")) serv = limpiar_texto(row.get(col_serv, "")) if rad_original and serv: # Guarda varias llaves para que cruce aunque Excel venga con espacios, guiones o .0 mapa[rad_original] = serv mapa[normalizar_clave_radicado(rad_original)] = serv solo_digitos = re.sub(r"\D", "", rad_original) if solo_digitos: mapa[solo_digitos] = serv return mapa def simplificar_observacion(obs): """Convierte observaciones largas a códigos rápidos para filtrar.""" obs = limpiar_texto(obs) if not obs: return "" partes_salida = [] obs_norm = normalizar_nombre_columna(obs) reglas = [ ("SIN_AUTORIZACION", ["autorizacion vacia", "sin autorizacion"]), ("AUTORIZACION_INCOMPLETA", ["autorizacion incompleta"]), ("AUTORIZACION_INVALIDA", ["autorizacion invalida"]), ("SIN_FECHA", ["sin fecha"]), ("FECHA_INVALIDA", ["fecha ingreso vacia", "fecha ingreso invalida", "fecha"]), ("DOCUMENTO_INVALIDO", ["documento invalido", "longitud invalida", "tipo de documento vacio", "tipo de documento no parametrizado"]), ("VALOR_INVALIDO", ["total cargos", "bruto"]), ] for codigo, palabras in reglas: if any(p in obs_norm for p in palabras): partes_salida.append(codigo) if not partes_salida: partes_salida.append(obs.upper()) # Quitar repetidos conservando orden. vistos = set() final = [] for p in partes_salida: if p not in vistos: vistos.add(p) final.append(p) return " | ".join(final) def preparar_consolidado(df_ok_total, df_err_total): columnas = [*COLUMNAS_FINAL_IT, "OBSERVACION", "archivo_origen", "tipo_archivo"] partes = [] if df_ok_total is not None and not df_ok_total.empty: ok = df_ok_total.copy() ok["OBSERVACION"] = "" partes.append(ok) if df_err_total is not None and not df_err_total.empty: err = df_err_total.copy() err["OBSERVACION"] = err.get("observacion", "").apply(simplificar_observacion) partes.append(err) if partes: cons = pd.concat(partes, ignore_index=True) else: cons = pd.DataFrame(columns=columnas) for c in COLUMNAS_FINAL_IT: if c not in cons.columns: cons[c] = "" # Detectar duplicado sobre TODO el consolidado, así esté bueno o malo. if not cons.empty and "radicado" in cons.columns: rad = cons["radicado"].astype(str).str.strip() conteo = rad.map(rad.value_counts()) mask_dup = rad.ne("") & conteo.ge(2) cons.loc[mask_dup, "OBSERVACION"] = cons.loc[mask_dup, "OBSERVACION"].apply( lambda x: "RADICADO_DUPLICADO" if not limpiar_texto(x) else f"{x} | RADICADO_DUPLICADO" ) cons["OBSERVACION"] = cons["OBSERVACION"].apply(lambda x: "PASO_BOT" if not limpiar_texto(x) else x) # Limpieza de columnas auxiliares. for c in ["observacion", "cantidad_duplicados"]: if c in cons.columns: cons.drop(columns=[c], inplace=True, errors="ignore") for c in columnas: if c not in cons.columns: cons[c] = "" return cons[columnas] def preparar_resumen(df_consolidado, resumen_archivos): resumen_obs = pd.DataFrame(columns=["OBSERVACION", "CANTIDAD"]) if df_consolidado is not None and not df_consolidado.empty: resumen_obs = ( df_consolidado["OBSERVACION"] .fillna("") .astype(str) .str.split(" | ") .explode() .str.strip() .replace("", pd.NA) .dropna() .value_counts() .rename_axis("OBSERVACION") .reset_index(name="CANTIDAD") ) total = pd.DataFrame([{"OBSERVACION": "TOTAL", "CANTIDAD": len(df_consolidado)}]) resumen_obs = pd.concat([resumen_obs, total], ignore_index=True) resumen_arch = pd.DataFrame(resumen_archivos) return resumen_obs, resumen_arch def autoajustar_excel(writer, hojas): try: from openpyxl.styles import Font, PatternFill, Alignment for nombre_hoja in hojas: ws = writer.book[nombre_hoja] ws.freeze_panes = "A2" ws.auto_filter.ref = ws.dimensions for cell in ws[1]: cell.font = Font(bold=True) cell.fill = PatternFill("solid", fgColor="D9EAF7") cell.alignment = Alignment(horizontal="center") for col in ws.columns: max_len = 0 letter = col[0].column_letter for cell in col[:200]: max_len = max(max_len, len(str(cell.value or ""))) ws.column_dimensions[letter].width = min(max(max_len + 2, 12), 45) except Exception: pass def escribir_excel_consolidado(path_salida, df_original, df_consolidado, resumen_obs, resumen_archivos): with pd.ExcelWriter(path_salida, engine="openpyxl") as writer: df_original.to_excel(writer, index=False, sheet_name="Original") df_consolidado.to_excel(writer, index=False, sheet_name="Consolidado") resumen_obs.to_excel(writer, index=False, sheet_name="Resumen", startrow=0) # Debajo del resumen por observación se deja el detalle por archivo. start = len(resumen_obs) + 3 pd.DataFrame([{"": "DETALLE POR ARCHIVO"}]).to_excel( writer, index=False, header=False, sheet_name="Resumen", startrow=start ) resumen_archivos.to_excel(writer, index=False, sheet_name="Resumen", startrow=start + 2) autoajustar_excel(writer, ["Original", "Consolidado", "Resumen"]) def procesar_archivo_unico(archivo_entrada, log_callback=None, ruta_origen_servicio=None): archivo_entrada = Path(archivo_entrada) carpeta_temp = archivo_entrada.parent / "__TEMP_ARCHIVO_UNICO__" carpeta_temp.mkdir(exist_ok=True) copia = carpeta_temp / archivo_entrada.name shutil.copy2(archivo_entrada, copia) return procesar_carpeta_incremental(carpeta_temp, log_callback=log_callback, ruta_origen_servicio=ruta_origen_servicio) def procesar_carpeta_incremental(carpeta, log_callback=None, ruta_origen_servicio=None): """ Genera un Excel único con 3 hojas: 1. Original 2. Consolidado 3. Resumen También conserva CSV_FINAL_IT.csv solo con registros PASO_BOT para el bot. """ carpeta = Path(carpeta) archivos = listar_archivos_entrada(carpeta) if not archivos: raise RuntimeError("No se encontraron archivos CSV o Excel en la carpeta seleccionada.") salida = carpeta / "SALIDA_NORMALIZADOR_IT" salida.mkdir(exist_ok=True) excel_consolidado = salida / "CONSOLIDADO_BOT_IT.xlsx" csv_final = salida / "CSV_FINAL_IT.csv" todos_original = [] todos_ok = [] todos_errores = [] resumen = [] def log(msg): if log_callback: log_callback(msg) log("Procesando archivos y generando Excel único: Original | Consolidado | Resumen...") mapa_servicios = cargar_origen_servicio(ruta_origen_servicio) if mapa_servicios: log(f"Origen_Servicio cargado: {len(mapa_servicios)} radicados con servicio.") else: log("No se seleccionó Origen Servicio o no tenía datos válidos.") for idx_archivo, path in enumerate(archivos, start=1): try: log(f"[{idx_archivo}/{len(archivos)}] Leyendo: {path.name}") df = leer_archivo_flexible(path) df.columns = [str(c).strip() for c in df.columns] filas_leidas = len(df) original = df.copy() original.insert(0, "archivo_origen", path.name) todos_original.append(original) if df.empty: resumen.append({ "archivo": path.name, "estado": "SIN_DATOS", "tipo": "", "filas_leidas": 0, "paso_bot": 0, "con_error": 0, "duplicados": 0, "detalle": "Archivo vacío", }) continue especial = es_archivo_especial(df) if especial: log(" → Detectado como ESPECIAL") base = construir_base_especial(df, path.name) else: log(" → Detectado como NORMAL") base = construir_base_normal(df, path.name) # Llenar tipo_servicio desde el Excel/CSV Origen Servicio usando el radicado if mapa_servicios and "radicado" in base.columns: encontrados_servicio = 0 sin_cruce_servicio = 0 for i, row in base.iterrows(): rad_original = limpiar_texto(row.get("radicado", "")) rad_norm = normalizar_clave_radicado(rad_original) rad_digitos = re.sub(r"\D", "", rad_original) servicio_excel = ( mapa_servicios.get(rad_original, "") or mapa_servicios.get(rad_norm, "") or mapa_servicios.get(rad_digitos, "") ) if servicio_excel: base.at[i, "tipo_servicio"] = servicio_excel encontrados_servicio += 1 else: sin_cruce_servicio += 1 log(f" → Origen Servicio aplicado: {encontrados_servicio} filas con servicio / {sin_cruce_servicio} sin cruce.") ok, err = normalizar_registros(base) todos_ok.append(ok) todos_errores.append(err) resumen.append({ "archivo": path.name, "estado": "PROCESADO", "tipo": "ESPECIAL" if especial else "NORMAL", "filas_leidas": filas_leidas, "paso_bot": 0, # se actualiza después "con_error": len(err), "duplicados": 0, # se actualiza después "detalle": "", }) log(f" → Leídas: {filas_leidas} | OK preliminar: {len(ok)} | Con error: {len(err)}") except Exception as e: detalle = f"{type(e).__name__}: {e}" log(f" [ERROR] {path.name}: {detalle}") resumen.append({ "archivo": path.name, "estado": "ERROR_LECTURA", "tipo": "", "filas_leidas": 0, "paso_bot": 0, "con_error": 0, "duplicados": 0, "detalle": detalle, }) df_original = pd.concat(todos_original, ignore_index=True, sort=False) if todos_original else pd.DataFrame() df_ok_total = pd.concat(todos_ok, ignore_index=True) if todos_ok else pd.DataFrame() df_err_total = pd.concat(todos_errores, ignore_index=True) if todos_errores else pd.DataFrame() df_consolidado = preparar_consolidado(df_ok_total, df_err_total) # CSV para el bot: solo PASO_BOT. df_final_bot = df_consolidado[df_consolidado["OBSERVACION"].eq("PASO_BOT")].copy() if df_final_bot.empty: pd.DataFrame(columns=CSV_FINAL_IT_HEADERS).to_csv(csv_final, sep="|", index=False, encoding="utf-8-sig") else: df_out = df_final_bot.reindex(columns=COLUMNAS_FINAL_IT).fillna("") df_out = df_out.rename(columns=CSV_FINAL_IT_RENAME) df_out = df_out.reindex(columns=CSV_FINAL_IT_HEADERS) df_out.to_csv(csv_final, sep="|", index=False, encoding="utf-8-sig") # Actualizar resumen por archivo desde el consolidado real. if not df_consolidado.empty and "archivo_origen" in df_consolidado.columns: paso_por_archivo = df_consolidado[df_consolidado["OBSERVACION"].eq("PASO_BOT")].groupby("archivo_origen").size().to_dict() dup_por_archivo = df_consolidado[df_consolidado["OBSERVACION"].str.contains("RADICADO_DUPLICADO", na=False)].groupby("archivo_origen").size().to_dict() error_por_archivo = df_consolidado[~df_consolidado["OBSERVACION"].eq("PASO_BOT")].groupby("archivo_origen").size().to_dict() for r in resumen: a = r.get("archivo", "") r["paso_bot"] = int(paso_por_archivo.get(a, 0)) r["duplicados"] = int(dup_por_archivo.get(a, 0)) r["con_error"] = int(error_por_archivo.get(a, 0)) resumen_obs, resumen_arch = preparar_resumen(df_consolidado, resumen) escribir_excel_consolidado(excel_consolidado, df_original, df_consolidado, resumen_obs, resumen_arch) total_duplicados = int(df_consolidado["OBSERVACION"].str.contains("RADICADO_DUPLICADO", na=False).sum()) if not df_consolidado.empty else 0 total_errores = int((~df_consolidado["OBSERVACION"].eq("PASO_BOT")).sum()) if not df_consolidado.empty else 0 return { "carpeta_salida": str(salida), "excel_consolidado": str(excel_consolidado), "csv_final": str(csv_final), "total_final": len(df_final_bot), "total_errores": total_errores, "total_duplicados": total_duplicados, "archivos": len(archivos), "modo": "excel_consolidado", } def procesar_carpeta(carpeta, log_callback=None, ruta_origen_servicio=None): # Se deja como alias para que el modo masivo también genere el nuevo formato. return procesar_carpeta_incremental(carpeta, log_callback=log_callback, ruta_origen_servicio=ruta_origen_servicio) # ============================================================ # INTERFAZ GRÁFICA # ============================================================ class NormalizadorITApp: def __init__(self, root): self.root = root self.root.title("Normalizador CSV IT - Salud Total") self.root.geometry("900x600") self.root.minsize(850, 550) self.carpeta_var = tk.StringVar() self.archivo_var = tk.StringVar() self.ruta_origen_servicio = tk.StringVar() self.modo_entrada = tk.StringVar(value="carpeta") self.modo_proceso = tk.StringVar(value="incremental") self._build_ui() def _build_ui(self): frame = ttk.Frame(self.root, padding=15) frame.pack(fill="both", expand=True) title = ttk.Label( frame, text="Normalizador CSV para Bot IT", font=("Segoe UI", 16, "bold") ) title.pack(anchor="w") subtitle = ttk.Label( frame, text="Genera Excel único: Original, Consolidado y Resumen; además crea CSV_FINAL_IT.csv", font=("Segoe UI", 10) ) subtitle.pack(anchor="w", pady=(0, 15)) entrada_frame = ttk.LabelFrame(frame, text="Tipo de entrada") entrada_frame.pack(fill="x", pady=5) ttk.Radiobutton( entrada_frame, text="Procesar carpeta completa", variable=self.modo_entrada, value="carpeta" ).pack(anchor="w", padx=8, pady=2) ttk.Radiobutton( entrada_frame, text="Procesar un solo archivo CSV", variable=self.modo_entrada, value="archivo" ).pack(anchor="w", padx=8, pady=2) row = ttk.Frame(frame) row.pack(fill="x", pady=5) ttk.Label(row, text="Ruta:").pack(side="left") self.entry_ruta = ttk.Entry(row, textvariable=self.carpeta_var) self.entry_ruta.pack(side="left", fill="x", expand=True, padx=8) ttk.Button(row, text="Seleccionar", command=self.seleccionar_entrada).pack(side="left") row_servicio = ttk.Frame(frame) row_servicio.pack(fill="x", pady=5) ttk.Label(row_servicio, text="Origen Servicio:").pack(side="left") self.entry_origen_servicio = ttk.Entry(row_servicio, textvariable=self.ruta_origen_servicio) self.entry_origen_servicio.pack(side="left", fill="x", expand=True, padx=8) ttk.Button( row_servicio, text="Seleccionar Excel", command=self.seleccionar_origen_servicio ).pack(side="left") ttk.Button( row_servicio, text="Limpiar", command=lambda: self.ruta_origen_servicio.set("") ).pack(side="left", padx=(8, 0)) modo_frame = ttk.LabelFrame(frame, text="Modo de procesamiento") modo_frame.pack(fill="x", pady=8) ttk.Radiobutton( modo_frame, text="Procesar uno por uno y consolidar (recomendado)", variable=self.modo_proceso, value="incremental" ).pack(anchor="w", padx=8, pady=2) ttk.Radiobutton( modo_frame, text="Procesar todo junto (modo anterior)", variable=self.modo_proceso, value="masivo" ).pack(anchor="w", padx=8, pady=2) buttons = ttk.Frame(frame) buttons.pack(fill="x", pady=10) self.btn_procesar = ttk.Button(buttons, text="Procesar carpeta", command=self.iniciar_proceso) self.btn_procesar.pack(side="left") ttk.Button(buttons, text="Abrir salida", command=self.abrir_salida).pack(side="left", padx=8) self.progress = ttk.Progressbar(frame, mode="indeterminate") self.progress.pack(fill="x", pady=8) ttk.Label(frame, text="Log del proceso:").pack(anchor="w") self.txt_log = tk.Text(frame, height=22, wrap="word") self.txt_log.pack(fill="both", expand=True) scrollbar = ttk.Scrollbar(self.txt_log, command=self.txt_log.yview) self.txt_log.configure(yscrollcommand=scrollbar.set) footer = ttk.Label( frame, text="Salidas: CONSOLIDADO_BOT_IT.xlsx (Original, Consolidado, Resumen) | CSV_FINAL_IT.csv", font=("Segoe UI", 9) ) footer.pack(anchor="w", pady=(8, 0)) def log(self, msg): self.txt_log.insert("end", f"{datetime.now().strftime('%H:%M:%S')} {msg}\n") self.txt_log.see("end") self.root.update_idletasks() def seleccionar_entrada(self): modo = self.modo_entrada.get() if modo == "archivo": archivo = filedialog.askopenfilename( title="Selecciona un archivo CSV o Excel", filetypes=[("CSV o Excel", "*.csv *.xlsx *.xlsm *.xls"), ("CSV", "*.csv"), ("Excel", "*.xlsx *.xlsm *.xls")] ) if archivo: self.archivo_var.set(archivo) self.carpeta_var.set(archivo) else: carpeta = filedialog.askdirectory( title="Selecciona carpeta con CSV o Excel" ) if carpeta: self.carpeta_var.set(carpeta) def seleccionar_origen_servicio(self): archivo = filedialog.askopenfilename( title="Seleccionar Excel/CSV de Origen Servicio", filetypes=[ ("Excel o CSV", "*.xlsx *.xlsm *.xls *.csv"), ("Excel", "*.xlsx *.xlsm *.xls"), ("CSV", "*.csv"), ("Todos", "*.*"), ] ) if archivo: self.ruta_origen_servicio.set(archivo) self.log(f"Origen Servicio seleccionado: {archivo}") def abrir_salida(self): carpeta = self.carpeta_var.get().strip() if not carpeta: messagebox.showwarning("Sin carpeta", "Selecciona primero una carpeta.") return salida = Path(carpeta) / "SALIDA_NORMALIZADOR_IT" if not salida.exists(): messagebox.showinfo("Sin salida", "Todavía no existe la carpeta de salida.") return try: os.startfile(str(salida)) except Exception as e: messagebox.showerror("Error", f"No se pudo abrir la salida:\n{e}") def iniciar_proceso(self): ruta = self.carpeta_var.get().strip() if not ruta: messagebox.showwarning("Sin ruta", "Selecciona carpeta o archivo.") return if not Path(ruta).exists(): messagebox.showerror("Error", "La ruta seleccionada no existe.") return self.btn_procesar.config(state="disabled") self.progress.start(10) self.txt_log.delete("1.0", "end") self.log("Iniciando normalización...") if self.ruta_origen_servicio.get().strip(): self.log(f"Usando Origen Servicio: {self.ruta_origen_servicio.get().strip()}") else: self.log("Origen Servicio: no seleccionado (se usará el servicio del archivo original si existe).") hilo = threading.Thread(target=self._procesar_thread, args=(ruta,), daemon=True) hilo.start() def _procesar_thread(self, ruta): try: modo_entrada = self.modo_entrada.get() modo_proceso = self.modo_proceso.get() if modo_entrada == "archivo": resultado = procesar_archivo_unico(ruta, log_callback=self.log, ruta_origen_servicio=self.ruta_origen_servicio.get().strip()) else: if modo_proceso == "incremental": resultado = procesar_carpeta_incremental(ruta, log_callback=self.log, ruta_origen_servicio=self.ruta_origen_servicio.get().strip()) else: resultado = procesar_carpeta(ruta, log_callback=self.log, ruta_origen_servicio=self.ruta_origen_servicio.get().strip()) self.log("") self.log("Proceso terminado correctamente.") self.log(f"Archivos leídos: {resultado['archivos']}") self.log(f"Registros finales IT: {resultado['total_final']}") self.log(f"Errores normalización: {resultado['total_errores']}") self.log(f"Duplicados: {resultado['total_duplicados']}") self.log(f"Salida: {resultado['carpeta_salida']}") if resultado.get("excel_consolidado"): self.log(f"Excel consolidado: {resultado['excel_consolidado']}") messagebox.showinfo( "Proceso terminado", "Normalización finalizada.\n\n" f"Registros finales IT: {resultado['total_final']}\n" f"Errores: {resultado['total_errores']}\n" f"Duplicados: {resultado['total_duplicados']}\n\n" f"Salida:\n{resultado['carpeta_salida']}" ) except Exception as e: self.log("[ERROR GENERAL]") self.log(str(e)) self.log(traceback.format_exc()) messagebox.showerror("Error", f"Ocurrió un error:\n{e}") finally: self.progress.stop() self.btn_procesar.config(state="normal") def main(): root = tk.Tk() app = NormalizadorITApp(root) root.mainloop() if __name__ == "__main__": main()