🔍 1688搜索商品列表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联动示例吗？可以直接说 👍