从事B2B选品或铺货系统开发的用户,必然需要对接1688这一核心数据源。本文重点介绍的搜索API(alibaba.offer.search),堪称整个数据链路的起点。其价值在于:通过关键词、类目和价格区间,即可批量获取商品摘要,随后借助详情API补充完整信息,形成一套高效的对接流程。以下直接提供可运行的Python封装代码及关键参数详解,简洁实用。

一、 接口基本信息
首先列出几个核心参数,便于快速掌握:
- 接口名称:alibaba.offer.search(部分文档也标记为cn.alibaba.open.search.offer)
- 协议:HTTPS GET/POST
- 网关:https://gw.open.1688.com/openapi/param2/2/alibaba.offer.search/2.0
- 鉴权:AppKey + AppSecret(签名)+ AccessToken(买家身份可选,公开搜索可不传)
- 用途:关键词搜索、按类目过滤、价格区间筛选、分页遍历
⚠️ 避坑提醒:搜索接口仅返回offer(供应信息)摘要,不含完整SKU数据。获取offerId后,需调用alibaba.item.get接口才能获得详情。
二、 Python封装:搜索与分页遍历
现在直接提供代码,展示一个封装完备的客户端,实现搜索与分页遍历功能。
# ali1688_search.py
import hashlib
import time
import requests
import urllib.parse
from typing import List, Dict, Optional
class Ali1688SearchClient:
"""1688 商品搜索API客户端"""
def __init__(self, app_key: str, app_secret: str, access_token: str = None):
self.app_key = app_key
self.app_secret = app_secret
self.access_token = access_token
# 公开搜索可不传,登录用户传session key
self.gateway = "https://gw.open.1688.com/openapi/param2/2/alibaba.offer.search/2.0"
# ────────────────────────────────────────────
# 1688 标准 MD5 签名
# ────────────────────────────────────────────
def _sign(self, params: Dict) -> str:
filtered = sorted((k, v) for k, v in params.items() if v is not None)
qs = ''.join(f"{k}{v}" for k, v in filtered)
raw = f"{self.app_secret}{qs}{self.app_secret}"
return hashlib.md5(raw.encode('utf-8')).hexdigest().upper()
def _call(self, biz_params: Dict) -> Dict:
"""发起请求"""
api_params = {
"method": "alibaba.offer.search",
"app_key": self.app_key,
"timestamp": str(int(time.time() * 1000)),
"format": "json",
"v": "2.0",
"sign_method": "md5",
}
if self.access_token:
api_params["session"] = self.access_token
# 业务参数做URL编码放入 param2
api_params["param2"] = urllib.parse.quote_plus(str(biz_params).replace("'", '"'))
api_params["sign"] = self._sign(api_params)
resp = requests.get(self.gateway, params=api_params, timeout=15)
resp.raise_for_status()
data = resp.json()
if "error_response" in data:
err = data["error_response"]
raise Exception(f"1688 Search Error [{err.get('code')}]: {err.get('msg')}")
return data.get("alibaba_offer_search_response", {})
# ────────────────────────────────────────────
# 核心:搜索商品
# ────────────────────────────────────────────
def search_offers(
self,
keyword: str,
price_start: Optional[float] = None,
price_end: Optional[float] = None,
category_id: Optional[int] = None,
page_no: int = 1,
page_size: int = 40
) -> Dict:
"""
Args:
keyword: 搜索关键词,如 "纯棉T恤 男"
price_start: 最低批发价(元)
price_end: 最高批发价(元)
category_id: 1688类目ID(可在后台查,或先调 offer.getCategory)
page_no: 页码,从1开始
page_size: 每页条数,最大50(推荐40)
Returns:
{offers: [...], totalResult: int, pageNo: int, pageSize: int}
"""
biz = {
"keywords": keyword,
"pageNo": page_no,
"pageSize": min(page_size, 50), # 1688上限50
"sortType": "booked", # booked=成交量 desc, price_asc, price_desc, new
}
if price_start is not None:
biz["beginPrice"] = str(int(price_start * 100)) # ⚠️ 单位是分!
if price_end is not None:
biz["endPrice"] = str(int(price_end * 100))
if category_id:
biz["categoryId"] = category_id
return self._call(biz)
# ────────────────────────────────────────────
# 自动翻页遍历(生成器,避免内存爆炸)
# ────────────────────────────────────────────
def iter_all(self, keyword: str, max_pages: int = 5, **kwargs):
"""逐页yield每条offer,max_pages控制最大翻页数防死循环"""
for p in range(1, max_pages + 1):
result = self.search_offers(keyword, page_no=p, **kwargs)
offers = result.get("offers", []) or []
total = result.get("totalResult", 0)
if not offers:
break
for offer in offers:
yield offer
if p * kwargs.get("page_size", 40) >= total:
break
time.sleep(0.3) # 友好限速
实际调用示例
if __name__ == "__main__":
client = Ali1688SearchClient(
app_key="YOUR_APP_KEY",
app_secret="YOUR_APP_SECRET",
access_token=None # 公开搜索可不传
)
try:
# ① 单次搜索
result = client.search_offers(
keyword="不锈钢保温杯 定制",
price_start=15.0, # ≥15元
price_end=50.0, # ≤50元
page_no=1,
page_size=20
)
offers = result.get("offers", [])
total = result.get("totalResult", 0)
print(f"✅ 共找到 {total} 个商品,当前页 {len(offers)} 条")
for offer in offers[:3]: # 预览前3条
print(f"• {offer.get('subject')}"
f"批发价:¥{offer.get('priceRange')}"
f"offerId:{offer.get('offerId')}")
# ② 翻页遍历(取前2页)
print("── 翻页遍历示例 ──")
count = 0
for offer in client.iter_all(
keyword="不锈钢保温杯 定制",
price_start=15.0,
price_end=50.0,
page_size=20,
max_pages=2
):
count += 1
if count <= 3:
print(f"offerId:{offer.get('offerId')} {offer.get('subject')}")
except Exception as e:
print(f"❌ 搜索失败: {e}")
三、 关键参数详解与避坑
价格区间单位 → 分(Cent)
这是最容易出错的环节之一。许多开发者在传入价格参数时,直接使用“15.0”、“50.0”等元单位,导致接口要么忽略价格过滤,要么返回空结果。
- 正确做法:15~50元,要传
1500和5000(分)。 - 错误示范:直接传
15.0/50.0,结果为空。
分页限制
分页参数遵循常规规则:
- pageNo:起始值为1,每次递增1。
- pageSize:最大支持50,超过将报错。建议设置为20~40,以平衡抓取速度和数据完整性。
- totalResult:响应中返回该字段,用于判断是否已达到最后一页。
排序 sortType
- booked:按成交量降序排列,适合选品场景推荐。
- price_asc / price_desc:按价格升序或降序排列。
- new:按发布时间降序排列。
返回摘要字段(常用)
- offerId:供应ID,后续需传入
alibaba.item.get(offerId)以获取详情。 - subject:商品标题。
- priceRange:起批价文本,仅用于展示。
- imageList[0]:主图URL。
- minOrderQuantity:最小起订量(MOQ),ERP采购校验时有用。
- supplierName:供应商店铺名。
四、 生产级建议
- 关键词编码:中文关键词直接采用UTF-8编码,requests库自动处理URL编码,无需手动转换。
- 类目约束:建议先调用
alibaba.category.get获取叶子类目ID后再搜索,可大幅降低数据噪音,尤其适用于服装、电子等大类目。 - 限流保护:搜索接口的QPS通常不超过10,建议每次请求后sleep(0.2秒)或使用令牌桶控制速率。
- 增量更新:部分搜索接口支持
modifiedStartTime/End参数,可结合gmtModified实现每日增量同步,避免每次全量翻页。
五、 完整对接链路(面试版)
关键词/类目筛选
│
▼
搜索API(alibaba.offer.search) ──▶ 获取offerId列表
│ │
│ 详情API(alibaba.item.get)
│ ▼
│ 得到SKU/批发价/库存 ──▶ 写入ERP商品主数据
│
▼
根据totalResult分页遍历 ──▶ 去重(跳过已存在的offerId) ──▶ 写入ES/MySQL
