逆向分析与 API 调用：另一种获取京东实时商品数据的技术方案探讨

https://p.3.cn/prices/mgets?skuIds=J_100008348542

import requests
import json
import time
import random
from user_agent import generate_user_agent  # 需要安装: pip install user_agent

class JDProductFetcher:
    def __init__(self):
        # 初始化请求头，模拟浏览器行为
        self.headers = {
            "Accept": "*/*",
            "Accept-Encoding": "gzip, deflate, br",
            "Accept-Language": "zh-CN,zh;q=0.9",
            "Connection": "keep-alive",
            "Host": "p.3.cn",
            "Referer": "https://item.jd.com/",
            "Sec-Fetch-Dest": "script",
            "Sec-Fetch-Mode": "no-cors",
            "Sec-Fetch-Site": "cross-site",
            "User-Agent": generate_user_agent()
        }
        
        # 商品详情API的请求头（主机不同）
        self.detail_headers = self.headers.copy()
        self.detail_headers["Host"] = "item-soa.jd.com"
    
    def get_price(self, product_id):
        """获取商品价格"""
        try:
            # 构造价格API的URL
            url = f"https://p.3.cn/prices/mgets?skuIds=J_{product_id}"
            
            # 随机延迟，避免请求过于频繁
            time.sleep(random.uniform(1, 3))
            
            # 发送请求
            response = requests.get(url, headers=self.headers)
            
            # 检查响应状态
            if response.status_code != 200:
                print(f"获取价格失败，状态码: {response.status_code}")
                return None
            
            # 解析JSON数据
            price_data = json.loads(response.text)
            
            if price_data and len(price_data) > 0:
                return {
                    "product_id": product_id,
                    "price": price_data[0].get("p"),
                    "original_price": price_data[0].get("m"),
                    "update_time": time.strftime("%Y-%m-%d %H:%M:%S")
                }
            else:
                print("未获取到价格数据")
                return None
                
        except Exception as e:
            print(f"获取价格时发生错误: {str(e)}")
            return None
    
    def get_product_detail(self, product_id):
        """获取商品详细信息"""
        try:
            # 构造商品详情API的URL
            url = f"https://item-soa.jd.com/getItemDetail?skuId={product_id}"
            
            # 随机延迟
            time.sleep(random.uniform(1, 3))
            
            # 发送请求
            response = requests.get(url, headers=self.detail_headers)
            
            # 检查响应状态
            if response.status_code != 200:
                print(f"获取商品详情失败，状态码: {response.status_code}")
                return None
            
            # 解析JSON数据
            detail_data = json.loads(response.text)
            
            # 提取需要的信息
            if detail_data.get("code") == 200 and "data" in detail_data:
                data = detail_data["data"]
                return {
                    "product_id": product_id,
                    "name": data.get("itemName"),
                    "brand": data.get("brandName"),
                    "shop_name": data.get("shopName"),
                    "category": data.get("category", {}).get("categoryName"),
                    "description": data.get("wDiscription"),
                    "stock": data.get("stockState")  # 1表示有货，0表示无货
                }
            else:
                print("未获取到商品详情数据")
                return None
                
        except Exception as e:
            print(f"获取商品详情时发生错误: {str(e)}")
            return None
    
    def get_product_full_info(self, product_id):
        """获取商品完整信息（价格+详情）"""
        price_info = self.get_price(product_id)
        if not price_info:
            return None
            
        detail_info = self.get_product_detail(product_id)
        if not detail_info:
            return None
            
        # 合并信息
        full_info = {**price_info,** detail_info}
        return full_info

if __name__ == "__main__":
    # 示例用法
    fetcher = JDProductFetcher()
    
    # 测试商品ID（可以替换为其他京东商品ID）
    product_ids = ["100008348542", "100012166285", "100023216438"]
    
    for pid in product_ids:
        print(f"\n获取商品 {pid} 的信息...")
        product_info = fetcher.get_product_full_info(pid)
        
        if product_info:
            print("商品完整信息:")
            print(json.dumps(product_info, ensure_ascii=False, indent=2))
        else:
            print(f"获取商品 {pid} 信息失败")