本文将详细介绍如何使用 Python 调用 1688 商品详情 API 接口，涵盖官方开放平台和第三方代理两种方案，包含完整的代码示例和最佳实践。
一、方案概述
1. 官方开放平台 API（推荐）
1688 开放平台提供官方 API 接口
alibaba.product.get，需企业认证申请。核心接口信息：
- 接口地址：
https://gw.open.1688.com/openapi/param2/1/com.alibaba.product/alibaba.product.get - 请求方式：POST
- 认证方式：OAuth2.0 + 签名验证
2. 第三方代理 API
通过第三方数据服务商（如
api-gw.onebound.cn）调用封装好的接口，无需企业资质，适合个人开发者。二、官方 API 方案详解
2.1 准备工作
- 创建应用：获取
AppKey和AppSecret - 申请权限：申请
alibaba.product.get接口权限 - 获取 Access Token：通过 OAuth2.0 授权流程获取
2.2 签名算法实现
1688 API 使用 MD5 签名验证，参数需按字典序排序：
Python
import hashlibimport urllib.parseimport timefrom collections import OrderedDictdef generate_sign(params: dict, app_secret: str) -> str:
"""
生成 1688 API 请求签名
规则：参数按key升序排序，拼接成字符串，首尾加AppSecret，MD5加密转大写
"""
# 按key升序排序
sorted_params = OrderedDict(sorted(params.items()))
# 拼接参数
sign_str = app_secret for key, value in sorted_params.items():
if key != 'sign' and value is not None:
sign_str += f"{key}{value}"
sign_str += app_secret
# MD5加密并转大写
return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()2.3 完整调用代码
Python
import requestsimport jsonimport timeclass Alibaba1688API:
def __init__(self, app_key: str, app_secret: str, access_token: str):
self.app_key = app_key
self.app_secret = app_secret
self.access_token = access_token
self.base_url = "https://gw.open.1688.com/openapi/param2/1/com.alibaba.product/alibaba.product.get"
def generate_sign(self, params: dict) -> str:
"""生成签名"""
from collections import OrderedDict
sorted_params = OrderedDict(sorted(params.items()))
sign_str = self.app_secret for key, value in sorted_params.items():
if key != 'sign' and value is not None:
sign_str += f"{key}{value}"
sign_str += self.app_secret
import hashlib return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
def get_product_detail(self, product_id: int, fields: str = None) -> dict:
"""
获取商品详情
Args:
product_id: 商品ID
fields: 指定返回字段，如 'subject,priceRanges,imageUrl,skuInfo'
"""
# 构建请求参数
params = {
'access_token': self.access_token,
'productID': product_id,
'_aop_timestamp': str(int(time.time() * 1000)),
}
if fields:
params['fields'] = fields
# 生成签名
params['_aop_signature'] = self.generate_sign(params)
try:
response = requests.post(self.base_url, data=params, timeout=30)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
return None
def parse_product_info(self, data: dict) -> dict:
"""解析商品信息"""
if not data or not data.get('success'):
error_msg = data.get('errorMsg', '未知错误') if data else '无数据返回'
print(f"API调用失败: {error_msg}")
return None
product = data.get('result', {})
# 提取关键字段
result = {
'product_id': product.get('productId'),
'title': product.get('subject'),
'main_image': product.get('imageUrl'),
'detail_url': product.get('detailPage'),
'status': product.get('status'),
'create_time': product.get('createTime'),
'last_update': product.get('lastUpdateTime'),
}
# 解析价格区间
price_ranges = product.get('priceRanges', [])
if price_ranges:
result['price_ranges'] = [
{
'start_quantity': pr.get('startQuantity'),
'price': pr.get('price')
}
for pr in price_ranges ]
# 解析SKU信息
sku_info = product.get('skuInfo', {})
if sku_info:
result['sku_map'] = sku_info.get('skuMap', {})
result['specs'] = sku_info.get('specs', [])
return result# 使用示例if __name__ == "__main__":
# 配置信息（实际应用中应从配置文件或环境变量读取）
APP_KEY = "your_app_key"
APP_SECRET = "your_app_secret"
ACCESS_TOKEN = "your_access_token"
PRODUCT_ID = 123456789012345 # 替换为实际商品ID
# 初始化API客户端
api = Alibaba1688API(APP_KEY, APP_SECRET, ACCESS_TOKEN)
# 获取商品详情
raw_data = api.get_product_detail(
product_id=PRODUCT_ID,
fields="subject,priceRanges,imageUrl,skuInfo,status"
)
if raw_data:
# 解析数据
product_info = api.parse_product_info(raw_data)
print(json.dumps(product_info, ensure_ascii=False, indent=2))三、第三方代理方案
3.1 方案特点
| 维度 | 官方 API | 第三方代理 |
|---|---|---|
| 认证门槛 | 需企业资质 | 无需资质 |
| 字段完整度 | 部分字段受限 | 字段更完整 |
| 调用成本 | 按量计费 | 按条计费 |
| 稳定性 | 高 | 中等 |
| 适用场景 | 企业级应用 | 个人/研究 |
3.2 第三方接口调用示例
Python
import requestsimport jsonclass ThirdParty1688API:
"""第三方 1688 商品详情 API 封装"""
def __init__(self, api_key: str, api_secret: str, base_url: str = "https://api-gw.onebound.cn/1688"):
self.api_key = api_key
self.api_secret = api_secret
self.base_url = base_url
def get_item_detail(self, num_iid: str) -> dict:
"""
获取商品详情
Args:
num_iid: 1688 商品ID
"""
url = f"{self.base_url}/item_get"
params = {
'key': self.api_key,
'secret': self.api_secret,
'num_iid': num_iid }
try:
response = requests.get(url, params=params, timeout=30)
response.raise_for_status()
data = response.json()
if data.get('error') != '0':
print(f"API错误: {data.get('error', '未知错误')}")
return None
return data.get('item', {})
except requests.exceptions.RequestException as e:
print(f"请求异常: {e}")
return None
def extract_key_fields(self, item_data: dict) -> dict:
"""提取关键字段"""
if not item_data:
return None
return {
'title': item_data.get('title'),
'price': item_data.get('price'),
'orginal_price': item_data.get('orginal_price'),
'num_iid': item_data.get('num_iid'),
'detail_url': item_data.get('detail_url'),
'pic_url': item_data.get('pic_url'),
'brand': item_data.get('brand'),
'root_cat_id': item_data.get('rootCatId'),
'cid': item_data.get('cid'),
'desc': item_data.get('desc'),
'item_imgs': item_data.get('item_imgs', []),
'sku': item_data.get('sku', []),
'props_list': item_data.get('props_list', {}),
'seller_id': item_data.get('seller_id'),
'shop_name': item_data.get('shop_name'),
'sales': item_data.get('sales'),
}# 使用示例if __name__ == "__main__":
API_KEY = "your_api_key"
API_SECRET = "your_api_secret"
NUM_IID = "610947572360" # 商品ID
api = ThirdParty1688API(API_KEY, API_SECRET)
item_data = api.get_item_detail(NUM_IID)
if item_data:
clean_data = api.extract_key_fields(item_data)
print(json.dumps(clean_data, ensure_ascii=False, indent=2))四、网页逆向方案（备选）
当 API 受限时，可通过逆向网页接口获取数据：
Python
import requestsimport reimport jsonclass WebCrawler1688:
"""通过网页接口获取商品详情"""
def __init__(self):
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Accept': 'application/json, text/javascript, */*; q=0.01',
'Accept-Language': 'zh-CN,zh;q=0.9',
})
def get_detail(self, offer_id: str):
"""获取商品详情"""
# 先访问商品页获取 Cookie
detail_url = f"https://detail.1688.com/offer/{offer_id}.html"
self.session.get(detail_url, timeout=10)
# 调用 laputa 接口
api_url = "https://laputa.1688.com/offer/ajax/WidgetOfferDetail.do"
params = {
'offerId': offer_id,
'callback': 'jQuery_callback'
}
response = self.session.get(api_url, params=params, timeout=10)
# 提取 JSONP 数据
json_str = re.search(r'jQuery_callback\((.*)\)', response.text)
if json_str:
return json.loads(json_str.group(1))
return None五、关键注意事项
5.1 频率控制
1688 对 API 调用有严格限流，建议：
- 官方 API：不超过 5-10 次/秒
- 网页接口：每次请求间隔 1-3 秒，配合代理池使用
Python
import timeimport randomdef safe_request(func):
"""请求装饰器，添加随机延迟"""
def wrapper(*args, **kwargs):
time.sleep(random.uniform(1, 3))
return func(*args, **kwargs)
return wrapper5.2 错误处理
| 错误码 | 含义 | 处理建议 |
|---|---|---|
ITEM_NOT_FOUND | 商品不存在 | 检查商品ID是否正确 |
INVALID_TOKEN | Token 失效 | 重新获取 Access Token |
FREQUENCY_LIMITED | 频率超限 | 降低请求频率，启用队列 |
isv.invalid-parameter | 参数错误 | 检查必填参数 |
5.3 合规建议
- 遵守 robots.txt：1688 禁止高频抓取
/offer/*.html - 数据使用范围：仅限内部使用，不得转售或公开发布
- 用户授权：获取店铺数据需获得商家明确授权
六、数据存储示例
Python
import sqlite3from datetime import datetimeclass DataStorage:
def __init__(self, db_path: str = "1688_products.db"):
self.conn = sqlite3.connect(db_path)
self._init_table()
def _init_table(self):
"""初始化数据表"""
self.conn.execute("""
CREATE TABLE IF NOT EXISTS products (
id INTEGER PRIMARY KEY AUTOINCREMENT,
product_id TEXT UNIQUE,
title TEXT,
price TEXT,
main_image TEXT,
detail_url TEXT,
raw_data TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
self.conn.commit()
def save_product(self, product: dict):
"""保存商品信息"""
try:
self.conn.execute("""
INSERT INTO products (product_id, title, price, main_image, detail_url, raw_data)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(product_id) DO UPDATE SET
title=excluded.title,
price=excluded.price,
main_image=excluded.main_image,
detail_url=excluded.detail_url,
raw_data=excluded.raw_data,
updated_at=CURRENT_TIMESTAMP
""", (
product.get('product_id'),
product.get('title'),
str(product.get('price', '')),
product.get('main_image'),
product.get('detail_url'),
json.dumps(product, ensure_ascii=False)
))
self.conn.commit()
except Exception as e:
print(f"保存失败: {e}")七、总结
本文介绍了三种获取 1688 商品详情的方案：
- 官方 API：适合企业级应用，数据稳定合规
- 第三方代理：适合快速验证，字段完整度高
- 网页逆向：适合研究学习，需注意反爬机制
