京东工业平台商品详情接口开发指南：工业级数据解析与实战实现

在工业供应链数字化转型过程中，京东工业平台作为专注于工业用品采购的 B2B 电商平台，其商品详情数据包含丰富的技术参数、合规认证和供应链信息，对企业采购系统、供应链管理工具的开发具有重要价值。本文将详细介绍京东工业平台商品详情接口的调用流程，包括 OAuth2.0 认证机制、签名生成、工业特性数据解析及完整代码实现，帮助开发者快速构建稳定高效的工业商品数据对接功能。

一、京东工业商品详情接口基础信息

京东工业开放平台提供的jd.industry.product.detail.get接口是获取商品完整信息的核心接口，专为工业场景设计，相比消费类商品接口增加了大量专业维度的数据。

接口特点：

采用 OAuth2.0 + 签名双重认证机制，安全性更高

支持获取工业商品特有的技术参数、规格型号、认证资质等信息

包含详细的包装、物流、售后服务等供应链关键数据

提供多维度价格体系（零售、批发、企业集采等）

接口端点：https://api.jd.com/routerjson

二、认证机制与核心参数解析

1. 完整认证流程

京东工业 API 采用业界标准的 OAuth2.0 认证框架，结合签名机制确保接口安全：

获取 Access Token：通过 client_id 和 client_secret 获取访问令牌（有效期 24 小时）

生成签名：对请求参数按规则进行签名计算

接口调用：在请求中携带 Access Token 和签名信息

2. 签名生成规则

收集所有请求参数（包括公共参数和业务参数）

按参数名 ASCII 码升序排序

拼接为 "key=value" 形式的字符串，用 & 连接

拼接上 Access Token 和 app_secret，形成待签名字符串

使用 MD5 算法对字符串进行加密，得到 32 位大写签名

3. 核心参数说明

公共参数：

method：接口名称，固定为jd.industry.product.detail.get

app_key：应用唯一标识

access_token：访问令牌

timestamp：时间戳（毫秒级）

format：响应格式，固定为 json

v：API 版本，固定为 1.0

sign：签名值

业务参数：

productId：商品 ID（必填，可从搜索接口获取）

needTechParam：是否需要技术参数（true/false，默认 false）

needCert：是否需要认证信息（true/false，默认 false）

needInventory：是否需要库存信息（true/false，默认 false）

三、完整代码实现

以下是 Python 实现的京东工业商品详情接口调用功能，包含完整的认证流程、签名生成和工业数据解析：

import requests
import time
import hashlib
import json
from typing import Dict, List, Optional, Any

class JDIndustryProductAPI:
    def __init__(self, app_key: str, app_secret: str, redirect_uri: str):
        """
        初始化京东工业商品API客户端
        
        :param app_key: 应用的App Key
        :param app_secret: 应用的App Secret
        :param redirect_uri: 授权回调地址
        """
        self.app_key = app_key
        self.app_secret = app_secret
        self.redirect_uri = redirect_uri
        self.base_url = "https://api.jd.com/routerjson"
        self.token_url = "https://oauth.jd.com/token"
        self.access_token = None
        self.token_expiry = 0  # 令牌过期时间戳(秒)
    
    def get_auth_url(self) -> str:
        """生成授权URL，用于获取code（首次授权时使用）"""
        params = {
            "response_type": "code",
            "client_id": self.app_key,
            "redirect_uri": self.redirect_uri,
            "state": "JD_INDUSTRY_STATE"
        }
        return f"https://oauth.jd.com/authorize?{requests.compat.urlencode(params)}"
    
    def get_access_token(self, code: Optional[str] = None, refresh_token: Optional[str] = None) -> bool:
        """
        获取或刷新访问令牌
        
        :param code: 授权码（首次获取时需要）
        :param refresh_token: 刷新令牌（令牌过期时使用）
        :return: 是否成功
        """
        params = {
            "client_id": self.app_key,
            "client_secret": self.app_secret,
            "grant_type": "authorization_code" if code else "refresh_token"
        }
        
        if code:
            params["code"] = code
            params["redirect_uri"] = self.redirect_uri
        elif refresh_token:
            params["refresh_token"] = refresh_token
        else:
            return False
        
        try:
            response = requests.post(self.token_url, data=params, timeout=10)
            result = response.json()
            
            if "access_token" in result:
                self.access_token = result["access_token"]
                expires_in = result.get("expires_in", 86400)
                self.token_expiry = time.time() + expires_in
                self.refresh_token = result.get("refresh_token")
                return True
            else:
                print(f"获取令牌失败: {result.get('error_description')}")
                return False
                
        except Exception as e:
            print(f"令牌请求异常: {str(e)}")
            return False
    
    def _generate_sign(self, params: Dict[str, str]) -> str:
        """生成API请求签名"""
        # 按参数名ASCII升序排序
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        
        # 拼接参数
        query_string = "&".join([f"{k}={v}" for k, v in sorted_params])
        
        # 拼接access_token和app_secret
        sign_str = f"{query_string}&access_token={self.access_token}{self.app_secret}"
        
        # 计算MD5签名并转为大写
        return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
    
    def get_product_detail(self,
                         product_id: str,
                         need_tech_param: bool = True,
                         need_cert: bool = True,
                         need_inventory: bool = True) -> Dict[str, Any]:
        """
        获取京东工业商品详情
        
        :param product_id: 商品ID
        :param need_tech_param: 是否需要技术参数
        :param need_cert: 是否需要认证信息
        :param need_inventory: 是否需要库存信息
        :return: 商品详情数据
        """
        # 检查令牌有效性
        if not self.access_token or time.time() >= self.token_expiry:
            if not self.refresh_token or not self.get_access_token(refresh_token=self.refresh_token):
                return {"success": False, "error_msg": "访问令牌无效或已过期"}
        
        # 业务参数
        biz_params = {
            "productId": product_id,
            "needTechParam": str(need_tech_param).lower(),
            "needCert": str(need_cert).lower(),
            "needInventory": str(need_inventory).lower()
        }
        
        # 公共参数
        params = {
            "method": "jd.industry.product.detail.get",
            "app_key": self.app_key,
            "timestamp": str(int(time.time() * 1000)),  # 毫秒级时间戳
            "format": "json",
            "v": "1.0",
            "param_json": json.dumps(biz_params, ensure_ascii=False)
        }
        
        # 生成签名
        params["sign"] = self._generate_sign(params)
        
        try:
            # 发送请求
            response = requests.post(
                self.base_url,
                data=params,
                timeout=15
            )
            response.raise_for_status()
            
            # 解析响应
            result = response.json()
            
            # 处理API错误
            if "error_response" in result:
                error = result["error_response"]
                return {
                    "success": False,
                    "error_code": error.get("code"),
                    "error_msg": error.get("msg")
                }
            
            # 处理正常响应
            detail_data = result.get("jd_industry_product_detail_get_response", {}).get("result", {})
            return self._parse_product_detail(detail_data)
            
        except requests.exceptions.RequestException as e:
            return {
                "success": False,
                "error_msg": f"请求异常: {str(e)}"
            }
        except Exception as e:
            return {
                "success": False,
                "error_msg": f"处理响应失败: {str(e)}"
            }
    
    def _parse_product_detail(self, raw_data: Dict[str, Any]) -> Dict[str, Any]:
        """解析原始商品详情数据为结构化格式"""
        if not raw_data:
            return {"success": False, "error_msg": "无商品详情数据"}
        
        # 基础信息解析
        base_info = {
            "product_id": raw_data.get("productId"),
            "sku_id": raw_data.get("skuId"),
            "name": raw_data.get("name"),
            "brand": {
                "id": raw_data.get("brandId"),
                "name": raw_data.get("brandName")
            },
            "category": {
                "id": raw_data.get("categoryId"),
                "name": raw_data.get("categoryName"),
                "parent_id": raw_data.get("parentCategoryId"),
                "parent_name": raw_data.get("parentCategoryName")
            },
            "model": raw_data.get("model"),  # 型号
            "spec": raw_data.get("spec"),    # 规格
            "production_date": raw_data.get("productionDate"),  # 生产日期
            "warranty_period": raw_data.get("warrantyPeriod"),  # 保修期
            "url": raw_data.get("productUrl")
        }
        
        # 价格信息解析
        price_info = {
            "retail_price": raw_data.get("retailPrice"),  # 零售价
            "wholesale_price": self._parse_wholesale_price(raw_data.get("wholesalePriceList", [])),  # 批发价
            "enterprise_price": raw_data.get("enterprisePrice"),  # 企业集采价
            "currency": raw_data.get("currency", "CNY"),
            "tax_included": raw_data.get("taxIncluded", True)  # 是否含税
        }
        
        # 技术参数解析
        tech_params = []
        for group in raw_data.get("techParamGroups", []):
            group_params = []
            for param in group.get("params", []):
                group_params.append({
                    "name": param.get("name"),
                    "value": param.get("value"),
                    "unit": param.get("unit"),
                    "standard": param.get("standard")  # 执行标准
                })
            tech_params.append({
                "group_name": group.get("groupName"),
                "params": group_params
            })
        
        # 认证信息解析
        certifications = []
        for cert in raw_data.get("certifications", []):
            certifications.append({
                "type": cert.get("type"),  # 认证类型
                "number": cert.get("number"),  # 认证编号
                "issuer": cert.get("issuer"),  # 发证机构
                "issue_date": cert.get("issueDate"),  # 发证日期
                "expiry_date": cert.get("expiryDate"),  # 有效期至
                "image_url": cert.get("imageUrl")  # 认证证书图片
            })
        
        # 库存信息解析
        inventory = {
            "total_stock": raw_data.get("totalStock"),
            "warehouses": []
        }
        for wh in raw_data.get("warehouseInventories", []):
            inventory["warehouses"].append({
                "id": wh.get("warehouseId"),
                "name": wh.get("warehouseName"),
                "location": wh.get("location"),  # 仓库位置
                "stock": wh.get("stock"),
                "lock_stock": wh.get("lockStock")  # 锁定库存
            })
        
        # 包装与物流信息
        logistics = {
            "packaging": {
                "length": raw_data.get("packageLength"),
                "width": raw_data.get("packageWidth"),
                "height": raw_data.get("packageHeight"),
                "weight": raw_data.get("packageWeight"),
                "unit": raw_data.get("packageUnit")
            },
            "delivery": {
                "min_delivery_days": raw_data.get("minDeliveryDays"),
                "max_delivery_days": raw_data.get("maxDeliveryDays"),
                "support_batch": raw_data.get("supportBatch", False),  # 是否支持批量发货
                "special_logistics": raw_data.get("specialLogistics", False)  # 是否需要特殊物流
            }
        }
        
        # 售后服务信息
        after_sales = {
            "support_return": raw_data.get("supportReturn", False),
            "return_days": raw_data.get("returnDays"),
            "maintenance_service": raw_data.get("maintenanceService", []),  # 维修服务
            "technical_support": raw_data.get("technicalSupport", False)  # 是否提供技术支持
        }
        
        return {
            "success": True,
            "base_info": base_info,
            "price_info": price_info,
            "tech_params": tech_params,
            "certifications": certifications,
            "inventory": inventory,
            "logistics": logistics,
            "after_sales": after_sales,
            "images": {
                "main_images": raw_data.get("mainImages", []),
                "detail_images": raw_data.get("detailImages", []),
                "tech_images": raw_data.get("techImages", []),  # 技术图纸
                "cert_images": raw_data.get("certImages", [])   # 认证证书图片
            }
        }
    
    def _parse_wholesale_price(self, wholesale_data: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """解析批发价格体系"""
        wholesale_prices = []
        for item in wholesale_data:
            wholesale_prices.append({
                "min_quantity": item.get("minQuantity"),
                "max_quantity": item.get("maxQuantity"),
                "price": item.get("price"),
                "discount": item.get("discount")  # 折扣比例
            })
        return wholesale_prices

# 使用示例
if __name__ == "__main__":
    # 替换为你的应用信息
    APP_KEY = "your_app_key"
    APP_SECRET = "your_app_secret"
    REDIRECT_URI = "your_redirect_uri"
    
    # 初始化API客户端
    jd_industry_api = JDIndustryProductAPI(APP_KEY, APP_SECRET, REDIRECT_URI)
    
    # 首次使用需要获取授权码
    # print("请访问以下URL进行授权:")
    # print(jd_industry_api.get_auth_url())
    # code = input("请输入授权后获取的code: ")
    # jd_industry_api.get_access_token(code=code)
    
    # 已获取过token的情况，可直接使用refresh_token刷新
    # jd_industry_api.refresh_token = "your_refresh_token"
    # jd_industry_api.get_access_token(refresh_token=jd_industry_api.refresh_token)
    
    # 示例：获取商品详情（替换为实际商品ID）
    if jd_industry_api.access_token:
        product_detail = jd_industry_api.get_product_detail(
            product_id="100012345678",
            need_tech_param=True,
            need_cert=True,
            need_inventory=True
        )
        
        if product_detail["success"]:
            print(f"商品名称: {product_detail['base_info']['name']}")
            print(f"型号: {product_detail['base_info']['model']}")
            print(f"零售价: {product_detail['price_info']['retail_price']}{product_detail['price_info']['currency']}")
            print(f"技术参数组数: {len(product_detail['tech_params'])}")
            print(f"认证数量: {len(product_detail['certifications'])}")
            print(f"总库存: {product_detail['inventory']['total_stock']}")
        else:
            print(f"获取失败: {product_detail['error_msg']}（错误码: {product_detail.get('error_code')}）")
    else:
        print("无法获取有效的访问令牌")
    

四、代码核心功能解析

1. 认证机制实现

完整实现 OAuth2.0 认证流程，支持首次授权和令牌刷新

自动处理令牌过期问题，确保接口调用连续性

提供授权 URL 生成方法，简化首次接入流程

2. 工业数据解析增强

针对工业商品特性设计数据结构，重点解析技术参数、认证信息等专业字段

提取多维度价格体系（零售 / 批发 / 企业集采），适配 B2B 采购场景

解析库存分布、物流信息和售后服务，满足供应链管理需求

3. 签名与安全优化

严格按照京东工业 API 规范实现签名生成逻辑

采用毫秒级时间戳，避免时间同步问题导致的签名失效

完整保留参数处理逻辑，确保签名准确性

4. 错误处理机制

统一返回格式，包含成功标识、业务数据及错误信息

处理令牌无效、过期等常见认证问题

捕获 HTTP 请求异常，提供详细的故障排查依据

五、实战注意事项

1. 接口权限与申请

京东工业 API 需企业资质申请，个人开发者无法接入

不同等级的企业账号拥有不同的接口权限，高级权限需单独申请

部分敏感数据（如详细库存分布）需要特殊权限审批

2. 调用策略优化

技术参数和认证信息等数据更新频率低，建议本地缓存（6-24 小时）

库存数据实时性要求高，建议按需实时获取

批量获取商品详情时，需控制请求频率（建议 QPS≤5）

3. 工业场景适配

型号和规格是工业商品的核心标识，需重点处理和存储

认证信息需验证有效期，对过期认证商品进行风险提示

批发价格存在阶梯区间，需完整解析用于采购量决策

4. 安全与合规

妥善保管 app_secret 和 access_token，避免泄露

生产环境建议部署在服务端，禁止客户端直接调用

数据使用需遵守京东工业开放平台的开发者协议

六、功能扩展方向

商品对比工具：基于详情数据实现多维度参数对比，辅助采购决策

合规检查系统：自动校验商品认证是否满足行业标准和项目要求

库存预警功能：结合库存数据和历史采购量，实现关键物料预警

价格趋势分析：定期获取价格数据，分析价格波动规律

通过本文提供的方案，开发者可以快速实现京东工业平台商品详情数据的对接，为工业采购系统、供应链管理工具等应用提供精准的数据支持。实际开发中，建议结合具体工业领域（如制造业、能源业等）的特性，进一步优化数据解析和应用逻辑。​​​​

审核编辑 黄宇