前言 供应链
ERP、货源比价系统、跨境铺货工具开发中，1688 商品详情是核心数据源。市面上绝大多数教程仅简单拉取标题、单价等基础字段，完全忽略 B2B
平台核心特性：阶梯批发价、多规格起订量、工厂商家资质、现货 /
代发标识。同时缺少空值容错、数据标准化、统一异常捕获逻辑，直接上线极易出现程序崩溃、数据错乱问题。本文以批发业务需求为核心重构调用逻辑，专门针对
1688 工业批发特有字段做解析封装，和通用零售商品接口教程区分开，适配长期批量同步商品数据场景。 一、差异化核心设计点 阶梯批发价自动解析：单独提取不同采购数量对应的梯度价格，解决普通脚本只取单一零售价，无法核算批量拿货成本的痛点。 商家资质结构化提取：自动识别实力商家、深度验厂、48 小时发货等工厂标签，用于货源质量筛选。 多层级空字段容错机制：对库存、规格、产地、代发服务等易空字段设置默认值，避免字典读取时报键不存在异常。 统一异常分级处理：区分签名错误、商品下架、接口限流、网络超时四种场景，限流自动休眠重试。 数据归一输出：统一字段命名规则，无需额外处理即可直接写入数据库，适配多系统数据对接。 二、接口基础规范 调用网关：https://gw.open.1688.com/openapi/router/rest 三、完整可运行代码 四、生产环境踩坑总结 1688 签名逻辑和淘宝完全不同，必须前后拼接 app_secret，直接复用零售平台签名函数会持续鉴权失败。 阶梯价格列表是 B2B 核心数据，不同采购量价格差异大，做成本测算必须完整解析分层价格。 部分小工厂商品无产地、代发服务字段，代码统一填充默认值，杜绝程序读取空键报错。 批量同步商品时必须设置请求间隔，高频调用会触发平台限流，限制周期可达 3-7 天。 商品下架后接口会返回特定错误码，业务层可捕获后标记商品失效，停止后续同步。
接口方法：alibaba.product.get
签名规则：1688 专属首尾拼接 MD5 加密，参数升序排序，时间戳为 10 位秒级数字
调用限制：企业认证账号 QPS 上限 3，单商品单次请求返回完整规格、资质数据
import requests
import hashlib
import time
import json
from urllib.parse import quote
class AlibabaProductDetailClient:
def __init__(self, app_key, app_secret):
self.app_key = app_key
self.app_secret = app_secret
self.gateway = "https://gw.open.1688.com/openapi/router/rest"
self.http_session = requests.Session()
def build_md5_sign(self, params):
# 1688标准首尾MD5签名算法
valid_params = dict(filter(lambda item: item[1], params.items()))
sorted_items = sorted(valid_params.items(), key=lambda x: x[0])
sign_str = self.app_secret
for k, v in sorted_items:
sign_str += f"{k}{quote(str(v))}"
sign_str += self.app_secret
return hashlib.md5(sign_str.encode()).hexdigest().upper()
def get_product_detail(self, product_id):
params = {
"app_key": self.app_key,
"method": "alibaba.product.get",
"format": "json",
"v": "2.0",
"sign_method": "md5",
"timestamp": str(int(time.time())),
"productId": product_id
}
params["sign"] = self.build_md5_sign(params)
try:
resp = self.http_session.post(self.gateway, data=params, timeout=15)
res_data = resp.json()
# 限流重试逻辑
error_info = res_data.get("error_response", {})
if error_info.get("code") == 429:
time.sleep(3)
return self.get_product_detail(product_id)
if error_info:
return {"code": -1, "msg": error_info.get("sub_msg", "接口调用失败"), "data": {}}
raw_data = res_data.get("alibaba_product_get_response", {}).get("product", {})
# 阶梯批发价解析
price_tiers = []
price_list = raw_data.get("priceInfoList", [])
for price_item in price_list:
price_tiers.append({
"min_num": int(price_item.get("quantityStart", 1)),
"max_num": int(price_item.get("quantityEnd", 99999)),
"batch_price": float(price_item.get("price", 0))
})
# 商家资质标签整理
shop_tags = []
tag_list = raw_data.get("sellerTagList", [])
for tag in tag_list:
shop_tags.append(tag.get("tagName", ""))
# 标准化返回数据，空值填充默认值
standard_info = {
"product_id": raw_data.get("productId"),
"title": raw_data.get("subject", ""),
"origin_place": raw_data.get("originPlace", "无"),
"total_stock": int(raw_data.get("stockQuantity", 0)),
"min_buy": int(raw_data.get("minOrderQuantity", 1)),
"batch_price_tier": price_tiers,
"main_image": raw_data.get("mainImageUrl", ""),
"seller_name": raw_data.get("sellerName", ""),
"seller_tags": shop_tags,
"support_dropship": bool(raw_data.get("supportDropShip", False))
}
time.sleep(1)
return {"code": 200, "data": standard_info}
except Exception as e:
return {"code": -2, "msg": f"网络异常：{str(e)}", "data": {}}
# 调用示例
if __name__ == "__main__":
client = AlibabaProductDetailClient(app_key="你的密钥", app_secret="你的密钥")
result = client.get_product_detail(product_id="商品ID")
print(json.dumps(result, ensure_ascii=False, indent=2))
