Selenium+Telethon自动选择QxBroker交易时长问题求助

你的问题核心有两个：一是异步事件处理器中调用同步Selenium操作时的await要求，二是交易时长下拉菜单的正确定位与选择逻辑。以下是修改后的完整代码及关键说明：

关键修改点 #

• 用asyncio.to_thread（Python 3.9+）将同步的Selenium操作包装为异步任务，避免阻塞Telethon的事件循环
• 添加显式等待（WebDriverWait）确保下拉菜单及选项加载完成，防止元素未找到错误
• 实现时长下拉菜单的展开与目标选项选择逻辑

修改后的完整代码 #

from telethon import TelegramClient, events
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import asyncio

# 配置信息
email = "****@gmail.com"
password = "*****"
api_id = 123456
api_hash = 'DFDGGGGGGGDGBGFBGFBFGB'

# 初始化Selenium驱动（全局初始化，避免重复创建）
driver = webdriver.Chrome()
driver.get("https://qxbroker.com/en/sign-in/")

# 登录逻辑（初始化阶段同步执行）
WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.NAME, "email"))).send_keys(email)
WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.NAME, "password"))).send_keys(password)
WebDriverWait(driver, 10).until(EC.element_to_be_clickable((By.XPATH, '//*[@id="tab-1"]/form/button/div'))).click()

# 异步设置交易时长函数
async def set_trade_duration():
    # 展开时长下拉菜单（需根据页面实际结构调整XPath）
    await asyncio.to_thread(
        lambda: WebDriverWait(driver, 10).until(
            EC.element_to_be_clickable((By.XPATH, "//*[contains(@class, 'duration-selector') or text()='交易时长']"))
        ).click()
    )
    # 选择5-10分钟选项（同样需根据页面实际文本调整XPath）
    await asyncio.to_thread(
        lambda: WebDriverWait(driver, 10).until(
            EC.element_to_be_clickable((By.XPATH, "//div[contains(text(), '5-10分钟')]"))
        ).click()
    )

# 初始化Telethon客户端
client = TelegramClient('anon', api_id, api_hash)

@client.on(events.NewMessage)
async def handler(event):
    if "EUR/USD" in event.raw_text:
        # 先设置交易时长
        await set_trade_duration()
        
        if "UP" in event.raw_text:
            print("UP")
            await asyncio.to_thread(
                lambda: WebDriverWait(driver, 10).until(
                    EC.element_to_be_clickable((By.XPATH, '//*[@id="root"]/div/div[1]/main/div[2]/div[1]/div/div[6]/div[1]/button'))
                ).click()
            )
        elif 'DOWN' in event.raw_text:
            print("DOWN")
            await asyncio.to_thread(
                lambda: WebDriverWait(driver, 10).until(
                    EC.element_to_be_clickable((By.XPATH, '//*[@id="root"]/div/div[1]/main/div[2]/div[1]/div/div[6]/div[4]/button'))
                ).click()
            )

client.start()
client.run_until_disconnected()

注意事项 #

1. XPath调整：如果时长下拉菜单的定位不准确，需用浏览器开发者工具（F12）查看页面实际HTML结构，修改对应的XPath属性（比如class、文本内容）
2. Python版本兼容：若使用Python 3.8及以下版本，替换asyncio.to_thread为以下代码：

   loop = asyncio.get_event_loop()
   await loop.run_in_executor(None, lambda: ...)
   

3. 显式等待：所有元素操作都用WebDriverWait包裹，超时时间可根据页面加载速度调整
4. 线程安全：Selenium WebDriver并非线程安全，但通过asyncio.to_thread确保每次操作在单线程执行，不会引发冲突

内容的提问来源于stack exchange，提问作者Mohamed Hussain Astro