commit ec5ebbd20fb9224605543bd1229c40103033abef Author: ldiaz Date: Thu Jun 25 13:29:47 2026 -0500 v3.6 - Envio automatizado WhatsApp Web con busqueda por nombre y telefono, deteccion de ticks via msg-meta/SVG title, y validacion integral de selectores diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..5e40055 --- /dev/null +++ b/.gitignore @@ -0,0 +1,18 @@ +# Bytecode compilado +__pycache__/ +*.pyc + +# Perfil Chrome (datos de sesion) +perfil_whatsapp/ + +# Evidencias de envio (capturas PNG) +evidencias/ + +# Diagnostico de selectores (reportes y capturas) +diagnosticos/ +diagnosticos_selectores/ + +# Salida de ejecucion +resultados_envio.csv +log_envios.txt +estado_sesion.txt diff --git a/README.md b/README.md new file mode 100644 index 0000000..53dfea5 --- /dev/null +++ b/README.md @@ -0,0 +1,26 @@ +# WhatsApp sender modular + +Esta carpeta contiene una version organizada de `whatsapp_fusionado_final_v3_6.py`. + +## Ejecutar + +Desde esta carpeta: + +```powershell +python whatsapp_fusionado_final_v3_6.py +``` + +## Estructura + +- `whatsapp_fusionado_final_v3_6.py`: entrada principal compatible con el nombre anterior. +- `whatsapp_sender/config.py`: rutas, tiempos y constantes. +- `whatsapp_sender/dependencies.py`: instalacion/verificacion de dependencias. +- `whatsapp_sender/utils.py`: logs, directorios y limpieza de sesiones. +- `whatsapp_sender/driver_session.py`: ChromeDriver y sesion de WhatsApp Web. +- `whatsapp_sender/whatsapp.py`: abrir chats, enviar mensajes y validar envio. +- `whatsapp_sender/messages.py`: carga y validacion del CSV/XLSX. +- `whatsapp_sender/processor.py`: procesamiento de cada fila. +- `whatsapp_sender/main.py`: flujo general del programa. + +Coloca `mensajes.csv` o `mensajes.xlsx` en esta carpeta, o cambia `MENSAJES_FILE` en +`whatsapp_sender/config.py`. diff --git a/mensajes.csv b/mensajes.csv new file mode 100644 index 0000000..6d96ac2 --- /dev/null +++ b/mensajes.csv @@ -0,0 +1,3 @@ +tipo,id_destino,mensaje,hora,minuto +contacto,Luyzer José Díaz Pacheco,"Hola, este mensaje es solo para ti 👋",09,35 +grupo,Díaz Pacheco INC,"¡Hola grupo! 😃\nEste es un mensaje automático 🚀",09,40 \ No newline at end of file diff --git a/scripts/validar_busqueda_whatsapp.py b/scripts/validar_busqueda_whatsapp.py new file mode 100644 index 0000000..6cb3430 --- /dev/null +++ b/scripts/validar_busqueda_whatsapp.py @@ -0,0 +1,225 @@ +#!/usr/bin/env python3 +import argparse +import os +import sys +sys.stdout.reconfigure(encoding='utf-8', errors='replace') +import time +from datetime import datetime + +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.ui import WebDriverWait + +from whatsapp_sender.driver_session import ensure_session_active, iniciar_driver +from whatsapp_sender.whatsapp import buscar_caja_mensaje, click_seguro, elementos_visibles, obtener_nombre_resultado + + +SEARCH_SELECTORS = [ + ("input data-tab 3", By.CSS_SELECTOR, "input[role='textbox'][data-tab='3']"), + ("input aria Buscar", By.CSS_SELECTOR, "input[aria-label*='Buscar' i]"), + ("input placeholder Buscar", By.CSS_SELECTOR, "input[placeholder*='Buscar' i]"), + ("input aria Search", By.CSS_SELECTOR, "input[aria-label*='Search' i]"), + ("input placeholder Search", By.CSS_SELECTOR, "input[placeholder*='Search' i]"), + ("data-testid chat-list-search", By.CSS_SELECTOR, "[data-testid='chat-list-search']"), + ("role search", By.CSS_SELECTOR, "div[role='search']"), + ("aria Buscar", By.CSS_SELECTOR, "div[aria-label*='Buscar' i]"), + ("aria Search", By.CSS_SELECTOR, "div[aria-label*='Search' i]"), + ("textbox data-tab 3", By.CSS_SELECTOR, "div[contenteditable='true'][data-tab='3']"), + ("textbox buscar", By.CSS_SELECTOR, "div[contenteditable='true'][aria-label*='buscar' i]"), + ("textbox search", By.CSS_SELECTOR, "div[contenteditable='true'][aria-label*='search' i]"), + ("textbox inside side", By.XPATH, "//*[@contenteditable='true' and ancestor::*[@id='side' or @id='pane-side']]"), +] + +RESULT_SELECTORS = [ + ("cell-frame-container", By.CSS_SELECTOR, "[data-testid='cell-frame-container']"), + ("pane-side tabindex", By.CSS_SELECTOR, "#pane-side div[tabindex='-1']"), + ("side tabindex", By.CSS_SELECTOR, "#side div[tabindex='-1']"), + ("grid row", By.CSS_SELECTOR, "div[role='grid'] div[role='row']"), + ("aria grid row", By.CSS_SELECTOR, "div[aria-label][role='grid'] div[role='row']"), +] + +SESSION_SELECTORS = [ + ("pane-side", By.ID, "pane-side"), + ("side", By.ID, "side"), + ("chat-list", By.CSS_SELECTOR, "[data-testid='chat-list']"), + ("role grid", By.CSS_SELECTOR, "div[role='grid']"), +] + + +def contar_visibles(driver, by, selector): + return len(elementos_visibles(driver, by, selector)) + + +def imprimir_estado_selectores(driver, titulo, selectores): + print(f"\n== {titulo} ==") + for nombre, by, selector in selectores: + try: + total = len(driver.find_elements(by, selector)) + visibles = contar_visibles(driver, by, selector) + print(f"{nombre}: total={total}, visibles={visibles}") + except Exception as exc: + print(f"{nombre}: ERROR {exc}") + + +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 encontrar_buscador(driver, timeout=15): + fin = time.time() + timeout + while time.time() < fin: + for nombre, by, selector in SEARCH_SELECTORS: + 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) + time.sleep(0.3) + + tag_name = (elemento.tag_name or "").lower() + if tag_name == "input" or elemento.get_attribute("contenteditable") == "true": + print(f"Buscador detectado por: {nombre}") + return elemento + + internos = elemento.find_elements( + By.CSS_SELECTOR, + "input[role='textbox'], input[type='text'], div[contenteditable='true'], div[role='textbox']", + ) + for interno in internos: + if interno.is_displayed(): + click_seguro(driver, interno) + print(f"Buscador interno detectado por: {nombre}") + return interno + except (StaleElementReferenceException, WebDriverException): + continue + except Exception: + continue + time.sleep(0.5) + return None + + +def obtener_resultados(driver): + resultados = [] + vistos = set() + for nombre_selector, by, selector in RESULT_SELECTORS: + for elemento in elementos_visibles(driver, by, selector): + try: + key = elemento.id + except Exception: + key = id(elemento) + + if key in vistos: + continue + + vistos.add(key) + resultados.append((nombre_selector, elemento)) + return resultados + + +def guardar_captura(driver, prefijo): + os.makedirs("diagnosticos", exist_ok=True) + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + nombre = os.path.join("diagnosticos", f"{prefijo}_{timestamp}.png") + driver.save_screenshot(nombre) + print(f"\nCaptura guardada: {nombre}") + + +def validar_busqueda(destino, abrir=False, cerrar=False): + driver = iniciar_driver() + driver.get("https://web.whatsapp.com") + + try: + if not ensure_session_active(driver): + print("No se pudo validar sesion activa. Revisa QR o captura diagnostica.") + guardar_captura(driver, "validar_busqueda_sin_sesion") + return 2 + + imprimir_estado_selectores(driver, "Selectores de sesion/panel", SESSION_SELECTORS) + imprimir_estado_selectores(driver, "Selectores de buscador antes de buscar", SEARCH_SELECTORS) + + buscador = encontrar_buscador(driver) + if not buscador: + print("\nNo se encontro el buscador de chats.") + guardar_captura(driver, "validar_busqueda_sin_buscador") + return 3 + + limpiar_campo(driver, buscador) + for caracter in destino: + buscador.send_keys(caracter) + time.sleep(0.04) + + try: + WebDriverWait(driver, 10).until(lambda d: len(obtener_resultados(d)) > 0) + except TimeoutException: + print("\nNo aparecieron resultados durante la espera.") + + time.sleep(1) + imprimir_estado_selectores(driver, "Selectores de resultados despues de buscar", RESULT_SELECTORS) + + destino_norm = destino.strip().lower() + resultados = obtener_resultados(driver) + print(f"\nResultados leidos: {len(resultados)}") + + coincidencia = None + for indice, (selector, elemento) in enumerate(resultados, start=1): + nombre = obtener_nombre_resultado(elemento) + if not nombre: + continue + + nombre_norm = nombre.strip().lower() + tipo_match = "" + if nombre_norm == destino_norm: + tipo_match = "EXACTO" + coincidencia = elemento + elif destino_norm in nombre_norm or nombre_norm in destino_norm: + tipo_match = "PARCIAL" + coincidencia = coincidencia or elemento + + print(f"{indice}. [{selector}] {nombre} {tipo_match}".rstrip()) + + if not coincidencia: + print("\nNo se detecto coincidencia exacta ni parcial.") + guardar_captura(driver, "validar_busqueda_sin_coincidencia") + return 4 + + print("\nCoincidencia encontrada.") + + if abrir: + click_seguro(driver, coincidencia) + if buscar_caja_mensaje(driver, timeout=10): + print("Chat abierto y caja de mensaje detectada.") + else: + print("Se hizo clic en la coincidencia, pero no se detecto caja de mensaje.") + + guardar_captura(driver, "validar_busqueda_ok") + return 0 + finally: + if cerrar: + driver.quit() + + +def main(): + parser = argparse.ArgumentParser( + description="Diagnostica selectores de busqueda de contactos/grupos en WhatsApp Web sin enviar mensajes." + ) + parser.add_argument("destino", nargs="?", help="Nombre del contacto o grupo a buscar.") + parser.add_argument("--abrir", action="store_true", help="Abre la coincidencia encontrada para validar la caja de mensaje.") + parser.add_argument("--cerrar", action="store_true", help="Cierra Chrome al terminar.") + args = parser.parse_args() + + destino = args.destino or input("Contacto o grupo a buscar: ").strip() + if not destino: + print("Debes ingresar un contacto o grupo.") + return 1 + + return validar_busqueda(destino, abrir=args.abrir, cerrar=args.cerrar) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/validar_selectores.py b/scripts/validar_selectores.py new file mode 100644 index 0000000..21a1ac0 --- /dev/null +++ b/scripts/validar_selectores.py @@ -0,0 +1,366 @@ +#!/usr/bin/env python3 +""" +Validador integral de selectores para WhatsApp Web. +Prueba todas las estrategias de localización usadas en whatsapp_sender/ +y reporta cuáles siguen funcionando y cuáles no. +""" + +import os +import time +from datetime import datetime +from collections import OrderedDict + +from selenium.webdriver.common.by import By +from selenium.webdriver.support import expected_conditions as EC +from selenium.webdriver.support.ui import WebDriverWait + +from whatsapp_sender.driver_session import ensure_session_active, iniciar_driver +from whatsapp_sender.utils import log_text + +DIAGNOSTICOS_DIR = "diagnosticos_selectores" +os.makedirs(DIAGNOSTICOS_DIR, exist_ok=True) + + +def captura(driver, nombre): + ts = datetime.now().strftime("%H%M%S") + path = os.path.join(DIAGNOSTICOS_DIR, f"{nombre}_{ts}.png") + try: + driver.save_screenshot(path) + except Exception: + pass + return path + + +def probar_selector(driver, by, selector, desc, categoria): + """ + Prueba un selector y devuelve un dict con resultados. + """ + resultado = { + "categoria": categoria, + "descripcion": desc, + "estrategia": str(by), + "selector": selector, + "total": 0, + "visibles": 0, + "coincide": False, + "error": None, + } + try: + elementos = driver.find_elements(by, selector) + resultado["total"] = len(elementos) + visibles = 0 + for elem in elementos: + try: + if elem.is_displayed(): + visibles += 1 + except Exception: + continue + resultado["visibles"] = visibles + resultado["coincide"] = visibles > 0 + except Exception as exc: + resultado["error"] = str(exc) + return resultado + + +def probar_selectores(driver, selectores, categoria): + resultados = [] + for item in selectores: + res = probar_selector(driver, item["by"], item["selector"], item["desc"], categoria) + resultados.append(res) + return resultados + + +def imprimir_resultados(resultados, titulo): + print(f"\n{'=' * 70}") + print(f" {titulo}") + print(f"{'=' * 70}") + ok = sum(1 for r in resultados if r["coincide"]) + total = len(resultados) + print(f" {ok}/{total} selectores funcionan\n") + + for r in resultados: + icono = "OK" if r["coincide"] else "---" + print(f" [{icono}] {r['descripcion']}") + print(f" Selector: {r['selector']}") + print(f" By: {r['estrategia']} | Total: {r['total']} | Visibles: {r['visibles']}") + if r["error"]: + print(f" ERROR: {r['error']}") + print() + + return ok, total + + +def construir_selectores_session(): + return [ + {"by": By.ID, "selector": "pane-side", "desc": "Panel lateral #pane-side"}, + {"by": By.ID, "selector": "side", "desc": "Panel lateral #side"}, + {"by": By.CSS_SELECTOR, "selector": "[data-testid='chat-list']", "desc": "Chat list data-testid"}, + {"by": By.CSS_SELECTOR, "selector": "[data-testid='cell-frame-container']", "desc": "Cell frame container"}, + {"by": By.CSS_SELECTOR, "selector": "div[role='grid']", "desc": "role=grid"}, + {"by": By.CSS_SELECTOR, "selector": "div[aria-label*='chat' i][role='grid']", "desc": "Grid con aria-label chat"}, + {"by": By.CSS_SELECTOR, "selector": "div[contenteditable='true'][role='textbox'][aria-label*='buscar' i]", "desc": "Busqueda contenteditable label=buscar"}, + {"by": By.CSS_SELECTOR, "selector": "div[contenteditable='true'][role='textbox'][aria-label*='search' i]", "desc": "Busqueda contenteditable label=search"}, + ] + + +def construir_selectores_qr(): + return [ + {"by": By.CSS_SELECTOR, "selector": "canvas[aria-label*='Scan']", "desc": "QR canvas aria-label Scan"}, + {"by": By.CSS_SELECTOR, "selector": "canvas[aria-label*='escan' i]", "desc": "QR canvas aria-label escan"}, + {"by": By.CSS_SELECTOR, "selector": "canvas[aria-label='Scan me!']", "desc": "QR canvas exacto Scan me!"}, + {"by": By.CSS_SELECTOR, "selector": "div[data-ref] canvas", "desc": "QR div[data-ref] canvas"}, + ] + + +def construir_selectores_busqueda(): + return [ + {"by": By.CSS_SELECTOR, "selector": "input[role='textbox'][data-tab='3']", "desc": "Input textbox data-tab=3"}, + {"by": By.CSS_SELECTOR, "selector": "input[aria-label*='Buscar' i]", "desc": "Input aria-label Buscar"}, + {"by": By.CSS_SELECTOR, "selector": "input[placeholder*='Buscar' i]", "desc": "Input placeholder Buscar"}, + {"by": By.CSS_SELECTOR, "selector": "input[aria-label*='Search' i]", "desc": "Input aria-label Search"}, + {"by": By.CSS_SELECTOR, "selector": "input[placeholder*='Search' i]", "desc": "Input placeholder Search"}, + {"by": By.CSS_SELECTOR, "selector": "[data-testid='chat-list-search']", "desc": "data-testid chat-list-search"}, + {"by": By.CSS_SELECTOR, "selector": "div[role='search']", "desc": "div role=search"}, + {"by": By.CSS_SELECTOR, "selector": "div[contenteditable='true'][data-tab='3']", "desc": "Contenteditable data-tab=3"}, + {"by": By.CSS_SELECTOR, "selector": "div[contenteditable='true'][role='textbox'][aria-label*='buscar' i]", "desc": "Contenteditable label buscar"}, + {"by": By.CSS_SELECTOR, "selector": "div[contenteditable='true'][role='textbox'][aria-label*='search' i]", "desc": "Contenteditable label search"}, + {"by": By.CSS_SELECTOR, "selector": "div[contenteditable='true'][role='textbox'][aria-placeholder*='buscar' i]", "desc": "Contenteditable placeholder buscar"}, + {"by": By.CSS_SELECTOR, "selector": "div[contenteditable='true'][role='textbox'][aria-placeholder*='search' i]", "desc": "Contenteditable placeholder search"}, + {"by": By.XPATH, "selector": "//*[@contenteditable='true' and ancestor::*[@id='side' or @id='pane-side']]", "desc": "XPath contenteditable dentro de #side/#pane-side"}, + ] + + +def construir_selectores_busqueda_botones(): + return [ + {"by": By.CSS_SELECTOR, "selector": "button[aria-label*='buscar' i]", "desc": "Boton aria-label buscar"}, + {"by": By.CSS_SELECTOR, "selector": "button[aria-label*='search' i]", "desc": "Boton aria-label search"}, + {"by": By.CSS_SELECTOR, "selector": "div[role='button'][aria-label*='buscar' i]", "desc": "Div role=button label buscar"}, + {"by": By.CSS_SELECTOR, "selector": "div[role='button'][aria-label*='search' i]", "desc": "Div role=button label search"}, + {"by": By.CSS_SELECTOR, "selector": "span[data-icon='search']", "desc": "Span data-icon=search"}, + ] + + +def construir_selectores_resultados(): + return [ + {"by": By.CSS_SELECTOR, "selector": "[data-testid='cell-frame-container']", "desc": "cell-frame-container"}, + {"by": By.CSS_SELECTOR, "selector": "#pane-side div[tabindex='-1']", "desc": "#pane-side div tabindex=-1"}, + {"by": By.CSS_SELECTOR, "selector": "#side div[tabindex='-1']", "desc": "#side div tabindex=-1"}, + {"by": By.CSS_SELECTOR, "selector": "div[role='grid'] div[role='row']", "desc": "role=grid > role=row"}, + {"by": By.CSS_SELECTOR, "selector": "div[aria-label][role='grid'] div[role='row']", "desc": "aria-label grid > role=row"}, + ] + + +def construir_selectores_caja_mensaje(): + return [ + {"by": By.CSS_SELECTOR, "selector": "[data-testid='conversation-compose-box-input']", "desc": "Compose box data-testid"}, + {"by": By.CSS_SELECTOR, "selector": "footer div[contenteditable='true'][role='textbox']", "desc": "Footer contenteditable textbox"}, + {"by": By.CSS_SELECTOR, "selector": "footer div[contenteditable='true'][data-lexical-editor='true']", "desc": "Footer lexical editor"}, + {"by": By.CSS_SELECTOR, "selector": "footer div[contenteditable='true']", "desc": "Footer contenteditable generico"}, + {"by": By.CSS_SELECTOR, "selector": "div[contenteditable='true'][data-tab='10']", "desc": "Contenteditable data-tab=10"}, + {"by": By.CSS_SELECTOR, "selector": "div[contenteditable='true'][data-tab='9']", "desc": "Contenteditable data-tab=9"}, + {"by": By.CSS_SELECTOR, "selector": "footer div[aria-placeholder*='mensaje' i]", "desc": "Footer placeholder mensaje"}, + {"by": By.CSS_SELECTOR, "selector": "footer div[aria-placeholder*='message' i]", "desc": "Footer placeholder message"}, + {"by": By.CSS_SELECTOR, "selector": "footer div[aria-label*='mensaje' i]", "desc": "Footer label mensaje"}, + {"by": By.CSS_SELECTOR, "selector": "footer div[aria-label*='message' i]", "desc": "Footer label message"}, + {"by": By.XPATH, "selector": "//footer//*[@contenteditable='true']", "desc": "XPath footer contenteditable"}, + ] + + +def construir_selectores_enviar(): + return [ + {"by": By.CSS_SELECTOR, "selector": "button[aria-label*='enviar' i]", "desc": "Boton aria-label enviar"}, + {"by": By.CSS_SELECTOR, "selector": "button[aria-label*='send' i]", "desc": "Boton aria-label send"}, + {"by": By.CSS_SELECTOR, "selector": "[data-icon*='send']", "desc": "Elemento con data-icon=send/wds-ic-send-filled"}, + {"by": By.CSS_SELECTOR, "selector": "span[data-testid*='send']", "desc": "Span con data-testid=send"}, + ] + + +def construir_selectores_ticks(): + return [ + {"by": By.CSS_SELECTOR, "selector": "[data-testid='msg-meta'] svg title", "desc": "SVG title dentro de msg-meta"}, + {"by": By.CSS_SELECTOR, "selector": "[data-testid='msg-meta']", "desc": "Contenedor msg-meta"}, + {"by": By.CSS_SELECTOR, "selector": "span[data-icon='message-fail']", "desc": "Icono de mensaje fallido"}, + ] + + +def construir_selectores_mensaje_saliente(): + return [ + {"by": By.CSS_SELECTOR, "selector": "div[data-testid='msg-container'] span.selectable-text", "desc": "msg-container span.selectable-text"}, + {"by": By.CSS_SELECTOR, "selector": "div[data-testid='msg-container'] [data-testid='selectable-text']", "desc": "msg-container testid selectable-text"}, + {"by": By.CSS_SELECTOR, "selector": "span.selectable-text", "desc": "span.selectable-text generico"}, + ] + + +def construir_selectores_nombre_chat(): + return [ + {"by": By.XPATH, "selector": "//header//span[@dir='auto' and @title]", "desc": "XPath header span title"}, + {"by": By.CSS_SELECTOR, "selector": "header span[title]", "desc": "Header span[title]"}, + {"by": By.CSS_SELECTOR, "selector": "header span[dir='auto']", "desc": "Header span[dir=auto]"}, + {"by": By.CSS_SELECTOR, "selector": "header div[role='button'] span", "desc": "Header role=button span"}, + ] + + +def construir_selectores_nombre_resultado(): + return [ + {"by": By.CSS_SELECTOR, "selector": "span[title]", "desc": "span[title] generico"}, + {"by": By.CSS_SELECTOR, "selector": "span[dir='auto']", "desc": "span[dir=auto] generico"}, + {"by": By.CSS_SELECTOR, "selector": "div[role='gridcell'] span[dir='auto']", "desc": "gridcell span[dir=auto]"}, + ] + + +def generar_reporte_markdown(resultados_por_categoria, archivo): + with open(archivo, "w", encoding="utf-8") as f: + f.write("# Reporte de validacion de selectores - WhatsApp Web\n\n") + f.write(f"Generado: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n") + + total_ok = 0 + total_all = 0 + + for categoria, resultados in resultados_por_categoria: + ok = sum(1 for r in resultados if r["coincide"]) + total_ok += ok + total_all += len(resultados) + pct = ok / len(resultados) * 100 if resultados else 0 + f.write(f"## {categoria}\n") + f.write(f"- **Selectores funcionales:** {ok}/{len(resultados)} ({pct:.1f}%)\n\n") + f.write("| Estado | Descripcion | Estrategia | Selector | Total | Visibles |\n") + f.write("|--------|-------------|------------|----------|-------|----------|\n") + for r in resultados: + estado = ":white_check_mark:" if r["coincide"] else ":x:" + error_str = f" ERROR: {r['error']}" if r["error"] else "" + f.write(f"| {estado} | {r['descripcion']}{error_str} | {r['estrategia']} | `{r['selector']}` | {r['total']} | {r['visibles']} |\n") + f.write("\n") + + f.write(f"---\n\n## Resumen global\n\n") + f.write(f"**{total_ok}/{total_all} selectores funcionales ({total_ok/total_all*100:.1f}%)**\n") + + +def validar_selectores(destino_busqueda=None, cerrar=False): + driver = iniciar_driver() + driver.get("https://web.whatsapp.com") + + try: + print("Esperando sesion activa de WhatsApp Web...") + if not ensure_session_active(driver): + print("No se detecto sesion activa. Escanea el QR.") + captura(driver, "sin_sesion") + return False + + captura(driver, "sesion_activa") + + categorias = [ + ("Sesion / Panel lateral", construir_selectores_session()), + ("QR detection", construir_selectores_qr()), + ("Buscador - campo principal", construir_selectores_busqueda()), + ("Buscador - botones alternativos", construir_selectores_busqueda_botones()), + ("Resultados de busqueda", construir_selectores_resultados()), + ("Caja de mensaje", construir_selectores_caja_mensaje()), + ("Boton enviar", construir_selectores_enviar()), + ("Ticks / estado de mensaje", construir_selectores_ticks()), + ("Texto mensaje saliente", construir_selectores_mensaje_saliente()), + ("Nombre de chat (header)", construir_selectores_nombre_chat()), + ("Nombre en resultados", construir_selectores_nombre_resultado()), + ] + + resultados_por_categoria = [] + + for categoria, selectores in categorias: + resultados = probar_selectores(driver, selectores, categoria) + ok, total = imprimir_resultados(resultados, categoria) + resultados_por_categoria.append((categoria, resultados)) + + # Si se proporcionó un destino, probar búsqueda real y apertura de chat + if destino_busqueda: + print(f"\n{'=' * 70}") + print(f" PRUEBA DE BUSQUEDA REAL con destino: '{destino_busqueda}'") + print(f"{'=' * 70}") + from whatsapp_sender.whatsapp import (abrir_buscador, buscar_resultados_chat, buscar_caja_mensaje, + click_seguro, limpiar_campo) + buscador = abrir_buscador(driver) + if buscador: + print(" Buscador abierto correctamente.") + limpiar_campo(driver, buscador) + for c in destino_busqueda: + buscador.send_keys(c) + time.sleep(0.05) + time.sleep(2) + resultados_busqueda = probar_selectores(driver, construir_selectores_resultados(), "Resultados post-busqueda") + ok_b, total_b = imprimir_resultados(resultados_busqueda, "Resultados de busqueda (despues de escribir)") + resultados_por_categoria.append((f"Resultados post-busqueda ('{destino_busqueda}')", resultados_busqueda)) + captura(driver, "post_busqueda") + + # Abrir chat por enlace directo (send?phone=) para validar + # selectores de caja de mensaje, header y boton enviar + import re + numeros = re.findall(r"\d{7,15}", destino_busqueda) + if numeros: + telefono = numeros[0] + print(f" Abriendo chat directo con numero {telefono}...") + driver.get(f"https://web.whatsapp.com/send?phone={telefono}") + time.sleep(5) + captura(driver, "chat_directo") + + categorias_chat = [ + ("Caja de mensaje (con chat abierto)", construir_selectores_caja_mensaje()), + ("Nombre de chat (header, con chat abierto)", construir_selectores_nombre_chat()), + ("Ticks / estado de mensaje", construir_selectores_ticks()), + ("Texto mensaje saliente", construir_selectores_mensaje_saliente()), + ] + for cat, selectores in categorias_chat: + resultados = probar_selectores(driver, selectores, cat) + ok, total = imprimir_resultados(resultados, cat) + resultados_por_categoria.append((cat, resultados)) + + # Type in compose box so send button replaces mic button + compose = driver.find_elements(By.CSS_SELECTOR, + "[data-testid='conversation-compose-box-input'], " + "footer div[contenteditable='true']") + if compose: + compose[0].click() + compose[0].send_keys("test") + time.sleep(1) + resultados_envio = probar_selectores(driver, construir_selectores_enviar(), "Boton enviar (con texto en caja)") + ok_e, total_e = imprimir_resultados(resultados_envio, "Boton enviar (con texto en caja)") + resultados_por_categoria.append(("Boton enviar (con texto en caja)", resultados_envio)) + if compose: + driver.execute_script("arguments[0].innerHTML = '';", compose[0]) + else: + print(" No se encontro numero en el destino para abrir chat directo.") + else: + print(" No se pudo abrir el buscador.") + + # Generar reporte + reporte_path = os.path.join(DIAGNOSTICOS_DIR, "reporte_selectores.md") + generar_reporte_markdown(resultados_por_categoria, reporte_path) + print(f"\nReporte guardado: {reporte_path}") + + # Resumen final + total_ok = 0 + total_all = 0 + for cat, res in resultados_por_categoria: + total_ok += sum(1 for r in res if r["coincide"]) + total_all += len(res) + print(f"\n{'=' * 70}") + print(f" RESUMEN GLOBAL: {total_ok}/{total_all} selectores funcionales ({total_ok/total_all*100:.1f}%)") + print(f"{'=' * 70}") + + return True + + finally: + if cerrar: + try: + driver.quit() + except Exception: + pass + + +if __name__ == "__main__": + import argparse + parser = argparse.ArgumentParser( + description="Valida todos los selectores CSS/XPath usados en whatsapp_sender/." + ) + parser.add_argument("destino", nargs="?", help="Contacto/grupo para prueba de busqueda real") + parser.add_argument("--cerrar", action="store_true", help="Cerrar Chrome al terminar") + args = parser.parse_args() + + validar_selectores(destino_busqueda=args.destino, cerrar=args.cerrar) diff --git a/whatsapp_fusionado_final_v3_6.py b/whatsapp_fusionado_final_v3_6.py new file mode 100644 index 0000000..6df9de7 --- /dev/null +++ b/whatsapp_fusionado_final_v3_6.py @@ -0,0 +1,6 @@ +#!/usr/bin/env python3 +from whatsapp_sender.main import main + + +if __name__ == "__main__": + main() diff --git a/whatsapp_sender/__init__.py b/whatsapp_sender/__init__.py new file mode 100644 index 0000000..62a2641 --- /dev/null +++ b/whatsapp_sender/__init__.py @@ -0,0 +1 @@ +"""Automatizacion modular de envios por WhatsApp Web.""" diff --git a/whatsapp_sender/config.py b/whatsapp_sender/config.py new file mode 100644 index 0000000..a8582dd --- /dev/null +++ b/whatsapp_sender/config.py @@ -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"} diff --git a/whatsapp_sender/dependencies.py b/whatsapp_sender/dependencies.py new file mode 100644 index 0000000..ef0536e --- /dev/null +++ b/whatsapp_sender/dependencies.py @@ -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) diff --git a/whatsapp_sender/driver_session.py b/whatsapp_sender/driver_session.py new file mode 100644 index 0000000..4e21c82 --- /dev/null +++ b/whatsapp_sender/driver_session.py @@ -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 diff --git a/whatsapp_sender/main.py b/whatsapp_sender/main.py new file mode 100644 index 0000000..8ba3232 --- /dev/null +++ b/whatsapp_sender/main.py @@ -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 diff --git a/whatsapp_sender/messages.py b/whatsapp_sender/messages.py new file mode 100644 index 0000000..cbaffb1 --- /dev/null +++ b/whatsapp_sender/messages.py @@ -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) diff --git a/whatsapp_sender/processor.py b/whatsapp_sender/processor.py new file mode 100644 index 0000000..43fc6c4 --- /dev/null +++ b/whatsapp_sender/processor.py @@ -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 diff --git a/whatsapp_sender/utils.py b/whatsapp_sender/utils.py new file mode 100644 index 0000000..39001e2 --- /dev/null +++ b/whatsapp_sender/utils.py @@ -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}") diff --git a/whatsapp_sender/whatsapp.py b/whatsapp_sender/whatsapp.py new file mode 100644 index 0000000..2cb0be5 --- /dev/null +++ b/whatsapp_sender/whatsapp.py @@ -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