一、核心设计思路

  1. 配置集中管理:把淘宝、京东的密钥、接口地址等配置统一维护,降低维护成本。
  2. 平台逻辑隔离:各自实现淘宝 / 京东的鉴权、参数拼接逻辑,对外屏蔽差异。
  3. 统一调用入口:封装成一个类,提供 call 方法,只需指定平台、接口名、参数即可调用。
  4. 稳定性增强:增加重试、异常处理机制,适配两个平台的调用规范。

二、完整实现代码

前置依赖安装

bash

运行

pip install requests hashlib hmac time uuid

统一 API 客户端代码

python

运行

import requests
import hashlib
import hmac
import time
import uuid
from typing import Dict, Any

class UnifiedEcommerceAPI:
    """统一的淘宝/京东电商API客户端"""
    
    def __init__(self, config: Dict[str, Dict[str, str]]):
        """
        初始化客户端
        :param config: 平台配置字典,格式示例:
        {
            "taobao": {
                "app_key": "淘宝AppKey",
                "app_secret": "淘宝AppSecret",
                "gateway": "https://eco.taobao.com/router/rest"
            },
            "jd": {
                "client_id": "京东ClientID",
                "client_secret": "京东ClientSecret",
                "token_url": "https://oauth.jd.com/oauth/token",
                "api_url": "https://api.jd.com/routerjson",
                "redirect_uri": "京东回调地址",
                "access_token": "京东Access Token(可先手动获取)"
            }
        }
        """
        self.config = config
        # 重试配置(可根据平台调整)
        self.retry_times = 3
        self.timeout = 10

    # -------------------------- 淘宝API核心逻辑 --------------------------
    def _taobao_sign(self, params: Dict[str, Any], app_secret: str) -> str:
        """生成淘宝API签名(MD5-HMAC)"""
        # 1. 按参数名ASCII升序排序
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        # 2. 拼接成 key=value 格式
        sign_str = app_secret + "".join([f"{k}{v}" for k, v in sorted_params]) + app_secret
        # 3. MD5加密并转大写
        sign = hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
        return sign

    def _call_taobao(self, method: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """调用淘宝API"""
        tb_config = self.config["taobao"]
        # 淘宝公共参数(必传)
        common_params = {
            "app_key": tb_config["app_key"],
            "method": method,          # 接口方法名(如taobao.item.get)
            "format": "json",
            "v": "2.0",
            "sign_method": "md5",
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
            "partner_id": "apidoc",
            "session": "",             # 如需用户授权则填写,开放接口可留空
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        }
        # 合并公共参数和业务参数
        all_params = {**common_params, **params}
        # 生成签名
        all_params["sign"] = self._taobao_sign(all_params, tb_config["app_secret"])
        
        # 发送请求(带重试)
        for retry in range(self.retry_times):
            try:
                response = requests.post(
                    tb_config["gateway"],
                    params=all_params,
                    timeout=self.timeout
                )
                result = response.json()
                # 检查淘宝返回的错误
                if "error_response" in result:
                    raise Exception(f"淘宝API错误: {result['error_response']}")
                return result
            except Exception as e:
                if retry == self.retry_times - 1:
                    raise Exception(f"淘宝API调用失败(重试{self.retry_times}次): {str(e)}")
                time.sleep(0.5)  # 重试间隔
        return {}

    # -------------------------- 京东API核心逻辑 --------------------------
    def _refresh_jd_token(self) -> str:
        """刷新京东Access Token(如需自动刷新可扩展)"""
        jd_config = self.config["jd"]
        params = {
            "grant_type": "refresh_token",
            "client_id": jd_config["client_id"],
            "client_secret": jd_config["client_secret"],
            "refresh_token": jd_config.get("refresh_token", "")  # 需提前获取
        }
        response = requests.post(jd_config["token_url"], params=params)
        token_data = response.json()
        if "access_token" in token_data:
            self.config["jd"]["access_token"] = token_data["access_token"]
            return token_data["access_token"]
        raise Exception(f"京东Token刷新失败: {token_data}")

    def _call_jd(self, method: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """调用京东API"""
        jd_config = self.config["jd"]
        access_token = jd_config["access_token"]
        
        # 京东公共参数
        common_params = {
            "method": method,          # 接口方法名(如jd.union.open.order.query)
            "app_key": jd_config["client_id"],
            "access_token": access_token,
            "format": "json",
            "v": "2.0",
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
            "sign_method": "md5",
            "uuid": str(uuid.uuid4()),
            "client_ip": "127.0.0.1"
        }
        all_params = {**common_params, **params}
        
        # 发送请求(带重试)
        for retry in range(self.retry_times):
            try:
                response = requests.post(
                    jd_config["api_url"],
                    params=all_params,
                    timeout=self.timeout
                )
                result = response.json()
                # 检查京东Token过期
                if result.get("code") == "1001":
                    if retry < self.retry_times - 1:
                        access_token = self._refresh_jd_token()
                        all_params["access_token"] = access_token
                        continue
                # 检查京东其他错误
                if result.get("code") != "0":
                    raise Exception(f"京东API错误: {result.get('msg', '未知错误')}")
                return result
            except Exception as e:
                if retry == self.retry_times - 1:
                    raise Exception(f"京东API调用失败(重试{self.retry_times}次): {str(e)}")
                time.sleep(0.5)
        return {}

    # -------------------------- 统一调用入口 --------------------------
    def call(self, platform: str, method: str, params: Dict[str, Any]) -> Dict[str, Any]:
        """
        统一调用接口
        :param platform: 平台标识(taobao/jd)
        :param method: 接口方法名(如淘宝taobao.item.get,京东jd.union.open.order.query)
        :param params: 业务参数(仅需传平台接口的业务参数,公共参数自动拼接)
        :return: 接口返回的JSON数据
        """
        if platform not in ["taobao", "jd"]:
            raise ValueError(f"不支持的平台:{platform},仅支持taobao/jd")
        
        if platform == "taobao":
            return self._call_taobao(method, params)
        elif platform == "jd":
            return self._call_jd(method, params)

# -------------------------- 使用示例 --------------------------
if __name__ == "__main__":
    # 1. 配置淘宝/京东密钥(替换为你自己的)
    api_config = {
        "taobao": {
            "app_key": "你的淘宝AppKey",
            "app_secret": "你的淘宝AppSecret",
            "gateway": "https://eco.taobao.com/router/rest"
        },
        "jd": {
            "client_id": "你的京东ClientID",
            "client_secret": "你的京东ClientSecret",
            "token_url": "https://oauth.jd.com/oauth/token",
            "api_url": "https://api.jd.com/routerjson",
            "redirect_uri": "你的京东回调地址",
            "access_token": "你的京东Access Token",  # 手动获取的token
            "refresh_token": "你的京东Refresh Token"  # 用于自动刷新token
        }
    }

    # 2. 初始化统一客户端
    client = UnifiedEcommerceAPI(api_config)

    # 3. 调用淘宝接口(示例:获取商品详情)
    try:
        tb_result = client.call(
            platform="taobao",
            method="taobao.item.get",  # 淘宝商品详情接口
            params={"num_iid": "123456789"}  # 淘宝商品ID
        )
        print("淘宝接口返回:", tb_result)
    except Exception as e:
        print("淘宝调用失败:", e)

    # 4. 调用京东接口(示例:查询订单)
    try:
        jd_result = client.call(
            platform="jd",
            method="jd.union.open.order.query",  # 京东订单查询接口
            params={
                "startTime": "2024-01-01 00:00:00",
                "endTime": "2024-01-31 23:59:59",
                "pageNo": 1,
                "pageSize": 20
            }
        )
        print("京东接口返回:", jd_result)
    except Exception as e:
        print("京东调用失败:", e)

三、关键代码解释

  1. 配置初始化UnifiedEcommerceAPI 类初始化时接收淘宝、京东的密钥配置,集中管理避免分散。
  2. 淘宝鉴权_taobao_sign 方法严格按照淘宝的 MD5 签名规则生成签名,确保请求合法。
  3. 京东 Token 处理_refresh_jd_token 实现 Token 自动刷新,解决京东 Access Token 过期问题。
  4. 统一调用入口call 方法只需传入 platform(平台)、method(接口名)、params(业务参数),无需关心底层差异。
  5. 重试机制:对两个平台的请求都增加了重试逻辑,应对网络波动或临时接口故障。

四、使用前置条件

  1. 淘宝侧

    • 注册淘宝开放平台开发者账号,创建应用并获取 AppKey/AppSecret
    • 申请目标接口(如商品详情、订单查询)的调用权限。
    • 测试阶段可使用淘宝沙箱环境:https://gw.api.tbsandbox.com/router/rest
  2. 京东侧

    • 注册京东开放平台开发者账号,创建应用并获取 ClientID/ClientSecret
    • 通过授权流程获取 Access Token(可先手动在京东开发者后台生成测试 token)。
    • 申请对应接口的权限(如京东联盟订单查询权限)。

总结

  1. 核心是通过统一封装屏蔽淘宝、京东的鉴权、参数差异,对外提供单一调用入口。
  2. 实现了重试机制Token 自动刷新,保障接口调用的稳定性。
  3. 使用时只需替换配置中的密钥 / Token,传入平台、接口名和业务参数即可调用,无需关注底层细节。
Logo

电商企业物流数字化转型必备!快递鸟 API 接口,72 小时快速完成物流系统集成。全流程实战1V1指导,营造开放的API技术生态圈。

更多推荐