一、接口概览
1.1 接口简介
item_search接口是汽车之家开放平台的核心接口之一，专门用于二手车信息检索。支持按地区、品牌、价格、车龄等多维度筛选二手车，返回结构化的车辆列表信息。
1.2 核心功能
✅ 地区筛选：按省市区精准定位二手车源
✅ 多条件搜索：品牌、车系、价格、里程、车龄等
✅ 分页查询：支持大数据量的分页加载
✅ 排序功能：按价格、里程、发布时间等排序
✅ 字段选择：可指定返回字段，优化网络传输
二、准备工作
2.1 环境配置
requirements.txt
requests>=2.28.0
python-dotenv>=1.0.0
pydantic>=2.0.0
aiohttp>=3.8.0
redis>=4.5.0
2.2 认证配置
config.py
import os
from dotenv import load_dotenv
load_dotenv()
class Config:
# 汽车之家API配置
AUTOHOME_APP_KEY = os.getenv('AUTOHOME_APP_KEY')
AUTOHOME_APP_SECRET = os.getenv('AUTOHOME_APP_SECRET')
AUTOHOME_API_BASE = os.getenv('AUTOHOME_API_BASE',
'https://openapi.autohome.com.cn/api/v1'
)
# 请求配置
REQUEST_TIMEOUT = 30
MAX_RETRIES = 3
DEFAULT_PAGE_SIZE = 20
MAX_PAGE_SIZE = 100
# 缓存配置
CACHE_TTL = 3600 # 1小时
三、接口详解
3.1 接口地址
GET /usedcar/search
3.2 请求参数详解
基础参数
参数名
类型
必填
说明
示例
app_key
string
是
应用标识
autohome_app_2024
timestamp
int
是
时间戳
1706774400
sign
string
是
请求签名
详见签名算法
format
string
否
返回格式
json（默认）
version
string
是
API版本
1.0
搜索参数
参数名
类型
必填
说明
示例
province_id
int
否
省份ID
11（北京）
city_id
int
否
城市ID
1101（北京市）
district_id
int
否
区县ID
110101（东城区）
brand_id
int
否
品牌ID
1（奥迪）
series_id
int
否
车系ID
10（A6L）
min_price
float
否
最低价格（万元）
10.0
max_price
float
否
最高价格（万元）
50.0
min_mileage
int
否
最低里程（万公里）
1
max_mileage
int
否
最高里程（万公里）
20
min_year
int
否
最低年份
2018
max_year
int
否
最高年份
2023
fuel_type
string
否
燃油类型
汽油/柴油/新能源
transmission
string
否
变速箱类型
自动/手动
body_type
string
否
车身类型
轿车/SUV/MPV
keyword
string
否
搜索关键词
"奥迪A6L 2020款"
sort_field
string
否
排序字段
price/mileage/year/publish_time
sort_order
string
否
排序方向
asc/desc（默认desc）
page_no
int
否
页码
1（默认）
page_size
int
否
每页条数
20（默认）
fields
string
否
返回字段
id,title,price,mileage,year
四、完整代码实现
4.1 Python完整实现
import requests
import time
import hashlib
import hmac
import json
from typing import Dict, Any, List, Optional
from datetime import datetime, timedelta
from dataclasses import dataclass
from urllib.parse import urlencode
import redis
@dataclass
class UsedCarBasicInfo:
"""二手车基本信息"""
car_id: int
title: str
price: float
original_price: float
mileage: float
year: int
month: int
city: str
district: str
brand: str
series: str
model: str
fuel_type: str
transmission: str
body_type: str
publish_time: str
thumbnail_url: str
source_type: str # 个人/商家
@dataclass
class UsedCarDetail:
"""二手车详细信息"""
basic_info: UsedCarBasicInfo
images: List[str]
specs: Dict[str, str]
seller_info: Dict[str, Any]
inspection_report: Dict[str, Any]
@dataclass
class SearchResult:
"""搜索结果"""
success: bool
code: int
message: str
data: Dict[str, Any]
used_cars: List[UsedCarBasicInfo]
pagination: Dict[str, Any]
class AutoHomeUsedCarAPI:
"""汽车之家二手车API客户端"""
def __init__(self, app_key: str, app_secret: str, sandbox: bool = True, redis_client=None):
self.app_key = app_key
self.app_secret = app_secret
self.base_url = "https://sandbox-openapi.autohome.com.cn" if sandbox else "https://openapi.autohome.com.cn"
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'AutoHome-UsedCar-API/1.0',
'Accept': 'application/json'
})
self.redis = redis_client
self._access_token = None
self._token_expires = None
def _generate_signature(self, params: Dict[str, Any], timestamp: int) -> str:
"""生成请求签名"""
# 排序参数
sorted_params = sorted(params.items())
param_str = '&'.join([f"{k}={v}" for k, v in sorted_params])
# 构建签名字符串
sign_str = f"{self.app_key}{param_str}{timestamp}{self.app_secret}"
# 计算HMAC-SHA256签名
signature = hmac.new(
self.app_secret.encode('utf-8'),
sign_str.encode('utf-8'),
hashlib.sha256
).hexdigest()
return signature
def _get_access_token(self) -> str:
"""获取访问令牌"""
# 检查token是否有效
if self._access_token and self._token_expires and self._token_expires > datetime.now():
return self._access_token
# 获取新token
timestamp = int(time.time())
params = {
'app_key': self.app_key,
'timestamp': timestamp,
'grant_type': 'client_credentials'
}
# 生成签名
signature = self._generate_signature(params, timestamp)
params['sign'] = signature
# 请求token
url = f"{self.base_url}/oauth/token"
response = self.session.post(url, data=params)
if response.status_code == 200:
data = response.json()
self._access_token = data['access_token']
self._token_expires = datetime.now() + timedelta(seconds=data['expires_in'] - 300) # 提前5分钟过期
return self._access_token
else:
raise Exception(f"获取token失败: {response.status_code} - {response.text}")
def search_used_cars(
self,
province_id: Optional[int] = None,
city_id: Optional[int] = None,
district_id: Optional[int] = None,
brand_id: Optional[int] = None,
series_id: Optional[int] = None,
min_price: Optional[float] = None,
max_price: Optional[float] = None,
min_mileage: Optional[float] = None,
max_mileage: Optional[float] = None,
min_year: Optional[int] = None,
max_year: Optional[int] = None,
fuel_type: Optional[str] = None,
transmission: Optional[str] = None,
body_type: Optional[str] = None,
keyword: Optional[str] = None,
sort_field: str = "publish_time",
sort_order: str = "desc",
page_no: int = 1,
page_size: int = 20,
fields: Optional[List[str]] = None
) -> SearchResult:
"""
搜索二手车
Args:
province_id: 省份ID
city_id: 城市ID
district_id: 区县ID
brand_id: 品牌ID
series_id: 车系ID
min_price: 最低价格（万元）
max_price: 最高价格（万元）
min_mileage: 最低里程（万公里）
max_mileage: 最高里程（万公里）
min_year: 最低年份
max_year: 最高年份
fuel_type: 燃油类型
transmission: 变速箱类型
body_type: 车身类型
keyword: 搜索关键词
sort_field: 排序字段
sort_order: 排序方向
page_no: 页码
page_size: 每页条数
fields: 返回字段列表
Returns:
搜索结果
"""
# 获取访问令牌
access_token = self._get_access_token()
# 构建请求参数
params = {
'app_key': self.app_key,
'timestamp': int(time.time()),
'format': 'json',
'version': '1.0',
'sort_field': sort_field,
'sort_order': sort_order,
'page_no': page_no,
'page_size': min(page_size, Config.MAX_PAGE_SIZE)
}
# 添加可选参数
if province_id:
params['province_id'] = province_id
if city_id:
params['city_id'] = city_id
if district_id:
params['district_id'] = district_id
if brand_id:
params['brand_id'] = brand_id
if series_id:
params['series_id'] = series_id
if min_price:
params['min_price'] = min_price
if max_price:
params['max_price'] = max_price
if min_mileage:
params['min_mileage'] = min_mileage
if max_mileage:
params['max_mileage'] = max_mileage
if min_year:
params['min_year'] = min_year
if max_year:
params['max_year'] = max_year
if fuel_type:
params['fuel_type'] = fuel_type
if transmission:
params['transmission'] = transmission
if body_type:
params['body_type'] = body_type
if keyword:
params['keyword'] = keyword
if fields:
params['fields'] = ','.join(fields)
# 生成签名
signature = self._generate_signature(params, params['timestamp'])
params['sign'] = signature
# 添加认证头
headers = {
'Authorization': f'Bearer {access_token}',
'Content-Type': 'application/json'
}
# 发送请求
url = f"{self.base_url}/api/v1/usedcar/search"
try:
response = self.session.get(
url,
params=params,
headers=headers,
timeout=Config.REQUEST_TIMEOUT
)
if response.status_code == 200:
result = response.json()
# 解析结果
used_cars = self._parse_used_cars(result.get('data', {}).get('list', []))
pagination = result.get('data', {}).get('pagination', {})
return SearchResult(
success=result.get('success', False),
code=result.get('code', 0),
message=result.get('message', ''),
data=result.get('data', {}),
used_cars=used_cars,
pagination=pagination
)
elif response.status_code == 401:
# Token过期，重新获取
self._access_token = None
return self.search_used_cars(
province_id=province_id, city_id=city_id, district_id=district_id,
brand_id=brand_id, series_id=series_id, min_price=min_price, max_price=max_price,
min_mileage=min_mileage, max_mileage=max_mileage, min_year=min_year, max_year=max_year,
fuel_type=fuel_type, transmission=transmission, body_type=body_type, keyword=keyword,
sort_field=sort_field, sort_order=sort_order, page_no=page_no, page_size=page_size,
fields=fields
)
else:
return SearchResult(
success=False,
code=response.status_code,
message=f"HTTP {response.status_code}",
data={},
used_cars=[],
pagination={}
)
except requests.exceptions.Timeout:
return SearchResult(
success=False,
code=408,
message="请求超时",
data={},
used_cars=[],
pagination={}
)
except requests.exceptions.RequestException as e:
return SearchResult(
success=False,
code=500,
message=f"网络请求异常: {str(e)}",
data={},
used_cars=[],
pagination={}
)
def _parse_used_cars(self, car_list: List[Dict[str, Any]]) -> List[UsedCarBasicInfo]:
"""解析二手车列表数据"""
used_cars = []
for car_data in car_list:
try:
basic_info = UsedCarBasicInfo(
car_id=car_data.get('id'),
title=car_data.get('title', ''),
price=car_data.get('price', 0),
original_price=car_data.get('original_price', 0),
mileage=car_data.get('mileage', 0),
year=car_data.get('year', 0),
month=car_data.get('month', 0),
city=car_data.get('city', ''),
district=car_data.get('district', ''),
brand=car_data.get('brand', {}).get('name', ''),
series=car_data.get('series', {}).get('name', ''),
model=car_data.get('model', ''),
fuel_type=car_data.get('fuel_type', ''),
transmission=car_data.get('transmission', ''),
body_type=car_data.get('body_type', ''),
publish_time=car_data.get('publish_time', ''),
thumbnail_url=car_data.get('thumbnail_url', ''),
source_type=car_data.get('source_type', '个人')
)
used_cars.append(basic_info)
except Exception as e:
print(f"解析车辆数据失败: {e}, 数据: {car_data}")
continue
return used_cars
def search_all_used_cars(
self,
max_pages: int = 10,
**search_params
) -> List[UsedCarBasicInfo]:
"""
获取所有符合条件的二手车（自动处理分页）
Args:
max_pages: 最大页数限制
**search_params: 搜索参数
Returns:
二手车列表
"""
all_cars = []
page_no = 1
while page_no <= max_pages:
result = self.search_used_cars(page_no=page_no, **search_params)
if not result.success:
print(f"第{page_no}页查询失败: {result.message}")
break
# 添加当前页数据
all_cars.extend(result.used_cars)
pagination = result.pagination
print(f"已获取第{page_no}页，共{len(result.used_cars)}条，总计{len(all_cars)}条")
# 检查是否还有下一页
has_next = pagination.get('has_next', False)
total_pages = pagination.get('total_pages', 0)
if not has_next or page_no >= total_pages:
break
page_no += 1
# 避免请求过于频繁
time.sleep(0.5)
return all_cars
def get_used_car_detail(self, car_id: int) -> Optional[UsedCarDetail]:
"""
获取二手车详细信息
"""
# 获取访问令牌
access_token = self._get_access_token()
# 构建请求
headers = {
'Authorization': f'Bearer {access_token}',
'Content-Type': 'application/json'
}
url = f"{self.base_url}/api/v1/usedcar/detail/{car_id}"
try:
response = self.session.get(url, headers=headers, timeout=30)
if response.status_code == 200:
data = response.json().get('data', {})
# 解析基础信息
basic_info = UsedCarBasicInfo(
car_id=data.get('id'),
title=data.get('title', ''),
price=data.get('price', 0),
original_price=data.get('original_price', 0),
mileage=data.get('mileage', 0),
year=data.get('year', 0),
month=data.get('month', 0),
city=data.get('city', ''),
district=data.get('district', ''),
brand=data.get('brand', {}).get('name', ''),
series=data.get('series', {}).get('name', ''),
model=data.get('model', ''),
fuel_type=data.get('fuel_type', ''),
transmission=data.get('transmission', ''),
body_type=data.get('body_type', ''),
publish_time=data.get('publish_time', ''),
thumbnail_url=data.get('thumbnail_url', ''),
source_type=data.get('source_type', '个人')
)
# 解析详细信息
images = data.get('images', [])
specs = data.get('specs', {})
seller_info = data.get('seller_info', {})
inspection_report = data.get('inspection_report', {})
return UsedCarDetail(
basic_info=basic_info,
images=images,
specs=specs,
seller_info=seller_info,
inspection_report=inspection_report
)
else:
return None
except Exception as e:
print(f"获取车辆详情失败: {e}")
return None
使用示例
def demo_used_car_api():
"""二手车API使用演示"""
# 初始化客户端
client = AutoHomeUsedCarAPI(
app_key=Config.AUTOHOME_APP_KEY,
app_secret=Config.AUTOHOME_APP_SECRET,
sandbox=True
)
print("=== 示例1：按地区搜索二手车 ===")
result = client.search_used_cars(
province_id=11, # 北京
city_id=1101, # 北京市
min_price=10,
max_price=30,
min_year=2018,
max_year=2022,
page_size=5
)
if result.success:
for car in result.used_cars:
print(f"{car.brand} {car.series} - {car.price}万 - {car.mileage}万公里 - {car.year}年")
print("\n=== 示例2：按品牌搜索 ===")
result = client.search_used_cars(
brand_id=1, # 奥迪
min_price=20,
max_price=50,
sort_field="price",
sort_order="asc"
)
print("\n=== 示例3：获取所有符合条件的二手车 ===")
all_cars = client.search_all_used_cars(
province_id=11,
min_price=15,
max_price=25,
min_year=2019,
max_mileage=15
)
print(f"共找到 {len(all_cars)} 辆符合条件的二手车")
if name == "main":
demo_used_car_api()
4.2 Java实现
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.databind.ObjectMapper;
import okhttp3.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.security.InvalidKeyException;
import java.security.NoSuchAlgorithmException;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.TimeUnit;
public class AutoHomeUsedCarClient {
private static final Logger logger = LoggerFactory.getLogger(AutoHomeUsedCarClient.class);
private final String appKey;
private final String appSecret;
private final String baseUrl;
private final OkHttpClient httpClient;
private final ObjectMapper objectMapper;
private String accessToken;
private LocalDateTime tokenExpires;
public AutoHomeUsedCarClient(String appKey, String appSecret, boolean sandbox) {
this.appKey = appKey;
this.appSecret = appSecret;
this.baseUrl = sandbox ?
"https://sandbox-openapi.autohome.com.cn" :
"https://openapi.autohome.com.cn";
this.httpClient = new OkHttpClient.Builder()
.connectTimeout(30, TimeUnit.SECONDS)
.readTimeout(30, TimeUnit.SECONDS)
.writeTimeout(30, TimeUnit.SECONDS)
.build();
this.objectMapper = new ObjectMapper();
this.objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
}
public UsedCarSearchResult searchUsedCars(UsedCarSearchParams params) throws IOException {
// 获取访问令牌
String token = getAccessToken();
// 构建请求参数
Map<String, Object> requestParams = new HashMap<>();
requestParams.put("app_key", appKey);
requestParams.put("timestamp", System.currentTimeMillis() / 1000);
requestParams.put("format", "json");
requestParams.put("version", "1.0");
// 添加搜索参数
if (params.getProvinceId() != null) {
requestParams.put("province_id", params.getProvinceId());
}
if (params.getCityId() != null) {
requestParams.put("city_id", params.getCityId());
}
// ... 其他参数
// 生成签名
String signature = generateSignature(requestParams, (Long) requestParams.get("timestamp"));
requestParams.put("sign", signature);
// 构建请求URL
HttpUrl.Builder urlBuilder = HttpUrl.parse(baseUrl + "/api/v1/usedcar/search").newBuilder();
for (Map.Entry<String, Object> param : requestParams.entrySet()) {
urlBuilder.addQueryParameter(param.getKey(), param.getValue().toString());
}
// 发送请求
Request request = new Request.Builder()
.url(urlBuilder.build())
.addHeader("Authorization", "Bearer " + token)
.addHeader("User-Agent", "AutoHome-Java-Client/1.0")
.build();
try (Response response = httpClient.newCall(request).execute()) {
if (response.isSuccessful()) {
String responseBody = response.body().string();
Map<String, Object> result = objectMapper.readValue(responseBody, Map.class);
return parseSearchResult(result);
} else {
throw new IOException("请求失败: " + response.code());
}
}
}
// 省略其他方法...
}
class UsedCarSearchParams {
private Integer provinceId;
private Integer cityId;
private Integer districtId;
private Integer brandId;
private Integer seriesId;
private Double minPrice;
private Double maxPrice;
private Double minMileage;
private Double maxMileage;
private Integer minYear;
private Integer maxYear;
private String fuelType;
private String transmission;
private String bodyType;
private String keyword;
private String sortField = "publish_time";
private String sortOrder = "desc";
private Integer pageNo = 1;
private Integer pageSize = 20;
// 省略getter/setter方法
}
4.3 PHP实现
<?php
class AutoHomeUsedCarService
{
private $appKey;
private $appSecret;
private $baseUrl;
private $accessToken;
private $tokenExpires;
public function __construct($appKey, $appSecret, $sandbox = true)
{
$this->appKey = $appKey;
$this->appSecret = $appSecret;
$this->baseUrl = $sandbox
? 'https://sandbox-openapi.autohome.com.cn'
: 'https://openapi.autohome.com.cn';
}
public function searchUsedCars($params = [])
{
// 获取访问令牌
$token = $this->getAccessToken();
// 构建请求参数
$requestParams = [
'app_key' => $this->appKey,
'timestamp' => time(),
'format' => 'json',
'version' => '1.0'
];
// 合并搜索参数
$requestParams = array_merge($requestParams, $params);
// 生成签名
$signature = $this->generateSignature($requestParams);
$requestParams['sign'] = $signature;
// 发送请求
$url = $this->baseUrl . '/api/v1/usedcar/search?' . http_build_query($requestParams);
$ch = curl_init();
curl_setopt_array($ch, [
CURLOPT_URL => $url,
CURLOPT_RETURNTRANSFER => true,
CURLOPT_TIMEOUT => 30,
CURLOPT_HTTPHEADER => [
'Authorization: Bearer ' . $token,
'User-Agent: AutoHome-PHP-Client/1.0'
]
]);
$response = curl_exec($ch);
$httpCode = curl_getinfo($ch, CURLINFO_HTTP_CODE);
curl_close($ch);
if ($httpCode === 200) {
return json_decode($response, true);
} else {
throw new Exception("请求失败: HTTP {$httpCode}");
}
}
private function getAccessToken()
{
// 检查token是否有效
if ($this->accessToken && $this->tokenExpires && $this->tokenExpires > time()) {
return $this->accessToken;
}
// 获取新token
$timestamp = time();
$params = [
'app_key' => $this->appKey,
'timestamp' => $timestamp,
'grant_type' => 'client_credentials'
];
$signature = $this->generateSignature($params);
$params['sign'] = $signature;
$ch = curl_init();
curl_setopt_array($ch, [
CURLOPT_URL => $this->baseUrl . '/oauth/token',
CURLOPT_RETURNTRANSFER => true,
CURLOPT_POST => true,
CURLOPT_POSTFIELDS => http_build_query($params),
CURLOPT_TIMEOUT => 30
]);
$response = curl_exec($ch);
$httpCode = curl_getinfo($ch, CURLINFO_HTTP_CODE);
curl_close($ch);
if ($httpCode === 200) {
$result = json_decode($response, true);
$this->accessToken = $result['access_token'];
$this->tokenExpires = time() + $result['expires_in'] - 300; // 提前5分钟过期
return $this->accessToken;
} else {
throw new Exception("获取token失败: HTTP {$httpCode}");
}
}
private function generateSignature($params)
{
// 移除sign参数并排序
unset($params['sign']);
ksort($params);
// 拼接参数字符串
$paramStr = '';
foreach ($params as $key => $value) {
$paramStr .= $key . '=' . $value . '&';
}
$paramStr = rtrim($paramStr, '&');
// 构建签名字符串
$signStr = $this->appKey . $paramStr . $params['timestamp'] . $this->appSecret;
// 计算HMAC-SHA256
return hash_hmac('sha256', $signStr, $this->appSecret);
}
}
// 使用示例
try {
$service = new AutoHomeUsedCarService('your_app_key', 'your_app_secret');
$result = $service->searchUsedCars([
'province_id' => 11,
'city_id' => 1101,
'min_price' => 10,
'max_price' => 30,
'min_year' => 2018,
'page_size' => 10
]);
if ($result['success']) {
foreach ($result['data']['list'] as $car) {
echo "{$car['brand']['name']} {$car['series']['name']} - {$car['price']}万\n";
}
}
} catch (Exception $e) {
echo "错误: " . $e->getMessage() . "\n";
}
?>
五、返回结果解析
5.1 成功响应示例
{
"success": true,
"code": 200,
"message": "成功",
"data": {
"list": [
{
"id": 12345,
"title": "2020款 奥迪A6L 45 TFSI 臻选致雅型",
"price": 28.5,
"original_price": 45.8,
"mileage": 5.2,
"year": 2020,
"month": 6,
"city": "北京市",
"district": "朝阳区",
"brand": {
"id": 1,
"name": "奥迪"
},
"series": {
"id": 10,
"name": "A6L"
},
"model": "45 TFSI 臻选致雅型",
"fuel_type": "汽油",
"transmission": "自动",
"body_type": "轿车",
"publish_time": "2026-01-15 14:30:00",
"thumbnail_url": "https://img.autohome.com.cn/usedcar/12345.jpg",
"source_type": "商家"
}
],
"pagination": {
"page_no": 1,
"page_size": 20,
"total_count": 125,
"total_pages": 7,
"has_next": true,
"has_previous": false
},
"summary": {
"avg_price": 25.8,
"avg_mileage": 6.5,
"avg_year": 2019.5,
"brand_distribution": {
"奥迪": 15,
"宝马": 12,
"奔驰": 10
}
}
}
}
5.2 错误响应示例
{
"success": false,
"code": 400,
"message": "参数错误：province_id无效",
"data": null
}
5.3 状态码说明
状态码
说明
处理建议
200
成功
-
400
参数错误
检查请求参数格式
401
认证失败
检查API密钥和签名
403
权限不足
检查API权限范围
404
数据不存在
检查搜索条件
429
请求频率超限
降低请求频率
500
服务器错误
稍后重试
六、高级功能实现
6.1 智能搜索建议
class IntelligentUsedCarSearch:
"""智能二手车搜索服务"""
def __init__(self, api_client):
self.client = api_client
self.search_history = []
def smart_search(self, query: str, location: str = None) -> SearchResult:
"""
智能搜索：自动识别搜索类型
Args:
query: 搜索查询字符串
location: 地区信息
Returns:
搜索结果
"""
# 解析查询类型
search_type = self._detect_search_type(query)
# 构建搜索参数
params = self._build_search_params(query, search_type, location)
# 执行搜索
result = self.client.search_used_cars(**params)
# 记录搜索历史
self._record_search_history(query, search_type, result)
return result
def _detect_search_type(self, query: str) -> str:
"""自动识别搜索类型"""
import re
# 检查是否为价格范围
if re.match(r'^\d+-\d+万$', query):
return 'price_range'
# 检查是否为年份范围
if re.match(r'^\d+-\d+年$', query):
return 'year_range'
# 检查是否为里程范围
if re.match(r'^\d+-\d+万公里$', query):
return 'mileage_range'
# 检查是否为品牌+车系
brand_patterns = ['奥迪', '宝马', '奔驰', '大众', '丰田', '本田']
for brand in brand_patterns:
if brand in query:
return 'brand_model'
# 默认为关键词搜索
return 'keyword'
def _build_search_params(self, query: str, search_type: str, location: str) -> Dict[str, Any]:
"""根据搜索类型构建参数"""
params = {}
import re
if search_type == 'price_range':
# 解析价格范围
match = re.match(r'^(\d+)-(\d+)万$', query)
if match:
params['min_price'] = float(match.group(1))
params['max_price'] = float(match.group(2))
elif search_type == 'year_range':
# 解析年份范围
match = re.match(r'^(\d+)-(\d+)年$', query)
if match:
params['min_year'] = int(match.group(1))
params['max_year'] = int(match.group(2))
elif search_type == 'mileage_range':
# 解析里程范围
match = re.match(r'^(\d+)-(\d+)万公里$', query)
if match:
params['min_mileage'] = float(match.group(1))
params['max_mileage'] = float(match.group(2))
elif search_type == 'brand_model':
# 解析品牌和车型
params['keyword'] = query
else:
params['keyword'] = query
# 添加地区信息
if location:
# 这里可以集成地理编码服务将地址转换为ID
pass
return params
6.2 数据缓存优化
import redis
from functools import lru_cache
class CachedUsedCarAPI(AutoHomeUsedCarAPI):
"""带缓存的二手车API"""
def __init__(self, app_key, app_secret, redis_client, sandbox=True):
super().__init__(app_key, app_secret, sandbox)
self.redis = redis_client
self.cache_prefix = "autohome:usedcar:"
def search_used_cars_cached(self, cache_key: str, **params) -> SearchResult:
"""
带缓存的二手车搜索
"""
# 检查缓存
cached = self.redis.get(cache_key)
if cached:
data = json.loads(cached)
return SearchResult(**data)
# 调用API
result = super().search_used_cars(**params)
# 缓存结果
if result.success:
# 根据数据量设置缓存时间
ttl = self._calculate_ttl(result)
self.redis.setex(
cache_key,
ttl,
json.dumps(result.__dict__)
)
return result
def _calculate_ttl(self, result: SearchResult) -> int:
"""根据搜索结果计算缓存时间"""
total_count = result.pagination.get('total_count', 0)
if total_count == 0:
return 1800 # 30分钟
elif total_count <= 100:
return 3600 # 1小时
else:
return 7200 # 2小时
def get_cache_key(self, **params) -> str:
"""生成缓存键"""
# 移除分页参数
cache_params = params.copy()
cache_params.pop('page_no', None)
cache_params.pop('page_size', None)
# 生成唯一键
param_str = json.dumps(cache_params, sort_keys=True)
return f"{self.cache_prefix}{hashlib.md5(param_str.encode()).hexdigest()}"
6.3 批量处理优化
from concurrent.futures import ThreadPoolExecutor, as_completed
class BatchUsedCarProcessor:
"""批量二手车处理器"""
def __init__(self, api_client):
self.client = api_client
def batch_search_by_regions(
self,
regions: List[Dict[str, int]],
search_params: Dict[str, Any],
max_workers: int = 3
) -> Dict[str, List[UsedCarBasicInfo]]:
"""
按地区批量搜索二手车
Args:
regions: 地区列表 [{"province_id": 11, "city_id": 1101}, ...]
search_params: 搜索参数
max_workers: 最大并发数
Returns:
按地区分组的搜索结果
"""
results = {}
with ThreadPoolExecutor(max_workers=max_workers) as executor:
# 提交所有地区搜索任务
future_to_region = {
executor.submit(
self.client.search_all_used_cars,
**{**search_params, **region}
): region
for region in regions
}
# 收集结果
for future in as_completed(future_to_region):
region = future_to_region[future]
try:
region_cars = future.result(timeout=300) # 5分钟超时
region_key = f"{region.get('province_id')}_{region.get('city_id')}"
results[region_key] = region_cars
except Exception as e:
print(f"地区 {region} 搜索失败: {e}")
return results
def analyze_regional_prices(
self,
regional_results: Dict[str, List[UsedCarBasicInfo]]
) -> Dict[str, Dict[str, float]]:
"""
分析地区价格差异
"""
analysis = {}
for region_key, cars in regional_results.items():
if not cars:
continue
prices = [car.price for car in cars if car.price > 0]
mileages = [car.mileage for car in cars if car.mileage > 0]
years = [car.year for car in cars if car.year > 0]
analysis[region_key] = {
'avg_price': sum(prices) / len(prices) if prices else 0,
'avg_mileage': sum(mileages) / len(mileages) if mileages else 0,
'avg_year': sum(years) / len(years) if years else 0,
'car_count': len(cars)
}
return analysis
七、实战应用场景
7.1 二手车比价平台
class UsedCarPriceComparator:
"""二手车比价平台"""
def __init__(self, api_client):
self.client = api_client
self.price_history = {}
def compare_prices_by_model(
self,
brand: str,
model: str,
regions: List[Dict[str, int]],
year_range: Tuple[int, int] = (2018, 2023)
) -> Dict[str, Any]:
"""
按车型比较地区价格差异
"""
# 搜索各地区的同款车型
search_params = {
'min_year': year_range[0],
'max_year': year_range[1],
'keyword': f"{brand} {model}"
}
regional_results = self.client.batch_search_by_regions(regions, search_params)
# 分析价格差异
price_analysis = self.client.analyze_regional_prices(regional_results)
# 生成价格对比报告
report = {
'brand': brand,
'model': model,
'year_range': year_range,
'regional_prices': price_analysis,
'recommendation': self._generate_recommendation(price_analysis)
}
return report
def track_price_trend(
self,
car_id: int,
duration_days: int = 30
) -> List[Dict[str, Any]]:
"""
跟踪二手车价格趋势
"""
price_history = []
# 模拟获取历史价格数据
for i in range(duration_days, 0, -1):
# 实际应用中可以从数据库获取历史数据
# 这里简化实现
price_point = {
'date': (datetime.now() - timedelta(days=i)).strftime('%Y-%m-%d'),
'price': 25.5 + random.uniform(-1, 1), # 模拟价格波动
'market_avg': 26.0 + random.uniform(-0.5, 0.5)
}
price_history.append(price_point)
return price_history
7.2 二手车推荐系统
class UsedCarRecommender:
"""二手车推荐系统"""
def __init__(self, api_client):
self.client = api_client
def recommend_cars_by_budget(
self,
budget: float,
user_preferences: Dict[str, Any],
region: Dict[str, int]
) -> List[UsedCarBasicInfo]:
"""
根据预算推荐二手车
"""
# 构建搜索参数
search_params = {
'province_id': region.get('province_id'),
'city_id': region.get('city_id'),
'min_price': max(1, budget * 0.7), # 预算的70%作为最低价
'max_price': budget * 1.1, # 预算的110%作为最高价
'min_year': user_preferences.get('min_year', 2018),
'max_mileage': user_preferences.get('max_mileage', 15),
'fuel_type': user_preferences.get('fuel_type'),
'body_type': user_preferences.get('body_type'),
'sort_field': 'publish_time',
'sort_order': 'desc'
}
# 执行搜索
result = self.client.search_used_cars(**search_params)
if not result.success:
return []
cars = result.used_cars
# 应用推荐算法
recommended_cars = self._apply_recommendation_algorithm(cars, user_preferences)
return recommended_cars[:10] # 返回前10个推荐
def _apply_recommendation_algorithm(
self,
cars: List[UsedCarBasicInfo],
preferences: Dict[str, Any]
) -> List[UsedCarBasicInfo]:
"""应用推荐算法"""
scored_cars = []
for car in cars:
score = 0
# 价格得分（越接近预算上限得分越高）
target_price = preferences.get('target_price', car.price)
price_diff = abs(car.price - target_price)
price_score = max(0, 100 - price_diff * 10)
score += price_score * 0.3
# 车龄得分
current_year = datetime.now().year
car_age = current_year - car.year
age_score = max(0, 100 - car_age * 5)
score += age_score * 0.25
# 里程得分
mileage_score = max(0, 100 - car.mileage * 2)
score += mileage_score * 0.2
# 品牌偏好得分
preferred_brands = preferences.get('preferred_brands', [])
if car.brand in preferred_brands:
score += 50
# 发布时间得分（越新越好）
publish_time = datetime.fromisoformat(car.publish_time.replace(' ', 'T'))
days_ago = (datetime.now() - publish_time).days
recency_score = max(0, 100 - days_ago)
score += recency_score * 0.1
scored_cars.append((car, score))
# 按得分排序
scored_cars.sort(key=lambda x: x[1], reverse=True)
return [car for car, score in scored_cars]
八、故障排查与优化
8.1 常见问题解决
问题1：签名验证失败
def debug_signature_generation(params, app_secret, timestamp):
"""调试签名生成过程"""
print("=== 签名调试信息 ===")
# 排序参数
sorted_params = sorted(params.items())
param_str = '&'.join([f"{k}={v}" for k, v in sorted_params])
print(f"参数字符串: {param_str}")
# 构建签名字符串
sign_str = f"{app_key}{param_str}{timestamp}{app_secret}"
print(f"签名字符串: {sign_str}")
# 计算签名
import hmac
signature = hmac.new(
app_secret.encode('utf-8'),
sign_str.encode('utf-8'),
hashlib.sha256
).hexdigest()
print(f"计算签名: {signature}")
return signature
问题2：分页数据异常
def stable_pagination_search(self, **params):
"""稳定的分页搜索"""
# 确保有排序字段
if 'sort_field' not in params:
params['sort_field'] = 'id' # 使用唯一字段排序
if 'sort_order' not in params:
params['sort_order'] = 'desc'
# 使用游标分页
all_cars = []
last_id = None
while True:
if last_id:
# 使用游标进行分页
params['cursor'] = last_id
result = self.search_used_cars(**params)
if not result.success or not result.used_cars:
break
all_cars.extend(result.used_cars)
last_id = result.used_cars[-1].car_id
# 检查是否还有更多数据
pagination = result.pagination
if not pagination.get('has_next', False):
break
# 避免频繁请求
time.sleep(0.5)
return all_cars
8.2 性能优化建议
合理使用缓存
多级缓存策略
class MultiLevelCache:
def init(self, redis_client):
self.memory_cache = {}
self.redis = redis_client
self.memory_ttl = 300 # 5分钟
self.redis_ttl = 3600 # 1小时
def get_used_car_data(self, cache_key):
# 1. 检查内存缓存
if cache_key in self.memory_cache:
data, expire_time = self.memory_cache[cache_key]
if time.time() < expire_time:
return data
# 2. 检查Redis缓存
if self.redis:
cached = self.redis.get(cache_key)
if cached:
data = json.loads(cached)
# 更新内存缓存
self.memory_cache[cache_key] = (data, time.time() + self.memory_ttl)
return data
return None
批量请求优化
import asyncio
import aiohttp
async def batch_search_async(api_client, search_queries):
"""异步批量搜索"""
async with aiohttp.ClientSession() as session:
tasks = []
for query in search_queries:
task = api_client.search_used_cars_async(session, **query)
tasks.append(task)
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
九、最佳实践总结
9.1 安全实践
密钥管理：使用环境变量存储API密钥
HTTPS强制：确保所有请求使用HTTPS
输入验证：验证所有输入参数
错误处理：不暴露敏感错误信息
9.2 性能实践
缓存策略：根据数据更新频率设置合适的缓存时间
批量操作：合并多个请求减少API调用次数
分页优化：使用游标分页提高稳定性
异步处理：批量操作使用异步方式提高吞吐量
9.3 业务实践
地区数据管理：维护地区ID映射表
品牌车系管理：缓存品牌车系信息
价格趋势分析：记录历史价格数据
推荐算法优化：基于用户行为优化推荐
附录：快速开始模板
quick_start.py
from autohome_usedcar import AutoHomeUsedCarAPI
1. 初始化客户端
client = AutoHomeUsedCarAPI(
app_key="your_app_key",
app_secret="your_app_secret",
sandbox=True
)
2. 简单搜索
result = client.search_used_cars(
province_id=11, # 北京
min_price=10,
max_price=30,
min_year=2018
)
if result.success:
for car in result.used_cars:
print(f"{car.brand} {car.series} - {car.price}万")
3. 批量获取
all_cars = client.search_all_used_cars(
province_id=11,
min_price=15,
max_price=25
)
print(f"共找到 {len(all_cars)} 辆二手车")
4. 获取详情
car_detail = client.get_used_car_detail(12345)
if car_detail:
print(f"车辆详情: {car_detail.basic_info.title}")
通过本攻略，您应该能够：
理解汽车之家二手车搜索接口的完整功能
实现安全的API认证和请求
处理各种搜索条件和分页逻辑
构建高性能的二手车搜索应用
在实际业务中灵活应用该接口
建议根据实际业务需求选择合适的实现方案，并遵循最佳实践确保系统的稳定性和可维护性。
