爬坑 10 年！京东店铺全量商品接口实战开发：从分页优化、SKU 关联到数据完整性闭环

import time
import hashlib
import requests
import json
import redis
from typing import Dict, List, Optional
from urllib.parse import quote  # 京东需URL编码
class JdSellerWareAPI:
    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.session = self._init_session()
        # 缓存SKU数据（京东SKU查询频繁，缓存1小时）
        self.redis = redis.Redis(host='localhost', port=6379, db=1)
        self.sku_cache_expire = 3600
    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_jd_sign(self, params: Dict) -> str:
        """生成京东签名：关键坑点——参数值要URL编码，中文不编码必错"""
        # 1. 过滤空值，按ASCII升序排序
        valid_params = {k: v for k, v in params.items() if v is not None}
        sorted_params = sorted(valid_params.items(), key=lambda x: x[0])
        # 2. 拼接：京东要求key=value&key=value格式，最后加secret
        sign_str = "&".join([f"{k}={quote(str(v), safe='')}" for k, v in sorted_params])
        sign_str += f"&app_secret={self.app_secret}"
        # 3. MD5加密转大写
        return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
    def _get_sku_detail(self, sku_id: str) -> Optional[Dict]:
        """补全SKU规格：早年没做这个，拉的商品没尺寸/颜色数据，白折腾3天"""
        cache_key = f"jd_sku:{sku_id}"
        # 先查缓存，减少重复调用
        if cached_sku := self.redis.get(cache_key):
            return json.loads(cached_sku.decode())
        
        # 缓存未命中，调用京东SKU接口
        params = {
            "method": "jd.ware.sku.get",
            "app_key": self.app_key,
            "timestamp": str(int(time.time() * 1000)),  # 京东要毫秒级时间戳
            "format": "json",
            "v": "2.0",
            "sku_id": sku_id,
            "fields": "sku_id,spec,price,stock"
        }
        params["sign"] = self._generate_jd_sign(params)
        try:
            response = self.session.post(self.api_url, data=params, timeout=(5, 15))
            result = response.json()
            if "error_response" in result:
                print(f"SKU查询失败: {result['error_response']['msg']}")
                return None
            sku_data = result["ware_sku_get_response"]["sku_info"]
            # 缓存结果
            self.redis.setex(cache_key, self.sku_cache_expire, json.dumps(sku_data))
            return sku_data
        except Exception as e:
            print(f"SKU查询异常: {str(e)}")
            return None

from concurrent.futures import ThreadPoolExecutor, as_completed
def _fetch_page_items(self, seller_id: str, page_no: int, start_time: str = None) -> List[Dict]:
    """拉取单页商品：处理京东分页超时与空数据"""
    params = {
        "method": "jd.seller.ware.list.get",
        "app_key": self.app_key,
        "timestamp": str(int(time.time() * 1000)),
        "format": "json",
        "v": "2.0",
        "seller_id": seller_id,
        "page_no": page_no,
        "page_size": 20,  # 京东最大20，别改大
        "ware_status": "ON_SALE",  # 只拉在售商品，减少无效数据
        "fields": "ware_id,title,price,sales,stock,sku_ids,modified_time"
    }
    # 按更新时间分段，解决100页限制
    if start_time:
        params["start_time"] = start_time
    params["sign"] = self._generate_jd_sign(params)
    try:
        # 京东图像接口耗时久，超时设长点
        response = self.session.post(self.api_url, data=params, timeout=(8, 20))
        result = response.json()
        if "error_response" in result:
            err_msg = result["error_response"]["msg"]
            print(f"分页{page_no}错误: {err_msg}")
            # 1001是参数错，直接返回；5002是系统忙，重试
            return [] if "1001" in err_msg else None
        # 解析商品，补全SKU
        raw_items = result.get("seller_ware_list_get_response", {}).get("wares", {}).get("ware", [])
        if not raw_items:
            return []
        # 补全SKU数据
        for item in raw_items:
            sku_list = []
            for sku_id in item["sku_ids"].split(","):
                if sku_detail := self._get_sku_detail(sku_id):
                    sku_list.append(sku_detail)
            item["sku_list"] = sku_list
        return raw_items
    except Exception as e:
        print(f"分页{page_no}异常: {str(e)}")
        return None
def get_all_ware_items(self, seller_id: str, start_time: str = None) -> List[Dict]:
    """全量拉取：按页码+时间分段，突破100页限制"""
    all_items = []
    page_no = 1
    max_page = 100  # 京东超过100页返回空，到100页就换时间分段
    while page_no <= max_page:
        # 拉取单页，重试3次（京东偶尔抽风）
        retry = 0
        page_items = None
        while retry < 3:
            page_items = self._fetch_page_items(seller_id, page_no, start_time)
            if page_items is not None:  # 不是系统忙，就继续
                break
            time.sleep(2)  # 系统忙，等2秒再试
            retry += 1
        if not page_items:  # 没数据了，退出
            break
        all_items.extend(page_items)
        # 到100页，更新start_time为最后一个商品的修改时间，重新拉
        if page_no == max_page:
            last_modified = all_items[-1]["modified_time"]
            print(f"突破100页限制，从{last_modified}开始继续拉取")
            page_no = 1
            start_time = last_modified
        else:
            page_no += 1
        time.sleep(0.5)  # 控制频率，避免限流
    return all_items

def verify_ware_completeness(self, seller_id: str, fetched_items: List[Dict]) -> Dict:
    """双重校验：用京东计数接口+SKU关联校验"""
    # 1. 调用京东计数接口拿官方总数
    official_count = 0
    try:
        params = {
            "method": "jd.seller.ware.count.get",
            "app_key": self.app_key,
            "timestamp": str(int(time.time() * 1000)),
            "format": "json",
            "v": "2.0",
            "seller_id": seller_id,
            "ware_status": "ON_SALE"
        }
        params["sign"] = self._generate_jd_sign(params)
        response = self.session.post(self.api_url, data=params, timeout=(5, 10))
        result = response.json()
        if "error_response" not in result:
            official_count = result["seller_ware_count_get_response"]["total_count"]
    except Exception as e:
        print(f"计数接口异常: {str(e)}")
    # 2. 校验SKU关联完整性（无SKU的商品占比不能超过5%）
    no_sku_count = sum(1 for item in fetched_items if not item.get("sku_list"))
    sku_complete_rate = 1 - (no_sku_count / len(fetched_items)) if fetched_items else 0
    # 3. 结果返回：允许5个误差，SKU完整率≥95%算合格
    fetched_count = len(fetched_items)
    return {
        "fetched_count": fetched_count,
        "official_count": official_count,
        "sku_complete_rate": round(sku_complete_rate * 100, 1),
        "is_complete": abs(fetched_count - official_count) <= 5 and sku_complete_rate >= 0.95
    }

if __name__ == "__main__":
    # 初始化客户端（替换成自己的key和secret）
    jd_api = JdSellerWareAPI("your_app_key", "your_app_secret")
    
    # 1. 全量拉取店铺商品（seller_id替换成目标店铺ID）
    print("===== 全量拉取商品 =====")
    all_wares = jd_api.get_all_ware_items(seller_id="12345678")
    print(f"拉取商品总数: {len(all_wares)}")
    
    # 2. 完整性校验
    print("\n===== 数据完整性校验 =====")
    verify_res = jd_api.verify_ware_completeness(seller_id="12345678", fetched_items=all_wares)
    print(f"官方总数: {verify_res['official_count']} | 拉取数: {verify_res['fetched_count']}")
    print(f"SKU完整率: {verify_res['sku_complete_rate']}% | 是否完整: {'是' if verify_res['is_complete'] else '否'}")
    
    # 3. 打印示例商品（带SKU）
    if all_wares:
        print("\n===== 示例商品数据 =====")
        sample = all_wares[0]
        print(f"商品ID: {sample['ware_id']} | 标题: {sample['title']}")
        print(f"售价: {sample['price']}元 | 销量: {sample['sales']}件")
        print(f"SKU数量: {len(sample['sku_list'])} | 第一个SKU规格: {sample['sku_list'][0]['spec']}")