一、接口概览
1.1 接口简介
item_get接口是汽车之家开放平台的核心接口之一，用于获取车辆的完整详细信息，包括车辆基本参数、配置信息、图片、价格、经销商信息等。1.2 接口特性
- 数据全面：返回车辆的全维度信息
- 实时更新：数据与汽车之家官网保持同步
- 结构化返回：JSON格式，字段定义清晰
- 权限控制：支持不同级别的数据访问权限
二、准备工作
2.1 环境配置
# requirements.txtrequests>=2.28.0python-dotenv>=1.0.0pydantic>=2.0.0aiohttp>=3.8.0
2.2 认证配置
# config.pyimport osfrom 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三、接口详解
3.1 接口地址
GET /vehicle/{vehicle_id}3.2 请求参数
参数名 | 类型 | 必填 | 说明 | 示例 |
|---|---|---|---|---|
vehicle_id | int | 是 | 车辆ID | 12345 |
include_images | bool | 否 | 是否包含图片 | true |
include_specs | bool | 否 | 是否包含详细配置 | true |
include_prices | bool | 否 | 是否包含价格信息 | true |
include_dealers | bool | 否 | 是否包含经销商信息 | true |
language | string | 否 | 返回语言 | zh-CN |
3.3 请求示例
import requestsfrom typing import Dict, Any, Optionaldef get_vehicle_detail(
vehicle_id: int,
include_images: bool = True,
include_specs: bool = True,
include_prices: bool = True,
include_dealers: bool = False) -> Dict[str, Any]: """
获取车辆详情
Args:
vehicle_id: 车辆ID
include_images: 是否包含图片
include_specs: 是否包含详细配置
include_prices: 是否包含价格信息
include_dealers: 是否包含经销商信息
Returns:
车辆详情数据
"""
# 构建请求参数
params = { 'include_images': str(include_images).lower(), 'include_specs': str(include_specs).lower(), 'include_prices': str(include_prices).lower(), 'include_dealers': str(include_dealers).lower()
}
# 添加认证信息
headers = { 'Authorization': f'Bearer {get_access_token()}', 'Content-Type': 'application/json'
}
# 发送请求
url = f"{Config.AUTOHOME_API_BASE}/vehicle/{vehicle_id}"
response = requests.get(
url,
params=params,
headers=headers,
timeout=Config.REQUEST_TIMEOUT
)
if response.status_code == 200: return response.json() else: raise Exception(f"请求失败: {response.status_code} - {response.text}")四、完整代码实现
4.1 Python完整实现
import requestsimport timeimport hashlibimport hmacfrom typing import Dict, Any, List, Optionalfrom datetime import datetime, timedeltafrom dataclasses import dataclassimport json@dataclassclass VehicleBasicInfo: """车辆基本信息"""
vehicle_id: int
brand: str
series: str
model: str
year: int
fuel_type: str
transmission: str
body_type: str
displacement: str
power: str
torque: str
fuel_consumption: str@dataclassclass VehicleImage: """车辆图片信息"""
image_id: int
url: str
title: str
type: str # exterior, interior, detail
size: str # small, medium, large@dataclassclass VehicleSpec: """车辆配置信息"""
category: str
name: str
value: str
description: str@dataclassclass VehiclePrice: """车辆价格信息"""
price_type: str # msrp, dealer_price, market_price
amount: float
currency: str
region: str
update_time: strclass AutoHomeAPI: """汽车之家API客户端"""
def __init__(self, app_key: str, app_secret: str, sandbox: bool = True): 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-API-Client/1.0', 'Accept': 'application/json'
}) self._access_token = None
self._token_expires = None
def _generate_signature(self, params: Dict[str, Any], timestamp: int) -> str: """生成请求签名"""
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
# 排序参数
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 get_vehicle_detail(
self,
vehicle_id: int,
include_images: bool = True,
include_specs: bool = True,
include_prices: bool = True,
include_dealers: bool = False,
language: str = 'zh-CN'
) -> Dict[str, Any]: """
获取车辆详细信息
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
Args:
vehicle_id: 车辆ID
include_images: 是否包含图片
include_specs: 是否包含详细配置
include_prices: 是否包含价格信息
include_dealers: 是否包含经销商信息
language: 返回语言
Returns:
车辆详情数据
"""
# 获取访问令牌
access_token = self._get_access_token()
# 构建请求参数
params = { 'include_images': str(include_images).lower(), 'include_specs': str(include_specs).lower(), 'include_prices': str(include_prices).lower(), 'include_dealers': str(include_dealers).lower(), 'language': language
}
# 添加认证头
headers = { 'Authorization': f'Bearer {access_token}', 'Content-Type': 'application/json'
}
# 发送请求
url = f"{self.base_url}/api/v1/vehicle/{vehicle_id}"
try:
response = self.session.get(
url,
params=params,
headers=headers,
timeout=Config.REQUEST_TIMEOUT
)
if response.status_code == 200: return response.json() elif response.status_code == 404: raise VehicleNotFoundException(f"车辆ID {vehicle_id} 不存在") elif response.status_code == 401: # Token可能过期，重新获取
self._access_token = None
return self.get_vehicle_detail(
vehicle_id, include_images, include_specs,
include_prices, include_dealers, language
) else: raise Exception(f"请求失败: {response.status_code} - {response.text}")
except requests.exceptions.Timeout: raise Exception("请求超时") except requests.exceptions.RequestException as e: raise Exception(f"网络请求异常: {str(e)}")
def get_vehicle_detail_structured(
self,
vehicle_id: int,
**kwargs ) -> Dict[str, Any]: """
获取结构化的车辆详情信息
"""
raw_data = self.get_vehicle_detail(vehicle_id, **kwargs)
# 解析基础信息
basic_info = VehicleBasicInfo(
vehicle_id=raw_data.get('id'),
brand=raw_data.get('brand', {}).get('name'),
series=raw_data.get('series', {}).get('name'),
model=raw_data.get('model'),
year=raw_data.get('year'),
fuel_type=raw_data.get('fuel_type'),
transmission=raw_data.get('transmission'),
body_type=raw_data.get('body_type'),
displacement=raw_data.get('displacement'),
power=raw_data.get('power'),
torque=raw_data.get('torque'),
fuel_consumption=raw_data.get('fuel_consumption')
)
# 解析图片信息
images = [] for img_data in raw_data.get('images', []):
images.append(VehicleImage(
image_id=img_data.get('id'),
url=img_data.get('url'),
title=img_data.get('title'), type=img_data.get('type'),
size=img_data.get('size')
))
# 解析配置信息
specs = [] for spec_data in raw_data.get('specs', []):
specs.append(VehicleSpec(
category=spec_data.get('category'),
name=spec_data.get('name'),
value=spec_data.get('value'),
description=spec_data.get('description')
))
# 解析价格信息
prices = [] for price_data in raw_data.get('prices', []):
prices.append(VehiclePrice(
price_type=price_data.get('type'),
amount=price_data.get('amount'),
currency=price_data.get('currency'),
region=price_data.get('region'),
update_time=price_data.get('update_time')
))
return { 'basic_info': basic_info, 'images': images, 'specs': specs, 'prices': prices, 'raw_data': raw_data
}
def batch_get_vehicle_details(
self,
vehicle_ids: List[int],
max_workers: int = 5,
**kwargs ) -> Dict[int, Dict[str, Any]]: """
批量获取车辆详情
"""
from concurrent.futures import ThreadPoolExecutor, as_completed
results = {}
with ThreadPoolExecutor(max_workers=max_workers) as executor: # 提交所有任务
future_to_id = {
executor.submit(self.get_vehicle_detail, vehicle_id, **kwargs): vehicle_id for vehicle_id in vehicle_ids
}
# 收集结果
for future in as_completed(future_to_id):
vehicle_id = future_to_id[future] try:
result = future.result(timeout=30)
results[vehicle_id] = result except Exception as e: print(f"获取车辆 {vehicle_id} 详情失败: {e}")
results[vehicle_id] = {'error': str(e)}
return resultsclass VehicleNotFoundException(Exception): """车辆不存在异常"""
pass# 使用示例def demo_vehicle_api(): """API使用演示"""
# 初始化客户端
client = AutoHomeAPI(
app_key=Config.AUTOHOME_APP_KEY,
app_secret=Config.AUTOHOME_APP_SECRET,
sandbox=True
)
# 获取单个车辆详情
print("=== 获取单个车辆详情 ===")
vehicle_detail = client.get_vehicle_detail(12345) print(json.dumps(vehicle_detail, ensure_ascii=False, indent=2))
# 获取结构化信息
print("\n=== 获取结构化车辆信息 ===")
structured_info = client.get_vehicle_detail_structured(12345)
basic_info = structured_info['basic_info'] print(f"车辆: {basic_info.brand} {basic_info.series} {basic_info.model}") print(f"排量: {basic_info.displacement}") print(f"功率: {basic_info.power}")
# 批量获取
print("\n=== 批量获取车辆详情 ===")
vehicle_ids = [12345, 12346, 12347, 12348]
batch_results = client.batch_get_vehicle_details(vehicle_ids) for vid, result in batch_results.items(): if 'error' not in result: print(f"车辆 {vid}: {result.get('model')}") else: print(f"车辆 {vid}: 获取失败 - {result['error']}")if __name__ == "__main__":
demo_vehicle_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.MessageDigest;import java.security.NoSuchAlgorithmException;import java.time.LocalDateTime;import java.time.format.DateTimeFormatter;import java.util.*;import java.util.concurrent.TimeUnit;public class AutoHomeApiClient { private static final Logger logger = LoggerFactory.getLogger(AutoHomeApiClient.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 AutoHomeApiClient(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 Map<String, Object> getVehicleDetail(int vehicleId, Map<String, Object> params) throws IOException { // 获取访问令牌
String token = getAccessToken();
// 构建请求URL
HttpUrl.Builder urlBuilder = HttpUrl.parse(baseUrl + "/api/v1/vehicle/" + vehicleId).newBuilder();
if (params != null) { for (Map.Entry<String, Object> param : params.entrySet()) {
urlBuilder.addQueryParameter(param.getKey(), param.getValue().toString());
}
}
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
// 构建请求
Request request = new Request.Builder()
.url(urlBuilder.build())
.addHeader("Authorization", "Bearer " + token)
.addHeader("Content-Type", "application/json")
.addHeader("User-Agent", "AutoHome-Java-Client/1.0")
.build();
// 发送请求
try (Response response = httpClient.newCall(request).execute()) { if (response.isSuccessful()) { String responseBody = response.body().string(); return objectMapper.readValue(responseBody, Map.class);
} else if (response.code() == 404) { throw new VehicleNotFoundException("车辆ID " + vehicleId + " 不存在");
} else if (response.code() == 401) { // Token过期，重新获取
this.accessToken = null; return getVehicleDetail(vehicleId, params);
} else { throw new IOException("请求失败: " + response.code() + " - " + response.message());
}
}
}
private String getAccessToken() throws IOException { // 检查token是否有效
if (accessToken != null && tokenExpires != null && tokenExpires.isAfter(LocalDateTime.now())) { return accessToken;
}
// 获取新token
long timestamp = System.currentTimeMillis() / 1000;
Map<String, Object> params = new HashMap<>();
params.put("app_key", appKey);
params.put("timestamp", timestamp);
params.put("grant_type", "client_credentials");
// 生成签名
String signature = generateSignature(params, timestamp);
params.put("sign", signature);
// 构建请求
FormBody.Builder formBuilder = new FormBody.Builder(); for (Map.Entry<String, Object> param : params.entrySet()) {
formBuilder.add(param.getKey(), param.getValue().toString());
}
Request request = new Request.Builder()
.url(baseUrl + "/oauth/token")
.post(formBuilder.build())
.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); this.accessToken = (String) result.get("access_token"); int expiresIn = (Integer) result.get("expires_in"); this.tokenExpires = LocalDateTime.now().plusSeconds(expiresIn - 300); // 提前5分钟过期
return accessToken;
} else { throw new IOException("获取token失败: " + response.code());
}
}
}
private String generateSignature(Map<String, Object> params, long timestamp) { try { // 排序参数
List<String> keys = new ArrayList<>(params.keySet());
Collections.sort(keys);
// 构建参数字符串
StringBuilder paramStr = new StringBuilder(); for (String key : keys) {
paramStr.append(key).append("=").append(params.get(key)).append("&");
} if (paramStr.length() > 0) {
paramStr.deleteCharAt(paramStr.length() - 1); // 移除最后一个&
}
// 构建签名字符串
String signStr = appKey + paramStr.toString() + timestamp + appSecret;
// 计算HMAC-SHA256
Mac sha256_HMAC = Mac.getInstance("HmacSHA256"); SecretKeySpec secret_key = new SecretKeySpec(appSecret.getBytes(StandardCharsets.UTF_8), "HmacSHA256");
sha256_HMAC.init(secret_key); byte[] hash = sha256_HMAC.doFinal(signStr.getBytes(StandardCharsets.UTF_8));
// 转换为十六进制
StringBuilder hexString = new StringBuilder(); for (byte b : hash) { String hex = Integer.toHexString(0xff & b); if (hex.length() == 1) hexString.append('0');
hexString.append(hex);
}
return hexString.toString();
} catch (NoSuchAlgorithmException | InvalidKeyException e) { throw new RuntimeException("生成签名失败", e);
}
}
}class VehicleNotFoundException extends RuntimeException { public VehicleNotFoundException(String message) { super(message);
}
}4.3 PHP实现
<?phpclass AutoHomeApiClient{ 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 getVehicleDetail($vehicleId, $params = []) { // 获取访问令牌
$token = $this->getAccessToken();
// 构建请求URL
$url = $this->baseUrl . '/api/v1/vehicle/' . $vehicleId; if (!empty($params)) { $url .= '?' . http_build_query($params);
}
// 发送请求
$ch = curl_init(); curl_setopt_array($ch, [
CURLOPT_URL => $url,
CURLOPT_RETURNTRANSFER => true,
CURLOPT_TIMEOUT => 30,
CURLOPT_HTTPHEADER => [ 'Authorization: Bearer ' . $token, 'Content-Type: application/json', '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);
} elseif ($httpCode === 404) { throw new Exception("车辆ID {$vehicleId} 不存在");
} elseif ($httpCode === 401) { // Token过期，重新获取
$this->accessToken = null; return $this->getVehicleDetail($vehicleId, $params);
} 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'
];
# 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex
// 生成签名
$signature = $this->generateSignature($params, $timestamp); $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, $timestamp) { // 排序参数
ksort($params); $paramStr = ''; foreach ($params as $key => $value) { $paramStr .= $key . '=' . $value . '&';
} $paramStr = rtrim($paramStr, '&');
// 构建签名字符串
$signStr = $this->appKey . $paramStr . $timestamp . $this->appSecret;
// 计算HMAC-SHA256
return hash_hmac('sha256', $signStr, $this->appSecret);
}
}// 使用示例try { $client = new AutoHomeApiClient('your_app_key', 'your_app_secret', true); $vehicleDetail = $client->getVehicleDetail(12345, [ 'include_images' => 'true', 'include_specs' => 'true'
]);
echo "车辆信息:\n"; echo "品牌: " . $vehicleDetail['brand']['name'] . "\n"; echo "车系: " . $vehicleDetail['series']['name'] . "\n"; echo "车型: " . $vehicleDetail['model'] . "\n";
} catch (Exception $e) { echo "错误: " . $e->getMessage() . "\n";
}?>五、返回结果解析
5.1 成功响应示例
{
"success": true,
"code": 200,
"message": "成功",
"data": {
"id": 12345,
"brand": {
"id": 1,
"name": "宝马",
"logo": "https://img.autohome.com.cn/brand/1.png"
},
"series": {
"id": 10,
"name": "3系",
"image": "https://img.autohome.com.cn/series/10.jpg"
},
"model": "2023款 325Li M运动套装",
"year": 2023,
"fuel_type": "汽油",
"transmission": "8挡手自一体",
"body_type": "轿车",
"displacement": "2.0T",
"power": "184马力",
"torque": "300牛·米",
"fuel_consumption": "6.2L/100km",
"msrp": 349900,
"images": [
{
"id": 1,
"url": "https://img.autohome.com.cn/vehicle/12345_1.jpg",
"title": "外观前脸",
"type": "exterior",
"size": "large"
}
],
"specs": [
{
"category": "车身",
"name": "长宽高",
"value": "4838×1827×1454mm",
"description": "车身尺寸"
}
],
"prices": [
{
"type": "msrp",
"amount": 349900,
"currency": "CNY",
"region": "全国",
"update_time": "2023-01-15 10:00:00"
}
]
}}5.2 错误响应示例
{
"success": false,
"code": 404,
"message": "车辆不存在",
"data": null}5.3 状态码说明
状态码 | 说明 | 处理建议 |
|---|---|---|
200 | 成功 | - |
400 | 参数错误 | 检查请求参数格式 |
401 | 认证失败 | 检查API密钥和签名 |
403 | 权限不足 | 检查API权限范围 |
404 | 车辆不存在 | 检查vehicle_id是否正确 |
429 | 请求频率超限 | 降低请求频率 |
500 | 服务器错误 | 稍后重试 |
六、高级功能实现
6.1 智能数据解析
class IntelligentVehicleParser: """智能车辆数据解析器"""
def __init__(self): self.spec_categories = { '车身': ['长宽高', '轴距', '整备质量'], '发动机': ['排量', '最大功率', '最大扭矩', '气缸数'], '变速箱': ['变速箱类型', '挡位个数'], '底盘转向': ['驱动方式', '前悬架', '后悬架'], '车轮制动': ['前制动器', '后制动器', '驻车制动']
}
def parse_vehicle_specs(self, specs: List[VehicleSpec]) -> Dict[str, Dict[str, str]]: """解析车辆配置信息"""
parsed_specs = {}
for category in self.spec_categories.keys():
parsed_specs[category] = {}
for spec in specs: for category, spec_names in self.spec_categories.items(): if spec.name in spec_names:
parsed_specs[category][spec.name] = spec.value break
else: # 未分类的配置
if '其他' not in parsed_specs:
parsed_specs['其他'] = {}
parsed_specs['其他'][spec.name] = spec.value
return parsed_specs
def extract_technical_data(self, vehicle_data: Dict[str, Any]) -> Dict[str, Any]: """提取技术数据"""
basic_info = vehicle_data.get('basic_info', {})
# 解析功率和扭矩
power_match = re.search(r'(\d+(\.\d+)?)', basic_info.power or '')
torque_match = re.search(r'(\d+(\.\d+)?)', basic_info.torque or '')
fuel_match = re.search(r'(\d+(\.\d+)?)', basic_info.fuel_consumption or '')
return { 'power_kw': float(power_match.group(1)) * 0.735 if power_match else None, 'torque_nm': float(torque_match.group(1)) if torque_match else None, 'fuel_consumption_l_100km': float(fuel_match.group(1)) if fuel_match else None, 'displacement_cc': self._parse_displacement(basic_info.displacement or ''), 'transmission_type': self._classify_transmission(basic_info.transmission or ''), 'body_type_code': self._classify_body_type(basic_info.body_type or '')
}
def _parse_displacement(self, displacement: str) -> int: """解析排量"""
match = re.search(r'(\d+(\.\d+)?)', displacement) if match: return int(float(match.group(1)) * 1000) return 0
def _classify_transmission(self, transmission: str) -> str: """分类变速箱类型"""
if '手自一体' in transmission: return 'AT'
elif '双离合' in transmission: return 'DCT'
elif '无级变速' in transmission: return 'CVT'
elif '手动' in transmission: return 'MT'
else: return 'UNKNOWN'
def _classify_body_type(self, body_type: str) -> str: """分类车身类型"""
mapping = { '轿车': 'SEDAN', 'SUV': 'SUV', 'MPV': 'MPV', '跑车': 'COUPE', '皮卡': 'PICKUP', '微面': 'VAN'
} return mapping.get(body_type, 'UNKNOWN')6.2 数据缓存优化
import redisfrom functools import lru_cacheclass CachedAutoHomeAPI(AutoHomeAPI): """带缓存的汽车之家API"""
def __init__(self, app_key, app_secret, redis_client=None, cache_ttl=3600): super().__init__(app_key, app_secret) self.redis = redis_client self.cache_ttl = cache_ttl self.memory_cache = {}
@lru_cache(maxsize=1000)
def get_vehicle_detail_cached(self, vehicle_id: int, **kwargs) -> Dict[str, Any]: """
带缓存的车辆详情获取
"""
cache_key = f"autohome:vehicle:{vehicle_id}"
# 检查内存缓存
if cache_key in self.memory_cache:
cached_time, data = self.memory_cache[cache_key] if time.time() - cached_time < 300: # 内存缓存5分钟
return data
# 检查Redis缓存
if self.redis:
cached = self.redis.get(cache_key) if cached:
data = json.loads(cached) # 更新内存缓存
self.memory_cache[cache_key] = (time.time(), data) return data
# 调用API
data = super().get_vehicle_detail(vehicle_id, **kwargs)
# 缓存到Redis
if self.redis and data.get('success'): # 根据数据更新频率设置不同TTL
ttl = self._calculate_ttl(data) self.redis.setex(cache_key, ttl, json.dumps(data)) self.memory_cache[cache_key] = (time.time(), data)
return data
def _calculate_ttl(self, vehicle_data: Dict[str, Any]) -> int: """根据车辆信息计算缓存时间"""
year = vehicle_data.get('year', 0)
current_year = datetime.now().year
# 老车型缓存时间更长
if year < current_year - 3: return 24 * 3600 # 24小时
elif year < current_year - 1: return 12 * 3600 # 12小时
else: return 3600 # 1小时6.3 批量处理优化
import asyncioimport aiohttpclass AsyncAutoHomeAPI: """异步汽车之家API客户端"""
def __init__(self, app_key, app_secret, sandbox=True): 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 = None
self.access_token = None
async def __aenter__(self): self.session = aiohttp.ClientSession() await self._get_access_token() return self
async def __aexit__(self, exc_type, exc_val, exc_tb): await self.session.close()
async def _get_access_token(self): """异步获取访问令牌"""
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
async with self.session.post( f"{self.base_url}/oauth/token",
data=params
) as response: if response.status == 200:
data = await response.json() self.access_token = data['access_token'] else: raise Exception(f"获取token失败: {response.status}")
async def get_vehicle_detail_async(self, vehicle_id: int, **kwargs) -> Dict[str, Any]: """异步获取车辆详情"""
params = { 'include_images': str(kwargs.get('include_images', True)).lower(), 'include_specs': str(kwargs.get('include_specs', True)).lower(), 'include_prices': str(kwargs.get('include_prices', True)).lower(), 'include_dealers': str(kwargs.get('include_dealers', False)).lower()
}
headers = { 'Authorization': f'Bearer {self.access_token}', 'Content-Type': 'application/json'
}
url = f"{self.base_url}/api/v1/vehicle/{vehicle_id}"
async with self.session.get(
url,
params=params,
headers=headers
) as response: if response.status == 200: return await response.json() else: raise Exception(f"请求失败: {response.status}")
async def batch_get_vehicle_details_async(
self,
vehicle_ids: List[int],
**kwargs ) -> Dict[int, Dict[str, Any]]: """异步批量获取车辆详情"""
tasks = [] for vehicle_id in vehicle_ids:
task = self.get_vehicle_detail_async(vehicle_id, **kwargs)
tasks.append(task)
results = await asyncio.gather(*tasks, return_exceptions=True)
return {
vehicle_id: result for vehicle_id, result in zip(vehicle_ids, results) if not isinstance(result, Exception)
}# 使用示例async def demo_async_api(): """异步API使用示例"""
async with AsyncAutoHomeAPI('your_app_key', 'your_app_secret') as client:
vehicle_ids = [12345, 12346, 12347, 12348, 12349]
results = await client.batch_get_vehicle_details_async(vehicle_ids)
for vehicle_id, result in results.items(): print(f"车辆 {vehicle_id}: {result.get('model')}")七、实战应用场景
7.1 汽车电商平台集成
class CarEcommercePlatform: """汽车电商平台集成"""
def __init__(self, autohome_client): self.client = autohome_client self.vehicle_cache = {}
def get_vehicle_display_info(self, vehicle_id: int) -> Dict[str, Any]: """获取车辆展示信息"""
# 获取车辆详情
vehicle_data = self.client.get_vehicle_detail_structured(vehicle_id)
basic_info = vehicle_data['basic_info']
images = vehicle_data['images']
prices = vehicle_data['prices']
# 构建展示信息
display_info = { 'vehicle_id': vehicle_id, 'title': f"{basic_info.brand} {basic_info.series} {basic_info.model}", 'year': basic_info.year, 'specs': { 'fuel_type': basic_info.fuel_type, 'transmission': basic_info.transmission, 'power': basic_info.power, 'fuel_consumption': basic_info.fuel_consumption
}, 'main_image': self._get_main_image(images), 'price_info': self._get_best_price(prices), 'features': self._extract_features(vehicle_data)
}
return display_info
def _get_main_image(self, images: List[VehicleImage]) -> str: """获取主图"""
for image in images: if image.type == 'exterior' and image.size == 'large': return image.url return images[0].url if images else ''
def _get_best_price(self, prices: List[VehiclePrice]) -> Dict[str, Any]: """获取最优价格"""
msrp_price = None
dealer_price = None
for price in prices: if price.price_type == 'msrp':
msrp_price = price.amount elif price.price_type == 'dealer_price': if not dealer_price or price.amount < dealer_price:
dealer_price = price.amount
return { 'msrp': msrp_price, 'best_price': dealer_price or msrp_price, 'discount': msrp_price - dealer_price if dealer_price and msrp_price else 0
}
def batch_update_vehicle_prices(self, vehicle_ids: List[int]) -> Dict[int, float]: """批量更新车辆价格"""
results = self.client.batch_get_vehicle_details(vehicle_ids, include_prices=True)
price_updates = {} for vehicle_id, result in results.items(): if 'error' not in result:
prices = result.get('prices', [])
best_price = self._get_best_price_from_raw(prices)
price_updates[vehicle_id] = best_price
return price_updates7.2 汽车对比工具
class VehicleComparisonTool: """车辆对比工具"""
def __init__(self, autohome_client): self.client = autohome_client
def compare_vehicles(self, vehicle_ids: List[int]) -> Dict[str, Any]: """对比多个车辆"""
if len(vehicle_ids) > 5: raise Exception("最多支持同时对比5辆车")
# 获取车辆详情
vehicle_details = self.client.batch_get_vehicle_details(vehicle_ids)
# 构建对比数据
comparison_data = { 'vehicles': [], 'common_specs': self._get_common_specs(vehicle_details), 'price_comparison': self._compare_prices(vehicle_details)
}
for vehicle_id, detail in vehicle_details.items(): if 'error' not in detail:
comparison_data['vehicles'].append({ 'vehicle_id': vehicle_id, 'basic_info': self._extract_basic_info(detail), 'key_features': self._extract_key_features(detail)
})
return comparison_data
def _get_common_specs(self, vehicle_details: Dict[int, Dict[str, Any]]) -> List[Dict[str, Any]]: """获取共同配置项"""
all_specs = [] for detail in vehicle_details.values(): if 'error' not in detail:
specs = detail.get('specs', [])
all_specs.extend([spec['name'] for spec in specs])
# 统计出现频率
from collections import Counter
spec_counter = Counter(all_specs)
# 返回出现次数大于1的配置项
common_specs = [spec for spec, count in spec_counter.items() if count > 1]
result = [] for spec_name in common_specs[:10]: # 限制返回数量
spec_values = {} for vehicle_id, detail in vehicle_details.items(): if 'error' not in detail: for spec in detail.get('specs', []): if spec['name'] == spec_name:
spec_values[vehicle_id] = spec['value'] break
result.append({ 'name': spec_name, 'values': spec_values
})
return result八、故障排查与优化
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：Token过期处理
def get_vehicle_detail_with_retry(self, vehicle_id: int, max_retries: int = 3, **kwargs): """带重试的车辆详情获取""" for attempt in range(max_retries): try: return self.get_vehicle_detail(vehicle_id, **kwargs) except Exception as e: if "401" in str(e) and attempt < max_retries - 1: # Token过期，重新获取 self._access_token = None time.sleep(1) continue else: raise
8.2 性能优化建议
- 合理使用缓存
# 多级缓存策略class MultiLevelCache: def __init__(self): self.memory_cache = {} self.redis_cache = redis.Redis()
def get_vehicle_data(self, vehicle_id): # 1. 检查内存缓存
if vehicle_id in self.memory_cache:
data, expire_time = self.memory_cache[vehicle_id] if time.time() < expire_time: return data
# 2. 检查Redis缓存
cache_key = f"vehicle:{vehicle_id}"
cached = self.redis_cache.get(cache_key) if cached:
data = json.loads(cached) # 更新内存缓存
self.memory_cache[vehicle_id] = (data, time.time() + 300) return data
return None- 批量请求优化
# 使用异步处理提高吞吐量async def process_vehicle_batch(vehicle_ids, batch_size=10): """批量处理车辆数据""" results = [] for i in range(0, len(vehicle_ids), batch_size): batch = vehicle_ids[i:i+batch_size] batch_results = await asyncio.gather(*[ get_vehicle_detail_async(vehicle_id) for vehicle_id in batch ]) results.extend(batch_results) await asyncio.sleep(0.1) # 避免频率限制 return results
九、最佳实践总结
9.1 安全实践
- 密钥保护：使用环境变量存储API密钥
- HTTPS强制：确保所有请求使用HTTPS
- 输入验证：验证所有输入参数
- 错误处理：不暴露敏感错误信息
9.2 性能实践
- 缓存策略：根据数据更新频率设置合适的缓存时间
- 批量操作：合并多个请求减少API调用次数
- 异步处理：使用异步IO提高并发性能
- 连接复用：使用连接池减少连接建立开销
9.3 代码质量
- 异常处理：完善的异常处理和重试机制
- 日志记录：详细记录API调用情况
- 单元测试：编写测试用例覆盖主要功能
- 类型注解：使用类型注解提高代码可读性
附录：快速开始模板
# quick_start.pyfrom autohome_api import AutoHomeAPI# 1. 初始化客户端client = AutoHomeAPI(
app_key="your_app_key",
app_secret="your_app_secret",
sandbox=True)# 2. 获取车辆详情vehicle_detail = client.get_vehicle_detail(12345)print(f"车辆: {vehicle_detail['brand']['name']} {vehicle_detail['model']}")# 3. 获取结构化信息structured_info = client.get_vehicle_detail_structured(12345)print(f"排量: {structured_info['basic_info'].displacement}")print(f"功率: {structured_info['basic_info'].power}")# 4. 批量获取vehicle_ids = [12345, 12346, 12347]
batch_results = client.batch_get_vehicle_details(vehicle_ids)for vid, result in batch_results.items(): if 'error' not in result: print(f"车辆 {vid}: {result.get('model')}")通过本攻略，您应该能够：
- 理解汽车之家
item_get接口的完整功能 - 实现安全的API认证和请求
- 处理各种错误情况和性能优化
- 在实际项目中灵活应用该接口
建议根据实际业务需求选择合适的实现方案，并遵循最佳实践确保系统的稳定性和可维护性。
