1688 开放平台店铺全商品接口实战：MemberId 解析 + HMAC 签名 + 全量分页 + 批发数据结构化（附 Python 代码）

import requests
import hmac
import hashlib
import base64
import time
import json
import re

class Ali1688ShopGoodsAPI:
    """1688店铺全商品接口：MemberId解析+签名+全量分页+批发数据结构化"""
    def __init__(self, app_key, app_secret):
        self.app_key = app_key
        self.app_secret = app_secret
        self.api_url = "https://gw.open.1688.com/openapi/param2/1/com.alibaba.product/alibaba.product.getBySellerId/"
        self.timeout = 15
        self.all_items = []
        self.offer_ids = set()

    def parse_member_id(self, shop_url):
        """从店铺URL解析16位MemberId（纯数字，精准匹配）"""
        pattern = r"sellerId=(\d{16})|shop/(\d{16})"
        match = re.search(pattern, shop_url)
        if match:
            return match.group(1) or match.group(2)
        return None

    def make_hmac_sign(self, params):
        """1688标准HMAC-SHA1签名：ASCII排序+Base64编码"""
        sorted_items = sorted(params.items())
        sign_str = "".join(f"{k}{v}" for k, v in sorted_items)
        hmac_obj = hmac.new(
            self.app_secret.encode("utf-8"),
            sign_str.encode("utf-8"),
            hashlib.sha1
        )
        return base64.b64encode(hmac_obj.digest()).decode("utf-8")

    def get_shop_goods(self, member_id, page_size=50):
        """单页获取店铺商品（批发字段结构化）"""
        timestamp = str(int(time.time() * 1000))
        params = {
            "app_key": self.app_key,
            "timestamp": timestamp,
            "memberId": member_id,
            "page": str(self.current_page),
            "pageSize": str(page_size)
        }
        params["sign"] = self.make_hmac_sign(params)

        try:
            resp = requests.post(self.api_url, data=params, timeout=self.timeout)
            res = resp.json()

            if res.get("error_response"):
                return None, res["error_response"].get("msg", "接口调用失败")

            data = res.get("alibaba_product_get_by_seller_id_response", {})
            goods_list = data.get("result", {}).get("itemList", [])
            total = data.get("result", {}).get("total", 0)
            return goods_list, total

        except Exception as e:
            return None, str(e)

    def get_all_shop_goods(self, shop_url, page_size=50):
        """全量获取店铺所有商品：自动分页+去重+防风控"""
        member_id = self.parse_member_id(shop_url)
        if not member_id:
            return {"code": -1, "msg": "MemberId解析失败，请检查店铺URL"}

        self.current_page = 1
        while True:
            goods_list, total = self.get_shop_goods(member_id, page_size)
            if not goods_list:
                break

            for g in goods_list:
                offer_id = g.get("offerId")
                if offer_id in self.offer_ids:
                    continue
                self.offer_ids.add(offer_id)

                self.all_items.append({
                    "offer_id": offer_id,
                    "title": g.get("subject"),
                    "price": g.get("price"),
                    "min_order": g.get("minOrderCount"),  # B2B起批量
                    "price_range": g.get("priceRange"),  # 批发价区间
                    "seller_nick": g.get("sellerNick"),
                    "is_gold_supplier": g.get("isGoldSupplier"),  # 实力商家
                    "main_img": g.get("imageUrl"),
                    "detail_url": g.get("detailUrl"),
                    "status": g.get("status")  # 上下架状态
                })

            # 分页终止条件
            if len(goods_list) < page_size or self.current_page >= 100:
                break
            self.current_page += 1
            time.sleep(1.2)  # 防风控间隔

        return {
            "code": 200,
            "msg": "success",
            "member_id": member_id,
            "total": len(self.all_items),
            "items": self.all_items
        }

# 调用示例
if __name__ == "__main__":
    API_KEY = "你的app_key"
    API_SECRET = "你的app_secret"
    SHOP_URL = "https://shop1234567890123456.1688.com/"  # 替换为目标店铺URL

    api = Ali1688ShopGoodsAPI(API_KEY, API_SECRET)
    result = api.get_all_shop_goods(SHOP_URL)
    print(json.dumps(result, ensure_ascii=False, indent=2))