以下是一个通用的电商平台 API 接口调用 Python 脚本框架，适用于 doubao 的特点，这个脚本支持多种主流电商平台（如淘宝、京东、拼多多等）的 API 调用，并并包含了签名验证、请求处理、异常处理等核心功能。
import requests
import time
import hashlib
import hmac
import json
from urllib.parse import urlencode, quote_plus
from typing import Dict, Optional, Any, List

class ApiException(Exception):
"""API调用异常基类"""
def init(self, code: str, message: str):
self.code = code
self.message = message
super().init(f"{code}: {message}")

class AuthenticationException(ApiException):
"""认证相关异常"""
pass

class RateLimitException(ApiException):
"""接口限流异常"""
def init(self, code: str, message: str, retry_after: int = 0):
super().init(code, message)
self.retry_after = retry_after

class EcommerceApiClient:
"""电商平台API通用客户端"""

def __init__(self, platform: str, app_key: str, app_secret: str, 
             access_token: Optional[str] = None, sandbox: bool = False):
    """
    初始化API客户端
    :param platform: 平台名称，如"taobao", "jd", "pinduoduo", "amazon"
    :param app_key: 应用Key
    :param app_secret: 应用密钥
    :param access_token: 访问令牌，部分平台需要
    :param sandbox: 是否使用沙箱环境
    """
    self.platform = platform.lower()
    self.app_key = app_key
    self.app_secret = app_secret
    self.access_token = access_token
    self.sandbox = sandbox
    self.session = requests.Session()
    self.session.headers.update({
        "Content-Type": "application/x-www-form-urlencoded;charset=utf-8",
        "User-Agent": "EcommerceApiClient/1.0.0"
    })

    # 各平台API配置
    self.api_configs = {
        "taobao": {
            "gateway": "https://eco.taobao.com/router/rest",
            "sandbox_gateway": "https://gw-api.tbsandbox.com/router/rest",
            "sign_method": "md5"
        },
        "jd": {
            "gateway": "https://api.jd.com/routerjson",
            "sandbox_gateway": "https://api.jd.com/routerjson",
            "sign_method": "md5"
        },
        "pinduoduo": {
            "gateway": "https://gw-api.pinduoduo.com/api/router",
            "sandbox_gateway": "https://gw-api.pinduoduo.com/api/router",
            "sign_method": "md5"
        },
        "amazon_sp": {
            "gateway": "https://sellingpartnerapi-na.amazon.com",
            "sandbox_gateway": "https://sandbox.sellingpartnerapi-na.amazon.com",
            "sign_method": "hmac-sha256"
        }
    }

    # 确保平台配置存在
    if self.platform not in self.api_configs:
        raise ValueError(f"不支持的平台: {platform}，支持的平台有: {list(self.api_configs.keys())}")

    self.config = self.api_configs[self.platform]
    self.gateway = self.config["sandbox_gateway"] if sandbox else self.config["gateway"]

    # 请求计数器，用于控制频率
    self.request_counter = 0
    self.last_reset_time = time.time()

def _get_sign(self, params: Dict[str, Any]) -> str:
    """
    根据平台规则生成签名
    :param params: 请求参数
    :return: 签名字符串
    """
    if self.platform == "taobao" or self.platform == "jd":
        # 淘宝/京东签名方式: 排序后拼接app_secret
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        sign_str = self.app_secret + "".join([f"{k}{v}" for k, v in sorted_params]) + self.app_secret
        return hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()

    elif self.platform == "pinduoduo":
        # 拼多多签名方式
        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
        return hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()

    elif self.platform == "amazon_sp":
        # 亚马逊SP-API签名较为复杂，这里简化处理
        # 实际使用建议参考官方SDK
        timestamp = time.strftime('%Y%m%dT%H%M%SZ', time.gmtime())
        return hmac.new(
            self.app_secret.encode('utf-8'),
            timestamp.encode('utf-8'),
            hashlib.sha256
        ).hexdigest()

    return ""

def _check_rate_limit(self):
    """检查并控制请求频率"""
    current_time = time.time()
    # 每分钟最多60次请求
    if current_time - self.last_reset_time > 60:
        self.request_counter = 0
        self.last_reset_time = current_time

    self.request_counter += 1
    if self.request_counter > 60:
        raise RateLimitException("RATE_LIMIT", "请求频率超过限制", 60 - (current_time - self.last_reset_time))

def _handle_response(self, response: requests.Response) -> Dict[str, Any]:
    """
    处理API响应
    :param response: 响应对象
    :return: 解析后的响应数据
    """
    try:
        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 10))
            raise RateLimitException("RATE_LIMIT", "请求过于频繁", retry_after)

        response.raise_for_status()

        # 不同平台返回格式可能不同
        if self.platform in ["taobao", "jd", "pinduoduo"]:
            result = response.json()

            # 处理淘宝API响应
            if self.platform == "taobao":
                error_response = result.get("error_response")
                if error_response:
                    raise ApiException(
                        error_response.get("code", "UNKNOWN_ERROR"),
                        error_response.get("msg", "未知错误")
                    )

            # 处理京东API响应
            elif self.platform == "jd":
                if result.get("code") != 0:
                    raise ApiException(
                        str(result.get("code", "UNKNOWN_ERROR")),
                        result.get("msg", "未知错误")
                    )

            # 处理拼多多API响应
            elif self.platform == "pinduoduo":
                if "error_response" in result:
                    error = result["error_response"]
                    raise ApiException(
                        str(error.get("code", "UNKNOWN_ERROR")),
                        error.get("error_msg", "未知错误")
                    )

            return result

        elif self.platform == "amazon_sp":
            # 亚马逊SP-API响应处理
            if response.headers.get("Content-Type", "").startswith("application/json"):
                return response.json()
            return {"content": response.text}

        return response.json()

    except requests.exceptions.JSONDecodeError:
        raise ApiException("INVALID_RESPONSE", f"无效的响应格式: {response.text}")
    except requests.exceptions.HTTPError as e:
        raise ApiException("HTTP_ERROR", f"HTTP错误: {str(e)}")

def call_api(self, method: str, params: Optional[Dict[str, Any]] = None, 
             http_method: str = "POST") -> Dict[str, Any]:
    """
    通用API调用方法
    :param method: API方法名，如"taobao.item.get"
    :param params: API参数
    :param http_method: HTTP请求方法，GET或POST
    :return: API返回结果
    """
    # 检查频率限制
    self._check_rate_limit()

    # 基础参数
    base_params = {
        "app_key": self.app_key,
        "method": method,
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
        "format": "json",
        "v": "2.0",
        "sign_method": self.config["sign_method"]
    }

    # 添加访问令牌（如果有）
    if self.access_token:
        base_params["access_token"] = self.access_token

    # 合并参数
    request_params = {** base_params, **(params or {})}

    # 生成签名
    request_params["sign"] = self._get_sign(request_params)

    # 构建请求URL和数据
    url = self.gateway
    data = None
    if http_method.upper() == "GET":
        url = f"{url}?{urlencode(request_params, quote_via=quote_plus)}"
    else:
        data = urlencode(request_params, quote_via=quote_plus)

    # 发送请求
    try:
        response = self.session.request(
            method=http_method,
            url=url,
            data=data,
            timeout=30
        )
        return self._handle_response(response)
    except requests.exceptions.RequestException as e:
        raise ApiException("REQUEST_FAILED", f"请求失败: {str(e)}")

使用示例

if name == "main":

# 初始化客户端（以淘宝为例）
taobao_client = EcommerceApiClient(
    platform="taobao",
    app_key="YOUR_TAOBAO_APP_KEY",
    app_secret="YOUR_TAOBAO_APP_SECRET",
    access_token="YOUR_TAOBAO_ACCESS_TOKEN",
    sandbox=True  # 测试环境
)

try:
    # 调用商品搜索API
    search_result = taobao_client.call_api(
        method="taobao.item.search",
        params={
            "q": "手机",
            "page_no": 1,
            "page_size": 20
        }
    )
    print("淘宝商品搜索结果:")
    print(json.dumps(search_result, ensure_ascii=False, indent=2))

    # 调用订单列表API（以拼多多为例）
    pdd_client = EcommerceApiClient(
        platform="pinduoduo",
        app_key="YOUR_PDD_CLIENT_ID",
        app_secret="YOUR_PDD_CLIENT_SECRET",
        access_token="YOUR_PDD_ACCESS_TOKEN"
    )

    order_result = pdd_client.call_api(
        method="pdd.order.list.get",
        params={
            "start_confirm_at": int(time.time()) - 86400 * 7,  # 7天前
            "end_confirm_at": int(time.time()),
            "page": 1,
            "page_size": 10
        }
    )
    print("\n拼多多订单列表结果:")
    print(json.dumps(order_result, ensure_ascii=False, indent=2))

except RateLimitException as e:
    print(f"请求被限流，建议{e.retry_after}秒后重试")
except AuthenticationException as e:
    print(f"认证失败: {e.message}")
except ApiException as e:
    print(f"API调用错误: {e.code} - {e.message}")
except Exception as e:
    print(f"发生未知错误: {str(e)}")

这个电商平台 API 调用脚本具有以下特点：
1. 多平台支持 ：内置了淘宝、京东、拼多多和亚马逊 SP-API 的配置，可轻松扩展支持其他平台
2. 完整的签名机制 ：根据不同平台的签名规则实现了对应的签名生成方法
3. 异常处理 ：定义了多种异常类型，包括认证异常、限流异常等，便于错误处理
4. 频率控制 ：内置请求频率控制机制，避免触发平台的 API 调用限制
5. 通用接口 ：提供统一的callapi方法，简化不同 API 的调用流程
使用方法：
替换示例中的YOUR*为实际的 API 密钥和令牌
根据需要调用的 API 方法，传入相应的参数
处理返回结果或捕获可能的异常
不同平台的 API 方法名和参数格式可能有所不同，使用时需要参考各平台的官方文档进行调整。对于更复杂的场景，建议结合平台提供的 SDK 进行开发。