技术深潜：从零开发接入淘宝商品 API 获取实时详情数据

pip install requests  # 用于发送HTTP请求

import requests
import hashlib
import time
import json
from urllib.parse import urlencode


class TaobaoAPI:
    def __init__(self, app_key, app_secret, session=None):
        """
        初始化淘宝API客户端
        :param app_key: 应用App Key
        :param app_secret: 应用App Secret
        :param session: 授权Session Key（可选）
        """
        self.app_key = app_key
        self.app_secret = app_secret
        self.session = session
        self.base_url = "https://eco.taobao.com/router/rest"  # 正式环境接口地址

    def _sign(self, params):
        """
        生成签名
        :param params: 请求参数
        :return: 签名结果
        """
        # 1. 去除空值和签名参数
        filtered_params = {k: v for k, v in params.items() if v is not None and k != 'sign'}
        # 2. 按参数名ASCII升序排序
        sorted_params = sorted(filtered_params.items(), key=lambda x: x[0])
        # 3. 拼接参数
        query_string = '&'.join([f"{k}={v}" for k, v in sorted_params])
        # 4. 拼接app_secret并MD5加密
        sign_str = query_string + self.app_secret
        sign = hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
        return sign

    def call(self, method, **kwargs):
        """
        调用API接口
        :param method: 接口方法名（如taobao.item.detail.get）
        :param kwargs: 接口私有参数
        :return: 接口返回数据（字典格式）
        """
        # 公共参数
        params = {
            "app_key": self.app_key,
            "method": method,
            "format": "json",  # 返回格式
            "v": "2.0",  # 接口版本
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),  # 时间戳
            "sign_method": "md5",  # 签名方法
            "session": self.session  # 可选，部分接口需要
        }

        # 合并私有参数
        params.update(kwargs)

        # 生成签名
        params["sign"] = self._sign(params)

        # 发送请求
        try:
            response = requests.get(self.base_url, params=params, timeout=10)
            response.raise_for_status()  # 抛出HTTP错误
            result = response.json()

            # 处理返回结果
            if "error_response" in result:
                error = result["error_response"]
                raise Exception(f"API错误: {error['msg']} (错误码: {error['code']})")
            return result
        except Exception as e:
            print(f"调用失败: {str(e)}")
            return None

    def get_item_detail(self, item_id):
        """
        获取商品详情
        :param item_id: 商品ID
        :return: 商品详情数据
        """
        return self.call(
            method="taobao.item.detail.get",
            item_id=item_id,
            fields="item:id,title,price,desc,pics,sales,stock"  # 需要返回的字段
        )


if __name__ == "__main__":
    # 配置参数（替换为自己的实际信息）
    APP_KEY = "你的AppKey"
    APP_SECRET = "你的AppSecret"
    SESSION_KEY = "你的SessionKey"  # 非必需，根据接口要求填写

    # 初始化API客户端
    taobao_api = TaobaoAPI(APP_KEY, APP_SECRET, SESSION_KEY)

    # 获取商品详情（示例商品ID，可替换为实际ID）
    item_id = "6543217890"  # 淘宝商品ID
    detail = taobao_api.get_item_detail(item_id)

    if detail:
        print("商品详情获取成功：")
        print(json.dumps(detail, ensure_ascii=False, indent=2))