Python实战：从零对接京东联盟API，实现商品查询与订单管理

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 必不可少的“粮草”申请应用与获取密钥理论懂了接下来是实操准备。你需要准备以下几样东西一个京东联盟账号这自然是前提如果没有就去注册一个。创建“网站/APP应用”登录京东联盟后台找到“推广管理” - “我的应用”。点击“创建应用”应用类型根据实际情况选择如“工具应用”。填写应用名称、描述等。最关键的一步在“API权限管理”中为你需要使用的API勾选上对应的权限。例如“商品查询”、“优惠券查询”、“订单查询”、“推广链接创建”等。不要漏选否则调用时会报“权限不足”的错误。拿到核心三要素应用创建成功后你会得到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_KEYyour_app_key_here JD_UNION_APP_SECRETyour_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(,, :)) # 去除空格减少传输量步骤二参数排序与拼接过滤掉sign参数本身如果有的话。将所有参数app_key,method,timestamp,param_json...按照参数名ASCII码从小到大排序字典序。将排序后的所有参数用key1value1key2value2...的格式拼接成一个字符串。在拼接字符串的首尾都加上你的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(), keylambda 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分钟。建议在服务器端获取当前时间而不是使用客户端可能不准确的时间。踩坑提示2param_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_keyNone, app_secretNone): 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, paramspublic_params, timeout10) response.raise_for_status() # 如果状态码不是200抛出HTTPError result response.json() except requests.exceptions.RequestException as e: # 网络请求异常处理 raise Exception(fNetwork request failed: {e}) except json.JSONDecodeError as e: # 响应不是合法JSON raise Exception(fInvalid 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(fAPI Error [{inner_result.get(code)}]: {error_msg}) # 返回真正的数据部分 return inner_result.get(data) # 如果没找到预期的响应结构 raise Exception(fUnexpected 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自营pPOP店。自营商品通常物流和服务更稳定。实操心得不要一次性拉取太多页。京东联盟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_index1, page_size100): 查询指定时间范围内的订单 :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最大支持100pageIndex从1开始。一定要根据totalCount来计算总页数循环拉取所有数据。5. 避坑大全与性能优化实战对接过程中会遇到各种意想不到的问题这里集中总结一下最常见的“坑”和优化方案。5.1 高频报错代码解析与应对策略错误码含义可能原因解决方案1001参数错误param_json格式不对、缺少必填参数、参数值类型错误。1. 检查param_json是否是合法JSON字符串。2. 对照官方文档确认所有必填参数已提供。3. 确认数字、字符串等类型是否正确。1002签名错误appSecret错误、签名算法实现有误、参数排序或拼接错误。1. 确认appSecret无误且未在代码中暴露。2.逐字核对签名生成函数特别是param_json作为整体字符串参与拼接。3. 使用官方提供的签名校验工具如果有或打印出待签名字符串进行比对。1003时间戳错误timestamp格式不对、与服务器时间差超过允许范围通常5分钟。1. 确保格式为YYYY-MM-DD HH:MM:SS。2. 确保服务器系统时间准确最好是NTP同步的时间。1004无权限应用未在后台勾选该API的权限。登录京东联盟后台在“我的应用”-“API权限管理”中补上对应权限。2001频率限制单位时间内调用次数超限。1. 降低调用频率增加请求间隔如sleep 0.5秒。2. 对于必须高频调用的任务考虑申请更高的频率限制部分API可能支持。3. 做好请求的缓存避免重复查询相同数据。5.2 提升稳定性的工程化实践个人项目或小规模使用可能直接写脚本就行但如果希望长期稳定运行尤其是作为服务的一部分就需要一些工程化考量。1. 请求重试与退避机制网络请求可能失败API也可能返回临时错误。一个健壮的客户端应该具备重试能力。import time from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type class RobustJDUnionClient(JDUnionClient): retry( stopstop_after_attempt(3), # 最多重试3次 waitwait_exponential(multiplier1, min2, max10), # 指数退避等待 retryretry_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, ttl3600): super().__init__() self.cache_dir cache_dir self.ttl ttl # 缓存存活时间单位秒 os.makedirs(cache_dir, exist_okTrue) def _get_cache_key(self, method, param_dict): 根据方法和参数生成缓存文件名 import hashlib key_str f{method}_{json.dumps(param_dict, sort_keysTrue)} return hashlib.md5(key_str.encode()).hexdigest() def execute_cached(self, method, param_dict, use_cacheTrue): 支持缓存的执行方法 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(fCache hit for {method}) return pickle.load(f) except: pass # 缓存文件损坏则重新请求 # 缓存不存在或已过期请求API print(fCache 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 result3. 异步化改造应对批量任务当你需要查询成千上万个商品的详情或生成大量推广链接时同步请求会非常慢。使用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, paramspublic_params, timeoutaiohttp.ClientTimeout(total10)) as resp: result await resp.json() # ... (错误处理和数据提取逻辑) return data async def batch_query_goods(self, keyword_list, max_concurrency5): 批量查询多个关键词的商品 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_exceptionsTrue) # 处理结果和异常 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)))模拟更自然的人类操作行为。