1396 lines
51 KiB
Python
1396 lines
51 KiB
Python
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()
|