[特殊字符]电商 API 接口选对了吗？1 个接口打通淘宝京东，效率提升 3 倍-洪萨配资

一、核心设计思路

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

二、完整实现代码

前置依赖安装

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)

三、关键代码解释

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

四、使用前置条件

淘宝侧：
- 注册淘宝开放平台开发者账号，创建应用并获取AppKey/AppSecret。
- 申请目标接口（如商品详情、订单查询）的调用权限。
- 测试阶段可使用淘宝沙箱环境：https://gw.api.tbsandbox.com/router/rest。
京东侧：
- 注册京东开放平台开发者账号，创建应用并获取ClientID/ClientSecret。
- 通过授权流程获取Access Token（可先手动在京东开发者后台生成测试 token）。
- 申请对应接口的权限（如京东联盟订单查询权限）。