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

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("无法获取有效的访问令牌")