1. 引言为什么我们需要汽车参数对比爬虫在2026年的今天汽车消费市场正处于百年未有之大变局。随着新能源技术的爆发式增长市场上充斥着琳琅满目的车型从传统的燃油车到纯电动车BEV、插电混动PHEV再到增程式和氢燃料电池车FCV。对于消费者而言做出一项购车决策变得越来越复杂。我们不仅需要对比传统的发动机参数还要对比电池容量、电机功率、自动驾驶级别L2 vs L3、百公里电耗等全新维度的数据 。在这种背景下数据成为了连接消费者与汽车厂商的桥梁。虽然像汽车之家、懂车帝、易车网等平台提供了详尽的参数配置表 但对于想要进行横向对比例如同时对比特斯拉Model Y、小鹏G7和理想L8或批量分析例如分析2026年所有新上市SUV的自动驾驶硬件配置的用户来说手动复制粘贴几百个Excel单元格简直是噩梦。汽车参数对比爬虫应运而生。它能够自动化地从多个数据源抓取结构化的车辆参数帮助我们构建私人定制的车型数据库。本文将带你使用Python 3.12及当前最新的技术栈从零开始构建一个高效、稳定、能够应对现代反爬机制的汽车参数爬虫。2. 项目准备与法律声明在开始编写代码之前有两点必须放在首位技术选型背景传统的requestsBeautifulSoup组合在面对如今普遍使用前端渲染SSR/CSR的网站时往往力不从心。因此本教程将引入httpx支持HTTP/2的异步客户端和playwright浏览器自动化作为核心抓取引擎。法律与道德声明遵守Robots协议在爬取前请检查目标网站的robots.txt如https://www.dongchedi.com/robots.txt。控制爬取频率通过在代码中设置asyncio.sleep()或time.sleep()来模拟人类操作避免对目标服务器造成压力。数据使用抓取的数据仅应用于个人学习、研究或非商业用途。如需商业使用请考虑使用官方API或联系MarkLines、JATO Dynamics等专业数据服务商 。3. 环境搭建与核心库安装我们将使用虚拟环境来管理项目依赖。打开终端执行以下命令bash# 创建虚拟环境 python -m venv venv_car_scraper # 激活虚拟环境 (Windows) venv_car_scraper\Scripts\activate # 激活虚拟环境 (MacOS/Linux) source venv_car_scraper/bin/activate # 安装核心库 pip install httpx0.27.0 parsel1.8.0 playwright1.42.0 pandas2.2.0 loguru0.7.2 # 安装 Playwright 浏览器驱动 playwright install chromiumhttpx比requests支持HTTP/2连接复用效率更高。parselScrapy框架的核心选择器支持CSS和XPath比bs4解析速度更快语法更强大。playwright用于处理由JavaScript动态加载的车型参数如通过点击“查看更多配置”才显示的底盘信息。loguru简化日志管理方便调试。4. 静态页面抓取实战以懂车帝为例懂车帝的车型参数页通常包含了大部分基础数据且结构较为清晰 。我们的目标是抓取指定车型的详细参数配置表。4.1 分析目标URL结构与数据接口打开浏览器开发者工具F12访问一款具体的车型页面如理想L8。观察网络请求Network Tab我们通常会找到两种数据来源直接HTML渲染页面源代码中直接包含参数表格。XHR/JSON接口页面加载后通过Ajax请求一个JSON文件然后前端动态填充表格。现代汽车网站为了前后端分离越来越多地采用第二种方式。在懂车帝的实践中我们可以按XHR过滤器查找名为config_spec或类似的数据接口。4.2 编写异步抓取函数利用httpx发起异步请求获取JSON数据。pythonimport httpx import asyncio from loguru import logger from typing import Dict, Any import json # 配置日志 logger.add(car_scraper.log, rotation500MB) class CarParamScraper: 汽车参数爬虫主类 使用 httpx 进行异步抓取 def __init__(self, concurrency: int 5): self.concurrency concurrency # 设置通用的请求头模拟浏览器 self.headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36, Accept: application/json, text/plain, */*, Accept-Language: zh-CN,zh;q0.9, Referer: https://www.dongchedi.com/, Origin: https://www.dongchedi.com, } # 创建支持 HTTP/2 的异步客户端 self.client httpx.AsyncClient( headersself.headers, http2True, timeout30.0, follow_redirectsTrue, limitshttpx.Limits(max_keepalive_connectionsself.concurrency) ) async def fetch_json(self, url: str) - Dict[str, Any]: 通用的 JSON 数据抓取方法 try: logger.info(f正在抓取: {url}) response await self.client.get(url) response.raise_for_status() # 抛出 4xx/5xx 错误 # 检查返回类型 if application/json in response.headers.get(content-type, ): return response.json() else: # 如果不是 JSON尝试解析 HTML 或者直接返回文本 logger.warning(f返回非JSON内容: {url}, 状态码: {response.status_code}) return {html_content: response.text} except httpx.HTTPStatusError as e: logger.error(fHTTP错误: {e} - URL: {url}) return {} except Exception as e: logger.error(f抓取异常: {e} - URL: {url}) return {} async def close(self): 关闭客户端连接 await self.client.aclose() # 示例获取特定车型的接口 # 注意此URL仅为演示格式实际接口需通过开发者工具查找 # async def demo(): # scraper CarParamScraper() # # 假设的车型数据接口 # api_url https://www.dongchedi.com/motor/config/v1/car/config/?series_idxxx # data await scraper.fetch_json(api_url) # print(json.dumps(data, indent2, ensure_asciiFalse)) # await scraper.close() # # if __name__ __main__: # asyncio.run(demo())4.3 使用Parsel解析复杂HTML如果目标网站返回的是HTML片段而非JSON我们就需要用到parsel。例如在抓取中国汽车质量网的参数表时其结构是典型的Table布局 。假设我们有如下HTML片段模拟自汽车质量网htmldiv classtab-pane table trtd发动机型号/tdtdCA4GC20TD-32/td/tr trtd排量(mL)/tdtd1989/td/tr trtd进气形式/tdtd涡轮增压/td/tr trtd最大马力(Ps)/tdtd224/td/tr /table /div解析代码pythonfrom parsel import Selector def parse_car_params_from_html(html_content: str) - dict: 使用 parsel 解析 HTML 中的参数表格 selector Selector(texthtml_content) params {} # 定位表格中的所有行 rows selector.css(div.tab-pane table tr) for row in rows: # 提取每一行的第一个td作为参数名第二个td作为参数值 key row.css(td::text).get() # 获取第一个文本 value row.css(td::text).getall() # 获取所有文本 if key and len(value) 1: # 清洗数据去除空格、换行 clean_key key.strip() clean_value value[1].strip() if len(value) 1 else if clean_key and clean_value: params[clean_key] clean_value return params # 使用示例 # with open(car_page.html, r, encodingutf-8) as f: # html f.read() # result parse_car_params_from_html(html) # print(result)5. 动态渲染页面抓取Playwright实战越来越多的汽车网站为了保护数据或实现炫酷的交互采用了大量的JavaScript渲染。例如MarkLines全球汽车产业平台在展示详细技术参数时可能需要用户点击“展开全部”才能看到完整数据 。这时httpx无法执行JS代码我们需要一个真正的浏览器——Playwright。5.1 场景设定我们需要模拟以下操作打开一个车型详情页。等待核心参数区域加载完成。点击“显示更多参数”按钮。滚动页面以加载懒加载的图片或数据。提取最终的HTML并传递给解析器。5.2 实现异步Playwright抓取器pythonfrom playwright.async_api import async_playwright, Browser, Page import asyncio from loguru import logger class DynamicCarScraper: 使用 Playwright 处理动态加载的汽车参数页面 def __init__(self, headless: bool True): self.headless headless self.playwright None self.browser: Browser None async def start(self): 启动浏览器实例 self.playwright await async_playwright().start() # 启动浏览器headlessFalse 可以观察操作过程 self.browser await self.playwright.chromium.launch( headlessself.headless, args[--disable-blink-featuresAutomationControlled] # 隐藏自动化特征 ) logger.info(Playwright 浏览器启动成功) async def get_page_content(self, url: str, wait_for_selector: str .config-table, click_selector: str None) - str: 获取经过JS渲染后的页面HTML Args: url: 目标网址 wait_for_selector: 等待该选择器出现表示核心内容加载完成 click_selector: 如果有需要点击展开的按钮提供其选择器 context await self.browser.new_context( viewport{width: 1920, height: 1080}, user_agentMozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36 ) page: Page await context.new_page() try: logger.info(f正在加载页面: {url}) # 导航到页面 await page.goto(url, wait_untilnetworkidle, timeout60000) # 等待核心选择器出现 await page.wait_for_selector(wait_for_selector, timeout10000) # 如果需要点击展开按钮 if click_selector: # 检查按钮是否存在且可见 button await page.query_selector(click_selector) if button: await button.scroll_into_view_if_needed() await button.click() logger.info(点击了展开按钮) # 等待数据加载 await page.wait_for_timeout(2000) # 等待2秒让新数据渲染 # 滚动到底部触发懒加载 await page.evaluate(window.scrollTo(0, document.body.scrollHeight)) await page.wait_for_timeout(1000) # 获取最终渲染的 HTML content await page.content() return content except Exception as e: logger.error(f页面加载失败: {e}) # 截图保存现场 await page.screenshot(patherror_screenshot.png) return finally: await page.close() await context.close() async def stop(self): 关闭浏览器 if self.browser: await self.browser.close() if self.playwright: await self.playwright.stop() logger.info(Playwright 浏览器已关闭) # 使用示例 # async def demo_dynamic(): # scraper DynamicCarScraper(headlessFalse) # 设为False看浏览器操作 # await scraper.start() # html await scraper.get_page_content( # urlhttps://www.marklines.com/cn/green_vehicles/detail/xxx, # wait_for_selector.vehicle-specs, # click_selectorbutton:has-text(显示全部规格) # ) # # 将 html 传递给 parsel 解析器 # if html: # params parse_car_params_from_html(html) # print(params) # await scraper.stop() # # if __name__ __main__: # asyncio.run(demo_dynamic())6. 并发调度与反爬策略大规模抓取时我们不能简单地使用for循环一个一个抓取那样效率太低。我们必须引入并发控制和反爬策略。6.1 信号量控制并发数使用asyncio.Semaphore来控制同时进行的请求数量防止本地资源耗尽或对目标服务器造成过大压力。6.2 IP代理与请求间隔针对像Cars.com这类对反爬虫要求严格的网站官方甚至建议使用住宅代理 。我们可以集成代理中间件。pythonimport random from typing import List class SmartScheduler: 智能调度器管理代理和抓取间隔 def __init__(self, proxy_list: List[str] None): self.proxy_list proxy_list or [] self.semaphore asyncio.Semaphore(10) # 最大并发10 self.last_request_time {} def get_random_proxy(self): 随机获取一个代理 if not self.proxy_list: return None proxy random.choice(self.proxy_list) return {http://: proxy, https://: proxy} async def controlled_fetch(self, scraper: CarParamScraper, url: str, domain: str): 带并发控制和请求间隔的抓取方法 async with self.semaphore: # 对同一域名实施请求间隔限制 now asyncio.get_event_loop().time() if domain in self.last_request_time: time_since_last now - self.last_request_time[domain] wait_time max(0, 2.0 - time_since_last) # 同一域名至少间隔2秒 if wait_time 0: logger.debug(f域名 {domain} 等待 {wait_time:.2f} 秒) await asyncio.sleep(wait_time) # 更新最后请求时间 self.last_request_time[domain] asyncio.get_event_loop().time() # 执行抓取 # 如果有代理可以在 httpx client 中设置 proxies # 这里简化为直接调用 fetch_json data await scraper.fetch_json(url) return data6.3 批量任务处理假设我们有一个包含多个车型ID的列表需要批量抓取pythonasync def batch_scrape(series_ids: List[str]): 批量抓取多个车型 scraper CarParamScraper(concurrency5) scheduler SmartScheduler(proxy_list[http://user:passproxy_ip:port]) # 示例代理 tasks [] base_url https://api.dongchedi.com/motor/config/v1/car/config/?series_id{} for sid in series_ids: url base_url.format(sid) # 从URL中提取域名用于限速 domain dongchedi.com task scheduler.controlled_fetch(scraper, url, domain) tasks.append(task) # 并发执行所有任务 results await asyncio.gather(*tasks, return_exceptionsTrue) # 处理结果 successful_data [] for i, result in enumerate(results): if isinstance(result, Exception): logger.error(f抓取车型 {series_ids[i]} 失败: {result}) elif result: successful_data.append({ series_id: series_ids[i], data: result }) await scraper.close() return successful_data7. 数据清洗与结构化存储抓取到的数据往往是杂乱的有的字段是字符串“暂无”有的是None有的包含单位“mm”需要单独提取数值。我们需要进行统一的清洗。7.1 使用Polars/Pandas进行清洗pythonimport pandas as pd import re def clean_price(value): 清洗价格字段提取数字 if isinstance(value, (int, float)): return float(value) if not value or value in [暂无, -, ]: return None # 匹配数字和小数点 numbers re.findall(r\d\.?\d*, str(value)) return float(numbers[0]) if numbers else None def clean_power(value): 清洗功率字段提取千瓦数 if not value: return None # 假设格式为 150kW 或 150 kW match re.search(r(\d(?:\.\d)?)\s*kW, str(value), re.IGNORECASE) return float(match.group(1)) if match else None def transform_to_dataframe(raw_data: List[dict]) - pd.DataFrame: 将原始数据转换为结构化的DataFrame # 扁平化处理 flattened [] for item in raw_data: # 假设 item[data] 是包含参数的字典 params item.get(data, {}).get(config, {}) base_info { series_id: item[series_id], car_name: params.get(car_name), price_min: clean_price(params.get(官方指导价, {}).get(min)), price_max: clean_price(params.get(官方指导价, {}).get(max)), engine_power: clean_power(params.get(发动机最大功率)), battery_capacity: clean_price(params.get(电池容量, ).replace(kWh, )), level: params.get(级别), length: clean_price(params.get(长*宽*高, ).split(*)[0]) if params.get(长*宽*高) else None, } flattened.append(base_info) df pd.DataFrame(flattened) return df # 使用示例 # df transform_to_dataframe(successful_data) # print(df.head()) # 保存到CSV # df.to_csv(cars_2026.csv, indexFalse, encodingutf-8-sig)8. 构建汽车知识图谱的初步尝试有了结构化的数据我们可以尝试构建一个简单的知识图谱以发现车型之间的关联。例如使用NetworkX库来展示“同价位区间”“同动力类型”的车型关系图 。pythonimport networkx as nx import matplotlib.pyplot as plt def build_simple_graph(df: pd.DataFrame): 构建一个简单的车型推荐图谱 如果两辆车价格区间相近±2万且动力类型相同则建立连接 G nx.Graph() # 添加节点 for idx, row in df.iterrows(): if pd.notna(row[car_name]): G.add_node(row[car_name], pricerow[price_min], levelrow[level]) # 添加边关联 cars list(G.nodes(dataTrue)) for i, (car1, attr1) in enumerate(cars): for j, (car2, attr2) in enumerate(cars): if i j: continue # 判断条件价格相近且级别相同 if (attr1.get(level) and attr1.get(level) attr2.get(level) and attr1.get(price) and attr2.get(price) and abs(attr1[price] - attr2[price]) 5): # 5万以内 G.add_edge(car1, car2, weight相近竞品) # 可视化 plt.figure(figsize(12, 8)) pos nx.spring_layout(G, k2, iterations50) nx.draw(G, pos, with_labelsTrue, node_colorlightblue, node_size1500, font_size8, font_familySimHei) plt.title(车型竞品关系图谱基于价格和级别) plt.show()9. 完整项目代码结构为了让整个项目更易于维护和扩展建议采用以下结构textcar_param_spider/ │ ├── main.py # 主程序入口 ├── config.py # 配置文件URL、代理列表、并发数等 ├── requirements.txt # 依赖列表 │ ├── spiders/ # 爬虫模块 │ ├── __init__.py │ ├── base.py # 基础爬虫类 │ ├── dongchedi_spider.py # 懂车帝专用爬虫 │ └── marklines_spider.py # MarkLines 专用爬虫需动态渲染 │ ├── pipelines/ # 数据处理管道 │ ├── __init__.py │ ├── cleaner.py # 数据清洗 │ └── storage.py # 存储到CSV/数据库 │ ├── middleware/ # 中间件 │ ├── __init__.py │ ├── scheduler.py # 请求调度与限速 │ └── proxy.py # 代理管理 │ ├── utils/ # 工具函数 │ ├── logger.py # 日志配置 │ └── parser.py # HTML/JSON解析通用方法 │ └── output/ # 输出文件夹 └── cars_data.csvmain.py示例pythonimport asyncio from spiders.dongchedi_spider import DongChediSpider from pipelines.storage import save_to_csv from utils.logger import logger async def run(): # 要抓取的车型ID列表 series_ids [1234, 5678, 91011] # 实际ID需要从官网获取 # 初始化爬虫 spider DongChediSpider(concurrency8) try: # 执行抓取 raw_data await spider.batch_fetch(series_ids) # 清洗数据 from pipelines.cleaner import clean_car_data cleaned_data clean_car_data(raw_data) # 保存结果 save_to_csv(cleaned_data, output/cars_2026.csv) logger.success(f抓取完成共获取 {len(cleaned_data)} 条有效数据) except Exception as e: logger.error(f程序运行失败: {e}) finally: await spider.close() if __name__ __main__: asyncio.run(run())10. 总结与展望通过本文的实践我们不仅掌握了一套基于Python最新技术栈httpxplaywrightasyncio的爬虫开发方法还深入了解了汽车垂直网站的数据结构和反爬特征。技术要点回顾使用httpx.AsyncClient实现高效的HTTP/2并发请求。利用parsel快速解析HTML/XML文档。借助playwright攻克动态渲染的难题模拟真实用户点击和滚动。通过信号量和域名限速实现友好抓取。使用pandas对抓取到的汽车参数进行清洗和结构化存储。