如何使用 Python + Selenium 实现多线程爬取懂车帝全国二手车数据

在本篇博客中，我们将介绍如何使用 Python 和 Selenium 库，结合 多线程 实现高效的网页爬取。我们的目标是从懂车帝（dongchedi.com）网站抓取全国二手车的数据，包括车源的详细信息并将其保存到 CSV 文件中。

一、项目概述

我们将构建一个程序来抓取懂车帝网站的全国二手车数据，获取以下信息：

• 车辆标题（例如：品牌、型号等）
• 新车指导价
• 省钱（即当前二手车的优惠金额）
• 售价
• 车源地
• 上牌时间
• 过户次数
• 车辆厂商、能源类型、发动机数据等详细参数

最终，我们将这些数据保存到 CSV 文件中，便于后续的数据分析和使用。

二、环境准备

在开始写代码之前，需要确保你已经安装了以下几个依赖：

1. Python（3.6及以上版本）
2. Selenium（用于与网页进行交互）
3. ChromeDriver（用于驱动Chrome浏览器）
4. ThreadPoolExecutor（Python内置模块，用于多线程）

安装这些依赖的命令如下：

pip install selenium

同时需要下载 ChromeDriver，下载地址：ChromeDriver，并确保版本与你的Chrome浏览器版本一致。

三、爬虫设计

我们的爬虫主要分为三个部分：

1. 获取网页链接：通过爬取多个分页，收集每一页上的二手车详情页的链接。
2. 抓取详情页数据：对于每个详情页，抓取具体的车辆信息。
3. 保存数据：将抓取到的数据保存到本地的CSV文件中。

为了提高效率，我们使用了 多线程 来并发处理多个页面的抓取，减少等待时间。

四、代码实现

下面是我们完整的代码实现。

1. 导入依赖库

import csv
import threading
import time
import random
from concurrent.futures import ThreadPoolExecutor
from selenium.webdriver.chrome.service import Service
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.common.exceptions import NoSuchElementException
import logging

2. 配置日志

我们使用 Python 的 logging 模块来记录程序的运行日志，方便调试。

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
lock = threading.Lock()

3. 数据转换函数

有些字段需要转换，比如“新车指导价”和“省钱”字段都带有“万”字，需要去除后进行数值计算。

def convert_to_number(price_str):
    price_str = price_str.replace("万", "")
    return float(price_str)

def calculate_price(xc_price, sq):
    cleaned_xc_price = convert_to_number(xc_price)
    cleaned_sq = convert_to_number(sq)
    price = round(cleaned_xc_price - cleaned_sq, 2)
    return price

4. 获取URL列表

通过解析每一页的二手车列表，获取每辆车的详情页链接。

def get_url(num1, num2):
    data = []
    service = Service("./chro_driver/chromedriver.exe")
    driver = webdriver.Chrome(service=service)

    for page_num in range(num1, num2):
        url = f"https://www.dongchedi.com/usedcar/x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-x-1-{page_num}-x-x-x-x-x"
        logging.info(f"正在访问页面: {url}")
        driver.get(url)
        time.sleep(random.uniform(1, 3))

        try:
            urlList = driver.find_elements(By.XPATH, '//*[@id="__next"]/div[1]/div[2]/div/div/div[2]/ul/li/a')
            urlList = [i.get_attribute("href") for i in urlList]

            for car_url in urlList:
                if car_url not in data:
                    logging.info(f"正在爬取 {car_url}")
                    car_data = get_data(car_url, driver)
                    if car_data:
                        data.append(car_data)
                        save_data(data)
                time.sleep(random.uniform(2, 4))
        except NoSuchElementException as e:
            logging.error(f"出现错误: {e}")
            break

5. 抓取详情页数据

对于每一辆车，我们获取其详细的价格、车源地、过户次数等信息。

def get_data(url, driver):
    driver.get(url)
    time.sleep(random.uniform(2, 4))
    try:
        title = driver.find_element(By.XPATH, '//*[@id="__next"]/div/div[2]/div/div[2]/div[2]/div[1]/h1/span').text
        xc_price = driver.find_element(By.XPATH, '//*[@id="__next"]/div/div[2]/div/div[2]/div[2]/div[3]/div/div/p[1]').text[6:11]
        sq = driver.find_element(By.XPATH, '//*[@id="__next"]/div/div[2]/div/div[2]/div[2]/div[3]/div/div/p[2]').text[5:10]
        price = calculate_price(xc_price, sq)
        cyd = driver.find_element(By.XPATH, '//*[@id="1"]/div[2]/div[1]/div/div[2]/p[1]').text
        spsj = driver.find_element(By.XPATH, '//*[@id="1"]/div[2]/div[1]/div/div[4]/p[1]').text
        ghcs = driver.find_element(By.XPATH, '//*[@id="1"]/div[2]/div[1]/div/div[3]/p[1]').text
        car_detail_url = driver.find_element(By.XPATH, '//*[@id="1"]/div[2]/div[1]/a').get_attribute("href")
        
        driver.get(car_detail_url)
        time.sleep(random.uniform(1, 3))
        
        cs = driver.find_element(By.XPATH, '//*[@id="__next"]/div/div/div/div[2]/div[2]/div[2]/div[3]/div[2]/div').text
        cljb = driver.find_element(By.XPATH, '//*[@id="__next"]/div/div/div/div[2]/div[2]/div[2]/div[4]/div[2]/div').text
        nylx = driver.find_element(By.XPATH, '//*[@id="__next"]/div/div/div/div[2]/div[2]/div[2]/div[5]/div[2]/div').text
        fdj = driver.find_element(By.XPATH, '//*[@id="__next"]/div/div/div/div[2]/div[2]/div[2]/div[7]/div[2]/div').text
        sssj = driver.find_element(By.XPATH, '//*[@id="__next"]/div/div/div/div[2]/div[2]/div[2]/div[6]/div[2]/div').text
        
        return {
            "标题": title,
            "售价": price,
            "新车指导价": xc_price,
            "省钱": sq,
            "厂商": cs,
            "能源类型": nylx,
            "发动机数据": fdj,
            "上市时间": sssj,
            "车源地": cyd,
            "上牌时间": spsj,
            "过户次数": ghcs
        }
    except Exception as e:
        logging.error(f"获取数据时出错: {e}")

6. 保存数据

每次抓取到新的数据后，我们将其保存到 CSV 文件中。

def save_data(data):
    with lock:
        with open('懂车帝全国二手车数据表.csv', 'a', newline='', encoding='utf-8') as f:
            writer = csv.writer(f)
            if f.tell() == 0:
                writer.writerow(['标题', '售价', '新车指导价', '省钱', '厂商', '能源类型', '发动机数据', '上市时间', '车源地', '上牌时间', '过户次数'])
            for item in data:
                writer.writerow([
                    item.get('标题'),
                    item.get('售价'),
                    item.get('新车指导价'),
                    item.get('省钱'),
                    item.get('厂商'),
                    item.get('能源类型'),
                    item.get('发动机数据'),
                    item.get('上市时间'),
                    item.get('车源地'),
                    item.get('上牌时间'),
                    item.get('过户次数')
                ])

7. 主函数

使用 ThreadPoolExecutor 来创建多个线程，并将爬取任务提交给线程池。

if __name__ == '__main__':
    pool = ThreadPoolExecutor(max_workers=8)
    future1 = pool.submit(get_url, 1, 21)
    future2 = pool.submit(get_url, 21, 41)
    future3 = pool

通过本篇博客，我们展示了如何使用 Python 和 Selenium 实现一个多线程的二手车数据爬虫。程序通过抓取懂车帝网站的全国二手车信息，获取车辆的基本信息及详细参数，并将数据保存为 CSV 文件。我们还利用 多线程（通过

ThreadPoolExecutor

）来提高爬取效率，缩短抓取时间。通过这种方式，可以有效地抓取大规模的网页数据，并且保持程序的高效性和稳定性。