☰
二手车按地区批量获取:item_search接口接入与数据落地全流程
2026/10/14 3:57:22 网站建设 项目流程

做二手车数据相关的开发,绕不开的一个核心需求就是:按地区拉取二手车列表。不管是做行情分析、车源聚合应用,还是给用户做智能推荐,第一步都得解决“某区域当前有哪些车源在售、大概什么价位”这个问题。我前段时间在对接某汽车信息平台的 item_search 接口时,也是从两眼一抹黑开始,中间踩了不少坑,最后才把一套稳定跑通的流程沉淀下来。这篇文章就把我验证过的接入路径完整写出来:从接口认知、申请接入、参数构造、返回解析,到批量获取、异常兜底、数据落地,一条龙讲清楚。适合正在做二手车数据对接、接口调用或者区域行情分析的开发者参考,也适合第一次接触这类 item_search 风格接口的新手快速上手。

1. 接口对接前的整体认知:item_search 是什么、能解决什么问题

1.1 为什么需要按地区获取二手车列表

二手车市场和一手车不一样,地域性极强。同样是某款家用轿车,在一线城市和三四线城市的车源数量、挂牌价、流转速度完全不同。做数据类应用的时候,如果不区分地区,直接拉全国数据,出来的统计结果没有业务参考意义,车商也没法根据自己所在城市的挂牌情况做定价决策。

所以很多业务场景都需要“地区维度”的数据:比如开发一个二手车估价小工具,需要看同城同款车源的平均挂牌价;做一个垂直搜索应用,用户一进来就选城市,然后只看这个城市在售的车源;或者做车商选品分析,关注的是某个区域近期新上架了什么车、什么价位段的车源供给是否充足。这些需求背后,都需要一个稳定的接口,能够根据地区参数返回结构化的二手车列表。

平台方提供的 item_search 接口就是干这个用的。它本质上是一个“商品搜索接口”,入参是你关心的条件组合(地区、关键词、页码、排序方式等),出参是符合条件的一批车源明细。相比自己去网页上逐个城市翻查,这种方式最大的优势是:数据是结构化的、字段是固定的、调用是可编程的,拿到就能直接用。

1.2 选择哪种接入方式更稳妥

真正在项目里做数据获取,通常有三条路可走。

第一是直接找平台方申请官方开放接口。这是最理想的路径,数据授权清晰、接口规范、稳定度高。但现实情况是,这类接口往往不是对个人开发者开放的,或者开放权限需要比较严格的资质审核,申请下来周期长,还有一些调用配额限制。

第二是通过第三方数据服务商中转。很多数据服务商把平台上公开的网页信息封装成了标准 API,item_search 就是这类接口里非常典型的一种。开发者只需要订阅对应服务、拿到 API Key,就可以通过统一方式调用。这种方案最大的优点是接入快、成本相对可控,而且服务商一般已经把登录态、风控等问题处理好了,个人开发者不需要处理复杂的前置逻辑。

第三是自己在业务系统里做采集程序。这条路技术上是可行的,但维护成本很高:平台页面结构调整、访问频率控制、反爬策略升级,任何一个变化都可能导致采集失效,而且法律合规风险也相对大。对于想快速做出产品验证的业务项目来说,性价比不高。

我自己最后走的是第二条路,也就是通过数据服务商提供的 item_search 接口来做地区维度的二手车数据获取。这篇文章后面讲的,就是这条路径中的完整实操,但接口接入的原理绝大多数也适用于其他类似的商品搜索型接口。

2. 前置准备与环境搭建:申请接入、了解核心文档、配置开发环境

2.1 申请 API Key 与读懂接入文档

无论你用的是哪家数据服务商,接入的第一步都是注册账号、完成认证、申请你要调用的接口服务。一般情况下,服务商会为每个项目提供一个 API Key 和一个 Secret Key。API Key 相当于你的账号标识,请求时会随参数一起提交;Secret Key 则是用来生成签名的,用来保证请求参数没有被篡改,服务商后台会核对签名是否正确。

申请到 Key 之后,不要急着写代码,先把接口文档完整读一遍。我习惯重点看四块内容:

  • 请求地址和请求方式:确定是 GET 还是 POST,参数放在 URL 还是 body 里,如果是 GET 请求,签名规则是怎样的;
  • 参数列表:哪些是必填项、哪些是选填项、参数的类型和取值范围、地区参数的编码规则;
  • 返回结构:成功时返回什么、失败时返回什么、数据在哪个字段下、字段之间的嵌套关系;
  • 调用限制:每秒最大请求次数、每日最大调用量、超出配额后的表现。

这一步看起来简单,但特别重要。多数对接失败的问题,都源于文档里的某一行小字,比如“地区参数默认编码规则是拼音”或者“排序字段传值时区分大小写”。宁可多花半小时看文档,也不要等代码跑不起来再去排查。

2.2 本地 Python 环境与依赖库

本次实操我全程用 Python 演示。如果你本地还没装 Python,建议直接装 3.9 以上的版本,指令也基本通用。为了不污染系统环境,我习惯先建一个虚拟环境再装依赖:

mkdir car_region_search cd car_region_search python -m venv venv source venv/bin/activate # Windows 下用 venv\Scripts\activate

然后安装下面几个库:

pip install requests pandas retrying openpyxl

这几个库分别负责:发送 HTTP 请求、做数据清洗与整理、实现失败重试、把结果导出为 Excel。其中retrying是可选依赖,不用它你可以手写一个简单的重试函数,效果是一样的。对于小规模的数据获取场景,requests一个库其实也够用,加上pandas纯粹是后面整理数据方便。

2.3 先把常量都抽出来

很多刚开始做接口对接的朋友,喜欢把 API Key、URL、地区编码直接写在业务代码里。我强烈不建议这样做。接口地址可能换版本,Key 可能过期换新,地区编码成百上千个,写死在逻辑里后面维护起来非常痛苦。

我推荐的工程化做法是建一个配置文件,把所有静态信息集中管理。比如建一个config.py:

API_KEY = "your_api_key_here" SECRET_KEY = "your_secret_key_here" BASE_URL = "https://api.example.com/item_search" # 演示用的域名,替换成你自己的 DEFAULT_HEADERS = { "Content-Type": "application/x-www-form-urlencoded;charset=UTF-8", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" } # 地区编码表建议单独维护,下面举两个例子 CITY_ID_MAP = { "北京": "110000", "上海": "310000", # ... 实际有几百条 }

API_KEY和SECRET_KEY这两项比较敏感,如果你的项目要提交到代码仓库,记得用环境变量读取,别硬编码到源码里。

3. 核心参数拆解与返回结构:把请求和响应弄明白

3.1 通用请求参数与签名逻辑

item_search 这类接口,通常每个请求都要带完整认证参数。这个项目采用的是一种非常常见的签名请求协议,请求 URL 的基础格式长这样:

https://api.example.com/item_search?key=API_KEY&city_id=310000&page=1&page_size=20&sort_type=default&sign=xxxxxxxx

其中sign的值不是随便写的,而是通过对请求参数做约定排序和字符串拼接,再用Secret Key做 MD5 计算得到的。具体规则各家略有差异,我遇到的标准流程是这样的:

  1. 把所有业务参数按参数名 ASCII 码升序排列;
  2. 按“参数名=值”的形式拼成一个长字符串;
  3. 在长字符串末尾拼接Secret Key;
  4. 对完整字符串做 MD5,得到 32 位小写字符串作为签名。

对应到代码,大概是这样的:

import hashlib import requests from urllib import parse def build_sign(params, secret_key): # 过滤掉值为空的参数 sorted_keys = sorted([k for k in params.keys() if params[k] != ""]) raw_string = "&".join([f"{k}={params[k]}" for k in sorted_keys]) raw_string = f"{raw_string}{secret_key}" return hashlib.md5(raw_string.encode("utf-8")).hexdigest() def search_second_hand_car(city_id, page=1, page_size=20, keyword="", sort_type="default"): params = { "key": API_KEY, "city_id": city_id, "page": page, "page_size": page_size, "keyword": keyword, "sort_type": sort_type, } params["sign"] = build_sign(params, SECRET_KEY) resp = requests.get(BASE_URL, params=params, headers=DEFAULT_HEADERS, timeout=15) return resp.json()

签名逻辑的核心目的是防篡改,所以拼接规则、排序规则哪怕差一个符号,服务端就能识别出来。遇到 401 或者“签名错误”的报错,第一步就是检查你的拼接规则和文档是否完全一致——尤其是排序用的分隔符、是否过滤空值这两个细节。

3.2 地区参数:用编码而不是城市名

地区参数通常不直接传中文城市名,而是传地区编码。编码可能是数字 ID,也可能是拼音缩写。具体项目用的是类似city_id的数字编码,全国省市的编码表在接口文档里会有附录。

这个编码表值得单独做成一个 JSON 文件或数据表维护起来,因为你做批量拉取的时候,大概率是要遍历多个城市的。把地区和编码的对应关系准备齐全,后面循环调用的代码可以写得很简洁:

import json with open("city_id_map.json", "r", encoding="utf-8") as f: CITY_ID_MAP = json.load(f) for city_name, city_id in CITY_ID_MAP.items(): print(city_name, city_id)

如果后续要做“华东区域车源分布”,直接在这个映射表上按行政区划过滤出目标城市列表,逻辑会非常清晰。

3.3 业务参数:分页、排序和过滤条件

除了地区字段,几个高频业务参数也值得提前理解透彻。

page和page_size是分页参数。page_size表示每页返回多少条,一般有上限,常见的是 20 或 50。不要试图把page_size一次性拉到几百条,平台方通常不允许,而且单次请求返回太大也会增加数据丢失风险。正常做法是设置一个合理页大小,然后根据总页数循环请求。

sort_type决定车辆的排序方式。常见取值包括:默认综合排序、按价格升序、按价格降序、按上架时间最新、按里程数排序。这个参数在做“区域最低价车源”“最新上架车源”等场景时非常有用,比如你要做“某城市最近 7 天新上架车源”的监控,就优先选择按上架时间排序,再过滤时间字段。

keyword是可选的搜索关键词,可以用来过滤车源。比如只要某品牌或某款车型,直接在keyword里传车型名即可。但要注意,关键词匹配规则并不可见,可能是标题模糊匹配,也可能是车型参数字段匹配。所以对结果做二次清洗还是必要的,不能完全依赖搜索词的过滤精度。

3.4 返回数据结构:逐层拆开再逐字段验收

一个典型的成功响应,JSON 结构大概长这样:

{ "code": 200, "msg": "success", "data": { "total": 1823, "page": 1, "page_size": 20, "total_pages": 92, "item_list": [ { "car_id": "CA12345678", "title": "某某品牌 2020款 家用轿车 1.5L 自动档", "price": "7.80万", "original_price": "10.40万", "brand": "某品牌", "series": "某车系", "model_year": 2020, "license_time": "2020-06", "mileage": "3.5万公里", "gearbox": "自动", "displacement": "1.5L", "city": "上海", "publish_time": "2024-01-15 10:32:00", "seller_type": "个人", "source_url": "https://...", "thumb_image": "https://..." } ] } }

拿到响应后,不要急着全部存库。先检查几个关键字段的值类型:price和mileage看起来是字符串,格式化可能需要转换;license_time是 “年-月”格式,如果你后面要算车龄,需要转成日期类型;publish_time是可用的时间戳,做增量更新的时候靠它。

提取total_pages的逻辑也要注意。有些接口给的是total,需要你自己用ceil(total / page_size)算总页数;有些接口直接把total_pages返回给你。两种情况代码里都要兼容,避免不同环境下直接报错。

4. 从单次请求到批量获取:完整实操链路

4.1 封装一个稳定的请求函数

前面的search_second_hand_car是最基础的单次请求。但是在真实环境里,网络抖动、服务端限流、临时超时都可能发生,所以我在实际项目里会再套一层“带重试的请求包装”。

基于retrying库,可以很快实现一个带指数退避的重试机制:

from retrying import retry def retry_if_result_none(result): return result is None or result.get("code") != 200 @retry(stop_max_attempt_number=3, wait_random_min=1000, wait_random_max=3000, retry_on_result=retry_if_result_none) def search_with_retry(city_id, page=1, page_size=20): try: data = search_second_hand_car(city_id, page, page_size) except Exception as e: print(f"请求异常: {e}") data = None if data is None or data.get("code") != 200: print(f"返回异常: city={city_id}, page={page}, resp={data}") return None return data

面这个函数会把“请求报错”和“业务返回非成功状态”统一当作失败处理,然后最多重试 3 次。每次重试之间随机等待 1 到 3 秒,避免固定频率请求把服务端的限流触发得更厉害。这个包装函数是后面所有批量操作的基础设施。

4.2 分页抓取:一页页把数据拉全

拿到第一页响应后,如果code == 200且total_pages > 1,就需要循环拉取剩余页面。我整理的通用分页代码如下:

import math import time def fetch_all_items_by_city(city_id, max_pages=100): all_items = [] first = search_with_retry(city_id, page=1, page_size=20) if first is None: return all_items, 0 data = first.get("data", {}) all_items.extend(data.get("item_list", [])) total = data.get("total", 0) page_size = data.get("page_size", 20) total_pages = data.get("total_pages") or math.ceil(total / page_size) total_pages = min(total_pages, max_pages) print(f"城市 {city_id} 共有 {total} 条车源,总页数 {total_pages},已抓取第 1 页") for page in range(2, total_pages + 1): resp = search_with_retry(city_id, page=page, page_size=20) if resp is None: print(f"第 {page} 页失败,跳过") continue item_list = resp.get("data", {}).get("item_list", []) all_items.extend(item_list) print(f"已抓取第 {page} 页,累计 {len(all_items)} 条") # 每页之间主动休眠,控制请求频率 time.sleep(0.8) return all_items, total

注意两个细节:

第一,max_pages这个参数是安全阀。如果一个城市车源几万条,循环请求可能耗时很长,还容易触发限流。设置上限可以防止程序长时间跑飞。真的需要全量数据时,可以再分段处理,比如按价格区间拆成多个子任务去跑。

第二,循环里的time.sleep(0.8)不是摆设。很多限流策略就是“滑动窗口内请求次数不能超过 N”,如果机器带宽好、跑得快,反而容易被封。0.8 秒只是一个参考值,实际要根据你的套餐配额调整。

4.3 多城市遍历:批量执行与断点续跑

先实现一个城市,再扩展到多城市就会顺利很多。遍历城市的思路很直接:从CITY_ID_MAP里取出目标城市,逐个调用fetch_all_items_by_city,把结果合并到总列表里。

不过多城市遍历容易遇到两个问题:一是某个城市数据量特别大,导致程序长时间卡住;二是跑到一半失败了,下次执行又要从头开始。我个人的解决方案是,为每个城市单独维护一个执行进度记录,处理完一个城市就把结果写入 CSV 文件,文件名直接带上城市和时间:

def fetch_all_cities(city_list): for city_name, city_id in city_list: print(f"开始获取: {city_name}({city_id})") items, total = fetch_all_items_by_city(city_id) if items: df = pd.DataFrame(items) file_name = f"data/car_items_{city_name}_{time.strftime('%Y%m%d')}.csv" df.to_csv(file_name, index=False, encoding="utf-8-sig") print(f"城市 {city_name} 数据保存完成: {file_name}, 共 {len(df)} 条") time.sleep(1)

这样即便中间某个城市失败了,下次跑的时候只跳过已经存在的 CSV 文件,或者只重跑没有输出的城市,就能实现简单的断点续跑。

4.4 数据清洗:把字符串变成可计算的字段

接口返回的原始字段里,价格、里程都是带单位的字符串,比如"7.80万"、"3.5万公里"。这种数据给前端展示没问题,但做分析和计算必须转成数值。

我封了一个简单的清洗函数:

def parse_wan_to_number(raw): """把 '7.80万' 转成 78000,把 '3.5万公里' 转成 35000""" if not raw: return None s = str(raw).replace("万公里", "").replace("万", "") try: return float(s) * 10000 except ValueError: return None def clean_item(item): return { "car_id": item.get("car_id"), "title": item.get("title"), "price_yuan": parse_wan_to_number(item.get("price")), "original_price_yuan": parse_wan_to_number(item.get("original_price")), "mileage_km": parse_wan_to_number(item.get("mileage")), "brand": item.get("brand"), "series": item.get("series"), "model_year": item.get("model_year"), "license_date": item.get("license_time"), "publish_date": item.get("publish_time"), "city": item.get("city"), "gearbox": item.get("gearbox"), "displacement": item.get("displacement"), "source_url": item.get("source_url"), }

价格转成price_yuan之后,按价格区间筛选、计算均价、做同比环比分析都变得很容易。另外建议做一次去重处理。车源可能有同样的car_id在不同页、不同批次里重复出现,所以入库前一定要按car_id去重:

df = df.drop_duplicates(subset="car_id", keep="last")

5. 数据落地与自动化扩展:让数据持续产生价值

5.1 用 SQLite 做增量存储

CSV 适合临时看数和交换数据,但长期做数据积累,我更推荐用 SQLite 存明细。单文件、零配置、随项目走,用 Python 自带模块就能操作。

建表逻辑非常简单:

import sqlite3 conn = sqlite3.connect("car_items.db") conn.execute(""" CREATE TABLE IF NOT EXISTS car_items ( car_id TEXT PRIMARY KEY, title TEXT, price_yuan REAL, original_price_yuan REAL, mileage_km REAL, brand TEXT, series TEXT, model_year INTEGER, license_date TEXT, publish_date TEXT, city TEXT, gearbox TEXT, displacement TEXT, source_url TEXT ) """)

写入时用INSERT OR REPLACE或INSERT OR IGNORE,天然就处理了重复数据的问题。我一般用INSERT OR IGNORE,保留下次任务里新出现的车源,同时把已有车源丢弃,避免覆盖历史记录里的价格变化。如果你想跟踪同一辆车挂牌价的波动,那就应该用INSERT OR REPLACE并额外加一个update_time字段。

5.2 增量更新:关注新上架的车源

二手车数据更新频率并不需要像库存系统那么实时,但为了做“新上车源监控”,建议每天或者每半天跑一次增量更新。

判断是否为新上架车源,优先用publish_date。比如每次抓取时,只保留publish_date大于上次执行时间的车源。但如果接口的publish_date并不是可靠的“新上架时间”,你也可以换一种思路:每天都全量拉取一遍,然后和库里已有的car_id对比,凡是库里面不存在的car_id,就是新上架车源。

这种对比方案更简单,实现成本低,因为二手车整个市场日增量通常也就是千条级别,全量拉取的成本完全可接受。需要注意的点是“下架车源”——某天接口里不再出现的car_id,可以在库里标记状态为已下架,而不是直接删除。这样后续可以统计某款车的平均挂牌周期,这个指标对车商非常有价值。

5.3 定时任务与告警

数据源跑起来之后,我习惯用一个轻量级的计划任务来驱动。Windows 下可以用任务计划程序,Linux 下用 crontab,或者直接在代码里用schedule库。核心的定时任务都绕不开两件事:执行抓取、写入数据库、生成日志。

# schedule_demo.py import schedule import time def job(): print("开始执行定时抓取...") # 这里放完整抓取逻辑 pass schedule.every().day.at("09:00").do(job) schedule.every().day.at("21:00").do(job) while True: schedule.run_pending() time.sleep(60)

另外强烈建议给抓取任务加一个“失败通知”逻辑。项目初期跑单机脚本,接口挂了可以不理会;但一旦形成了定时任务,就得考虑在连续失败若干次之后发个通知出来。最简单的做法是在异常分支里调用一个钉钉群机器人 Webhook 或者邮件 SMTP,把错误信息推送出来。选哪种渠道不重要,重要的是不能等问题堆积到第二天才发现。

6. 常见问题与排查实录:我把踩过的坑都列在这里

6.1 请求返回 401 或者提示签名错误

签名错误是新手对接 item_search 接口时最常遇到的问题。如果确认API_KEY没有写错,那大概率是构建签名的拼接顺序与平台要求不一致。

常见原因有三类:

  • 参数排序时没有按 ASCII 码升序排,或者某些语言默认的排序规则(比如对下划线、大小写的处理)和平台不一致;
  • 拼接字符串时使用了&分隔,但平台要求直接以key=valuekey=value形式拼接,没有&;
  • 参数值为空时没有过滤,把空字符串也参与了签名。

我建议写一个单元测试,把文档里的示例请求参数原样带入签名函数,对比输出结果是否与文档给出的示例签名字符串一致。这一步可以提前暴露绝大多数签名问题。

6.2 返回 data 为空,但接口状态正常

这种情况通常不是接口故障,而是参数本身没有匹配到数据。常见原因:

  • city_id传错了,比如把城市名字传进去了,而接口只认数字编码;
  • 分页超出范围,比如你传page=999,而总页数只有 20,接口可能返回空列表而不是错误;
  • 关键词过窄,平台的车源分类字段和你的关键词不一致,导致搜索无结果。

排查时先不带keyword试一次,确认当前城市最基本的分页请求能正常拿到数据;再逐步加上过滤条件,定位是哪类参数把结果过滤掉了。

6.3 请求频率过高被限流,报 429 或提示风控

被限流最常见的表现是:前面的请求都正常,突然某个请求返回 429,后面连续失败。这里没什么黑魔法,就是请求太频了。

处理办法按顺序来:

  • 降低请求频率,把每页休眠从 0.3 秒提升到 1 秒甚至 2 秒;
  • 增加随机延迟,让请求间隔不完全固定,避免明显的周期规律;
  • 配好重试策略,且重试间隔必须逐步拉长,不要固定间隔快速重试;
  • 如果业务并发要求高,考虑升级套餐或者申请更高的 QPS 配额。

还有一个小技巧:把多城市遍历的逻辑改成“串行 + 小批量暂停”,而不是一个城市接着一个城市高速跑完。宁可总耗时多个十几分钟,也不要让账号被风控盯上。

6.4 返回的部分字段与线上页面不一致

有些接口为了防止数据过度采集,会做一定程度的延迟缓存,或者对部分字段脱敏。比如刚上架的车源,可能在 API 里要滞后一两个小时才能检索到;某些高价车辆的卖家联系方式,也会被隐藏,或者只在页面展示而不在接口明细里返回。

遇到这种情况,先看接口文档的字段说明是否提到了“延迟更新时间”或者“部分字段不保证实时性”。如果业务确实需要更实时的数据,可以考虑在页面上合适的位置手动确认数据更新时间,并在文档中注明数据非实时。

6.5 常见问题速查表

现象可能原因优先排查项
401 / 签名错误签名拼接规则不对检查排序、分隔符、空值过滤
403Key 无权限或 IP 不在白名单检查账号权限配置
429请求过于频繁降低频率、加随机等待时间
code=200 但 item_list 为空参数过滤太严格先去掉 keyword 再测试
有数据但价格为 0字段解析类型错误检查原始 JSON 中 price 字段
城市编码填了中文报错地区参数只接受编码核对 city_id 映射表
程序长时间卡住没有设置请求超时给 requests 加 timeout 参数
CSV 打开乱码编码问题使用 utf-8-sig 导出

我个人在实际项目里实测下来,把这套流程串起来之后,单机脚本跑完全国主要城市的二手车列表获取,大概只需要几十分钟,数据量级在几万条左右,完全能满足绝大多数分析和展示类业务的需求。最后再分享一个小技巧:如果你在对接过程中遇到报错,先别急着改代码,把返回体的原始 JSON 完整打印出来,花两分钟逐层去看,很多时候答案早就在响应信息里写清楚了。接口对接这种事,慢就是快,把基础逻辑捋顺了,后面折腾的路就会少很多。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询