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