API 接口调用详解与最佳实践：无缝集成 1688 商品数据

import time
import hashlib
import requests
import json

class AlibabaAPI:
    def __init__(self, app_key, app_secret, access_token):
        self.app_key = app_key
        self.app_secret = app_secret
        self.access_token = access_token
        self.gateway = "https://gw.open.1688.com/openapi/param2/1/"
    
    def _generate_sign(self, params):
        """生成签名"""
        # 按参数名排序
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        # 拼接参数
        sign_str = self.app_secret
        for k, v in sorted_params:
            sign_str += f"{k}{v}"
        sign_str += self.app_secret
        # MD5加密并转为大写
        return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
    
    def call_api(self, api_name, biz_params):
        """调用API接口"""
        # 系统参数
        sys_params = {
            "app_key": self.app_key,
            "method": api_name,
            "format": "json",
            "v": "2.0",
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
            "sign_method": "md5",
            "access_token": self.access_token
        }
        
        # 合并系统参数和业务参数
        all_params = {**sys_params,** biz_params}
        
        # 生成签名
        all_params["sign"] = self._generate_sign(all_params)
        
        # 发送请求
        try:
            response = requests.get(self.gateway + api_name.replace('.', '/'), params=all_params, timeout=10)
            result = json.loads(response.text)
            
            # 处理错误响应
            if "error_response" in result:
                error = result["error_response"]
                raise Exception(f"API Error: {error['msg']} (code: {error['code']})")
            
            return result
        except Exception as e:
            print(f"调用失败: {str(e)}")
            return None

# 使用示例
if __name__ == "__main__":
    # 替换为实际的App Key、App Secret和Access Token
    APP_KEY = "your_app_key"
    APP_SECRET = "your_app_secret"
    ACCESS_TOKEN = "your_access_token"
    
    # 初始化API客户端
    alibaba_api = AlibabaAPI(APP_KEY, APP_SECRET, ACCESS_TOKEN)
    
    # 1. 搜索商品（示例接口：alibaba.product.search）
    search_params = {
        "keywords": "女装",  # 搜索关键词
        "page_no": 1,       # 页码
        "page_size": 20     # 每页数量
    }
    search_result = alibaba_api.call_api("alibaba.product.search", search_params)
    if search_result:
        print(f"搜索到{search_result['total_results']}个商品")
        for item in search_result.get("products", []):
            print(f"商品ID: {item['product_id']}, 标题: {item['title']}, 价格: {item['price']}")
    
    # 2. 获取商品详情（示例接口：alibaba.product.get）
    if search_result and search_result.get("products"):
        product_id = search_result["products"][0]["product_id"]
        detail_params = {
            "product_id": product_id,
            "fields": "product_id,title,price,specs,pics,description"  # 需要返回的字段
        }
        detail_result = alibaba_api.call_api("alibaba.product.get", detail_params)
        if detail_result:
            product = detail_result["product"]
            print(f"\n商品详情: {product['title']}")
            print(f"价格: {product['price']}")
            print(f"规格: {product['specs']}")