Python Selenium并行查找多元素始终失败问题求助

问题背景 #

使用Python Selenium时，需要根据页面中存在的特定元素执行对应代码。原循环逐个定位元素的代码可正常运行，但为提升效率尝试用asyncio实现多元素并行查找后，代码始终返回“No element was found”，但确认页面中至少存在一个目标元素。

原正常运行的循环查找代码：

elements = [
    (By.XPATH, "//div[@class='element_1'][contains(text(),'No data')]"),
    (By.XPATH,
     '//div[@ng-repeat="element_2"]/adns-data-module/div[@class="cls_1"]'),
    (By.CSS_SELECTOR, "table.table_1"),
    (By.CLASS_NAME, "Border_1")]

for locator in elements:
    try:
        print("Trying to verify locator" + str(locator))
        WebDriverWait(driver, 10).until(EC.presence_of_element_located(locator))
        print("Locator is present")
    except:
        print("Locator was not found")

核心错误原因分析 #

1. asyncio误用Selenium同步API：Selenium是纯同步阻塞库，WebDriverWait并非异步可等待对象，对其使用await完全无效，无法实现真正的异步等待，直接导致协程逻辑错误。
2. 代码缩进错误：main_2和run_main_2函数被错误缩进在find_first_element函数内部，外部调用时无法访问这两个函数，实际代码根本没有正确执行并行查找逻辑。
3. Driver线程安全问题：Selenium的WebDriver实例不支持多协程/多线程并发操作，多个协程同时操作同一个driver会引发竞争条件，导致元素查找失败。

修正方案（线程池实现并行查找） #

由于Selenium的同步特性，使用concurrent.futures.ThreadPoolExecutor实现并行查找是更适配的方案，既能达到“找到第一个存在的元素即停止”的需求，又能避免asyncio的误用问题。

修正后的并行查找代码 #

from concurrent.futures import ThreadPoolExecutor, FIRST_COMPLETED, wait
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import threading

# 为driver加锁，避免多线程并发操作引发异常
driver_lock = threading.Lock()

def check_element(driver, locator):
    try:
        with driver_lock:
            print(f"Trying to verify locator {locator}")
            element = WebDriverWait(driver, 10).until(EC.presence_of_element_located(locator))
        print(f"Locator {locator} is present")
        return locator, element
    except Exception as e:
        print(f"Locator {locator} was not found: {str(e)}")
        return None

def find_first_element(driver, elements):
    with ThreadPoolExecutor(max_workers=len(elements)) as executor:
        futures = [executor.submit(check_element, driver, locator) for locator in elements]
        done, _ = wait(futures, return_when=FIRST_COMPLETED)
        for future in done:
            result = future.result()
            if result is not None:
                return result
    return None

def run_parallel_check(driver, elements):
    result = find_first_element(driver, elements)
    if result is not None:
        print(f"The first element found is {result[0]}")
        print(f"The element's text is: {result[1].text}")
    else:
        print("No element was found")

修正后的调用代码 #

def start_button3_callback(self):      
    t = threading.Thread(target=self.button3_callback)
    t.start()

def button3_callback(self):
    options = webdriver.ChromeOptions()
    options.add_argument("--disable-exit-on-background")
    driver = webdriver.Chrome(options=options)
    
    # 先打开目标页面
    # driver.get("你的目标页面URL")
    
    run_parallel_check(driver, elements)

方案说明 #

• 用线程池替代asyncio，适配Selenium的同步阻塞特性，真正实现并行等待多个元素。
• 添加driver锁，避免多线程并发操作同一driver实例引发的异常。
• 通过wait方法的FIRST_COMPLETED参数，实现找到第一个存在的元素即停止所有任务，提升效率。

内容的提问来源于stack exchange，提问作者Tomáš Ulrich