1. 项目缘起：为什么需要自己动手对接京东联盟API？

最近在做一个电商数据聚合的小项目，需要整合多个平台的商品和订单信息，京东联盟自然是绕不开的一环。一开始，我也想着偷懒，去网上找找有没有现成的SDK或者封装好的库，结果发现情况有点尴尬。市面上能找到的一些第三方封装，要么是年久失修，文档缺失，要么就是功能不全，只实现了基础的“商品查询”，对于“订单查询”、“推广链接生成”这些核心功能支持得很弱。更关键的是，这些库的维护状态堪忧，京东联盟的API本身就在迭代，一旦接口有变动，等第三方库更新可能黄花菜都凉了。所以，与其把项目的稳定性寄托在别人身上，不如自己动手，从零开始理解并实现一套对接逻辑。这个过程虽然前期会多花点时间，但换来的是对接口的完全掌控、更高的定制灵活性，以及未来排查问题时清晰的底层认知。今天，我就把自己从零搭建Python对接京东联盟API的完整过程、踩过的坑和总结的经验，毫无保留地分享出来。

2. 战前准备：理解京东联盟API的核心机制与必备物料

在写第一行代码之前，我们必须先把京东联盟API的“游戏规则”搞清楚。这就像打仗前要看懂地图和武器说明书一样重要。

2.1 API的两种关键认证方式：Sign与OAuth2

京东联盟API主要采用两种认证方式，适用于不同的场景，理解它们的区别是成功对接的第一步。

1. 通用API签名（Sign）这是最常用、也是最基础的方式。几乎所有的“工具型”API，比如查询商品、生成链接、查询订单、查询佣金，都使用这种签名认证。它的核心流程是：你（开发者）需要先在京东联盟后台创建一个“应用”，拿到appKey和appSecret。每次调用API时，你需要将一堆参数（包括appKey、时间戳、API方法名等）按照特定规则拼接成一个字符串，然后用appSecret通过MD5算法生成一个签名（Sign）。服务器收到请求后，会用同样的规则再算一遍签名，如果一致，就认为请求是合法且未被篡改的。

注意：这里的appSecret是最高机密，绝对不能出现在前端代码、客户端或者任何可能被用户看到的地方。它只应该存在于你的服务器后端环境变量或安全的配置中心里。

2. OAuth2 授权这种方式用于需要“代表”某个京东联盟会员进行操作的场景。比如，如果你开发了一个工具，让推广者登录后可以查看他自己的订单、佣金明细，这时候就需要OAuth2。流程是：用户点击授权，跳转到京东的授权页面，同意后，京东会回调你指定的地址并传回一个code，你用这个code加上你的appKey和appSecret去交换access_token。后续调用用户相关的API时，就带上这个token。

对于大多数数据聚合、后台跑脚本的场景，我们主要使用第一种“应用级”的签名认证。本文也将重点围绕这种方式展开。

2.2 必不可少的“粮草”：申请应用与获取密钥

理论懂了，接下来是实操准备。你需要准备以下几样东西：

1. 一个京东联盟账号：这自然是前提，如果没有就去注册一个。
2. 创建“网站/APP应用”：
   • 登录京东联盟后台，找到“推广管理” -> “我的应用”。
   • 点击“创建应用”，应用类型根据实际情况选择（如“工具应用”）。填写应用名称、描述等。
   • 最关键的一步：在“API权限管理”中，为你需要使用的API勾选上对应的权限。例如，“商品查询”、“优惠券查询”、“订单查询”、“推广链接创建”等。不要漏选，否则调用时会报“权限不足”的错误。
3. 拿到核心三要素：应用创建成功后，你会得到：
   • appKey: 应用的唯一标识。
   • appSecret: 用于签名的密钥。
   • accessToken: 注意，这里后台显示的是一个“默认的”或“测试用的”accessToken。对于签名认证方式，我们暂时用不到它。它主要用于OAuth2流程中，或者某些特定接口。我们签名认证主要靠appKey和appSecret。

2.3 开发环境搭建：简约而不简单

我的选择是Python 3.8+，版本不要太老即可。库方面，我们追求轻量化和明确性：

• requests: 用于发送HTTP请求，这是绝对的核心。
• pandas: 非必须，但强烈推荐。用于处理和分析返回的表格数据，非常方便。
• python-dotenv: 非必须，但最佳实践。用于从.env文件加载环境变量，安全地管理你的appKey和appSecret。

安装命令很简单：

pip install requests pandas python-dotenv

我个人的习惯是在项目根目录创建一个.env文件，内容如下：

JD_UNION_APP_KEY=your_app_key_here JD_UNION_APP_SECRET=your_app_secret_here

然后在代码中通过os.getenv来读取。这样做的最大好处是代码里没有明文密钥，方便团队协作和不同环境（开发、测试、生产）的配置切换。

3. 核心攻坚：自研签名生成与通用请求函数

这是整个对接过程中最核心、也最容易出错的部分。我们将自己实现签名的生成逻辑，并封装一个健壮的通用请求函数。

3.1 解密签名（Sign）生成算法

京东联盟的签名算法其实是一种常见的HMAC-MD5的变体。官方文档有详细说明，但我们可以将其提炼为以下几个清晰步骤。假设我们要调用jd.union.open.goods.query这个API，查询关键词为“手机”的商品。

步骤一：准备所有参数将所有请求参数（包括公共参数和业务参数）放入一个字典。公共参数是每次请求都必须的：

params = { 'method': 'jd.union.open.goods.query', # API方法名 'app_key': '你的appKey', # 从环境变量读取 'timestamp': '2023-10-27 14:00:00', # 格式必须为'YYYY-MM-DD HH:MM:SS' 'format': 'json', # 返回格式 'v': '1.0', # API版本 'sign_method': 'md5', # 签名方法 # 以下是业务参数，需要封装在 `param_json` 字符串中 }

注意，所有业务参数（如商品ID、关键词、页码等）需要封装成一个JSON字符串，赋值给一个叫param_json的参数。这是京东联盟API的一个特殊规定。

import json business_params = { 'goodsReq': { 'keyword': '手机', 'pageIndex': 1, 'pageSize': 20, # ... 其他业务参数 } } params['param_json'] = json.dumps(business_params, separators=(',', ':')) # 去除空格，减少传输量

步骤二：参数排序与拼接

1. 过滤掉sign参数本身（如果有的话）。
2. 将所有参数（app_key,method,timestamp,param_json...）按照参数名ASCII码从小到大排序（字典序）。
3. 将排序后的所有参数，用key1value1key2value2...的格式拼接成一个字符串。
4. 在拼接字符串的首尾都加上你的appSecret。

用代码表示这个过程：

def generate_sign(params, app_secret): # 1. 过滤并排序 filtered_params = {k: v for k, v in params.items() if k != 'sign' and v is not None} sorted_params = sorted(filtered_params.items(), key=lambda x: x[0]) # 2. 拼接键值对 concatenated_str = '' for k, v in sorted_params: concatenated_str += f'{k}{v}' # 3. 首尾加上app_secret sign_str = app_secret + concatenated_str + app_secret # 4. 计算MD5并转为大写 import hashlib m = hashlib.md5() m.update(sign_str.encode('utf-8')) return m.hexdigest().upper()

步骤三：计算MD5并赋值将上一步得到的字符串进行MD5哈希计算，然后将结果转换为大写十六进制字符串。这个字符串就是最终的sign值。将其加入到请求参数中：params['sign'] = sign_value。

踩坑提示1：时间戳格式。timestamp的格式必须严格是'YYYY-MM-DD HH:MM:SS'，并且是东八区时间。很多请求失败是因为时间格式不对或者误差太大（服务器允许一定的时间漂移，但通常不超过5分钟）。建议在服务器端获取当前时间，而不是使用客户端可能不准确的时间。

踩坑提示2：param_json的引号。param_json是一个字符串，里面是JSON格式。在拼接签名原始字符串时，param_json的值就是这个包含引号的字符串本身。不要把它解析成字典后再拼接，那样签名一定会失败。

3.2 封装万能的请求函数

有了签名函数，我们就可以封装一个通用的请求函数来处理所有签名认证的API调用。

import requests import json import time from datetime import datetime from .sign_utils import generate_sign # 假设签名函数放在单独模块 import os from dotenv import load_dotenv load_dotenv() # 加载.env文件中的环境变量 class JDUnionClient: def __init__(self, app_key=None, app_secret=None): self.app_key = app_key or os.getenv('JD_UNION_APP_KEY') self.app_secret = app_secret or os.getenv('JD_UNION_APP_SECRET') self.base_url = 'https://router.jd.com/api' # API网关地址 if not self.app_key or not self.app_secret: raise ValueError("app_key and app_secret must be provided or set in environment variables.") def _get_timestamp(self): """生成符合要求的东八区时间戳""" return datetime.now().strftime('%Y-%m-%d %H:%M:%S') def execute(self, method, param_dict): """ 执行API调用 :param method: API方法名，如 'jd.union.open.goods.query' :param param_dict: 业务参数字典，会被转换为param_json :return: API响应数据的字典 """ # 1. 准备公共参数 public_params = { 'method': method, 'app_key': self.app_key, 'timestamp': self._get_timestamp(), 'format': 'json', 'v': '1.0', 'sign_method': 'md5', 'param_json': json.dumps(param_dict, separators=(',', ':')) } # 2. 生成签名 sign = generate_sign(public_params, self.app_secret) public_params['sign'] = sign # 3. 发送请求 try: response = requests.get(self.base_url, params=public_params, timeout=10) response.raise_for_status() # 如果状态码不是200，抛出HTTPError result = response.json() except requests.exceptions.RequestException as e: # 网络请求异常处理 raise Exception(f"Network request failed: {e}") except json.JSONDecodeError as e: # 响应不是合法JSON raise Exception(f"Invalid JSON response: {e}. Response text: {response.text}") # 4. 解析响应 # 京东联盟API的响应通常包裹在 `jd_union_open_xxx_response` 这样的键里 # 我们需要找到包含实际数据的那个键 for key in result.keys(): if key.endswith('_response'): inner_result = result[key] if inner_result.get('code') != '200': # 注意，这里是字符串'200' error_msg = inner_result.get('message', 'Unknown error') raise Exception(f"API Error [{inner_result.get('code')}]: {error_msg}") # 返回真正的数据部分 return inner_result.get('data') # 如果没找到预期的响应结构 raise Exception(f"Unexpected API response structure: {result}")

这个execute方法封装了参数组装、签名、请求发送、错误处理和数据提取的全过程。以后调用任何API，只需要关心方法名和业务参数即可。

4. 实战演练：常用API调用示例与深度解析

有了强大的客户端，我们就可以轻松调用各种API了。下面通过几个最常用的场景，展示如何调用并深入解析返回结果。

4.1 商品查询：从海量数据中精准捞取

商品查询API (jd.union.open.goods.query) 是最基础的API，参数众多，功能强大。

基础调用示例：

client = JDUnionClient() params = { 'goodsReq': { 'keyword': '蓝牙耳机', # 关键词 'pageIndex': 1, 'pageSize': 50, # 每页数量，最大100 'sortName': 'price', # 排序字段：price, commissionShare, inOrderCount30Days等 'sort': 'asc', # 排序方式：asc升序，desc降序 'isCoupon': 1, # 是否只查有券商品，1是，0否 } } try: data = client.execute('jd.union.open.goods.query', params) goods_list = data.get('list', []) print(f"查询到 {len(goods_list)} 个商品") for goods in goods_list[:3]: # 打印前3个 print(f"商品名: {goods.get('skuName')}") print(f"价格: {goods.get('price')}") print(f"佣金比例: {goods.get('commissionShare')}%") print(f"券后价: {goods.get('couponPrice')}") print("-" * 30) except Exception as e: print(f"查询失败: {e}")

参数深度解析与技巧：

• cid1,cid2,cid3: 通过类目ID筛选，可以极大提升查询精准度。如何获取类目ID？可以通过jd.union.open.category.goods.get这个API查询类目树，或者更简单，在京东联盟后台的“商品推广”页面，通过筛选类目，观察浏览器地址栏或网络请求中的cid参数。
• isPG: 是否只查询拼购商品。拼购价通常更有竞争力。
• isHot: 是否只查询爆品。对于追热点很有用。
• commissionShareStart/End: 佣金比例区间筛选。做高佣筛选的利器。
• owner: 商品归属，g=自营，p=POP店。自营商品通常物流和服务更稳定。

实操心得：不要一次性拉取太多页。京东联盟API对高频调用有限流。建议根据业务需要合理设置pageSize（比如50），并做好请求间隔控制（例如每秒1-2次）。对于需要大量数据的场景，考虑在凌晨等低峰期分批跑任务。

4.2 高效转链：将商品ID转化为推广链接

获取到商品列表后，下一步就是生成包含你推广位的购买链接。这里主要用到jd.union.open.promotion.common.get(通用推广链接创建)。

def generate_promotion_url(client, material_id, site_id): """ 生成推广链接 :param material_id: 商品ID (skuId) :param site_id: 推广位ID (你在联盟后台创建的) """ params = { 'promotionCodeReq': { 'materialId': str(material_id), # 注意转为字符串 'siteId': str(site_id), 'positionId': None, # 子推广位ID，可选 'couponUrl': None, # 如有关联优惠券，可传入券链接 } } try: data = client.execute('jd.union.open.promotion.common.get', params) # 返回数据中包含了短链接、长链接等信息 click_url = data.get('clickURL') # 推广长链接，用于嵌入网页 short_url = data.get('shortURL') # 推广短链接，用于文案、社交媒体 return {'click_url': click_url, 'short_url': short_url} except Exception as e: print(f"生成推广链接失败: {e}") return None # 使用示例 sku_id = '100012345678' # 示例商品ID site_id = '1234567' # 你的推广位ID url_info = generate_promotion_url(client, sku_id, site_id) if url_info: print(f"长链接: {url_info['click_url']}") print(f"短链接: {url_info['short_url']}")

重要提示：materialId可以是商品ID (skuId)，也可以是活动URL、内容频道ID等。siteId是你在京东联盟后台“推广管理”-“推广位管理”中创建的。不同推广位用于区分不同的流量来源，便于后期数据统计。

4.3 订单与佣金查询：数据核对的命脉

订单查询API (jd.union.open.order.query) 是进行佣金结算和数据核对的核心。它的参数设计主要围绕时间维度。

def query_orders(client, start_time, end_time, page_index=1, page_size=100): """ 查询指定时间范围内的订单 :param start_time/end_time: 格式 '2023-10-01 00:00:00' """ params = { 'orderReq': { 'pageIndex': page_index, 'pageSize': page_size, 'type': 1, # 订单时间类型：1-下单时间，2-完成时间，3-更新时间 'time': f'{start_time},{end_time}', # 'childUnionId': 0, # 子推客ID，如果你发展了下级，可以查下级的订单 } } try: data = client.execute('jd.union.open.order.query', params) order_list = data.get('data', []) total_count = data.get('totalCount', 0) print(f"时间范围[{start_time} - {end_time}]内共有 {total_count} 条订单，本页返回 {len(order_list)} 条") # 使用pandas进行数据分析非常方便 import pandas as pd if order_list: df = pd.DataFrame(order_list) # 计算预估总佣金 estimated_total_commission = df['estimateCosPrice'].astype(float).sum() print(f"本页订单预估总佣金: {estimated_total_commission:.2f} 元") # 筛选已结算的订单 settled_orders = df[df['validCode'] == 17] # 17代表已结算 print(f"其中已结算订单: {len(settled_orders)} 条") return order_list except Exception as e: print(f"订单查询失败: {e}") return []

订单状态 (validCode) 解读（部分关键状态）：

• 3: 已付款（等待发货）
• 11: 已完成（用户确认收货）
• 16: 已收货（订单完成，进入结算流程）
• 17:已结算（佣金已结算，可提现）
• 18: 已失效（订单取消、退款等导致佣金无效）

踩坑提示3：时间范围与翻页。订单查询API的时间范围time参数是必填的，且单次查询时间跨度不能超过24小时。这是官方限制。如果需要查更长时间的数据，必须分成多个24小时段循环查询。另外，pageSize最大支持100，pageIndex从1开始。一定要根据totalCount来计算总页数，循环拉取所有数据。

5. 避坑大全与性能优化实战

对接过程中会遇到各种意想不到的问题，这里集中总结一下最常见的“坑”和优化方案。

5.1 高频报错代码解析与应对策略

5.2 提升稳定性的工程化实践

个人项目或小规模使用可能直接写脚本就行，但如果希望长期稳定运行，尤其是作为服务的一部分，就需要一些工程化考量。

1. 请求重试与退避机制网络请求可能失败，API也可能返回临时错误。一个健壮的客户端应该具备重试能力。

import time from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type class RobustJDUnionClient(JDUnionClient): @retry( stop=stop_after_attempt(3), # 最多重试3次 wait=wait_exponential(multiplier=1, min=2, max=10), # 指数退避等待 retry=retry_if_exception_type((requests.exceptions.ConnectionError, requests.exceptions.Timeout)) ) def execute_with_retry(self, method, param_dict): """带重试机制的execute方法""" return self.execute(method, param_dict)

这里使用了tenacity库来实现优雅的重试。对于网络错误（连接超时、断开）进行重试，并采用指数退避策略（等待2秒、4秒...），避免对服务器造成冲击。

2. 结果缓存策略对于不经常变化的数据，如商品类目、某些静态配置，或者短时间内重复查询的商品信息，可以使用缓存来减少API调用，提升响应速度并避免限流。

from functools import lru_cache import pickle import os class CachedJDUnionClient(JDUnionClient): def __init__(self, cache_dir='./jd_cache', ttl=3600): super().__init__() self.cache_dir = cache_dir self.ttl = ttl # 缓存存活时间，单位秒 os.makedirs(cache_dir, exist_ok=True) def _get_cache_key(self, method, param_dict): """根据方法和参数生成缓存文件名""" import hashlib key_str = f"{method}_{json.dumps(param_dict, sort_keys=True)}" return hashlib.md5(key_str.encode()).hexdigest() def execute_cached(self, method, param_dict, use_cache=True): """支持缓存的执行方法""" if not use_cache: return self.execute(method, param_dict) cache_key = self._get_cache_key(method, param_dict) cache_file = os.path.join(self.cache_dir, f"{cache_key}.pkl") # 检查缓存是否存在且未过期 if os.path.exists(cache_file): file_mtime = os.path.getmtime(cache_file) if time.time() - file_mtime < self.ttl: try: with open(cache_file, 'rb') as f: print(f"Cache hit for {method}") return pickle.load(f) except: pass # 缓存文件损坏，则重新请求 # 缓存不存在或已过期，请求API print(f"Cache miss for {method}, requesting API...") result = self.execute(method, param_dict) # 将结果写入缓存 try: with open(cache_file, 'wb') as f: pickle.dump(result, f) except: pass # 缓存写入失败不影响主流程 return result

3. 异步化改造应对批量任务当你需要查询成千上万个商品的详情或生成大量推广链接时，同步请求会非常慢。使用asyncio和aiohttp进行异步化改造可以成倍提升效率。

import aiohttp import asyncio class AsyncJDUnionClient(JDUnionClient): async def execute_async(self, session, method, param_dict): """异步执行单个请求""" # ... (异步版本的参数组装和签名逻辑，与同步版类似) public_params = self._prepare_params(method, param_dict) sign = generate_sign(public_params, self.app_secret) public_params['sign'] = sign async with session.get(self.base_url, params=public_params, timeout=aiohttp.ClientTimeout(total=10)) as resp: result = await resp.json() # ... (错误处理和数据提取逻辑) return data async def batch_query_goods(self, keyword_list, max_concurrency=5): """批量查询多个关键词的商品""" async with aiohttp.ClientSession() as session: semaphore = asyncio.Semaphore(max_concurrency) # 控制并发数，避免被封 tasks = [] for keyword in keyword_list: task = asyncio.create_task(self._bounded_execute(session, semaphore, keyword)) tasks.append(task) all_results = await asyncio.gather(*tasks, return_exceptions=True) # 处理结果和异常 return all_results async def _bounded_execute(self, session, semaphore, keyword): async with semaphore: params = {'goodsReq': {'keyword': keyword, 'pageSize': 20}} await asyncio.sleep(0.5) # 每个请求之间稍微停顿 return await self.execute_async(session, 'jd.union.open.goods.query', params)

性能优化核心：异步化的关键在于使用信号量 (Semaphore) 控制并发上限。京东联盟API对频率敏感，盲目开几百个并发很快就会被限流。建议将并发数控制在5-10个，并在每个请求间加入少量随机延迟 (asyncio.sleep(random.uniform(0.1, 0.5)))，模拟更自然的人类操作行为。