Converting SeleniumWire's undetected_chromedriver to use multi-threading in SeleniumBase.

[ismail-amouma]

Hello, everyone. Please, I need help with this. I have a code that uses Seleniumwire uc WebDriver, and it runs normally when I execute it for just one link. However, when I attempt to run it with multiple links using threading, it generates a lot of errors related to proxies due to the libraries I heard that the best way to resolve these errors is to convert it to a Selenium-based solution. Can anyone please help me with how I can do that

My Code

import requests
from fake_useragent import UserAgent
from bs4 import BeautifulSoup
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import seleniumwire.undetected_chromedriver as uc
from concurrent.futures import ThreadPoolExecutor

import csv

import time 
start=time.time()
def get_leads():
    url="https://www.zillow.com/xml/sitemaps/us/hdp/for-sale-by-agent/latest.xml.gz"
    ua = UserAgent()
    header = {'User-Agent':str(ua.random)}
    req=requests.get(url, headers=header)
    soup=BeautifulSoup(req.content, "html.parser")
    links=soup.find_all("loc")
    links_list=[link.text for link in links]
    print(len(links_list))
    return links_list


def initialize_driver():

    options = {     
        'exclude_hosts': ['google-analytics.com'],
        'disable_capture': True,
        'disable_encoding': True ,
        'verify_ssl': False ,
        'proxy': {
            'http': '',
            'https': ''
        }
    }
    driver = uc.Chrome(seleniumwire_options=options,headless=True,no_sandbox=False)
    return driver

def sen_email(link):
    
    try:
        driver = initialize_driver()

        driver.get(link)
        print("getting driver")
        button = WebDriverWait(driver, 10).until(EC.visibility_of_element_located((By.XPATH,  '//button[@data-cft-name="contact-button-message"]')))
        button.click()
        print("button clicked")
        
        name_input = WebDriverWait(driver, 10).until(EC.visibility_of_element_located((By.XPATH, '(//input[@aria-invalid="false"])[1]')))
        phone_input = WebDriverWait(driver, 10).until(EC.visibility_of_element_located((By.XPATH, '(//input[@aria-invalid="false"])[2]')))
        email_input = WebDriverWait(driver, 10).until(EC.visibility_of_element_located((By.XPATH, '(//input[@aria-invalid="false"])[3]')))
        message_input = WebDriverWait(driver, 10).until(EC.visibility_of_element_located((By.XPATH, '//textarea[@aria-invalid="false"]')))
        name_input.send_keys("amoulamaa")

        phone_input.send_keys("(585) 282-8888")

        email_input.send_keys('amoumaia@gmail.com')

        message_input.clear()
        message_input.send_keys("hi i want to buy this proporty ")
        driver.execute_script("window.stop();")
        submit_button = WebDriverWait(driver, 10).until(EC.visibility_of_element_located((By.XPATH, '//div[@class="StyledFormField-c11n-8-84-3__sc-24oslp-0 kUPoTD"]/button')))
        driver.execute_script("arguments[0].click();", submit_button)
        time.sleep(2)
        try:
            update_data_button = driver.find_element(By.XPATH, '//button[@data-cft-name="tour-upgrade-date-form-reject-button"]')
            update_data_button.click()
            print("update_data_button")
            time.sleep(2)

            try:
                time_to_buy = driver.find_element(By.XPATH, '(//input[@name="timelineToBuy"])[1]')
                time_to_buy.click()
                availble_to_chat = driver.find_element(By.XPATH, '(//input[@name="isBethAvailable"])[1]')
                availble_to_chat.click()

                update_data_button = driver.find_element(By.XPATH, '//div[@class="StyledLoadingMask-c11n-8-84-3__sc-felv1f-0 dWALPG StyledLoadingButton-c11n-8-84-3__sc-1rron1i-0 jAEgyz"]/div[1]/button')
                update_data_button.click()
                time.sleep(2)
                print("update_data_button with time_to_buy")
            except:
                try:

                    button_zbi = driver.find_element(By.XPATH, '//button[@data-cft-name="post-submit-abad-next-button"]')
                    button_zbi.click()
                    time.sleep(2)
                except:
                    pass
        except:
            try:
                time_to_buy = driver.find_element(By.XPATH, '(//input[@name="timelineToBuy"])[1]')
                time_to_buy.click()
                availble_to_chat = driver.find_element(By.XPATH, '(//input[@name="isBethAvailable"])[1]')
                availble_to_chat.click()

                update_data_button = driver.find_element(By.XPATH, '//div[@class="StyledLoadingMask-c11n-8-84-3__sc-felv1f-0 dWALPG StyledLoadingButton-c11n-8-84-3__sc-1rron1i-0 jAEgyz"]/div[1]/button')
                update_data_button.click()
                time.sleep(2)
            except:
                try:
                    button_zbi = driver.find_element(By.XPATH, '//button[@data-cft-name="post-submit-abad-next-button"]')
                    button_zbi.click()
                    time.sleep(2)
                except:
                    pass
        dic = {
            "link": link,
            "email": "sent successfully"
        }
    except Exception as e:
        dic = {
            "link": link,
            "email": f"Error: {str(e)}"
        }
    finally:
        try:
            driver.close()
        except:
            pass
    return dic
def contact_agent():
    leads_list = get_leads()
    data_list=[]
    test_leads=leads_list[:10]
    with ThreadPoolExecutor() as executor:
        for item in executor.map(sen_email, test_leads):
            data_list.append(item)
    return data_list

list=contact_agent()
end=time.time()
print(end-start)
with open('leads.csv', 'w', newline='') as csvfile:
    fieldnames = ['link', 'email']
    writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
    writer.writeheader()
    for item in list:
        writer.writerow(item)

[ismail-amouma]

I just need some help with the '''initialize_driver()''' function and setting it up with SeleniumBase. Can anyone assist a fellow newbie here?

[ismail-amouma]

i tried this
'''
def initialize_driver():
driver = Driver(uc=True)
return driver
'''
but i need a way to add the proxy to it

[ismail-amouma]

If you're encountering issues with Selenium wire  and need to use a proxy, you can solve it by following these steps:

Create an instance of the `Driver` class with the desired proxy settings. Replace `'username:password@ip:port'` with your actual proxy credentials and address:

   ```python
   from seleniumbase import Driver

   driver = Driver(uc=True, proxy='username:password@ip:port')

[mdmintz]

If you're looking to run multi-threaded tests in UC Mode with a proxy, here's the scoop:

Multi-threading in UC Mode is possible if you use pytest multi-threading provided by pytest-xdist.

You'll need to use command-line options for this format, eg --uc to activate UC Mode, -n4 (for 4 parallel processes, etc), and --proxy=user:pass@host:port to set proxy settings.

Below is a sample run command:

pytest --uc -n4

(Add --proxy=user:pass@host:port to include proxy settings.)

Here's a sample file that uses @pytest.mark.parametrize() to turn one test into four tests when run with pytest:

import pytest

@pytest.mark.parametrize("", [[]] * 4)
def test_multi_threaded(sb):
    sb.driver.get("https://nowsecure.nl/#relax")
    try:
        sb.assert_text("OH YEAH, you passed!", "h1", timeout=5.25)
        sb.post_message("Selenium wasn't detected!", duration=2.8)
        sb._print("\n Success! Website did not detect Selenium! ")
    except Exception:
        sb.fail('Selenium was detected! Try using: "pytest --uc"')

Here's the output when running that file with pytest --uc -n4:

pytest test_multi_uc.py --uc -n4
============================ test session starts =============================
platform darwin -- Python 3.11.4, pytest-7.4.0, pluggy-1.2.0
rootdir: ~/github/SeleniumBase/examples
configfile: pytest.ini
plugins: html-2.0.1, rerunfailures-12.0, cov-4.1.0, metadata-3.0.0, ordering-0.6, xdist-3.3.1, seleniumbase-4.17.13
4 workers [4 items]     

 Success! Website did not detect Selenium! 

 Success! Website did not detect Selenium! 
..
 Success! Website did not detect Selenium! 
.
 Success! Website did not detect Selenium! 
.
============================= 4 passed in 9.38s ==============================

Some websites may block you if they detect multiple simultaneous connections like that. Be careful where you go.

Note that there are different syntax formats. See: https://github.com/seleniumbase/SeleniumBase/blob/master/help_docs/syntax_formats.md

Also note that pytest-xdist (the multi-processing library for pytest) is more advanced than https://docs.python.org/3/library/concurrent.futures.html for complex multi-threading of this nature.

[ismail-amouma]

Can I do this without using the command line? I have around 50,000 links to visit with different drivers, and I want to eventually turn this application into web app to track the progress is there a way to pass the commaned as arguments.

[mdmintz]

That's what there is if you want multithreading to work. pytest-xdist is the key.

[ismail-amouma]

Okay, thank you, mdmintz, for the library. It worked perfectly.