使用SeleniumBase与不同代理的多线程

Question

我的任务：

我的任务是为每个页面创建一个驱动程序，每个驱动程序都有自己的代理，并且要和其他驱动程序一起并行解析这些页面。此外，在从页面收集到链接后，我需要销毁所有驱动程序，然后重新创建新的驱动程序，确保每个驱动程序都有自己独特的代理，并且它们能够并行解析来自不同页面的产品。

问题：

问题是，每个创建的驱动程序都使用了来自 pool_proxies 的相同 IP 地址。

代码：

import concurrent.futures
import sys
from selenium.webdriver import ActionChains
import time
from selenium.webdriver.common.keys import Keys

sys.argv.append("-n")

pages = 3
pool_proxies_for_pages = ['proxy0', 'proxy1', 'proxy2']
pool_proxies_for_products = ['proxy5', 'proxy6', 'proxy7']

def create_undetected_webdriver(proxy):
    driver = Driver(uc=True, proxy=proxy, agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36')
    return driver

def parsing_products(page, list_for_pool, links):
    driver = create_undetected_webdriver(list_for_pool[page])
    #go through the products
    for link in links:
        driver.get(link)

def parsing_pages(page, proxy):
    driver = create_undetected_webdriver(proxy)
    url = f'https://www.ebay.com/e/_electronics/shop-all-ebay-refurbished-cell-phones?_pgn={page}'
    driver.get(url)

    #scrolling down
    ActionChains(driver).send_keys(Keys.END).perform()
    time.sleep(7)

    # get links
    elements = driver.find_elements("xpath", "//a[@tabindex='-1']")
    links = []
    for i in elements:
        link = i.get_attribute("href")
        links.append(link)

    #destroy driver
    driver.quit()
    #parsing products
    parsing_products(page, pool_proxies_for_products, links)

with concurrent.futures.ThreadPoolExecutor(max_workers=pages) as executor:
    for page in range(pages):
        executor.submit(parsing_pages, page, pool_proxies_for_pages[page])

我很乐意接受任何建议。

Answer 1

使用SeleniumBase进行带认证的代理时，采用了一个解决方案，基本上是创建一个包含代理凭证的压缩文件，这些凭证会用来进行代理，然后SeleniumBase会把这个扩展加载到Chrome浏览器中。默认情况下，它假设只使用一个代理，因此如果压缩文件已经存在，它会被覆盖，这样可以节省空间和内存。如果你需要同时使用多个代理，就需要设置一个参数：multi_proxy=True，这样每个使用代理的测试都会创建一个独特命名的压缩文件。

下面是一个使用这个功能的示例脚本（采用pytest格式）：

from parameterized import parameterized
from seleniumbase import BaseCase
BaseCase.main(__name__, __file__, "-n3")

class ProxyTests(BaseCase):
    @parameterized.expand(
        [
            ["user1:pass1@host1:port1"],
            ["user2:pass2@host2:port2"],
            ["user3:pass3@host3:port3"],
        ]
    )
    def test_multiple_proxies(self, proxy_string):
        self.get_new_driver(
            undetectable=True, proxy=proxy_string, multi_proxy=True
        )
        self.driver.get("https://browserleaks.com/webrtc")
        self.sleep(30)

如果你没有通过pytest的多线程功能pytest-xdist在使用，那么你可以查看这个链接，了解如何防止线程资源冲突：https://github.com/seleniumbase/SeleniumBase/issues/2478#issuecomment-1981699298。