从零封装淘宝 SDK：以商品详情接口为例的客户端库设计与实现

pip install requests python-dotenv

taobao_sdk/
├── __init__.py       # SDK入口，暴露核心客户端类
├── client.py         # 核心客户端，整合所有层，对外提供API
├── exceptions.py     # 自定义异常体系
└── utils.py          # 工具层，实现签名、时间戳等通用方法

# 所有自定义异常的基类，方便统一捕获
class TaobaoBaseException(Exception):
    """淘宝SDK基础异常"""
    pass


class TaobaoNetworkException(TaobaoBaseException):
    """网络异常：超时、连接失败、无响应等"""
    def __init__(self, msg: str = "淘宝接口网络请求失败"):
        self.msg = msg
        super().__init__(self.msg)


class TaobaoApiException(TaobaoBaseException):
    """淘宝API业务异常：开放平台返回错误码"""
    def __init__(self, error_code: str, error_msg: str):
        self.error_code = error_code
        self.error_msg = error_msg
        super().__init__(f"淘宝API错误[{error_code}]：{error_msg}")


class TaobaoSignException(TaobaoBaseException):
    """签名异常：签名生成失败、签名验证不通过"""
    def __init__(self, msg: str = "淘宝接口签名错误"):
        self.msg = msg
        super().__init__(self.msg)

import time
import hashlib
from typing import Dict, Any

# 淘宝接口默认编码
DEFAULT_ENCODING = "utf-8"
# GMT+8时区偏移（秒）
GMT8_OFFSET = 8 * 3600


def generate_gmt8_timestamp() -> str:
    """生成淘宝要求的GMT+8时间戳，格式：yyyy-MM-dd HH:mm:ss"""
    # 获取时间戳并转换为GMT+8时间
    gmt8_time = time.gmtime(time.time() + GMT8_OFFSET)
    return time.strftime("%Y-%m-%d %H:%M:%S", gmt8_time)


def sort_params(params: Dict[str, Any]) -> Dict[str, Any]:
    """按参数名首字母升序排序，淘宝签名的必备步骤"""
    return dict(sorted(params.items(), key=lambda x: x[0]))


def generate_sign(sorted_params: Dict[str, Any], app_secret: str) -> str:
    """
    生成淘宝接口的MD5签名
    :param sorted_params: 已排序的所有请求参数（公共+私有）
    :param app_secret: 淘宝开放平台的app_secret
    :return: 大写的MD5签名串
    """
    try:
        # 拼接参数为：app_secret + k1v1k2v2... + app_secret
        sign_str = app_secret
        for k, v in sorted_params.items():
            # 跳过空值参数，淘宝不参与签名
            if v is None or v == "":
                continue
            sign_str += f"{k}{v}"
        sign_str += app_secret
        # MD5加密并转大写
        md5 = hashlib.md5()
        md5.update(sign_str.encode(DEFAULT_ENCODING))
        return md5.hexdigest().upper()
    except Exception as e:
        from .exceptions import TaobaoSignException
        raise TaobaoSignException(f"签名生成失败：{str(e)}") from e

import requests
from typing import Dict, Optional, Any
from .utils import generate_gmt8_timestamp, sort_params, generate_sign
from .exceptions import TaobaoBaseException, TaobaoNetworkException, TaobaoApiException

# 淘宝开放平台REST接口地址
TAOBAO_API_URL = "https://eco.taobao.com/router/rest"
# 默认请求超时时间（秒）
DEFAULT_TIMEOUT = 10
# 默认响应格式
DEFAULT_FORMAT = "json"
# 默认API版本
DEFAULT_API_VERSION = "2.0"
# 默认签名方式
DEFAULT_SIGN_METHOD = "md5"


class TaobaoClient:
    """淘宝SDK核心客户端"""
    def __init__(self, app_key: str, app_secret: str, timeout: int = DEFAULT_TIMEOUT):
        """
        初始化淘宝客户端
        :param app_key: 淘宝开放平台申请的app_key
        :param app_secret: 淘宝开放平台申请的app_secret
        :param timeout: 请求超时时间，默认10秒
        """
        if not all([app_key, app_secret]):
            raise ValueError("app_key和app_secret不能为空")
        self.app_key = app_key
        self.app_secret = app_secret
        self.timeout = timeout
        # 初始化requests会话，提升请求效率
        self.session = requests.Session()

    def _get_common_params(self, method: str) -> Dict[str, str]:
        """
        获取淘宝接口的公共参数
        :param method: 接口方法名，如taobao.item.get
        :return: 公共参数字典
        """
        return {
            "app_key": self.app_key,
            "method": method,
            "timestamp": generate_gmt8_timestamp(),
            "format": DEFAULT_FORMAT,
            "v": DEFAULT_API_VERSION,
            "sign_method": DEFAULT_SIGN_METHOD,
            "charset": "utf-8"
        }

    def _request(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """
        通用请求方法：发送POST请求，处理响应和通用异常
        :param params: 拼接好的所有请求参数（公共+私有）
        :return: 解析后的JSON响应数据
        """
        try:
            # 发送POST请求，淘宝推荐使用form-data格式
            response = self.session.post(
                url=TAOBAO_API_URL,
                data=params,
                timeout=self.timeout,
                headers={"Content-Type": "application/x-www-form-urlencoded;charset=utf-8"}
            )
            # 检查HTTP状态码
            response.raise_for_status()
            # 解析JSON响应
            result = response.json()
            # 处理淘宝API业务错误
            if "error_response" in result:
                error_info = result["error_response"]
                raise TaobaoApiException(
                    error_code=error_info.get("code", "UNKNOWN"),
                    error_msg=error_info.get("msg", "未知错误")
                )
            return result
        except requests.exceptions.RequestException as e:
            # 捕获所有网络相关异常，封装为自定义网络异常
            raise TaobaoNetworkException(f"网络请求异常：{str(e)}") from e
        except Exception as e:
            # 其他未知异常，封装为基础异常
            raise TaobaoBaseException(f"请求处理失败：{str(e)}") from e

    def get_item_detail(self, num_iid: str, fields: Optional[str] = None) -> Dict[str, Any]:
        """
        淘宝商品详情接口（taobao.item.get）
        官方文档：https://open.taobao.com/api.htm?docId=2451&docType=2
        :param num_iid: 商品ID，必填
        :param fields: 需要返回的商品字段，如title,price,pic_url，默认返回核心字段
        :return: 解析后的商品详情数据
        """
        # 1. 定义接口方法名
        api_method = "taobao.item.get"
        # 2. 获取公共参数
        common_params = self._get_common_params(api_method)
        # 3. 定义接口私有参数
        private_params = {
            "num_iid": num_iid
        }
        # 4. 可选字段拼接
        if fields:
            private_params["fields"] = fields
        # 5. 合并公共参数和私有参数
        all_params = {**common_params, **private_params}
        # 6. 参数排序（签名必备）
        sorted_params = sort_params(all_params)
        # 7. 生成签名并添加到参数中
        sorted_params["sign"] = generate_sign(sorted_params, self.app_secret)
        # 8. 发送请求并获取响应
        result = self._request(sorted_params)
        # 9. 解析并返回商品详情核心数据（剥离外层响应）
        return result.get("item_get_response", {}).get("item", {})

    def close(self):
        """关闭requests会话，释放资源"""
        self.session.close()

    # 上下文管理器支持，方便with语句使用
    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

from .client import TaobaoClient
from .exceptions import (
    TaobaoBaseException,
    TaobaoNetworkException,
    TaobaoApiException,
    TaobaoSignException
)

# 定义SDK版本
__version__ = "1.0.0"

# 导出核心对象
__all__ = [
    "TaobaoClient",
    "TaobaoBaseException",
    "TaobaoNetworkException",
    "TaobaoApiException",
    "TaobaoSignException"
]

from taobao_sdk import TaobaoClient, TaobaoBaseException, TaobaoApiException, TaobaoNetworkException

# 配置淘宝开放平台的app_key和app_secret
APP_KEY = "你的淘宝app_key"
APP_SECRET = "你的淘宝app_secret"

def main():
    # 方式1：直接初始化，手动关闭
    # client = TaobaoClient(APP_KEY, APP_SECRET, timeout=15)
    # 方式2：使用with语句，自动关闭会话（推荐）
    with TaobaoClient(APP_KEY, APP_SECRET, timeout=15) as client:
        try:
            # 调用商品详情接口，指定商品ID和需要返回的字段
            item_detail = client.get_item_detail(
                num_iid="123456789",  # 替换为真实的淘宝商品ID
                fields="num_iid,title,price,pic_url,item_desc,sell_point"
            )
            # 打印商品详情
            print("商品详情：", item_detail)
            print("商品标题：", item_detail.get("title"))
            print("商品价格：", item_detail.get("price"))
        except TaobaoApiException as e:
            # 捕获淘宝API业务错误（如商品不存在、app_key无权限）
            print(f"API业务错误：{e.error_code} - {e.error_msg}")
        except TaobaoNetworkException as e:
            # 捕获网络错误（如超时、连接失败）
            print(f"网络错误：{e.msg}")
        except TaobaoBaseException as e:
            # 捕获所有淘宝SDK异常
            print(f"SDK错误：{str(e)}")
        except Exception as e:
            # 捕获其他未知错误
            print(f"系统错误：{str(e)}")

if __name__ == "__main__":
    main()

# 应用1的客户端
client1 = TaobaoClient("app_key_1", "app_secret_1")
# 应用2的客户端
client2 = TaobaoClient("app_key_2", "app_secret_2", timeout=20)

# 分别调用接口
item1 = client1.get_item_detail("123456")
item2 = client2.get_item_detail("987654")

def search_item(self, q: str, page_no: int = 1, page_size: int = 20, fields: Optional[str] = None) -> Dict[str, Any]:
    """
    淘宝商品搜索接口（taobao.item.search）
    官方文档：https://open.taobao.com/api.htm?docId=2450&docType=2
    :param q: 搜索关键词，必填
    :param page_no: 页码，默认1
    :param page_size: 每页条数，默认20（淘宝有上限）
    :param fields: 返回字段，默认核心字段
    :return: 商品搜索结果
    """
    api_method = "taobao.item.search"
    common_params = self._get_common_params(api_method)
    private_params = {
        "q": q,
        "page_no": page_no,
        "page_size": page_size
    }
    if fields:
        private_params["fields"] = fields
    # 复用通用逻辑：合并参数→排序→生成签名→请求
    all_params = {**common_params, **private_params}
    sorted_params = sort_params(all_params)
    sorted_params["sign"] = generate_sign(sorted_params, self.app_secret)
    result = self._request(sorted_params)
    # 解析搜索结果
    return result.get("item_search_response", {}).get("items", {})