基于Python-Selenium的Google Patents并行爬取问题求助

核心问题分析 #

• Selenium的WebDriver实例不是线程安全的，多线程共用同一个driver会导致请求上下文混乱，出现数据获取失败或冲突
• 每个线程单独初始化driver会大量消耗系统资源（浏览器进程），效率低下
• 索引（index）操作未做线程同步，多线程同时修改会引发异常

可行解决方案 #

方案1：使用线程本地存储（Thread Local）管理WebDriver #

给每个线程分配独立的driver实例，避免共用冲突，同时复用线程内的driver：

import threading
from concurrent.futures import ThreadPoolExecutor
from selenium import webdriver
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import csv
import time
import random

# 线程本地存储，每个线程有自己的driver实例
thread_local = threading.local()

def get_driver():
    if not hasattr(thread_local, "driver"):
        # 配置Chrome选项，无头模式节省资源
        options = webdriver.ChromeOptions()
        options.add_argument("--headless=new")
        options.add_argument("--disable-gpu")
        thread_local.driver = webdriver.Chrome(options=options)
    return thread_local.driver

def scrape_patent(patent_url, index):
    driver = get_driver()
    try:
        driver.get(patent_url)
        # 随机延迟，规避反爬
        time.sleep(random.uniform(1.5, 3))
        
        # 显式等待元素加载，提升稳定性
        assignee = WebDriverWait(driver, 10).until(
            EC.presence_of_element_located(("xpath", "//span[@itemprop='assignee']"))
        ).text
        status = WebDriverWait(driver, 10).until(
            EC.presence_of_element_located(("xpath", "//div[@class='status']"))
        ).text
        app_date = WebDriverWait(driver, 10).until(
            EC.presence_of_element_located(("xpath", "//time[@itemprop='filingDate']"))
        ).get_attribute("datetime")
        grant_date = WebDriverWait(driver, 10).until(
            EC.presence_of_element_located(("xpath", "//time[@itemprop='grantDate']"))
        ).get_attribute("datetime")
        
        # 返回带索引的数据，保证后续写入顺序正确
        return (index, assignee, status, app_date, grant_date)
    except Exception as e:
        print(f"爬取专利[{index}]失败: {str(e)}")
        return (index, None, None, None, None)

def main():
    # 替换为你的专利URL列表
    patent_urls = ["https://patents.google.com/patent/USxxxxxxx", "https://patents.google.com/patent/USyyyyyyy"]
    results = [None] * len(patent_urls)  # 预分配结果列表，避免线程顺序混乱
    
    # max_workers建议设为3-5，根据机器配置调整
    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = [executor.submit(scrape_patent, url, idx) for idx, url in enumerate(patent_urls)]
        for future in futures:
            idx, assignee, status, app_date, grant_date = future.result()
            results[idx] = [assignee, status, app_date, grant_date]
    
    # 写入CSV文件
    with open("patents.csv", "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(["当前权利人", "状态", "申请日", "授权日"])
        writer.writerows(results)
    
    # 关闭所有线程的driver
    if hasattr(thread_local, "driver"):
        thread_local.driver.quit()

if __name__ == "__main__":
    main()

关键优化点 #

• 用threading.local()为每个线程绑定独立的driver，彻底避免多线程上下文冲突
• 预分配结果列表，通过索引写入数据，保证最终CSV顺序与原URL列表一致，解决index异常问题
• 加入显式等待和随机延迟，提升爬取稳定性，规避反爬机制

方案2：改用进程池（ProcessPoolExecutor）替代线程池 #

如果线程池仍存在资源冲突问题，可改用进程池，每个进程拥有独立的driver实例：

from concurrent.futures import ProcessPoolExecutor
from selenium import webdriver
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import csv
import time
import random

def scrape_patent(patent_url, index):
    options = webdriver.ChromeOptions()
    options.add_argument("--headless=new")
    options.add_argument("--disable-gpu")
    driver = webdriver.Chrome(options=options)
    try:
        driver.get(patent_url)
        time.sleep(random.uniform(1.5, 3))
        
        assignee = WebDriverWait(driver, 10).until(
            EC.presence_of_element_located(("xpath", "//span[@itemprop='assignee']"))
        ).text
        status = WebDriverWait(driver, 10).until(
            EC.presence_of_element_located(("xpath", "//div[@class='status']"))
        ).text
        app_date = WebDriverWait(driver, 10).until(
            EC.presence_of_element_located(("xpath", "//time[@itemprop='filingDate']"))
        ).get_attribute("datetime")
        grant_date = WebDriverWait(driver, 10).until(
            EC.presence_of_element_located(("xpath", "//time[@itemprop='grantDate']"))
        ).get_attribute("datetime")
        
        return (index, assignee, status, app_date, grant_date)
    except Exception as e:
        print(f"爬取专利[{index}]失败: {str(e)}")
        return (index, None, None, None, None)
    finally:
        driver.quit()

def main():
    patent_urls = ["https://patents.google.com/patent/USxxxxxxx", "https://patents.google.com/patent/USyyyyyyy"]
    results = [None] * len(patent_urls)
    
    # 进程数建议设为2-3，避免浏览器进程过多耗尽资源
    with ProcessPoolExecutor(max_workers=2) as executor:
        futures = [executor.submit(scrape_patent, url, idx) for idx, url in enumerate(patent_urls)]
        for future in futures:
            idx, assignee, status, app_date, grant_date = future.result()
            results[idx] = [assignee, status, app_date, grant_date]
    
    with open("patents.csv", "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(["当前权利人", "状态", "申请日", "授权日"])
        writer.writerows(results)

if __name__ == "__main__":
    main()

注意事项 #

• 进程池的max_workers要远小于线程池，因为每个进程会启动独立的浏览器，资源消耗更大
• 进程间数据完全隔离，无需额外做索引同步，同样通过预分配列表保证数据顺序

内容的提问来源于stack exchange，提问作者Ignacio Galara