时间戳

首页 > 解决方案 > 在 selenium 中一次执行多个搜索

python - 在 selenium 中一次执行多个搜索

问题描述

我有一个长度不等的搜索查询列表。目前我正在做一个又一个的搜索。这显然需要一段时间。所以我希望我可以同时使用多个窗口/浏览器。

每次搜索必须执行的代码如下所示：

for query in queries:
    browser.get(url)
    search_bar = browser.find_element('xpath', XPATH).send_keys(query)

    time.sleep(wait_time)

    try:
        ...
        return_data.append(...)

我的一个想法是创建一个这样的浏览器列表：

browsers = [webdriver.Chrome(r'file_path') for i in range(num_of_browsers)]

但我不知道如何在不阻止其他浏览器的情况下向浏览器分配问题。也许使用线程？

标签： pythonselenium

解决方案

过了一段时间，我得到了一些工作：

自定义浏览器类

class Browser(Chrome):
    def __init__(self, path: str, identifier: int):
        super().__init__(path)
        self.identifier = identifier
        self.ready = True
        self.thread: threading.Thread = threading.Thread()

主要的 BulkSearch 类

class BulkSearch:
    threads = []
    return_data = []  # Not ordered
    url = URL

    def __init__(self, queries, num_of_browsers=4):
        self.queries = (i for i in queries)
        self.num_of_queries = len(queries)
        self.browsers = [Browser(PATH, i) for i in range(num_of_browsers)]

    def start_search(self):
        for _ in range(self.num_of_queries):
            current_query = next(self.queries)
            ready_browsers = [b for b in self.browsers if b.ready]

            while not ready_browsers:
                time.sleep(1)
                ready_browsers = [b for b in self.browsers if b.ready]

            current_browser = ready_browsers[0]

            self.browsers[current_browser.identifier].ready = False
            self.browsers[current_browser.identifier].thread = threading.Thread(target=self.search, args=[current_browser, current_query])
            self.browsers[current_browser.identifier].thread.start()

        while not all([b.ready for b in self.browsers]):
            time.sleep(0.5)

        quit_thread = threading.Thread(target=self.quit_browsers)
        quit_thread.start()

        return self.return_data

    def search(self, browser, query):
        browser.get(self.url)
        search_bar = browser.find_element('xpath', XPATH)
        search_bar.clear()
        search_bar.send_keys(query)


        try:
            self.return_data.append(DATA)

        except (EXCEPTIONS):
            self.return_data.append(NO RESULT)

        finally:
            self.browsers[browser.identifier].ready = True

    def quit_browsers(self):
        for browser in self.browsers:
            browser.quit()

这是通过扩展 Chrome-Driver 类来实现的，它有一个线程和一个分配的状态。当浏览器工作时，状态设置为未准备好，这样我就可以浏览已经完成工作的浏览器。

这样，当查询少于浏览器时，它也可以工作。

推荐阅读