diff --git a/.gitignore b/.gitignore deleted file mode 100644 index 5e40055..0000000 --- a/.gitignore +++ /dev/null @@ -1,18 +0,0 @@ -# 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 deleted file mode 100644 index 53dfea5..0000000 --- a/README.md +++ /dev/null @@ -1,26 +0,0 @@ -# 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 deleted file mode 100644 index 6d96ac2..0000000 --- a/mensajes.csv +++ /dev/null @@ -1,3 +0,0 @@ -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 deleted file mode 100644 index 6cb3430..0000000 --- a/scripts/validar_busqueda_whatsapp.py +++ /dev/null @@ -1,225 +0,0 @@ -#!/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 deleted file mode 100644 index 21a1ac0..0000000 --- a/scripts/validar_selectores.py +++ /dev/null @@ -1,366 +0,0 @@ -#!/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 deleted file mode 100644 index 6df9de7..0000000 --- a/whatsapp_fusionado_final_v3_6.py +++ /dev/null @@ -1,6 +0,0 @@ -#!/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 deleted file mode 100644 index 62a2641..0000000 --- a/whatsapp_sender/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Automatizacion modular de envios por WhatsApp Web.""" diff --git a/whatsapp_sender/config.py b/whatsapp_sender/config.py deleted file mode 100644 index a8582dd..0000000 --- a/whatsapp_sender/config.py +++ /dev/null @@ -1,23 +0,0 @@ -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 deleted file mode 100644 index ef0536e..0000000 --- a/whatsapp_sender/dependencies.py +++ /dev/null @@ -1,19 +0,0 @@ -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 deleted file mode 100644 index 4e21c82..0000000 --- a/whatsapp_sender/driver_session.py +++ /dev/null @@ -1,210 +0,0 @@ -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 deleted file mode 100644 index 8ba3232..0000000 --- a/whatsapp_sender/main.py +++ /dev/null @@ -1,94 +0,0 @@ -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 deleted file mode 100644 index cbaffb1..0000000 --- a/whatsapp_sender/messages.py +++ /dev/null @@ -1,23 +0,0 @@ -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 deleted file mode 100644 index 43fc6c4..0000000 --- a/whatsapp_sender/processor.py +++ /dev/null @@ -1,133 +0,0 @@ -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 deleted file mode 100644 index 39001e2..0000000 --- a/whatsapp_sender/utils.py +++ /dev/null @@ -1,46 +0,0 @@ -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 deleted file mode 100644 index 2cb0be5..0000000 --- a/whatsapp_sender/whatsapp.py +++ /dev/null @@ -1,434 +0,0 @@ -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