基于Selenium的无限滚动页面商品详情高效爬取方案问询

核心需求 #

• 采集商品详情页的图片链接、名称、分类、短描述、价格、库存状态、SKU及附加信息
• 处理商品链接点击时弹出的相似商品弹窗
• 完成详情采集后高效继续爬取剩余商品，采用"先爬所有商品链接+多线程处理详情页"的方案

最优实现思路 #

1. 批量获取商品详情页URL：用Selenium处理列表页无限滚动，一次性收集所有商品的详情页链接，避免重复滚动操作
2. 绕过弹窗直接访问详情页：直接通过商品URL跳转详情页，无需在列表页点击，自然规避相似商品弹窗问题，比点击后关闭弹窗效率更高
3. 多线程并行爬取：用concurrent.futures.ThreadPoolExecutor创建多线程，每个线程独立启动Chrome实例（无头模式）并行处理详情页采集，大幅提升整体爬取速度
4. 稳定性保障：添加异常捕获避免单个商品爬取失败中断流程，用显式等待替代固定sleep提升响应效率

具体实现代码 #

1. 依赖导入 #

from selenium import webdriver
from selenium.webdriver.chrome.service import Service
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 pandas as pd
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

2. 列表页批量获取商品链接 #

def get_all_product_links():
    chrome_options = Options()
    chrome_options.add_argument("--no-sandbox")
    chrome_options.add_argument("--disable-notifications")
    chrome_options.add_argument("--window-size=1280,720")
    chrome_options.add_argument("--headless=new")  # 无头模式减少资源占用

    webdriver_service = Service("chromedriver.exe")
    browser = webdriver.Chrome(service=webdriver_service, options=chrome_options)
    wait = WebDriverWait(browser, 20)
    url = 'https://iremedy.com/search?query=Vital%20Signs%20Monitors'
    browser.get(url)

    product_links = set()  # 集合自动去重，避免重复链接
    last_link_count = 0

    while True:
        # 等待商品卡片加载完成
        item_elements = wait.until(EC.presence_of_all_elements_located((By.CSS_SELECTOR, '[class^="card"] a')))
        # 提取当前页面所有商品链接
        current_links = {item.get_attribute('href') for item in item_elements}
        product_links.update(current_links)
        current_count = len(product_links)

        print(f"已收集 {current_count} 个商品链接")

        # 判断是否加载完成：连续两次链接数量无变化则停止滚动
        if current_count == last_link_count:
            print("所有商品链接已收集完成")
            break
        last_link_count = current_count

        # 滚动到页面底部加载更多商品
        browser.execute_script("window.scrollTo(0, document.body.scrollHeight);")
        time.sleep(2)  # 等待加载，可根据网站响应速度调整

    browser.quit()
    return list(product_links)

3. 详情页数据采集函数 #

def scrape_product_detail(product_url):
    chrome_options = Options()
    chrome_options.add_argument("--no-sandbox")
    chrome_options.add_argument("--disable-notifications")
    chrome_options.add_argument("--window-size=1280,720")
    chrome_options.add_argument("--headless=new")
    chrome_options.add_argument("--disable-gpu")

    webdriver_service = Service("chromedriver.exe")
    browser = webdriver.Chrome(service=webdriver_service, options=chrome_options)
    wait = WebDriverWait(browser, 15)
    product_data = {}

    try:
        browser.get(product_url)

        # 采集核心字段
        product_data['商品名称'] = wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, 'h1.product-title'))).text.strip()
        product_data['分类'] = wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, 'div.breadcrumb a:last-child'))).text.strip()
        product_data['短描述'] = wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, 'div.short-description'))).text.strip()
        product_data['价格'] = wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, 'span.price'))).text.strip()
        product_data['库存状态'] = wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, 'span.stock'))).text.strip()
        product_data['SKU'] = wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, 'span.sku'))).text.strip()
        product_data['主图链接'] = wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, 'div.product-main-image img'))).get_attribute('src')

        # 采集附加信息（表格形式）
        additional_info = {}
        info_rows = wait.until(EC.presence_of_all_elements_located((By.CSS_SELECTOR, 'table.additional-info tr')))
        for row in info_rows:
            key = row.find_element(By.CSS_SELECTOR, 'th').text.strip()
            value = row.find_element(By.CSS_SELECTOR, 'td').text.strip()
            additional_info[key] = value
        product_data['附加信息'] = str(additional_info)  # 转字符串方便CSV存储

        print(f"完成采集：{product_data['商品名称']}")

    except Exception as e:
        print(f"采集失败 {product_url}: {str(e)}")
        product_data['商品链接'] = product_url
        product_data['错误信息'] = str(e)
    finally:
        browser.quit()

    return product_data

4. 多线程执行与数据保存 #

if __name__ == "__main__":
    # 第一步：获取所有商品链接
    product_links = get_all_product_links()
    print(f"共获取到 {len(product_links)} 个商品链接")

    # 第二步：多线程爬取详情页（线程数根据机器性能调整，建议3-5个）
    all_product_data = []
    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = [executor.submit(scrape_product_detail, link) for link in product_links]
        for future in as_completed(futures):
            result = future.result()
            all_product_data.append(result)

    # 第三步：保存为CSV文件
    df = pd.DataFrame(all_product_data)
    df.to_csv('iremedy_vital_signs_monitors.csv', index=False, encoding='utf-8-sig')
    print("数据已保存到 iremedy_vital_signs_monitors.csv")

关键优化说明 #

• 无头模式：开启--headless=new减少浏览器UI资源消耗，提升爬取速度
• 链接去重：用集合存储链接，避免重复爬取同一商品
• 动态加载判断：通过对比链接数量变化判断是否加载完成，替代原代码的固定数量停止逻辑
• 异常隔离：每个详情页采集独立捕获异常，单个商品失败不影响整体流程
• 线程数控制：限制线程数避免触发网站反爬机制，可根据实际情况调整

内容的提问来源于stack exchange，提问作者Muhammad Talha Zeb Khan