import tkinter as tk from tkinter import filedialog, messagebox, scrolledtext import threading import csv import time import ctypes import os from datetime import datetime from pywinauto.application import Application from pywinauto import mouse try: import cv2 import numpy as np OPENCV_AVAILABLE = True except ImportError: OPENCV_AVAILABLE = False try: ctypes.windll.shcore.SetProcessDpiAwareness(2) except Exception: try: ctypes.windll.user32.SetProcessDPIAware() except Exception: pass DOC_COORDS = { "CC": (796, 343), "TI": (796, 549), "CE": (796, 329), "AS": (796, 296), "CD": (796, 314), "CN": (796, 371), "ME": (796, 378), "NI": (796, 412), "NU": (796, 453), "PS": (796, 450), "PE": (796, 467), "PT": (796, 483), "PJ": (796, 503), "RC": (796, 520), "SC": (796, 536), } DOC_STRATEGY = [ "CC", "TI", "CE", "RC", "PS", "PT", "PE", "AS", "ME", "CN", "SC", "CD", "NI", "NU", "PJ" ] SERVICES_COORDS = { "ambulancias y elem.": (1033, 367), "ambulatorio": (1033, 387), "c. externa": (1033, 405), "capacitacion": (1033, 417), "cirugia ambulatoria": (1033, 432), "fomento y prevencion": (1033, 449), "honorarios": (1033, 467), } COORD_CLOSE_ERROR = (1223, 704) TEMPLATE_ERROR_PATH = os.path.join("template", "error.png") TEMPLATE_SUCCESS_PATH = os.path.join("template", "success.png") TEMPLATE_BIEN_PATH = os.path.join("template", "bien.png") TEMPLATE_DIFERENCIAS_PATH = os.path.join("template", "flecha_diferencias_abierto.png") TEMPLATE_DIFERENCIAS_CERRADO_PATH = os.path.join("template", "flecha_diferencias_cerrado.png") TEMPLATE_FECHA_VACIA_SLASH_PATH = os.path.join("template", "fecha_vacia_slash.png") TEMPLATE_FECHA_VACIA_BLANCO_PATH = os.path.join("template", "fecha_vacia_blanco.png") TEMPLATE_POPUP_NOMBRES_PATH = os.path.join("template", "popup_nombres.png") TEMPLATE_RADICADO_EXISTENTE_PATH = os.path.join("template", "radicado_existente.png") # ========================= # CONFIGURACIÓN REINICIO COMPLETO APP # ========================= APP_PATH = r"C:\Users\Pysctas185\Documents\Integracion Total.lnk" PROCESS_NAME = "SaludTotal.Health.WinClient.TotalIntegration.exe" REINICIO_CADA = 6 TIMEOUT_REGISTRO = 120 REPORTS_DIR_NAME = "reportes_bot" def click_abs(window, x: int, y: int): try: window.wrapper_object().set_focus() except Exception: pass ctypes.windll.user32.SetCursorPos(x, y) time.sleep(0.05) ctypes.windll.user32.SetCursorPos(x + 1, y) time.sleep(0.05) ctypes.windll.user32.SetCursorPos(x, y) time.sleep(0.2) mouse.click(button="left", coords=(x, y)) class BotSaludTotal: def __init__(self, root): self.first_time = True self.root = root self.root.title("Bot Salud Total - Automatización con OpenCV") self.root.geometry("600x700") self.csv_path = tk.StringVar() self.date_format = tk.StringVar(value="dd/mm/yyyy") self.delimiter = tk.StringVar(value=",") self.is_running = False self.omitted_records = [] self.log_entries = [] self.resultados = [] # ========================= # CONTROL DE EJECUCIÓN / OVERLAY / REPORTES # ========================= self.run_id = datetime.now().strftime("%Y%m%d_%H%M%S") self.output_dir = os.path.join(os.getcwd(), REPORTS_DIR_NAME, self.run_id) self.total_registros = 0 self.procesados_ok = 0 self.errores = 0 self.omitidos = 0 self.existentes = 0 self.reinicios = 0 self.registro_actual = "-" self.estado_actual = "Listo" self.inicio_proceso = None self.inicio_registro = None self.overlay = None self.overlay_vars = {} tk.Label( root, text="Seleccione el archivo CSV de entrada:", font=("Arial", 10, "bold") ).pack(pady=10) frame_file = tk.Frame(root) frame_file.pack(pady=5, padx=20, fill="x") self.entry_file = tk.Entry(frame_file, textvariable=self.csv_path, state="readonly") self.entry_file.pack(side="left", fill="x", expand=True, padx=(0, 5)) tk.Button(frame_file, text="Buscar CSV", command=self.browse_file).pack(side="right") frame_delim = tk.LabelFrame(root, text="Delimitador de CSV", padx=10, pady=5) frame_delim.pack(pady=10, padx=20, fill="x") tk.Radiobutton(frame_delim, text="Coma (,)", variable=self.delimiter, value=",").pack(anchor="w") tk.Radiobutton(frame_delim, text="Pipe (|)", variable=self.delimiter, value="|").pack(anchor="w") frame_date = tk.LabelFrame(root, text="Formato de Fecha en CSV", padx=10, pady=5) frame_date.pack(pady=10, padx=20, fill="x") tk.Radiobutton( frame_date, text="Día/Mes/Año (Ej: 31/12/2025)", variable=self.date_format, value="dd/mm/yyyy" ).pack(anchor="w") tk.Radiobutton( frame_date, text="Mes/Día/Año (Ej: 12/31/2025)", variable=self.date_format, value="mm/dd/yyyy" ).pack(anchor="w") self.btn_start = tk.Button( root, text="INICIAR BOT", command=self.start_thread, bg="green", fg="white", font=("Arial", 11, "bold") ) self.btn_start.pack(pady=20) self.log_area = scrolledtext.ScrolledText(root, height=10, state="disabled") self.log_area.pack(padx=20, pady=10, fill="both", expand=True) self.crear_overlay_fantasma() self.actualizar_overlay() def crear_overlay_fantasma(self): """Crea un mini panel flotante, pequeño y liviano, para monitorear el bot.""" try: self.overlay = tk.Toplevel(self.root) self.overlay.title("BOT ST") self.overlay.geometry("210x235+15+555") self.overlay.overrideredirect(True) self.overlay.attributes("-topmost", True) self.overlay.attributes("-alpha", 0.88) self.overlay.configure(bg="#101010") def add_label(key, text, font=("Consolas", 8), fg="#00ff88"): var = tk.StringVar(value=text) lbl = tk.Label( self.overlay, textvariable=var, bg="#101010", fg=fg, font=font, anchor="w", justify="left" ) lbl.pack(fill="x", padx=8, pady=1) self.overlay_vars[key] = var return lbl add_label("titulo", "BOT SALUD TOTAL", ("Consolas", 9, "bold"), "#00ff88") add_label("estado", "ESTADO: Listo", ("Consolas", 8, "bold"), "#ffffff") add_label("progreso", "0/0 0%", ("Consolas", 8), "#ffffff") add_label("ok", "OK: 0", ("Consolas", 8), "#00ff88") add_label("err", "ERR: 0 | OMIT: 0", ("Consolas", 8), "#ff6666") add_label("rei", "REI: 0", ("Consolas", 8), "#66ccff") add_label("rad", "RAD: -", ("Consolas", 8), "#ffffff") add_label("vel", "VEL: 0 rad/min", ("Consolas", 8), "#ffff99") add_label("tiempo", "TIME: 00:00:00", ("Consolas", 8), "#ffffff") add_label("ultimo", "ULT: -", ("Consolas", 8), "#cccccc") except Exception as e: print(f"No se pudo crear overlay: {e}") def set_estado_overlay(self, estado, ultimo=None): self.estado_actual = estado if ultimo: self.ultimo_evento = ultimo self.actualizar_overlay() def actualizar_overlay(self): """Actualiza el overlay sin usar Textbox pesado.""" try: if not self.overlay or not self.overlay_vars: return total_procesados = self.procesados_ok + self.errores + getattr(self, "existentes", 0) porcentaje = int((total_procesados / self.total_registros) * 100) if self.total_registros else 0 if self.inicio_proceso: segundos = int(time.time() - self.inicio_proceso) h = segundos // 3600 m = (segundos % 3600) // 60 s = segundos % 60 tiempo_txt = f"{h:02d}:{m:02d}:{s:02d}" minutos = max(segundos / 60, 0.01) velocidad = total_procesados / minutos else: tiempo_txt = "00:00:00" velocidad = 0 self.overlay_vars["estado"].set(f"ESTADO: {self.estado_actual[:18]}") self.overlay_vars["progreso"].set(f"{total_procesados}/{self.total_registros} {porcentaje}%") self.overlay_vars["ok"].set(f"OK: {self.procesados_ok}") self.overlay_vars["err"].set(f"ERR: {self.errores} | EX: {getattr(self, 'existentes', 0)}") self.overlay_vars["rei"].set(f"REI: {self.reinicios}") self.overlay_vars["rad"].set(f"RAD: {str(self.registro_actual)[:18]}") self.overlay_vars["vel"].set(f"VEL: {velocidad:.1f} rad/min") self.overlay_vars["tiempo"].set(f"TIME: {tiempo_txt}") self.overlay_vars["ultimo"].set(f"ULT: {getattr(self, 'ultimo_evento', '-')[:18]}") self.overlay.update_idletasks() except Exception: pass def preparar_reportes_ejecucion(self): """Crea una carpeta única por ejecución para no sobrescribir reportes anteriores.""" self.run_id = datetime.now().strftime("%Y%m%d_%H%M%S") base_dir = os.path.dirname(self.csv_path.get()) if self.csv_path.get() else os.getcwd() self.output_dir = os.path.join(base_dir, REPORTS_DIR_NAME, self.run_id) os.makedirs(self.output_dir, exist_ok=True) def ruta_reporte(self, nombre_base, extension): return os.path.join(self.output_dir, f"{nombre_base}_{self.run_id}.{extension}") def registrar_omitido(self, row, tipo_error, detalle_error, accion_bot="Omitido y reinicio de flujo"): """Registra omitidos con estructura estándar y error exacto.""" radicado = row.get("radicado", "Desconocido") if row else "Desconocido" duracion = int(time.time() - self.inicio_registro) if self.inicio_registro else 0 registro = dict(row) if row else {} registro.update({ "fecha_hora_error": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "estado": "OMITIDO", "tipo_error": tipo_error, "detalle_error": detalle_error, "accion_bot": accion_bot, "duracion_segundos": duracion, }) self.omitted_records.append(registro) self.omitidos += 1 self.errores += 1 self.guardar_resultado( radicado, "ERROR", tipo_error, detalle_error, accion_bot, duracion_segundos=duracion, tipo_error=tipo_error ) self.log(f"OMITIDO: Radicado {radicado} | {tipo_error} | {detalle_error}") self.set_estado_overlay("ERROR", tipo_error) def browse_file(self): filename = filedialog.askopenfilename( title="Seleccionar archivo CSV", filetypes=(("Archivos CSV", "*.csv"), ("Todos los archivos", "*.*")) ) if filename: self.csv_path.set(filename) def limpiar_clipboard(self): """Limpia el portapapeles para evitar leer valores antiguos.""" try: self.root.clipboard_clear() self.root.update() except Exception: pass def leer_clipboard(self): """Lee el texto actual del portapapeles de Windows.""" try: return self.root.clipboard_get().strip() except Exception: return "" def fecha_tiene_numeros(self, texto): """Valida si el campo copiado parece tener una fecha real. Ejemplos: - '09/08/2025' => True - '//' o '' => False """ texto = str(texto).strip() return any(c.isdigit() for c in texto) def fecha_esta_vacia_opencv(self, window): """ Detecta si el campo Fecha del servicio está vacío usando OpenCV + templates. Requiere estos archivos en la carpeta template/: - fecha_vacia_slash.png -> recorte del campo cuando muestra // o __/__/____ - fecha_vacia_blanco.png -> recorte del campo amarillo totalmente vacío Si detecta alguno de los dos, el bot debe digitar la fecha del CSV. Si no detecta ninguno, asume que ya hay una fecha precargada y no la toca. """ if not OPENCV_AVAILABLE: self.log("OpenCV no disponible para detectar fecha. Se asume fecha vacía.") return True detecto_slash = False detecto_blanco = False if os.path.exists(TEMPLATE_FECHA_VACIA_SLASH_PATH): detecto_slash = self.check_template_opencv( window, TEMPLATE_FECHA_VACIA_SLASH_PATH, "fecha_vacia_slash.png" ) else: self.log(f"Template no encontrado: {TEMPLATE_FECHA_VACIA_SLASH_PATH}") if os.path.exists(TEMPLATE_FECHA_VACIA_BLANCO_PATH): detecto_blanco = self.check_template_opencv( window, TEMPLATE_FECHA_VACIA_BLANCO_PATH, "fecha_vacia_blanco.png" ) else: self.log(f"Template no encontrado: {TEMPLATE_FECHA_VACIA_BLANCO_PATH}") if detecto_slash: self.log("Fecha vacía detectada por template slash.") if detecto_blanco: self.log("Fecha vacía detectada por template blanco.") return detecto_slash or detecto_blanco def check_popup_nombres_opencv(self, window): """Detecta el popup azul de búsqueda por nombres/apellidos.""" if not OPENCV_AVAILABLE: return False if not os.path.exists(TEMPLATE_POPUP_NOMBRES_PATH): return False return self.check_template_opencv( window, TEMPLATE_POPUP_NOMBRES_PATH, "popup_nombres.png" ) def check_radicado_existente_opencv(self, window): """Detecta el popup que indica que el radicado ya fue procesado/validado.""" if not OPENCV_AVAILABLE: return False if not os.path.exists(TEMPLATE_RADICADO_EXISTENTE_PATH): return False return self.check_template_opencv( window, TEMPLATE_RADICADO_EXISTENTE_PATH, "radicado_existente.png" ) def cerrar_popup_especial(self, window): """Cierra popups especiales con teclado para evitar depender de coordenadas.""" try: window.wrapper_object().type_keys("{ENTER}", set_foreground=True) time.sleep(0.3) except Exception: pass try: window.wrapper_object().type_keys("{ESC}", set_foreground=True) time.sleep(0.3) except Exception: pass def manejar_radicado_existente(self, window, row, val_radicado): """Registra radicado existente y vuelve al menú seguro sin tratarlo como error técnico.""" duracion = int(time.time() - self.inicio_registro) if self.inicio_registro else 0 self.guardar_resultado( val_radicado, "RADICADO_EXISTENTE", "Radicado previamente procesado", "La aplicación indicó que ya se realizó validación del NIT y valor de la factura.", "Cerró popup y volvió al menú principal", duracion_segundos=duracion, tipo_error="" ) self.existentes += 1 self.log(f"RADICADO EXISTENTE: {val_radicado}. Se registra y vuelve al menú principal.") self.set_estado_overlay("EXISTENTE", "Radicado existente") self.save_reports() self.cerrar_popup_especial(window) # Volver al menú seguro, sin registrar como omitido técnico. time.sleep(0.2) click_abs(window, 366, 172) time.sleep(0.2) click_abs(window, 341, 247) self.first_time = True self.reinicios += 1 self.set_estado_overlay("REINICIO", "Menú principal") def log(self, message): timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") full_msg = f"[{timestamp}] {message}" self.log_entries.append(full_msg) self.log_area.config(state="normal") self.log_area.insert(tk.END, message + "\n") self.log_area.see(tk.END) self.log_area.config(state="disabled") def guardar_resultado(self, radicado, estado, motivo, detalle, accion_bot, duracion_segundos=None, tipo_error=""): """Guarda una trazabilidad del resultado de cada radicado.""" if duracion_segundos is None: duracion_segundos = int(time.time() - self.inicio_registro) if self.inicio_registro else 0 self.resultados.append({ "radicado": radicado, "estado": estado, "tipo_error": tipo_error, "motivo": motivo, "detalle": detalle, "fecha_hora": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "duracion_segundos": duracion_segundos, "accion_bot": accion_bot, }) def save_reports(self): try: os.makedirs(self.output_dir, exist_ok=True) log_filename = self.ruta_reporte("ejecucion_bot", "log") with open(log_filename, "a", encoding="utf-8") as f: for entry in self.log_entries: f.write(entry + "\n") self.log_entries = [] if self.omitted_records: omitted_filename = self.ruta_reporte("OMITIDOS", "csv") fieldnames = [] for registro in self.omitted_records: for key in registro.keys(): if key not in fieldnames: fieldnames.append(key) with open(omitted_filename, "w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() writer.writerows(self.omitted_records) self.log(f"Archivo de omitidos actualizado: {omitted_filename}") if self.resultados: resultados_filename = self.ruta_reporte("RESULTADOS", "csv") with open(resultados_filename, "w", newline="", encoding="utf-8-sig") as f: fieldnames = [ "radicado", "estado", "tipo_error", "motivo", "detalle", "fecha_hora", "duracion_segundos", "accion_bot" ] writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(self.resultados) self.log(f"Archivo de resultados actualizado: {resultados_filename}") except Exception as e: print(f"Error al guardar reportes: {e}") def restart_bot(self, window, row=None, tipo_error="ERROR_VALIDACION", detalle_error="Se detectó una validación/error en pantalla."): if row: self.registrar_omitido( row=row, tipo_error=tipo_error, detalle_error=detalle_error, accion_bot="Omitido y reinicio de flujo" ) self.save_reports() time.sleep(0.05) try: click_abs(window, COORD_CLOSE_ERROR[0], COORD_CLOSE_ERROR[1]) except Exception: pass time.sleep(0.2) click_abs(window, 366, 172) time.sleep(0.2) click_abs(window, 341, 247) self.first_time = True self.reinicios += 1 self.log("Bot reiniciado exitosamente.") self.set_estado_overlay("REINICIO", "Flujo reiniciado") def navegar_menu_inicial(self, window): """Navega al menú principal después de abrir Integración Total.""" self.log("Navegando al menú inicial...") # 1. Seleccionar Cuentas médicas click_abs(window, 67, 167) time.sleep(0.3) # 2. Desplegar Cuentas médicas window.wrapper_object().type_keys("{RIGHT}", set_foreground=True) time.sleep(0.3) # 3. Bajar a Digitación / Digitalización window.wrapper_object().type_keys("{DOWN}", set_foreground=True) time.sleep(0.3) # 4. Desplegar Digitación / Digitalización window.wrapper_object().type_keys("{RIGHT}", set_foreground=True) time.sleep(1) # 5. Clic en Digitar radicación click_abs(window, 385, 177) time.sleep(1) # 6. Clic en Digitación documentos click_abs(window, 338, 234) time.sleep(1) # IMPORTANTE: después de reiniciar, el siguiente registro debe comportarse como primera vez self.first_time = True self.log("Menú listo para radicar.") def reinicio_completo_app(self): """Cierra completamente la app, la abre nuevamente, navega al menú y reconecta pywinauto.""" try: self.reinicios += 1 self.set_estado_overlay("REINICIO APP", "Reinicio completo") self.log("Reinicio completo preventivo de la aplicación...") # Cerrar Integración Total os.system(f'taskkill /f /im "{PROCESS_NAME}"') time.sleep(3) # Abrir desde el acceso directo validado os.startfile(APP_PATH) self.log("Esperando carga de la aplicación...") time.sleep(12) # Reconectar con la ventana nueva app = Application(backend="uia").connect(title_re=".*Integracion Total Fase 2.*") window = app.window(title_re=".*Integracion Total Fase 2.*") window.wait("visible enabled ready", timeout=20) window.wrapper_object().set_focus() # Dejar la app en el menú correcto self.navegar_menu_inicial(window) self.first_time = True self.log("Aplicación reiniciada correctamente.") return app, window except Exception as e: self.log(f"ERROR reiniciando aplicación: {e}") return None, None def convert_date(self, date_str): if not date_str: return "" input_formats = [ "%d/%m/%Y", "%m/%d/%Y", "%Y-%m-%d", "%Y/%m/%d", "%d-%m-%Y", "%m-%d-%Y", ] user_fmt = "%d/%m/%Y" if self.date_format.get() == "dd/mm/yyyy" else "%m/%d/%Y" if user_fmt in input_formats: input_formats.remove(user_fmt) input_formats.insert(0, user_fmt) date_clean = str(date_str).strip() for fmt in input_formats: try: dt_obj = datetime.strptime(date_clean, fmt) return dt_obj.strftime("%m%d%Y") except ValueError: continue self.log(f"ADVERTENCIA: No se pudo convertir la fecha '{date_str}'. Se usará tal cual.") return date_str def limpiar_valor_monetario(self, valor): if valor is None: return 0.0 texto = str(valor).strip() if texto == "": return 0.0 texto = texto.replace("$", "").replace(" ", "") if "." in texto and "," not in texto: partes = texto.split(".") if len(partes[-1]) == 3: texto = texto.replace(".", "") texto = texto.replace(",", ".") try: return float(texto) except Exception: return 0.0 def formatear_valor_para_digitar(self, valor): numero = self.limpiar_valor_monetario(valor) if numero == int(numero): return str(int(numero)) return str(numero).replace(".", ",") def valor_mayor_a_cero(self, valor): return self.limpiar_valor_monetario(valor) > 0 def get_valid_doc_types(self, doc_value): doc_value = str(doc_value).strip() if not doc_value: return [] length = len(doc_value) is_numeric = doc_value.isdigit() valid_types = [] if length in [10, 12]: valid_types.append("AS") if 3 <= length <= 11: valid_types.append("CD") if 3 <= length <= 7 and is_numeric: valid_types.append("CE") if 3 <= length <= 10 and length != 9 and is_numeric: valid_types.append("CC") if 9 <= length <= 14 and is_numeric: valid_types.append("CN") if length in [10, 12]: valid_types.append("ME") if length == 9 and is_numeric: valid_types.append("NI") if length == 9 and is_numeric: valid_types.append("NU") if 3 <= length <= 16: valid_types.append("PS") if length == 15 and is_numeric: valid_types.append("PE") if 7 <= length <= 8 and is_numeric: valid_types.append("PT") if length == 9 and is_numeric: valid_types.append("PJ") if length == 10 and is_numeric: valid_types.append("RC") if 1 <= length <= 8 and is_numeric: valid_types.append("SC") if length == 10 and is_numeric: valid_types.append("TI") return valid_types def check_error_opencv(self, window): if not OPENCV_AVAILABLE: self.log("OpenCV no disponible. Asumiendo OK.") return False if not os.path.exists(TEMPLATE_ERROR_PATH): self.log(f"Template no encontrado en {TEMPLATE_ERROR_PATH}. Asumiendo OK.") return False try: screenshot_path = "temp_check.png" img_cap = window.capture_as_image() img_cap.save(screenshot_path) img_rgb = cv2.imread(screenshot_path) template = cv2.imread(TEMPLATE_ERROR_PATH) if img_rgb is None or template is None: self.log("Error al cargar imágenes para OpenCV.") return False img_gray = cv2.cvtColor(img_rgb, cv2.COLOR_BGR2GRAY) template_gray = cv2.cvtColor(template, cv2.COLOR_BGR2GRAY) res = cv2.matchTemplate(img_gray, template_gray, cv2.TM_CCOEFF_NORMED) threshold = 0.7 loc = np.where(res >= threshold) if len(loc[0]) > 0: self.log("Error detectado mediante Template Matching.") return True return False except Exception as e: self.log(f"Error en chequeo OpenCV: {e}") return False def check_template_opencv(self, window, template_path, nombre="template"): """ Detecta una imagen template en pantalla usando OpenCV. Retorna True si el score supera el umbral. """ if not OPENCV_AVAILABLE: return False if not os.path.exists(template_path): self.log(f"Template no encontrado: {template_path}") return False try: screenshot_path = "temp_check_visual.png" img_cap = window.capture_as_image() img_cap.save(screenshot_path) img_rgb = cv2.imread(screenshot_path) template = cv2.imread(template_path) if img_rgb is None or template is None: self.log(f"No se pudo leer imagen/template: {nombre}") return False img_gray = cv2.cvtColor(img_rgb, cv2.COLOR_BGR2GRAY) template_gray = cv2.cvtColor(template, cv2.COLOR_BGR2GRAY) res = cv2.matchTemplate(img_gray, template_gray, cv2.TM_CCOEFF_NORMED) _, max_val, _, _ = cv2.minMaxLoc(res) self.log(f"Score {nombre}: {max_val:.2f}") return max_val >= 0.60 except Exception as e: self.log(f"Error en chequeo OpenCV de {nombre}: {e}") return False def check_diferencias_abierto_opencv(self, window): """ Detecta si está abierto el panel Diferencias carátula-detalle usando el template flecha_diferencias_abierto.png. """ return self.check_template_opencv( window, TEMPLATE_DIFERENCIAS_PATH, "flecha_diferencias_abierto.png" ) def check_diferencias_cerrado_opencv(self, window): return self.check_template_opencv( window, TEMPLATE_DIFERENCIAS_CERRADO_PATH, "flecha_diferencias_cerrado.png" ) def normalizar_pantalla_radicacion(self, window): try: score_abierto = self.get_template_score( window, TEMPLATE_DIFERENCIAS_PATH, "flecha_diferencias_abierto.png" ) score_cerrado = self.get_template_score( window, TEMPLATE_DIFERENCIAS_CERRADO_PATH, "flecha_diferencias_cerrado.png" ) self.log(f"Score abierto: {score_abierto:.2f} | Score cerrado: {score_cerrado:.2f}") if score_abierto > score_cerrado and score_abierto >= 0.60: self.log("Panel confirmado ABIERTO por score. Cerrando...") click_abs(window, 558, 192) time.sleep(1) else: self.log("Panel confirmado CERRADO o incierto. No se toca.") except Exception as e: self.log(f"Error normalizando: {e}") def get_template_score(self, window, template_path, nombre="template"): if not OPENCV_AVAILABLE or not os.path.exists(template_path): return 0.0 try: screenshot_path = "temp_score_visual.png" img_cap = window.capture_as_image() img_cap.save(screenshot_path) img_rgb = cv2.imread(screenshot_path) template = cv2.imread(template_path) if img_rgb is None or template is None: return 0.0 img_gray = cv2.cvtColor(img_rgb, cv2.COLOR_BGR2GRAY) template_gray = cv2.cvtColor(template, cv2.COLOR_BGR2GRAY) res = cv2.matchTemplate(img_gray, template_gray, cv2.TM_CCOEFF_NORMED) _, max_val, _, _ = cv2.minMaxLoc(res) return max_val except Exception: return 0.0 def check_success_opencv(self, window): return self.check_template_opencv( window, TEMPLATE_SUCCESS_PATH, "success.png" ) def check_bien_opencv(self, window): return self.check_template_opencv( window, TEMPLATE_BIEN_PATH, "bien.png" ) def esperar_resultado_registro(self, window, timeout=TIMEOUT_REGISTRO): """ Espera hasta que la app responda después de registrar. Retorna: OK, BIEN, ERROR o TIMEOUT. IMPORTANTE: Se deja un tiempo de gracia inicial para evitar falsos positivos, porque durante la carga la pantalla se oscurece y puede parecerse visualmente al template de error. """ self.log(f"Esperando respuesta de registro hasta {timeout} segundos...") if not os.path.exists(TEMPLATE_SUCCESS_PATH): self.log(f"ADVERTENCIA: No existe {TEMPLATE_SUCCESS_PATH}. No se podrá detectar success.png.") if not os.path.exists(TEMPLATE_BIEN_PATH): self.log(f"ADVERTENCIA: No existe {TEMPLATE_BIEN_PATH}. No se podrá detectar bien.png.") # Ventana de gracia: no detectar error mientras la pantalla está cargando. self.log("Tiempo de gracia inicial para evitar falso positivo de error (4s)...") time.sleep(4) inicio = time.time() while time.time() - inicio < timeout: if self.check_success_opencv(window): return "OK" if self.check_bien_opencv(window): return "BIEN" if self.check_error_opencv(window): return "ERROR" time.sleep(1) return "TIMEOUT" def start_thread(self): if not self.csv_path.get(): messagebox.showwarning("Error", "Por favor seleccione un archivo CSV primero.") return if self.is_running: return self.is_running = True self.btn_start.config(state="disabled", text="Ejecutando...") threading.Thread(target=self.run_process, daemon=True).start() def run_process(self): try: self.log("Conectando a la aplicación...") app = Application(backend="uia").connect(title_re=".*Integracion Total Fase 2.*") window = app.window(title_re=".*Integracion Total Fase 2.*") self.preparar_reportes_ejecucion() self.omitted_records = [] self.log_entries = [] self.resultados = [] self.procesados_ok = 0 self.errores = 0 self.omitidos = 0 self.existentes = 0 self.reinicios = 0 self.registro_actual = "-" self.estado_actual = "Conectando" self.ultimo_evento = "Inicio" self.inicio_proceso = time.time() self.first_time = True self.actualizar_overlay() self.log(f"Iniciando nuevo proceso. Reportes en: {self.output_dir}") try: window.wait("visible enabled ready", timeout=5) except Exception as e: self.log(f"Advertencia en navegación inicial: {e}") data = [] with open(self.csv_path.get(), newline="", encoding="utf-8") as csvfile: reader = csv.DictReader(csvfile, delimiter=self.delimiter.get()) columnas_requeridas = { "radicado", "documento", "ingreso", "total_cargos", "autorizacion", "tipo_documento", "tipo_servicio", "copago", "cuota_moderadora", } headers = [h.strip() for h in reader.fieldnames] if reader.fieldnames else [] reader.fieldnames = headers if not columnas_requeridas.issubset(headers): missing = columnas_requeridas - set(headers) raise ValueError( "El CSV no tiene las columnas requeridas.\n" f"Faltan: {', '.join(missing)}\n" "Verifique si el delimitador es correcto." ) for row in reader: data.append(row) self.total_registros = len(data) self.actualizar_overlay() self.log(f"Se encontraron {len(data)} registros para procesar.") for i, row in enumerate(data): self.inicio_registro = time.time() self.registro_actual = row.get('radicado', 'N/A') self.set_estado_overlay("PROCESANDO", f"Registro {i + 1}") self.log(f"--- Procesando registro {i + 1}/{len(data)}: {row.get('radicado', 'N/A')} ---") val_radicado = row.get("radicado", "").strip() val_total_cargos = row.get("total_cargos", "0").strip() val_identificacion = row.get("documento", "").strip() val_fecha = row.get("ingreso", "").strip() val_auth = row.get("autorizacion", "").strip() val_tipo_doc = row.get("tipo_documento", "").strip().upper() val_service = row.get("tipo_servicio", "").strip().lower() val_copago = row.get("copago", "").strip() val_cuota_mod = row.get("cuota_moderadora", "").strip() val_fecha_final = self.convert_date(val_fecha) click_abs(window, 926, 309) time.sleep(0.2) window.wrapper_object().type_keys( val_radicado, with_spaces=True, set_foreground=True ) time.sleep(0.2) click_abs(window, 960, 531) # Validación inmediata después del botón de búsqueda del radicado time.sleep(0.5) if self.check_radicado_existente_opencv(window): self.manejar_radicado_existente(window, row, val_radicado) continue if self.check_error_opencv(window): self.log("Error al procesar el registro.") self.restart_bot(window, row, "ERROR_RADICADO", "Error visual después de consultar el radicado.") continue time.sleep(0.2) click_abs(window, 1375, 305) time.sleep(0.2) click_abs(window, 1041, 329) if val_service: if val_service in SERVICES_COORDS: self.log(f"Tipo de servicio: {val_service}") time.sleep(0.2) click_abs( window, SERVICES_COORDS[val_service][0], SERVICES_COORDS[val_service][1] ) else: self.log(f"ERROR: El tipo de servicio '{val_service}' no es válido.") self.restart_bot(window, row, "ERROR_TIPO_SERVICIO", f"Tipo de servicio no configurado: {val_service}") continue else: self.log("ERROR: El campo 'tipo_servicio' está vacío.") self.restart_bot(window, row, "ERROR_TIPO_SERVICIO", "El campo tipo_servicio está vacío.") continue # IMPORTANTE: # Antes aquí se hacía clic en "Cuota a cobrar" y se digitaba un 0. # Eso dañaba el valor/foco cuando la aplicación ya traía la cuota o cuando # el campo no debía tocarse. Ahora NO se escribe nada en cuota a cobrar. # Después de seleccionar el tipo de servicio, se ejecuta la búsqueda directamente. time.sleep(0.2) click_abs(window, 1300, 302) self.log("Esperando carga (7s)...") time.sleep(7) if self.check_error_opencv(window): self.log("Error al procesar el registro.") self.restart_bot(window, row, "ERROR_CARGA_DETALLE", "Error visual después de cargar el detalle del radicado.") continue self.normalizar_pantalla_radicacion(window) time.sleep(0.2) click_abs(window, 730, 321) time.sleep(0.2) click_abs(window, 700, 359) candidatos_validos = self.get_valid_doc_types(val_identificacion) candidatos_ordenados = [t for t in DOC_STRATEGY if t in candidatos_validos] tipos_a_probar = [] if val_tipo_doc and val_tipo_doc in DOC_COORDS: tipos_a_probar.append(val_tipo_doc) for t in candidatos_ordenados: if t != val_tipo_doc: tipos_a_probar.append(t) else: if not candidatos_ordenados: self.log( f"ADVERTENCIA: El documento {val_identificacion} " "no cumple reglas conocidas. Se probará estrategia completa." ) tipos_a_probar = DOC_STRATEGY else: tipos_a_probar = candidatos_ordenados self.log(f"Estrategia de documentos a probar: {tipos_a_probar}") documento_exitoso = False error_documento_fatal = False for tipo_doc in tipos_a_probar: self.log(f"Probando tipo de documento: {tipo_doc}") time.sleep(1) click_abs(window, 729, 279) coords_doc = DOC_COORDS.get(tipo_doc, (0, 0)) if coords_doc == (0, 0): self.log(f"Advertencia: Coordenadas para {tipo_doc} no configuradas. Saltando.") continue time.sleep(0.5) click_abs(window, coords_doc[0], coords_doc[1]) time.sleep(0.2) click_abs(window, 1015, 280) time.sleep(0.2) window.wrapper_object().type_keys("^a{DELETE}", set_foreground=True) time.sleep(0.1) window.wrapper_object().type_keys( val_identificacion, with_spaces=True, set_foreground=True ) # ========================= # FECHA SERVICIO INTELIGENTE POR OPENCV # ========================= # La fecha puede venir precargada. Para no sobrescribirla: # - Si detecta template de fecha vacía (slash o blanco), digita la fecha del CSV. # - Si NO detecta fecha vacía, asume que ya hay fecha cargada y no la toca. time.sleep(0.05) click_abs(window, 1172, 276) time.sleep(0.05) if self.fecha_esta_vacia_opencv(window): self.log(f"Fecha vacía detectada por OpenCV. Digitando fecha CSV: {val_fecha_final}") time.sleep(0.1) window.wrapper_object().type_keys( val_fecha_final, pause=0.1, with_spaces=False, set_foreground=True ) time.sleep(0.1) window.wrapper_object().type_keys("{TAB}{ENTER}", set_foreground=True) else: self.log("Fecha precargada detectada por OpenCV. No se modifica.") time.sleep(0.1) window.wrapper_object().type_keys("{TAB}{ENTER}", set_foreground=True) time.sleep(1) if self.check_popup_nombres_opencv(window): self.log("Popup de búsqueda por nombres detectado. Se omite el registro.") self.cerrar_popup_especial(window) self.restart_bot( window, row, "ERROR_BUSQUEDA_DOCUMENTO", "La aplicación abrió el popup de búsqueda por nombres/apellidos al validar el documento." ) error_documento_fatal = True break if self.check_error_opencv(window): self.log(f"Error detectado con {tipo_doc}. Intentando siguiente...") click_abs(window, COORD_CLOSE_ERROR[0], COORD_CLOSE_ERROR[1]) time.sleep(1) continue else: self.log("Validación OpenCV exitosa.") documento_exitoso = True break if error_documento_fatal: continue if not documento_exitoso: self.log( f"ERROR: No se pudo procesar el documento {val_identificacion} " "con ningún tipo. Omitiendo y reiniciando bot." ) self.restart_bot(window, row, "ERROR_DOCUMENTO", f"No validó documento {val_identificacion} con tipos probados: {', '.join(tipos_a_probar)}") continue #posible clic de antes de autorizar (xd) time.sleep(0.6) click_abs(window, 1126, 352) time.sleep(0.2) window.wrapper_object().type_keys( val_auth, with_spaces=True, set_foreground=True ) time.sleep(0.2) window.wrapper_object().type_keys("{TAB}{ENTER}", set_foreground=True) if self.check_error_opencv(window): self.log("Error al procesar el registro.") self.restart_bot(window, row, "ERROR_AUTORIZACION", "Error visual después de ingresar autorización.") continue time.sleep(0.3) click_abs(window, 741, 421) time.sleep(0.2) window.wrapper_object().type_keys("{SPACE}", set_foreground=True) time.sleep(0.3) click_abs(window, 740, 535) # ========================= # TOTAL CARGOS + COPAGO/MODERADORA EXCLUYENTES + AGREGAR # ========================= time.sleep(0.3) self.log(f"Digitando total_cargos: {val_total_cargos}") click_abs(window, 1105, 658) time.sleep(0.2) window.wrapper_object().type_keys( val_total_cargos, with_spaces=True, set_foreground=True ) copago_txt = self.formatear_valor_para_digitar(val_copago) cuota_mod_txt = self.formatear_valor_para_digitar(val_cuota_mod) copago_mayor = self.valor_mayor_a_cero(val_copago) cuota_mod_mayor = self.valor_mayor_a_cero(val_cuota_mod) if copago_mayor: self.log(f"Digitando copago: {copago_txt}") # 2 TAB → Vlr Copago time.sleep(0.3) window.wrapper_object().type_keys("{TAB}{TAB}", set_foreground=True) time.sleep(0.3) window.wrapper_object().type_keys( copago_txt, with_spaces=True, set_foreground=True ) # Como cuota moderadora se bloquea, NO se toca. # 3 TAB → botón Agregar, ENTER time.sleep(0.3) window.wrapper_object().type_keys("{TAB}{TAB}{TAB}", set_foreground=True) time.sleep(0.3) window.wrapper_object().type_keys("{ENTER}", set_foreground=True) elif cuota_mod_mayor: self.log(f"Digitando cuota moderadora: {cuota_mod_txt}") # 2 TAB → Vlr Copago time.sleep(0.3) window.wrapper_object().type_keys("{TAB}{TAB}", set_foreground=True) # 1 TAB → Vlr Cuota Mod time.sleep(0.3) window.wrapper_object().type_keys("{TAB}", set_foreground=True) time.sleep(0.3) window.wrapper_object().type_keys( cuota_mod_txt, with_spaces=True, set_foreground=True ) # 2 TAB → botón Agregar, ENTER time.sleep(0.3) window.wrapper_object().type_keys("{TAB}{TAB}", set_foreground=True) time.sleep(0.3) window.wrapper_object().type_keys("{ENTER}", set_foreground=True) else: self.log("Copago y cuota moderadora en 0. Va directo a agregar.") # total_cargos → 2 TAB copago → 1 TAB moderadora → 2 TAB agregar time.sleep(0.2) window.wrapper_object().type_keys("{TAB}{TAB}{TAB}{TAB}{TAB}", set_foreground=True) time.sleep(0.2) window.wrapper_object().type_keys("{ENTER}", set_foreground=True) self.log("Esperando confirmación visual antes de continuar al siguiente radicado...") resultado_registro = self.esperar_resultado_registro(window, TIMEOUT_REGISTRO) if resultado_registro in ("OK", "BIEN"): self.log(f"Registro exitoso/confirmación detectada: {resultado_registro}.") self.guardar_resultado( val_radicado, "OK", "Cuenta registrada", f"Mensaje visual detectado en pantalla: {resultado_registro}.", "Continuó" ) # Mantiene los clics originales para cerrar confirmación y volver al menú seguro. click_abs(window, 1229, 979) time.sleep(1) click_abs(window, 1223, 666) time.sleep(0.5) click_abs(window, 941, 305) self.procesados_ok += 1 self.set_estado_overlay("OK", "Registro OK") self.log(f"Registro {val_radicado} completado.") time.sleep(1) self.first_time = False elif resultado_registro == "ERROR": self.log("Error detectado después de registrar. Se omite el registro.") self.restart_bot(window, row, "ERROR_REGISTRO", "Error visual después de presionar agregar/registrar.") continue else: self.log("TIMEOUT: No apareció éxito ni error después de registrar.") self.errores += 1 self.guardar_resultado( val_radicado, "TIMEOUT", "ERROR_TIMEOUT", f"No apareció mensaje de éxito ni error en {TIMEOUT_REGISTRO} segundos.", "Reinicio completo y continúa", tipo_error="ERROR_TIMEOUT" ) self.set_estado_overlay("TIMEOUT", "Sin respuesta") self.save_reports() app, window = self.reinicio_completo_app() if window is None: self.log("No se pudo reiniciar después del timeout. Deteniendo proceso.") break continue # Reinicio completo preventivo para evitar lentitud acumulada de la app. # No se ejecuta si ya era el último registro. if (i + 1) % REINICIO_CADA == 0 and (i + 1) < len(data): app, window = self.reinicio_completo_app() if window is None: self.log("No se pudo reiniciar la aplicación. Deteniendo proceso.") break self.set_estado_overlay("FINALIZADO", "Proceso finalizado") self.log("--- PROCESO FINALIZADO ---") self.save_reports() messagebox.showinfo("Fin", "El proceso ha terminado exitosamente.") except Exception as e: self.log(f"ERROR: {str(e)}") messagebox.showerror("Error", f"Ocurrió un error:\n{str(e)}") finally: self.is_running = False try: self.btn_start.config(state="normal", text="INICIAR BOT") except Exception: pass if __name__ == "__main__": root = tk.Tk() app = BotSaludTotal(root) root.mainloop()