實(shí)時(shí)獲取京東指定商品的當(dāng)前價(jià)格，示例實(shí)戰(zhàn)

管理

編輯

刪除

鍵盤(pán)上的螞蟻 2025-07-09 11:23:47

暫不處理

以下是實(shí)時(shí)獲取京東指定商品當(dāng)前價(jià)格的實(shí)戰(zhàn)示例，包含接口分析、代碼實(shí)現(xiàn)及反爬處理，適用于開(kāi)發(fā)者快速落地價(jià)格監(jiān)控功能：

一、京東商品價(jià)格接口分析

京東商品價(jià)格數(shù)據(jù)通過(guò)獨(dú)立 API 接口返回，核心特點(diǎn)：

接口地址：https://p.3.cn/prices/mgets（公開(kāi)可訪問(wèn)，登錄）
請(qǐng)求方式：GET
關(guān)鍵參數(shù)：skuIds=J_商品ID（商品 ID 需拼接前綴J_）
返回格式：JSON，包含原價(jià)、當(dāng)前價(jià)、促銷(xiāo)價(jià)等信息

二、實(shí)戰(zhàn)代碼實(shí)現(xiàn)（Python）

1. 基礎(chǔ)版：?jiǎn)紊唐穬r(jià)格獲取

import requests
import json

def get_jd_price(sku_id):
    """
    獲取京東指定商品當(dāng)前價(jià)格
    :param sku_id: 商品ID（如100012345678）
    :return: 價(jià)格字典（含原價(jià)、當(dāng)前價(jià)等）
    """
    # 構(gòu)造請(qǐng)求參數(shù)
    url = "https://p.3.cn/prices/mgets"
    params = {
        "skuIds": f"J_{sku_id}",  # 必須拼接J_前綴
        "type": 1,
        "area": "1_72_2799_0",    # 地區(qū)編碼（可固定，不影響價(jià)格獲取）
        "pduid": "1234567890",   # 隨機(jī)用戶ID，可固定
        "_": "1688888888888"     # 時(shí)間戳（可選，增加隨機(jī)性）
    }
    
    # 請(qǐng)求頭（模擬瀏覽器）
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36",
        "Referer": f"https://item.jd.com/{sku_id}.html",  #  Referer必須與商品頁(yè)一致
        "Accept": "application/json, text/plain, */*"
    }
    
    try:
        # 發(fā)送請(qǐng)求
        response = requests.get(url, params=params, headers=headers, timeout=10)
        response.encoding = "utf-8"
        
        # 解析響應(yīng)
        if response.status_code == 200:
            price_data = json.loads(response.text)
            if price_data and len(price_data) > 0:
                # 提取核心價(jià)格字段
                return {
                    "sku_id": sku_id,
                    "current_price": float(price_data[0]["p"]),  # 當(dāng)前售價(jià)
                    "original_price": float(price_data[0]["m"]), # 原價(jià)（市場(chǎng)價(jià)）
                    "price_time": response.headers.get("Date")   # 價(jià)格更新時(shí)間
                }
            else:
                return {"error": "未獲取到價(jià)格數(shù)據(jù)"}
        else:
            return {"error": f"請(qǐng)求失敗，狀態(tài)碼：{response.status_code}"}
    
    except Exception as e:
        return {"error": f"接口調(diào)用異常：{str(e)}"}

# 使用示例
if __name__ == "__main__":
    # 京東商品ID（可從商品詳情頁(yè)URL獲取，如https://item.jd.com/100032608854.html中的100032608854）
    sku_id = "100032608854"
    price_info = get_jd_price(sku_id)
    
    if "error" in price_info:
        print(f"獲取失?。簕price_info['error']}")
    else:
        print(f"商品ID：{price_info['sku_id']}")
        print(f"當(dāng)前價(jià)格：￥{price_info['current_price']}")
        print(f"原價(jià)：￥{price_info['original_price']}")
        print(f"價(jià)格時(shí)間：{price_info['price_time']}")

2. 進(jìn)階版：批量獲取 + 反爬優(yōu)化

import requests
import json
import time
import random
from fake_useragent import UserAgent

class JDPriceFetcher:
    def __init__(self):
        self.ua = UserAgent()
        self.proxies = self._get_proxies()  # 代理池（可選，高頻請(qǐng)求時(shí)使用）
        self.area_codes = [
            "1_72_2799_0",  # 北京地區(qū)
            "2_281_0_0",    # 上海地區(qū)
            "3_340_0_0"     # 廣州地區(qū)
        ]
    
    def _get_proxies(self):
        """獲取代理IP（示例：實(shí)際需對(duì)接代理服務(wù)）"""
        # 免費(fèi)代理示例（不穩(wěn)定，生產(chǎn)環(huán)境建議使用付費(fèi)代理）
        return {
            # "http": "http://123.45.67.89:8080",
            # "https": "https://123.45.67.89:8080"
        }
    
    def _get_headers(self, sku_id):
        """動(dòng)態(tài)生成請(qǐng)求頭"""
        return {
            "User-Agent": self.ua.random,  # 隨機(jī)User-Agent
            "Referer": f"https://item.jd.com/{sku_id}.html",
            "Accept": "application/json, text/plain, */*",
            "Connection": "keep-alive",
            "Cache-Control": "no-cache"
        }
    
    def fetch_price(self, sku_id):
        """獲取單個(gè)商品價(jià)格（帶反爬處理）"""
        url = "https://p.3.cn/prices/mgets"
        params = {
            "skuIds": f"J_{sku_id}",
            "type": 1,
            "area": random.choice(self.area_codes),  # 隨機(jī)地區(qū)編碼
            "pduid": str(random.randint(1000000000, 9999999999)),  # 隨機(jī)用戶ID
            "_": int(time.time() * 1000)  # 實(shí)時(shí)時(shí)間戳
        }
        
        try:
            # 隨機(jī)延遲（1-3秒），避免高頻請(qǐng)求
            time.sleep(random.uniform(1, 3))
            
            response = requests.get(
                url,
                params=params,
                headers=self._get_headers(sku_id),
                proxies=self.proxies,
                timeout=10,
                verify=False  # 忽略SSL證書(shū)驗(yàn)證（可選，部分代理需要）
            )
            
            if response.status_code == 200:
                price_data = json.loads(response.text)
                if price_data and len(price_data) > 0:
                    return {
                        "sku_id": sku_id,
                        "current_price": float(price_data[0]["p"]),
                        "original_price": float(price_data[0]["m"]),
                        "status": "success"
                    }
                else:
                    return {"sku_id": sku_id, "error": "價(jià)格數(shù)據(jù)為空", "status": "fail"}
            else:
                return {
                    "sku_id": sku_id,
                    "error": f"狀態(tài)碼：{response.status_code}",
                    "status": "fail"
                }
        
        except Exception as e:
            return {
                "sku_id": sku_id,
                "error": str(e),
                "status": "fail"
            }
    
    def batch_fetch(self, sku_ids):
        """批量獲取多個(gè)商品價(jià)格"""
        results = []
        for sku_id in sku_ids:
            print(f"正在獲取商品 {sku_id} 價(jià)格...")
            result = self.fetch_price(sku_id)
            results.append(result)
        return results

# 批量獲取示例
if __name__ == "__main__":
    fetcher = JDPriceFetcher()
    # 待查詢的商品ID列表
    sku_ids = [
        "100032608854",  # 示例商品1
        "100012345678",  # 示例商品2
        "100008348544"   # 示例商品3
    ]
    
    # 批量獲取價(jià)格
    prices = fetcher.batch_fetch(sku_ids)
    
    # 打印結(jié)果
    print("\n===== 價(jià)格查詢結(jié)果 =====")
    for item in prices:
        if item["status"] == "success":
            print(f"商品 {item['sku_id']}：當(dāng)前價(jià) ￥{item['current_price']}，原價(jià) ￥{item['original_price']}")
        else:
            print(f"商品 {item['sku_id']}：獲取失敗，原因：{item['error']}")

三、關(guān)鍵技術(shù)點(diǎn)解析

參數(shù)構(gòu)造：skuIds必須以J_為前綴（京東商品 ID 格式要求）area參數(shù)為地區(qū)編碼（如1_72_2799_0代表北京朝陽(yáng)區(qū)），不影響價(jià)格但增加請(qǐng)求真實(shí)性
反爬處理：隨機(jī) User-Agent：使用fake_useragent庫(kù)生成不同設(shè)備的瀏覽器標(biāo)識(shí)Referer 驗(yàn)證：必須與商品詳情頁(yè) URL 一致，否則可能返回 403請(qǐng)求延遲：每次請(qǐng)求間隔 1-3 秒，避免觸發(fā)頻率限制代理 IP：高頻請(qǐng)求時(shí)使用代理池，防止 IP 被封
數(shù)據(jù)解析：核心字段：p（當(dāng)前價(jià)）、m（原價(jià)），返回值為字符串需轉(zhuǎn)為浮點(diǎn)數(shù)異常處理：針對(duì)網(wǎng)絡(luò)超時(shí)、JSON 解析失敗等情況返回錯(cuò)誤信息

四、注意事項(xiàng)

合規(guī)性：非商業(yè)用途合理調(diào)用（建議≤100 次 / 小時(shí)）遵守京東robots.txt規(guī)則（https://www.jd.com/robots.txt）數(shù)據(jù)不得用于商業(yè)售賣(mài)或惡意競(jìng)爭(zhēng)
接口穩(wěn)定性：京東可能不定期調(diào)整接口參數(shù)，需定期測(cè)試代碼若返回 403/429，需增加延遲或更換 IP
通過(guò)以上示例，可快速實(shí)現(xiàn)京東商品價(jià)格的實(shí)時(shí)獲取，結(jié)合定時(shí)任務(wù)和數(shù)據(jù)存儲(chǔ)，可進(jìn)一步構(gòu)建價(jià)格監(jiān)控系統(tǒng)，為用戶提供降價(jià)提醒、趨勢(shì)分析等增值服務(wù)。

請(qǐng)登錄后查看

鍵盤(pán)上的螞蟻最后編輯于2025-07-09 11:23:47

快捷回復(fù)