1688 商品 API 实战指南：B2B 场景下的合规对接与批量运营方案

import hashlibimport timeimport urllib.parseimport requestsimport osdef generate_1688_sign(params, app_secret):    """生成1688 API签名（关键：参数需URL编码）"""    # 1. 排除sign参数，按参数名ASCII升序排序    sorted_params = sorted([(k, v) for k, v in params.items() if k != "sign"])    # 2. 对参数值进行URL编码（1688特有要求，区别于淘宝）    encoded_params = []    for k, v in sorted_params:        # 确保参数值为字符串，特殊字符编码（如中文、&）        encoded_v = urllib.parse.quote_plus(str(v), encoding="utf-8")        encoded_params.append((k, encoded_v))    # 3. 拼接参数字符串    sign_str = "&".join([f"{k}={v}" for k, v in encoded_params])    # 4. 末尾拼接AppSecret，MD5加密后转大写    sign_str += "&secret=" + app_secret  # 1688签名需加"&secret="前缀    return hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()def get_1688_single_product(product_id, fields=None):    """查询单个1688商品详情（企业账号版）"""    # 从环境变量获取凭证（合规安全）    app_key = os.getenv("1688_APP_KEY")    app_secret = os.getenv("1688_APP_SECRET")    access_token = os.getenv("1688_ACCESS_TOKEN")        # 默认返回B2B核心字段，可按需扩展    default_fields = "productId,title,moq,priceRange,stockInfo,sellerInfo,batchPolicy"    fields = fields or default_fields        # 构造请求参数    params = {        "app_key": app_key,        "method": "alibaba.product.get",        "access_token": access_token,        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),  # 格式严格匹配        "format": "json",        "v": "1.0",        "productId": product_id,        "fields": fields    }        # 生成签名    params["sign"] = generate_1688_sign(params, app_secret)        # 发送HTTPS请求（1688强制要求HTTPS）    try:        response = requests.get(            url="https://gw.open.1688.com/openapi/param2/1/com.alibaba.product/alibaba.product.get",            params=params,            timeout=15,  # B2B数据量较大，适当延长超时            verify=True  # 开启SSL验证，避免安全风险        )        response.raise_for_status()  # 捕获HTTP错误（如403、500）        result = response.json()    except requests.exceptions.RequestException as e:        raise Exception(f"接口请求异常：{str(e)}")        # 处理错误响应    if "error_response" in result:        error = result["error_response"]        raise Exception(f"API错误({error['code']})：{error['msg']}（可能是权限不足或商品ID无效）")        # 返回商品核心数据    return result["product_get_response"]["product"]def get_1688_batch_products(product_ids, fields=None):    """批量查询1688商品详情（最多100个ID）"""    if len(product_ids) > 100:        raise Exception("批量查询最多支持100个商品ID")        app_key = os.getenv("1688_APP_KEY")    app_secret = os.getenv("1688_APP_SECRET")    access_token = os.getenv("1688_ACCESS_TOKEN")        default_fields = "productId,title,moq,priceRange,stockInfo,sellerInfo"    fields = fields or default_fields        params = {        "app_key": app_key,        "method": "alibaba.product.batch.get",        "access_token": access_token,        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),        "format": "json",        "v": "1.0",        "productIds": ",".join(product_ids),  # 商品ID用英文逗号分隔        "fields": fields    }        params["sign"] = generate_1688_sign(params, app_secret)        try:        response = requests.get(            url="https://gw.open.1688.com/openapi/param2/1/com.alibaba.product/alibaba.product.batch.get",            params=params,            timeout=20,            verify=True        )        response.raise_for_status()        result = response.json()    except requests.exceptions.RequestException as e:        raise Exception(f"批量请求异常：{str(e)}")        if "error_response" in result:        error = result["error_response"]        raise Exception(f"批量API错误({error['code']})：{error['msg']}")        return result["product_batch_get_response"]["products"]["product"]# 实战调用示例if __name__ == "__main__":    try:        # 1. 单商品查询        single_product = get_1688_single_product(product_id="694567890123")        print(f"商品标题：{single_product['title']}")        print(f"最小起订量：{single_product['moq']}{single_product['moqUnit']}")        print(f"起订价：{single_product['priceRange']['minPrice']}元")        print(f"供应商类型：{single_product['sellerInfo']['factoryVerify']?'源头工厂':'贸易商'}")                # 2. 批量商品查询        batch_ids = ["694567890123", "694567890456", "694567890789"]  # 示例ID        batch_products = get_1688_batch_products(product_ids=batch_ids)        print(f"\n批量查询到{len(batch_products)}个商品")        for p in batch_products:            print(f"- {p['title']}：{p['priceRange']['minPrice']}元起订")    except Exception as e:        print(f"调用失败：{str(e)}")