京东商品 API 集成指南：从认证授权到获取实时价格与库存

import requests

class JDClient:
    def __init__(self, app_key, app_secret):
        self.app_key = app_key
        self.app_secret = app_secret
        self.token_url = "https://oauth.jd.com/oauth/token"
        self.access_token = None
        self.expires_in = 0  # 令牌有效期（秒）

    def get_access_token(self):
        """获取访问令牌"""
        params = {
            "grant_type": "client_credentials",
            "client_id": self.app_key,
            "client_secret": self.app_secret
        }
        
        try:
            response = requests.post(self.token_url, params=params)
            result = response.json()
            
            if "access_token" in result:
                self.access_token = result["access_token"]
                self.expires_in = result["expires_in"]
                print(f"令牌获取成功，有效期：{self.expires_in}秒")
                return self.access_token
            else:
                print(f"令牌获取失败：{result.get('error_description')}")
                return None
                
        except Exception as e:
            print(f"请求异常：{str(e)}")
            return None

import time
import hashlib
import json

class JDProductAPI(JDClient):
    def __init__(self, app_key, app_secret):
        super().__init__(app_key, app_secret)
        self.api_url = "https://api.jd.com/routerjson"

    def sign_params(self, params):
        """生成签名"""
        # 1. 排序参数
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        # 2. 拼接参数
        sign_str = self.app_secret
        for k, v in sorted_params:
            sign_str += f"{k}{v}"
        sign_str += self.app_secret
        # 3. MD5加密
        return hashlib.md5(sign_str.encode()).hexdigest().upper()

    def get_product_price(self, sku_ids):
        """获取商品实时价格"""
        # 检查令牌是否有效（简化版，实际需判断过期时间）
        if not self.access_token:
            if not self.get_access_token():
                return None

        # 构建请求参数
        params = {
            "method": "jingdong.price.get",
            "app_key": self.app_key,
            "access_token": self.access_token,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "format": "json",
            "v": "1.0",
            "skuIds": sku_ids
        }

        # 生成签名
        params["sign"] = self.sign_params(params)

        try:
            response = requests.post(self.api_url, data=params)
            result = response.json()
            
            # 解析价格数据
            if "price_get_response" in result:
                return result["price_get_response"]["result"]
            else:
                print(f"价格查询失败：{result.get('error_response', {}).get('zh_desc')}")
                return None
                
        except Exception as e:
            print(f"请求异常：{str(e)}")
            return None

    def get_product_stock(self, sku_ids, area="1_2800_2855_0"):
        """获取商品库存信息"""
        if not self.access_token:
            if not self.get_access_token():
                return None

        params = {
            "method": "jingdong.stock.get",
            "app_key": self.app_key,
            "access_token": self.access_token,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "format": "json",
            "v": "1.0",
            "skuIds": sku_ids,
            "area": area
        }

        params["sign"] = self.sign_params(params)

        try:
            response = requests.post(self.api_url, data=params)
            result = response.json()
            
            if "stock_get_response" in result:
                return result["stock_get_response"]["result"]
            else:
                print(f"库存查询失败：{result.get('error_response', {}).get('zh_desc')}")
                return None
                
        except Exception as e:
            print(f"请求异常：{str(e)}")
            return None

if __name__ == "__main__":
    # 替换为你的appKey和appSecret
    APP_KEY = "你的appKey"
    APP_SECRET = "你的appSecret"
    
    # 初始化客户端
    jd_api = JDProductAPI(APP_KEY, APP_SECRET)
    
    # 测试商品ID（可替换为实际商品ID）
    test_sku_ids = "100012345678,100008348544"
    
    # 获取价格
    price_data = jd_api.get_product_price(test_sku_ids)
    if price_data:
        print("价格信息：")
        for item in json.loads(price_data):
            print(f"SKU: {item['skuId']}, 价格: {item['price']}元")
    
    # 获取库存
    stock_data = jd_api.get_product_stock(test_sku_ids)
    if stock_data:
        print("\n库存信息：")
        for item in json.loads(stock_data):
            stock_status = "有货" if item["stockState"] == 3 else "无货"
            print(f"SKU: {item['skuId']}, 状态: {stock_status}")

import time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_retry_session():
    """创建带重试机制的会话"""
    session = requests.Session()
    retry = Retry(
        total=3,
        backoff_factor=1,  # 重试间隔：1s, 2s, 4s...
        status_forcelist=[429, 500, 502, 503, 504]
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount("https://", adapter)
    return session