ldiaz/Agent-IA
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add whatsapp_modular_v3_6 - envio automatizado WhatsApp Web con busqueda por nombre, deteccion de ticks via msg-meta/SVG title, y validacion de selectores

Browse Source
This commit is contained in:
ldiaz
2026-06-25 14:25:59 -05:00
parent 3d42e4b84b
commit c1bcd57c4d
15 changed files with 1627 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
Python/whatsapp_modular_v3_6/whatsapp_sender/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Automatizacion modular de envios por WhatsApp Web."""

23
Python/whatsapp_modular_v3_6/whatsapp_sender/config.py Normal file
View File

@@ -0,0 +1,23 @@
CHROMEDRIVER_PATH = r"C:\MensajeriaW_CES\chromedriver-win64\chromedriver.exe"
CHROME_BINARY = r"C:\MensajeriaW_CES\chrome-win64\chrome-win64\chrome.exe"
MENSAJES_FILE = "mensajes.csv"
LOG_FILE = "log_envios.txt"
RESULTADOS_FILE = "resultados_envio.csv"
EVIDENCIAS_DIR = "evidencias"
DIAGNOSTICOS_DIR = "diagnosticos"
ESTADO_FILE = "estado_sesion.txt"
MODO_HEADLESS = False
WAIT_CHAT = 10
MAX_REINTENTOS = 2
TIMEOUT_SESION = 60
PAUSA_ENTRE_FILAS = 2
MINUTOS_ANTICIPACION = 1
REINTENTOS_RECONEXION = 2
REINTENTO_RELOAD_WAIT = 5
CLEAN_OLD_SESSIONS_DAYS = 30
IDIOMA_INTERFACE = "auto"
REQUIRED_COLUMNS = {"tipo", "id_destino", "mensaje", "hora", "minuto"}

19
Python/whatsapp_modular_v3_6/whatsapp_sender/dependencies.py Normal file
View File

@@ -0,0 +1,19 @@
import subprocess
import sys
DEPENDENCIAS = ["pandas", "openpyxl", "selenium", "pyautogui", "pyperclip"]
def instalar_paquete_si_falta(paquete):
try:
__import__(paquete)
except ImportError:
print(f"Instalando '{paquete}'...")
subprocess.check_call([sys.executable, "-m", "pip", "install", paquete, "--quiet"])
print(f"'{paquete}' instalado.")
def asegurar_dependencias():
for paquete in DEPENDENCIAS:
instalar_paquete_si_falta(paquete)

210
Python/whatsapp_modular_v3_6/whatsapp_sender/driver_session.py Normal file
View File

@@ -0,0 +1,210 @@
import time
from selenium import webdriver
from selenium.common.exceptions import NoSuchElementException
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import WebDriverWait
from .config import (
CHROME_BINARY,
CHROMEDRIVER_PATH,
CLEAN_OLD_SESSIONS_DAYS,
DIAGNOSTICOS_DIR,
IDIOMA_INTERFACE,
MODO_HEADLESS,
REINTENTO_RELOAD_WAIT,
REINTENTOS_RECONEXION,
TIMEOUT_SESION,
)
from .utils import asegurar_directorio, cleanup_old_chrome_data, log_text, save_session_state
def hay_elemento_visible(driver, by, selector):
try:
elementos = driver.find_elements(by, selector)
except Exception:
return False
for elemento in elementos:
try:
if elemento.is_displayed():
return True
except Exception:
continue
return False
def iniciar_driver():
options = Options()
options.binary_location = CHROME_BINARY
argumentos = [
"--user-data-dir=./perfil_whatsapp",
"--profile-directory=Default",
"--no-sandbox",
"--disable-dev-shm-usage",
"--disable-gpu",
"--disable-software-rasterizer",
"--disable-blink-features=AutomationControlled",
"--disable-extensions",
"--disable-infobars",
"--remote-debugging-port=9222",
"--window-size=1920,1080",
"--log-level=3",
"--start-maximized",
"--disable-features=RendererCodeIntegrity",
"--disable-web-security",
"--disable-session-crashed-bubble",
"--disable-popup-blocking",
"--disable-notifications",
]
for argumento in argumentos:
options.add_argument(argumento)
if MODO_HEADLESS:
options.add_argument("--headless=new")
options.add_experimental_option(
"prefs",
{
"credentials_enable_service": False,
"profile.password_manager_enabled": False,
},
)
options.add_experimental_option("detach", True)
options.add_experimental_option("excludeSwitches", ["enable-automation"])
options.add_experimental_option("useAutomationExtension", False)
cleanup_old_chrome_data(dirpath="./perfil_whatsapp", days=CLEAN_OLD_SESSIONS_DAYS)
return webdriver.Chrome(service=Service(CHROMEDRIVER_PATH), options=options)
def construir_condiciones_xpath(idioma):
textos_chat = {
"es": ["Lista de chats", "Lista de conversaciones", "Chats", "Conversaciones"],
"en": ["Chat list", "Chats", "Conversation list"],
"pt": ["Lista de conversas", "Conversas"],
}
if idioma == "auto":
return sum(textos_chat.values(), [])
return textos_chat.get(idioma, textos_chat["en"])
def detectar_qr(driver):
qr_selectores = [
"canvas[aria-label*='Scan']",
"canvas[aria-label*='escan' i]",
"canvas[aria-label='Scan me!']",
"div[data-ref] canvas",
]
for selector in qr_selectores:
try:
qr = driver.find_element(By.CSS_SELECTOR, selector)
if qr.is_displayed():
return True
except NoSuchElementException:
continue
except Exception:
continue
return False
def esperar_sesion_activa(driver, tiempo_maximo=TIMEOUT_SESION, idioma=IDIOMA_INTERFACE):
print("Verificando sesion activa de WhatsApp Web...")
inicio = time.time()
textos = construir_condiciones_xpath(idioma)
condiciones = " or ".join([f"contains(@aria-label, '{text}')" for text in textos])
xpath_combinado = f"//div[{condiciones}] | //div[@data-testid='chat-list']"
selectores_sesion = [
(By.ID, "pane-side"),
(By.ID, "side"),
(By.CSS_SELECTOR, "[data-testid='chat-list']"),
(By.CSS_SELECTOR, "[data-testid='cell-frame-container']"),
(By.CSS_SELECTOR, "div[role='grid']"),
(By.CSS_SELECTOR, "div[aria-label*='chat' i][role='grid']"),
(By.CSS_SELECTOR, "div[contenteditable='true'][role='textbox'][aria-label*='buscar' i]"),
(By.CSS_SELECTOR, "div[contenteditable='true'][role='textbox'][aria-label*='search' i]"),
]
while time.time() - inicio < tiempo_maximo:
try:
elems = driver.find_elements(By.XPATH, xpath_combinado)
for elem in elems:
try:
if elem.is_displayed():
print("Sesion activa detectada.")
save_session_state(True)
return True
except Exception:
continue
for by, selector in selectores_sesion:
if hay_elemento_visible(driver, by, selector):
print("Sesion activa detectada.")
save_session_state(True)
return True
if detectar_qr(driver):
print("QR detectado. Escanea el QR en el navegador...")
time.sleep(2)
continue
try:
WebDriverWait(driver, 3).until(
EC.presence_of_element_located(
(
By.CSS_SELECTOR,
"div[contenteditable='true'][role='textbox'], div[contenteditable='true'][data-tab]",
)
)
)
print("Interfaz detectada.")
save_session_state(True)
return True
except Exception:
pass
except Exception as exc:
print(f"Error verificando sesion: {exc}")
time.sleep(1)
print("No se detecto sesion activa.")
save_session_state(False)
try:
asegurar_directorio(DIAGNOSTICOS_DIR)
nombre = f"{DIAGNOSTICOS_DIR}/sesion_fallida_{int(time.time())}.png"
driver.save_screenshot(nombre)
print(f"Captura diagnostica guardada: {nombre}")
log_text(f"SESION_FALLIDA captura:{nombre}")
except Exception as exc:
print(f"No se pudo guardar captura diagnostica: {exc}")
return False
def ensure_session_active(driver, retries=REINTENTOS_RECONEXION):
if esperar_sesion_activa(driver):
return True
for intento in range(1, retries + 1):
try:
print(f"Intento reconexion {intento}/{retries}")
driver.get("https://web.whatsapp.com")
time.sleep(REINTENTO_RELOAD_WAIT)
if esperar_sesion_activa(driver, tiempo_maximo=TIMEOUT_SESION):
print("Sesion recuperada.")
return True
except Exception as exc:
print(f"Error reconectando: {exc}")
time.sleep(2)
print("No se pudo recuperar la sesion.")
return False

94
Python/whatsapp_modular_v3_6/whatsapp_sender/main.py Normal file
View File

@@ -0,0 +1,94 @@
import time
from .dependencies import asegurar_dependencias
asegurar_dependencias()
import pandas as pd
from .config import MENSAJES_FILE, PAUSA_ENTRE_FILAS, RESULTADOS_FILE
from .driver_session import ensure_session_active, iniciar_driver
from .messages import cargar_mensajes
from .processor import procesar_row
from .utils import log_text
def actualizar_contadores(estado, nota, contadores):
if estado == "NO_ENCONTRADO":
contadores["no_encontrados"] += 1
elif estado in ["FALLIDO", "ERROR_ENVIO"]:
contadores["fallidos"] += 1
elif estado == "ERROR_SESSION":
contadores["error_sesion"] += 1
elif estado.startswith(("ENVIADO", "ENTREGADO", "LEIDO", "PENDIENTE")) or nota.upper().startswith("ENVIADO"):
contadores["exitos"] += 1
else:
contadores["fallidos"] += 1
def main():
print("Inicio del script v3.6 modular (reconexion + validacion de envio).")
try:
df = cargar_mensajes(MENSAJES_FILE)
except Exception as exc:
print(f"Error cargando archivo de mensajes: {exc}")
return
total = len(df)
print(f"{total} filas validas cargadas desde '{MENSAJES_FILE}'")
if total == 0:
print("No hay filas validas (tipo 'contacto' o 'grupo'). Saliendo.")
return
driver = iniciar_driver()
driver.get("https://web.whatsapp.com")
if not ensure_session_active(driver):
print("No se detecto sesion activa. Abre WhatsApp Web y escanea el QR en la ventana del navegador.")
try:
driver.quit()
except Exception:
pass
return
resultados = []
contadores = {"exitos": 0, "fallidos": 0, "no_encontrados": 0, "error_sesion": 0}
for idx, row in df.iterrows():
print(f"\n--- Procesando fila {idx + 1}/{total} ---")
resultado = procesar_row(driver, row)
resultados.append(resultado)
actualizar_contadores(resultado.get("estado", ""), resultado.get("nota", ""), contadores)
time.sleep(PAUSA_ENTRE_FILAS)
pct = (idx + 1) / total * 100
print(
f"Progreso: {idx + 1}/{total} ({pct:.1f}%) - "
f"Exitos: {contadores['exitos']} - "
f"Fallidos: {contadores['fallidos']} - "
f"No encontrados: {contadores['no_encontrados']} - "
f"ErrSes: {contadores['error_sesion']}"
)
pd.DataFrame(resultados).to_csv(RESULTADOS_FILE, index=False, encoding="utf-8-sig")
print(f"\nResultados guardados en '{RESULTADOS_FILE}'")
log_text(
f"FIN: {total} procesados. Exitos: {contadores['exitos']}. "
f"Fallidos: {contadores['fallidos']}. NoEncontrados: {contadores['no_encontrados']}. "
f"ErrSes: {contadores['error_sesion']}"
)
print("\nRESUMEN FINAL")
print(f"Total filas: {total}")
print(f"Exitos: {contadores['exitos']}")
print(f"Fallidos: {contadores['fallidos']}")
print(f"No encontrados: {contadores['no_encontrados']}")
print(f"Errores de sesion: {contadores['error_sesion']}")
if total:
print(f"Porcentaje exitos: {contadores['exitos'] / total * 100:.1f}%")
try:
driver.quit()
except Exception:
pass

23
Python/whatsapp_modular_v3_6/whatsapp_sender/messages.py Normal file
View File

@@ -0,0 +1,23 @@
import os
import pandas as pd
from .config import REQUIRED_COLUMNS
def cargar_mensajes(path):
ext = os.path.splitext(path)[1].lower()
if ext == ".csv":
df = pd.read_csv(path, dtype=str)
elif ext in [".xls", ".xlsx"]:
df = pd.read_excel(path, dtype=str)
else:
raise ValueError("Formato no soportado. Usa .csv o .xlsx")
df.columns = [column.strip() for column in df.columns]
missing = REQUIRED_COLUMNS - set(df.columns)
if missing:
raise ValueError(f"Columnas faltantes en '{path}': {missing}")
df["tipo"] = df["tipo"].str.strip().str.lower()
return df[df["tipo"].isin(["contacto", "grupo"])].reset_index(drop=True)

133
Python/whatsapp_modular_v3_6/whatsapp_sender/processor.py Normal file
View File

@@ -0,0 +1,133 @@
import os
import time
from datetime import datetime, timedelta
from .config import EVIDENCIAS_DIR, MAX_REINTENTOS, MINUTOS_ANTICIPACION, WAIT_CHAT
from .driver_session import ensure_session_active
from .utils import asegurar_directorio, log_text
from .whatsapp import abrir_chat, enviar_mensaje, obtener_nombre_chat_actual, validate_sent, verificar_ticks
def construir_resultado(tipo, destino, mensaje, hora, minuto):
return {
"tipo": tipo,
"destino": destino,
"mensaje": mensaje,
"hora_programada": f"{hora if hora is not None else ''}:{minuto if minuto is not None else ''}",
"estado": "NO_INTENTADO",
"intento_exitoso": None,
"timestamp": None,
"captura": None,
"nota": "",
}
def obtener_hora_minuto(row):
try:
return int(float(row["hora"])), int(float(row["minuto"]))
except Exception:
return None, None
def esperar_hora_programada(hora, minuto, destino):
if hora is None or minuto is None:
return
ahora = datetime.now()
hora_envio = datetime(ahora.year, ahora.month, ahora.day, hora, minuto)
if hora_envio <= ahora + timedelta(minutes=MINUTOS_ANTICIPACION):
hora_envio = ahora + timedelta(minutes=2)
espera = (hora_envio - datetime.now()).total_seconds()
if espera > 0:
print(f"Esperando {int(espera)}s hasta {hora_envio.strftime('%H:%M')} para destino {destino}")
time.sleep(espera)
def guardar_captura_evidencia(driver, destino):
nombre_chat = obtener_nombre_chat_actual(driver)
asegurar_directorio(EVIDENCIAS_DIR)
safe = "".join(c for c in nombre_chat if c.isalnum() or c in (" ", "_", "-")).strip().replace(" ", "_")[:40]
if nombre_chat == "chat_desconocido":
nombre_captura = os.path.join(EVIDENCIAS_DIR, f"envio_{destino[:12]}_{int(time.time())}.png")
else:
nombre_captura = os.path.join(EVIDENCIAS_DIR, f"envio_{safe}_{int(time.time())}.png")
try:
driver.save_screenshot(nombre_captura)
return nombre_captura
except Exception:
return None
def procesar_row(driver, row):
tipo = str(row["tipo"]).strip().lower()
destino = str(row["id_destino"]).strip()
mensaje = str(row["mensaje"])
hora, minuto = obtener_hora_minuto(row)
resultado = construir_resultado(tipo, destino, mensaje, hora, minuto)
esperar_hora_programada(hora, minuto, destino)
for intento in range(1, MAX_REINTENTOS + 1):
print(f"Intento {intento}/{MAX_REINTENTOS} -> {tipo} '{destino}'")
if not ensure_session_active(driver):
resultado["estado"] = "ERROR_SESSION"
resultado["nota"] = "No se pudo recuperar sesion"
resultado["timestamp"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
log_text(f"{destino} | ERROR_SESSION")
return resultado
if not abrir_chat(driver, tipo, destino):
resultado["estado"] = "NO_ENCONTRADO"
resultado["nota"] = "No se encontro chat/grupo"
resultado["timestamp"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
log_text(f"{destino} | NO_ENCONTRADO")
return resultado
try:
time.sleep(WAIT_CHAT)
if not enviar_mensaje(driver, mensaje):
raise Exception("Error al ejecutar accion de envio")
time.sleep(1)
validacion = validate_sent(driver, mensaje, timeout=15)
if validacion is True:
resultado["estado"] = verificar_ticks(driver)
resultado["nota"] = "ENVIADO_CONFIRMED"
elif validacion is None:
print("La sesion se perdio durante validacion. Intentando recuperacion...")
if ensure_session_active(driver):
print("Sesion recuperada. Reintentando envio.")
time.sleep(2)
continue
resultado["estado"] = "ERROR_SESSION"
resultado["nota"] = "Sesion perdida y no recuperada"
resultado["timestamp"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
log_text(f"{destino} | ERROR_SESSION durante validacion")
return resultado
else:
resultado["estado"] = "ENVIADO (SIN CONFIRMACION)"
resultado["nota"] = "ENVIADO_SIN_CONFIRMACION"
resultado["captura"] = guardar_captura_evidencia(driver, destino)
resultado["intento_exitoso"] = intento
resultado["timestamp"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
log_text(f"{destino} | {resultado['estado']} | intento {intento} | captura:{resultado['captura']}")
print(f"Resultado para {destino}: {resultado['estado']}")
return resultado
except Exception as exc:
print(f"Error en intento {intento} con {destino}: {exc}")
log_text(f"{destino} | ERROR intento {intento}: {exc}")
resultado["nota"] = str(exc)
if intento < MAX_REINTENTOS:
print("Reintentando en 3s...")
time.sleep(3)
resultado["estado"] = "FALLIDO"
resultado["timestamp"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
log_text(f"{destino} | FALLIDO despues de {MAX_REINTENTOS} intentos")
return resultado

46
Python/whatsapp_modular_v3_6/whatsapp_sender/utils.py Normal file
View File

@@ -0,0 +1,46 @@
import os
import shutil
from datetime import datetime, timedelta
from pathlib import Path
from .config import CLEAN_OLD_SESSIONS_DAYS, ESTADO_FILE, LOG_FILE
def log_text(line):
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
with open(LOG_FILE, "a", encoding="utf-8") as file:
file.write(f"{ts} | {line}\n")
def asegurar_directorio(path):
os.makedirs(path, exist_ok=True)
def save_session_state(active):
try:
with open(ESTADO_FILE, "w", encoding="utf-8") as file:
file.write("SESION_ACTIVA" if active else "SESION_INACTIVA")
except Exception as exc:
print(f"No se pudo guardar estado de sesion: {exc}")
def cleanup_old_chrome_data(dirpath="./perfil_whatsapp", days=CLEAN_OLD_SESSIONS_DAYS):
if days <= 0:
return
try:
base = Path(dirpath)
if not base.exists():
return
cutoff = datetime.now() - timedelta(days=days)
for child in base.iterdir():
try:
mtime = datetime.fromtimestamp(child.stat().st_mtime)
if mtime < cutoff and child.is_dir():
shutil.rmtree(child)
print(f"Eliminada sesion antigua: {child}")
except Exception:
continue
except Exception as exc:
print(f"Error limpiando sesiones antiguas: {exc}")

434
Python/whatsapp_modular_v3_6/whatsapp_sender/whatsapp.py Normal file
View File

@@ -0,0 +1,434 @@
import re
import time
import pyperclip
from selenium.common.exceptions import StaleElementReferenceException, TimeoutException, WebDriverException
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import WebDriverWait
def es_numero_telefono(destino):
return bool(re.fullmatch(r"\+?\d{7,15}", destino.strip()))
def elementos_visibles(driver, by, selector):
try:
elementos = driver.find_elements(by, selector)
except Exception:
return []
visibles = []
for elemento in elementos:
try:
if elemento.is_displayed():
visibles.append(elemento)
except (StaleElementReferenceException, WebDriverException):
continue
return visibles
def primer_visible(driver, by, selector):
elementos = elementos_visibles(driver, by, selector)
return elementos[0] if elementos else None
def esperar_primer_visible(driver, candidatos, timeout=15):
fin = time.time() + timeout
while time.time() < fin:
for by, selector in candidatos:
elemento = primer_visible(driver, by, selector)
if elemento:
return elemento
time.sleep(0.3)
return None
def click_seguro(driver, elemento):
try:
driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", elemento)
time.sleep(0.1)
elemento.click()
except Exception:
try:
driver.execute_script("arguments[0].click();", elemento)
except Exception:
pass
def limpiar_campo(driver, elemento):
click_seguro(driver, elemento)
ActionChains(driver).key_down(Keys.CONTROL).send_keys("a").key_up(Keys.CONTROL).send_keys(Keys.DELETE).perform()
time.sleep(0.2)
def buscar_caja_mensaje(driver, timeout=20):
candidatos = [
(By.CSS_SELECTOR, "[data-testid='conversation-compose-box-input']"),
(By.CSS_SELECTOR, "footer div[contenteditable='true'][role='textbox']"),
(By.CSS_SELECTOR, "footer div[contenteditable='true'][data-lexical-editor='true']"),
(By.CSS_SELECTOR, "footer div[contenteditable='true']"),
(By.CSS_SELECTOR, "div[contenteditable='true'][data-tab='10']"),
(By.CSS_SELECTOR, "div[contenteditable='true'][data-tab='9']"),
(By.CSS_SELECTOR, "footer div[aria-placeholder*='mensaje' i]"),
(By.CSS_SELECTOR, "footer div[aria-placeholder*='message' i]"),
(By.CSS_SELECTOR, "footer div[aria-label*='mensaje' i]"),
(By.CSS_SELECTOR, "footer div[aria-label*='message' i]"),
(By.XPATH, "//footer//*[@contenteditable='true']"),
]
return esperar_primer_visible(driver, candidatos, timeout=timeout)
def abrir_buscador(driver, timeout=15):
try:
WebDriverWait(driver, 15).until(
EC.presence_of_element_located(
(
By.CSS_SELECTOR,
"#side, #pane-side, [data-testid='chat-list'], div[role='grid']",
)
)
)
except Exception:
pass
candidatos = [
(By.CSS_SELECTOR, "input[role='textbox'][data-tab='3']"),
(By.CSS_SELECTOR, "input[aria-label*='Buscar' i]"),
(By.CSS_SELECTOR, "input[placeholder*='Buscar' i]"),
(By.CSS_SELECTOR, "input[aria-label*='Search' i]"),
(By.CSS_SELECTOR, "input[placeholder*='Search' i]"),
(By.CSS_SELECTOR, "[data-testid='chat-list-search']"),
(By.CSS_SELECTOR, "div[contenteditable='true'][data-tab='3']"),
(By.CSS_SELECTOR, "div[contenteditable='true'][role='textbox'][aria-label*='buscar' i]"),
(By.CSS_SELECTOR, "div[contenteditable='true'][role='textbox'][aria-label*='search' i]"),
(By.CSS_SELECTOR, "div[contenteditable='true'][role='textbox'][aria-placeholder*='buscar' i]"),
(By.CSS_SELECTOR, "div[contenteditable='true'][role='textbox'][aria-placeholder*='search' i]"),
(
By.XPATH,
"//*[@contenteditable='true' and ancestor::*[@id='side' or @id='pane-side']]",
),
(
By.XPATH,
"//*[@contenteditable='true' and contains(translate(@aria-label, 'ABCDEFGHIJKLMNOPQRSTUVWXYZ', 'abcdefghijklmnopqrstuvwxyz'), 'buscar')]",
),
(
By.XPATH,
"//*[@contenteditable='true' and contains(translate(@aria-label, 'ABCDEFGHIJKLMNOPQRSTUVWXYZ', 'abcdefghijklmnopqrstuvwxyz'), 'search')]",
),
]
fin = time.time() + timeout
while time.time() < fin:
for by, selector in candidatos:
for elemento in elementos_visibles(driver, by, selector):
try:
rect = elemento.rect
if rect and rect.get("y", 9999) > 350:
continue
click_seguro(driver, elemento)
if (elemento.tag_name or "").lower() == "input":
return elemento
return elemento
except (StaleElementReferenceException, WebDriverException):
continue
time.sleep(0.5)
botones_busqueda = [
(By.CSS_SELECTOR, "button[aria-label*='buscar' i]"),
(By.CSS_SELECTOR, "button[aria-label*='search' i]"),
(By.CSS_SELECTOR, "div[role='button'][aria-label*='buscar' i]"),
(By.CSS_SELECTOR, "div[role='button'][aria-label*='search' i]"),
(By.CSS_SELECTOR, "span[data-icon='search']"),
]
boton = esperar_primer_visible(driver, botones_busqueda, timeout=3)
if boton:
click_seguro(driver, boton)
return esperar_primer_visible(driver, candidatos, timeout=5)
return None
def obtener_nombre_resultado(chat):
selectores_nombre = [
"span[title]",
"span[dir='auto']",
"div[role='gridcell'] span[dir='auto']",
]
for selector in selectores_nombre:
try:
for span in chat.find_elements(By.CSS_SELECTOR, selector):
texto = (span.get_attribute("title") or span.text or "").strip()
if texto:
return texto
except (StaleElementReferenceException, WebDriverException):
continue
return ""
def buscar_resultados_chat(driver):
candidatos = [
"[data-testid='cell-frame-container']",
"#pane-side div[tabindex='-1']",
"#side div[tabindex='-1']",
"div[role='grid'] div[role='row']",
"div[aria-label][role='grid'] div[role='row']",
]
resultados = []
vistos = set()
for selector in candidatos:
for elemento in elementos_visibles(driver, By.CSS_SELECTOR, selector):
try:
key = elemento.id
except Exception:
key = id(elemento)
if key not in vistos:
resultados.append(elemento)
vistos.add(key)
return resultados
def abrir_chat(driver, tipo, destino):
try:
if es_numero_telefono(destino):
numero = destino.replace("+", "").strip()
print(f"Abriendo chat directo con numero {numero}...")
driver.get(f"https://web.whatsapp.com/send?phone={numero}")
if buscar_caja_mensaje(driver, timeout=35):
print("Chat directo abierto.")
return True
print("No se pudo confirmar la apertura del chat directo.")
return False
print(f"Buscando {tipo} '{destino}'...")
buscador = abrir_buscador(driver)
if not buscador:
print("No se encontro buscador.")
return False
limpiar_campo(driver, buscador)
time.sleep(0.5)
for caracter in destino:
buscador.send_keys(caracter)
time.sleep(0.05)
try:
WebDriverWait(driver, 15).until(lambda d: len(buscar_resultados_chat(d)) > 0)
except TimeoutException:
print("No aparecieron resultados de busqueda.")
time.sleep(1)
destino_norm = destino.strip().lower()
chat_exacto = None
chat_parcial = None
for chat in buscar_resultados_chat(driver):
try:
nombre_chat = obtener_nombre_resultado(chat)
if not nombre_chat:
continue
nombre_norm = nombre_chat.strip().lower()
print(f"Resultado encontrado: {nombre_chat}")
if nombre_norm == destino_norm:
chat_exacto = chat
break
if destino_norm in nombre_norm or nombre_norm in destino_norm:
chat_parcial = chat
except (StaleElementReferenceException, WebDriverException):
continue
chat_objetivo = chat_exacto or chat_parcial
if chat_objetivo:
click_seguro(driver, chat_objetivo)
if buscar_caja_mensaje(driver, timeout=15):
print(f"{tipo.capitalize()} abierto.")
return True
print("Se selecciono el chat pero no se detecto caja mensaje.")
return False
try:
buscador.send_keys(Keys.ENTER)
if buscar_caja_mensaje(driver, timeout=8):
print(f"{tipo.capitalize()} '{destino}' abierto usando el primer resultado.")
return True
except Exception:
pass
print(f"No se encontro {tipo} '{destino}'.")
return False
except TimeoutException:
print(f"Timeout abriendo chat '{destino}'.")
return False
except WebDriverException as exc:
print(f"Error navegador: {exc}")
return False
except Exception as exc:
print(f"Error inesperado: {exc}")
return False
def enviar_mensaje(driver, mensaje):
try:
caja = buscar_caja_mensaje(driver, timeout=20)
if not caja:
raise Exception("No se encontro caja texto")
pyperclip.copy(mensaje.replace("\\n", "\n"))
limpiar_campo(driver, caja)
time.sleep(0.2)
caja.send_keys(Keys.CONTROL, "v")
time.sleep(0.3)
botones_envio = [
(By.CSS_SELECTOR, "button[aria-label*='enviar' i]"),
(By.CSS_SELECTOR, "button[aria-label*='send' i]"),
(By.CSS_SELECTOR, "[data-icon*='send']"),
(By.CSS_SELECTOR, "span[data-testid*='send']"),
]
boton = esperar_primer_visible(driver, botones_envio, timeout=2)
if boton:
click_seguro(driver, boton)
else:
caja.send_keys(Keys.ENTER)
print("Mensaje enviado.")
return True
except Exception as exc:
print(f"No se pudo enviar: {exc}")
return False
def obtener_nombre_chat_actual(driver):
try:
nombre_elem = driver.find_element(By.XPATH, "//header//span[@dir='auto' and @title]")
return nombre_elem.get_attribute("title").strip()
except Exception:
pass
try:
encabezado = driver.find_element(By.CSS_SELECTOR, "header")
for selector in ["span[title]", "span[dir='auto']", "div[role='button'] span"]:
for elemento in encabezado.find_elements(By.CSS_SELECTOR, selector):
texto = (elemento.get_attribute("title") or elemento.text or "").strip()
if texto:
return texto
except Exception:
pass
return "chat_desconocido"
def verificar_ticks(driver):
try:
metas = driver.find_elements(By.CSS_SELECTOR, "[data-testid='msg-meta']")
ultimo_con_status = None
for meta in reversed(metas):
if meta.find_elements(By.CSS_SELECTOR, "svg title, span[aria-label], [data-icon]"):
ultimo_con_status = meta
break
if not ultimo_con_status:
containers = driver.find_elements(By.CSS_SELECTOR, "[data-testid='msg-container']")
if containers and containers[-1].find_elements(By.CSS_SELECTOR, "[data-icon='message-fail']"):
return "FALLIDO"
return "ENVIADO (SIN CONFIRMACION)"
titles = ultimo_con_status.find_elements(By.CSS_SELECTOR, "svg title")
if titles:
titulo = (titles[-1].get_attribute("textContent") or titles[-1].text).strip()
mapping = {
"wds-ic-read": "LEIDO (2 ticks azules)",
"wds-ic-status-pending": "PENDIENTE (reloj)",
"wds-ic-status-sent": "ENVIADO (1 tick)",
"wds-ic-sent": "ENVIADO (1 tick)",
"wds-ic-status-delivered": "ENTREGADO (2 ticks)",
"wds-ic-delivered": "ENTREGADO (2 ticks)",
}
return mapping.get(titulo, f"DESCONOCIDO ({titulo})")
aria_spans = ultimo_con_status.find_elements(By.CSS_SELECTOR, "span[aria-label]")
if aria_spans:
label = aria_spans[-1].get_attribute("aria-label").strip()
aria_map = {
"Pendiente": "PENDIENTE (reloj)",
"Enviado": "ENVIADO (1 tick)",
"Entregado": "ENTREGADO (2 ticks)",
"Leído": "LEIDO (2 ticks azules)",
"Read": "LEIDO (2 ticks azules)",
}
return aria_map.get(label, f"DESCONOCIDO ({label})")
status_icons = ultimo_con_status.find_elements(By.CSS_SELECTOR, "[data-icon]")
if status_icons:
icono = status_icons[-1].get_attribute("data-icon")
legacy = {
"msg-time": "PENDIENTE (reloj)",
"msg-check": "ENVIADO (1 tick)",
"msg-dblcheck": "ENTREGADO (2 ticks)",
"msg-dblcheck-ack": "LEIDO (2 ticks azules)",
}
return legacy.get(icono, f"DESCONOCIDO ({icono})")
return "ENVIADO (SIN DETALLE)"
except Exception:
return "ERROR_VERIFICACION_STATUS"
def get_last_outgoing_text(driver):
try:
elems = driver.find_elements(
By.CSS_SELECTOR,
"[data-testid='msg-container'] span.selectable-text, "
"[data-testid='msg-container'] [data-testid='selectable-text']",
)
if elems:
texto = elems[-1].text.strip()
if texto:
return texto
elems = driver.find_elements(By.CSS_SELECTOR, "span.selectable-text")
for elem in reversed(elems[-10:]):
texto = elem.text.strip()
if texto:
return texto
except Exception:
pass
return ""
def validate_sent(driver, mensaje, timeout=8):
short = (mensaje.strip().replace("\n", " "))[:40].strip().lower()
if not short:
return False
inicio = time.time()
while time.time() - inicio < timeout:
try:
if driver.find_elements(
By.CSS_SELECTOR,
"canvas[aria-label*='Scan'], canvas[aria-label*='escan' i], div[data-ref] canvas",
):
print("Se detecto QR durante la validacion: sesion perdida.")
return None
except Exception:
pass
last = get_last_outgoing_text(driver)
if last:
last_norm = last.replace("\n", " ").strip().lower()
if short in last_norm or last_norm in short:
return True
if last_norm.startswith(short[:20]):
return True
time.sleep(0.8)
return False