一、技術(shù)準(zhǔn)備

1. 注冊京東開放平臺賬號 訪問京東開放平臺完成注冊，需提供企業(yè)資質(zhì)（營業(yè)執(zhí)照等）。 創(chuàng)建應(yīng)用：登錄后進(jìn)入「開發(fā)者中心」→「應(yīng)用管理」→「創(chuàng)建應(yīng)用」，填寫應(yīng)用名稱、描述等信息，提交審核通過后獲取AppKey和AppSecret。
2. API權(quán)限申請 在應(yīng)用管理頁面，為應(yīng)用添加「商品查詢」相關(guān)權(quán)限（如jd.item.sku.info.get、jd.union.open.goods.query等）。 提交權(quán)限申請后，等待京東審核（通常1-3個工作日）。

二、核心API接口解析

1. SKU信息查詢接口（推薦jd.item.sku.info.get） 功能：通過商品ID或SKU ID列表獲取SKU詳情（價格、庫存、規(guī)格等）。 請求參數(shù)： csharp 體驗AI代碼助手 代碼解讀復(fù)制代碼python params = { "method": "jd.item.sku.info.get", "app_key": "YOUR_APP_KEY", "item_id": "商品ID", # 或通過skuIds批量查詢 "skuIds": "SKU_ID_1,SKU_ID_2", # 多個SKU用逗號分隔 "timestamp": int(time.time() * 1000), # 毫秒級時間戳 "v": "2.0", "sign_method": "md5" # 簽名算法 } 響應(yīng)字段： skuInfo：包含skuId、price（價格）、stockQuantity（庫存）、specifications（規(guī)格）等。
2. 商品列表批量查詢接口（推薦jd.union.open.goods.query） 功能：按條件篩選商品（類目、價格區(qū)間、銷量排序等），支持分頁。 請求參數(shù)： csharp 體驗AI代碼助手 代碼解讀復(fù)制代碼python params = { "method": "jd.union.open.goods.query", "app_key": "YOUR_APP_KEY", "pageIndex": 1, # 頁碼 "pageSize": 20, # 每頁數(shù)量 "sortType": 3, # 按銷量排序（3為銷量降序） "timestamp": int(time.time() * 1000), "v": "1.0", "sign_method": "md5" } 響應(yīng)字段： wareInfoVos：包含商品名稱、價格、銷量、主圖URL等。

三、認(rèn)證與簽名流程

python
import hashlib
import time
 
def generate_sign(params, app_secret):
    # 按字母順序排序參數(shù)
    sorted_params = sorted(params.items(), key=lambda x: x[0])
    # 拼接查詢字符串
    query_str = ''.join([f"{k}{v}" for k, v in sorted_params])
    # 拼接AppSecret并生成MD5簽名
    sign_str = query_str + app_secret
    return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
 
# 示例：生成簽名
params = {
    "method": "jd.item.sku.info.get",
    "app_key": "YOUR_APP_KEY",
    "item_id": "123456",
    "timestamp": int(time.time() * 1000),
    "v": "2.0"
}
params['sign'] = generate_sign(params, 'YOUR_APP_SECRET')

四、完整代碼示例（Python）

python
import requests
import json
import time
 
class JDSKUFetcher:
    def __init__(self, app_key, app_secret):
        self.app_key = app_key
        self.app_secret = app_secret
        self.base_url = "https://api.jd.com/routerjson"
 
    def fetch_sku_info(self, item_id, sku_ids=None):
        """獲取單個商品或多個SKU的詳情"""
        params = {
            "method": "jd.item.sku.info.get",
            "app_key": self.app_key,
            "item_id": item_id,
            "timestamp": int(time.time() * 1000),
            "v": "2.0",
            "sign_method": "md5"
        }
        if sku_ids:
            params["skuIds"] = ','.join(map(str, sku_ids))
        params['sign'] = self.generate_sign(params)
        response = requests.post(self.base_url, data=params)
        return self.parse_response(response)
 
    def generate_sign(self, params):
        """生成MD5簽名"""
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        query_str = ''.join([f"{k}{v}" for k, v in sorted_params])
        sign_str = query_str + self.app_secret
        return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
 
    def parse_response(self, response):
        """解析京東API響應(yīng)"""
        if response.status_code != 200:
            raise Exception(f"HTTP錯誤: {response.status_code}")
        data = response.json()
        if data.get('error_code') != '0':
            raise Exception(f"API錯誤: {data.get('error_msg')}")
        return data.get('jd_item_sku_info_get_response', {}).get('skuInfo', [])
 
# 使用示例
if __name__ == "__main__":
    fetcher = JDSKUFetcher("YOUR_APP_KEY", "YOUR_APP_SECRET")
    # 獲取單個商品的所有SKU
    sku_list = fetcher.fetch_sku_info(item_id="123456")
    print(json.dumps(sku_list, indent=2))
    
    # 獲取多個指定SKU的詳情
    # sku_details = fetcher.fetch_sku_info(sku_ids=[1001, 1002, 1003])

五、批量處理與數(shù)據(jù)存儲

1. 分布式爬取架構(gòu)（使用線程池） python 體驗AI代碼助手 代碼解讀復(fù)制代碼python from concurrent.futures import ThreadPoolExecutor def batch_fetch(sku_ids): with ThreadPoolExecutor(max_workers=5) as executor: futures = [executor.submit(fetcher.fetch_sku_info, sku_id) for sku_id in sku_ids] results = [f.result() for f in futures] return results
2. 數(shù)據(jù)存儲到SQLite python 體驗AI代碼助手 代碼解讀復(fù)制代碼python import sqlite3 class DataStorage: def __init__(self): self.conn = sqlite3.connect('jd_sku.db') self._create_table() def _create_table(self): cursor = self.conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS sku_data ( sku_id INTEGER PRIMARY KEY, item_id TEXT, price REAL, stock INTEGER, specs TEXT ) ''') self.conn.commit() def save_data(self, sku_info): cursor = self.conn.cursor() for sku in sku_info: cursor.execute(''' INSERT OR REPLACE INTO sku_data VALUES (?, ?, ?, ?, ?) ''', ( sku.get('skuId'), sku.get('itemId'), sku.get('price'), sku.get('stockQuantity'), json.dumps(sku.get('specifications')) )) self.conn.commit()

六、注意事項與合規(guī)要求

1. 調(diào)用頻率限制：京東API對未認(rèn)證應(yīng)用限制10次/秒，認(rèn)證后提升至50次/秒，需通過time.sleep()控制請求間隔。
2. 數(shù)據(jù)隱私保護：禁止抓取用戶手機號、地址等敏感信息，僅用于合法用途（如價格監(jiān)控、庫存分析）。
3. 錯誤重試機制： python 體驗AI代碼助手 代碼解讀復(fù)制代碼python def safe_request(func): def wrapper(*args, **kwargs): for _ in range(3): try: return func(*args, **kwargs) except (requests.ConnectionError, requests.Timeout): time.sleep(2) continue except Exception as e: logging.error(f"請求失敗: {str(e)}") break return None return wrapper
4. 遵守京東開放平臺規(guī)則：定期檢查API文檔更新，避免使用已廢棄接口。

通過本指南，您可以高效、合規(guī)地批量獲取京東商品SKU信息，適用于價格監(jiān)控、庫存管理、競品分析等場景。