1688搜索商品列表API详解：关键词、价格区间与分页参数配置（附Python源码）

🔍 1688搜索商品列表API详解：关键词、价格区间与分页参数配置（附Python源码）

1688搜索商品列表API（alibaba.offer.search）是B2B选品、比价、铺货系统的入口接口。它的核心价值是让你用关键词+类目+价格筛选批量获取商品快照，再结合商品详情API完成全量同步。下面直接给你可落地的Python封装和参数解析。

一、 接口基本信息

⚠️避坑：搜索接口返回的是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 # 封装好API供应商demo url=https://console.open.onebound.cn/console/?i=Lex 注册链接 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("\n── 翻页遍历示例 ──") 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}")

三、 关键参数详解与避坑

1. 价格区间单位 →分（Cent）

这是最多人踩的坑！

# ✅ 正确：15~50元 → 传1500 和 5000 biz["beginPrice"] = "1500" biz["endPrice"] = "5000" # ❌ 错误：直接传 15.0 / 50.0 → 接口忽略或查无结果

2. 分页限制

3. 排序sortType

4. 返回摘要字段（常用）

四、 生产级建议

1. 关键词编码：中文关键词直接传UTF-8，requests会自动URL编码，无需手动处理。

2. 类目约束：先通过alibaba.category.get拿到叶子类目ID再搜，可显著减少噪音（尤其服装/电子类）。

3. 限流保护：搜索接口QPS通常≤10，建议sleep(0.2)或使用令牌桶控制。

4. 增量更新：结合gmtModified筛选（部分搜索接口支持modifiedStartTime/End）做每日增量同步，不必全量翻页。

五、 完整对接链路（面试版）

关键词/类目 │ ▼ 搜索API(alibaba.offer.search) ──▶ offerId列表 │ │ │ 详情API(alibaba.item.get) │ ▼ │ SKU/批发价/库存 ──▶ ERP商品主数据 │ ▼ 按totalResult翻页 ──▶ 去重(offerId已存在跳过) ──▶ 写ES/MySQL

需要我补充类目树获取API封装或商品详情API联动示例吗？可以直接说 👍