京东商品详情接口实战指南：从并发优化到数据完整性闭环

import time
import hashlib
import requests
import json
from typing import Dict, List, Optional
from concurrent.futures import ThreadPoolExecutor, as_completed
class JdProductAPI:
    def __init__(self, app_key: str, app_secret: str):
        self.app_key = app_key
        self.app_secret = app_secret
        self.api_url = "京东开放平台接口地址"  # 按官方文档配置
        self.session = self._init_session()
    def _init_session(self) -> requests.Session:
        """初始化会话池，减少连接开销"""
        session = requests.Session()
        adapter = requests.adapters.HTTPAdapter(
            pool_connections=15, pool_maxsize=80, max_retries=3
        )
        session.mount('https://', adapter)
        return session
    def _generate_sign(self, params: Dict) -> str:
        """生成京东签名（处理毫秒级时间戳的坑）"""
        # 坑点1：必须按参数名ASCII升序排序，普通dict会乱序
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        # 坑点2：拼接格式为"key=value&key=value"，首尾加app_secret
        sign_str = "&".join([f"{k}={v}" for k, v in sorted_params])
        sign_str = f"{self.app_secret}{sign_str}{self.app_secret}"
        # 坑点3：MD5加密后需转大写
        return hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
    def _fetch_page_items(self, start_time: str, end_time: str, page: int = 1, page_size: int = 30) -> List[Dict]:
        """单页商品拉取（基础方法）"""
        params = {
            "method": "jingdong.ware.get",
            "app_key": self.app_key,
            "timestamp": str(int(time.time() * 1000)),  # 京东需毫秒级时间戳
            "format": "json",
            "v": "2.0",
            "sign_method": "md5",
            "page": str(page),
            "pageSize": str(page_size),
            "startModified": start_time,
            "endModified": end_time,
            "fields": "wareId,title,price,stock,specList,modifiedTime,wareStatus"
        }
        params["sign"] = self._generate_sign(params)
        try:
            response = self.session.get(self.api_url, params=params, timeout=(5, 18))
            result = response.json()
            if result.get("code") != 0:
                print(f"单页拉取失败: {result.get('message', '未知错误')}")
                return []
            return result.get("data", {}).get("wareList", [])
        except Exception as e:
            print(f"单页拉取异常: {str(e)}")
            return []

def parse_spec_structure(self, raw_specs: List[Dict]) -> List[Dict]:
        """
        递归解析京东多规格结构
        :param raw_specs: 接口返回的原始specList
        :return: 扁平化的规格列表
        """
        parsed_specs = []
        for spec in raw_specs:
            # 基础规格信息
            base_spec = {
                "specId": spec.get("specId", ""),
                "specName": spec.get("specName", ""),
                "specValue": spec.get("specValue", ""),
                "children": []  # 子规格容器
            }
            # 递归处理子规格（若有）
            if child_specs := spec.get("childSpecList"):
                base_spec["children"] = self.parse_spec_structure(child_specs)
            parsed_specs.append(base_spec)
        return parsed_specs
    def get_ware_with_parsed_spec(self, ware_id: str) -> Optional[Dict]:
        """获取商品详情+解析后规格"""
        params = {
            "method": "jingdong.ware.get",
            "app_key": self.app_key,
            "timestamp": str(int(time.time() * 1000)),
            "format": "json",
            "v": "2.0",
            "sign_method": "md5",
            "wareId": ware_id,
            "fields": "wareId,title,price,stock,specList,modifiedTime,wareStatus"
        }
        params["sign"] = self._generate_sign(params)
        try:
            response = self.session.get(self.api_url, params=params, timeout=(5, 15))
            result = response.json()
            if result.get("code") != 0:
                print(f"商品详情拉取失败: {result.get('message')}")
                return None
            ware_data = result.get("data", {}).get("ware", {})
            # 解析规格并替换原始字段
            if raw_specs := ware_data.get("specList"):
                ware_data["parsedSpecList"] = self.parse_spec_structure(raw_specs)
                del ware_data["specList"]  # 删除原始嵌套字段
            # 处理价格区间（拆分为起始价/最高价）
            if price_range := ware_data.get("price"):
                price_split = price_range.split("-")
                ware_data["startPrice"] = float(price_split[0])
                ware_data["endPrice"] = float(price_split[1]) if len(price_split) > 1 else ware_data["startPrice"]
                del ware_data["price"]
            return ware_data
        except Exception as e:
            print(f"商品详情异常: {str(e)}")
            return None

def verify_ware_completeness(self, fetched_wares: List[Dict], start_time: str, end_time: str) -> Dict:
        """
        双重校验商品完整性：官方计数+字段核对
        :param fetched_wares: 已拉取商品列表
        :param start_time/end_time: 时间范围
        :return: 校验结果
        """
        # 1. 获取官方总计数
        official_count = 0
        try:
            params = {
                "method": "jingdong.ware.count.get",
                "app_key": self.app_key,
                "timestamp": str(int(time.time() * 1000)),
                "format": "json",
                "v": "2.0",
                "sign_method": "md5",
                "startModified": start_time,
                "endModified": end_time
            }
            params["sign"] = self._generate_sign(params)
            response = self.session.get(self.api_url, params=params, timeout=(3, 10))
            result = response.json()
            if result.get("code") == 0:
                official_count = result.get("data", {}).get("totalCount", 0)
        except Exception as e:
            print(f"官方计数获取异常: {str(e)}")
        # 2. 字段完整性核对（必选字段：wareId、title、startPrice、wareStatus）
        required_fields = ["wareId", "title", "startPrice", "wareStatus"]
        incomplete_wares = []
        for ware in fetched_wares:
            missing_fields = [f for f in required_fields if f not in ware or not ware[f]]
            if missing_fields:
                incomplete_wares.append({
                    "wareId": ware.get("wareId", "未知ID"),
                    "missingFields": missing_fields
                })
        # 3. 生成校验结果
        fetched_count = len(fetched_wares)
        return {
            "officialCount": official_count,
            "fetchedCount": fetched_count,
            "completenessRate": round(fetched_count / official_count * 100, 1) if official_count != 0 else 0,
            "incompleteWares": incomplete_wares,
            "isAcceptable": abs(fetched_count - official_count) <= 3 and len(incomplete_wares) <= 2
        }

# tasks.py（Celery分布式任务）
from celery import Celery
import json
from jd_api import JdProductAPI  # 导入上文的JdProductAPI类
app = Celery('jd_ware_tasks', broker='redis://localhost:6379/0')
@app.task(bind=True, max_retries=3)
def fetch_time_segment(self, start_time: str, end_time: str, config: dict) -> int:
    """按时间分段拉取商品的分布式任务"""
    api = JdProductAPI(config["app_key"], config["app_secret"])
    try:
        # 分页拉取当前时间段商品
        page = 1
        total_fetched = 0
        while True:
            wares = api._fetch_page_items(start_time, end_time, page=page, page_size=30)
            if not wares:
                break
            # 存储结果（按时间分段存文件）
            with open(f"jd_wares_{start_time.replace(' ', '_')}_{end_time.replace(' ', '_')}_page{page}.json", "w") as f:
                json.dump(wares, f, ensure_ascii=False)
            total_fetched += len(wares)
            page += 1
            time.sleep(0.4)  # 控制频率
        return total_fetched
    except Exception as e:
        # 失败6秒后重试，最多3次
        self.retry(exc=e, countdown=6)

if __name__ == "__main__":
    # 初始化客户端（替换为合规申请的key/secret）
    jd_api = JdProductAPI(app_key="your_jd_app_key", app_secret="your_jd_app_secret")
    # 1. 按时间分段拉取（示例：2024年1月1日-2024年1月7日）
    print("===== 按时间分段拉取 =====")
    time_segments = [
        ("2024-01-01 00:00:00", "2024-01-02 23:59:59"),
        ("2024-01-03 00:00:00", "2024-01-04 23:59:59"),
        ("2024-01-05 00:00:00", "2024-01-07 23:59:59")
    ]
    total_wares = []
    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = [executor.submit(jd_api._fetch_page_items, s, e) for s, e in time_segments]
        for future in as_completed(futures):
            total_wares.extend(future.result())
    print(f"总拉取商品数: {len(total_wares)}")
    # 2. 完整性校验
    print("\n===== 完整性校验 =====")
    verify_result = jd_api.verify_ware_completeness(
        total_wares, 
        start_time="2024-01-01 00:00:00", 
        end_time="2024-01-07 23:59:59"
    )
    print(f"校验结果: 官方计数{verify_result['officialCount']} | 拉取计数{verify_result['fetchedCount']} | 完整率{verify_result['completenessRate']}%")
    if verify_result["incompleteWares"]:
        print(f"不完整商品数: {len(verify_result['incompleteWares'])}（示例：{verify_result['incompleteWares'][0]}）")
    # 3. 单商品详情+规格解析
    print("\n===== 单商品详情+规格解析 =====")
    sample_ware = jd_api.get_ware_with_parsed_spec(ware_id="100012014970")  # 示例商品ID
    if sample_ware:
        print(f"商品ID: {sample_ware['wareId']}")
        print(f"商品标题: {sample_ware['title']}")
        print(f"价格区间: {sample_ware['startPrice']}-{sample_ware['endPrice']}元")
        print(f"解析后规格（前2层）: {json.dumps(sample_ware['parsedSpecList'][:2], ensure_ascii=False, indent=2)}")