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']"), ] 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