实时数据抓取核心技术解析：京东商品详情 API 接入指南

import hashlib
import time
from urllib.parse import urlencode


class SignGenerator:
    @staticmethod
    def generate(params: dict, app_secret: str) -> str:
        """
        生成京东API签名
        :param params: 接口请求参数（不含sign）
        :param app_secret: 应用密钥
        :return: 签名字符串
        """
        # 1. 按参数名ASCII升序排序
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        # 2. 拼接为key=value&key=value格式
        param_str = urlencode(sorted_params)
        # 3. 追加app_secret并MD5加密
        sign_str = f"{param_str}{app_secret}"
        sign = hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
        return sign

    @staticmethod
    def get_timestamp() -> str:
        """生成符合格式的时间戳"""
        return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())

import requests
import json
from typing import Dict, Optional
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


class JDGoodsAPI:
    def __init__(self, app_key: str, app_secret: str):
        self.app_key = app_key
        self.app_secret = app_secret
        self.api_url = "https://api.jd.com/routerjson"
        self.sign_generator = SignGenerator()
        # 初始化带重试的会话
        self.session = self._init_session()

    def _init_session(self) -> requests.Session:
        """初始化HTTP会话，配置重试策略"""
        session = requests.Session()
        retry_strategy = Retry(
            total=3,  # 总重试次数
            backoff_factor=1,  # 重试间隔（1, 2, 4秒...）
            status_forcelist=[429, 500, 502, 503, 504]  # 需要重试的状态码
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        session.mount("https://", adapter)
        return session

    def fetch_goods_detail(self, ware_id: str) -> Optional[Dict]:
        """
        调用京东API获取商品详情
        :param ware_id: 商品ID
        :return: 解析后的商品数据字典，失败返回None
        """
        # 1. 组装基础参数
        base_params = {
            "method": "jingdong.ware.detail.get",
            "app_key": self.app_key,
            "timestamp": self.sign_generator.get_timestamp(),
            "v": "2.0",
            "format": "json",
            "ware_id": ware_id
        }

        # 2. 生成签名并添加到参数
        base_params["sign"] = self.sign_generator.generate(base_params, self.app_secret)

        # 3. 发送POST请求
        try:
            response = self.session.post(
                url=self.api_url,
                data=base_params,
                timeout=8  # 超时时间8秒
            )
            response.raise_for_status()  # 抛出HTTP错误
        except requests.exceptions.RequestException as e:
            print(f"API请求失败：{str(e)}")
            return None

        # 4. 解析响应数据
        try:
            result = json.loads(response.text)
            # 提取核心数据（京东API返回格式固定嵌套）
            response_key = "jingdong_ware_detail_get_response"
            if response_key in result:
                return result[response_key].get("result", {})
            else:
                print(f"API返回格式错误：{result}")
                return None
        except json.JSONDecodeError as e:
            print(f"JSON解析失败：{str(e)}")
            return None

class GoodsDataParser:
    @staticmethod
    def parse(raw_data: Dict) -> Dict:
        """
        解析原始商品数据，提取核心字段
        :param raw_data: API返回的原始数据
        :return: 结构化的商品信息
        """
        if not raw_data:
            return {}

        # 基础信息解析
        base_info = raw_data.get("base", {})
        # 价格信息解析（多价格场景取京东价）
        price_info = raw_data.get("price", {})
        # 库存信息解析
        stock_info = raw_data.get("stock", {})
        # SKU信息解析
        sku_list = raw_data.get("skuList", [])

        return {
            "商品ID": base_info.get("wareId", ""),
            "商品名称": base_info.get("wareName", ""),
            "主图URL": base_info.get("mainImgUrl", ""),
            "品牌名称": base_info.get("brandName", ""),
            "京东价": price_info.get("jdPrice", {}).get("price", 0.0),
            "市场价": price_info.get("marketPrice", {}).get("price", 0.0),
            "库存数量": stock_info.get("stockNum", 0),
            "是否现货": stock_info.get("isStock", False),
            "SKU数量": len(sku_list),
            "SKU列表": [{"skuId": sku.get("skuId"), "规格": sku.get("skuName")} for sku in sku_list],
            "更新时间": SignGenerator.get_timestamp()
        }

def main():
    # 配置参数（替换为实际值）
    APP_KEY = "your_app_key"
    APP_SECRET = "your_app_secret"
    TEST_WARE_ID = "100012345678"  # 测试商品ID

    # 1. 初始化API客户端
    jd_api = JDGoodsAPI(APP_KEY, APP_SECRET)

    # 2. 调用API获取原始数据
    raw_goods_data = jd_api.fetch_goods_detail(TEST_WARE_ID)
    if not raw_goods_data:
        print("获取商品数据失败")
        return

    # 3. 解析数据
    parsed_data = GoodsDataParser.parse(raw_goods_data)
    print("解析后的商品详情：")
    import pprint
    pprint.pprint(parsed_data)

    # 4. （可选）存储数据到数据库/缓存
    # 示例：保存到JSON文件
    import json
    with open(f"jd_goods_{TEST_WARE_ID}.json", "w", encoding="utf-8") as f:
        json.dump(parsed_data, f, ensure_ascii=False, indent=2)
    print(f"数据已保存到本地文件")


if __name__ == "__main__":
    main()

import redis
import json

class GoodsCache:
    def __init__(self, redis_config: Dict):
        self.client = redis.Redis(**redis_config)
    
    def set_cache(self, ware_id: str, data: Dict, expire=900):
        """缓存商品数据，默认15分钟过期"""
        self.client.set(f"jd_goods:{ware_id}", json.dumps(data), ex=expire)
    
    def get_cache(self, ware_id: str) -> Optional[Dict]:
        """获取缓存数据"""
        data = self.client.get(f"jd_goods:{ware_id}")
        return json.loads(data) if data else None