商品详情页实时数据采集实现逻辑：京东 API 接口底层技术拆解

pip install requests  # 网络请求
pip install fake_useragent  # 随机User-Agent
pip install python-dotenv  # 环境变量管理

import requests
import json
import time
from fake_useragent import UserAgent
from typing import Dict, Optional
import os
from dotenv import load_dotenv

# 加载环境变量（代理配置等）
load_dotenv()

class JDProductSpider:
    def __init__(self):
        # 初始化请求配置
        self.ua = UserAgent()
        self.proxies = self._get_proxies()  # 代理配置
        self.timeout = 10  # 请求超时时间
        self.retry_count = 3  # 重试次数

    def _get_proxies(self) -> Optional[Dict]:
        """从环境变量获取代理配置（可选）"""
        proxy = os.getenv("PROXY")
        if proxy:
            return {"http": proxy, "https": proxy}
        return None

    def _get_headers(self) -> Dict:
        """生成随机请求头，模拟浏览器行为"""
        return {
            "User-Agent": self.ua.random,
            "Referer": "https://www.jd.com/",
            "Accept": "application/json, text/plain, */*",
            "Accept-Language": "zh-CN,zh;q=0.9",
            "Connection": "keep-alive",
            # 可添加Cookie提升稳定性（从浏览器复制）
            "Cookie": os.getenv("JD_COOKIE", "")
        }

    def _parse_product_data(self, raw_data: Dict) -> Dict:
        """解析API返回的原始数据，提取核心字段"""
        try:
            # 从原始数据中提取商品基本信息
            product_info = raw_data.get("wareInfo", {}).get("basicInfo", {})
            price_info = raw_data.get("price", {}).get("p", "0")  # 价格
            stock_info = raw_data.get("stock", {}).get("StockState", 0)  # 库存状态（1=有货）
            
            # 结构化输出
            return {
                "product_id": product_info.get("wareId", ""),  # 商品ID
                "name": product_info.get("name", ""),  # 商品名称
                "price": float(price_info) if price_info else 0.0,  # 价格
                "shop_name": product_info.get("shopName", ""),  # 店铺名称
                "shop_id": product_info.get("shopId", ""),  # 店铺ID
                "brand": product_info.get("brandName", ""),  # 品牌
                "stock": "有货" if stock_info == 1 else "无货",  # 库存状态
                "url": f"https://item.jd.com/{product_info.get('wareId', '')}.html",  # 商品链接
                "timestamp": int(time.time())  # 采集时间戳
            }
        except Exception as e:
            print(f"数据解析失败: {str(e)}")
            return {}

    def get_product_detail(self, product_id: str) -> Optional[Dict]:
        """
        获取商品详情数据
        :param product_id: 京东商品ID（如https://item.jd.com/100012345678.html中的100012345678）
        :return: 结构化商品数据
        """
        # 京东商品详情内部API（经逆向分析获取）
        api_url = f"https://item-soa.jd.com/getWareBusiness?skuId={product_id}"
        
        for i in range(self.retry_count):
            try:
                headers = self._get_headers()
                response = requests.get(
                    url=api_url,
                    headers=headers,
                    proxies=self.proxies,
                    timeout=self.timeout
                )
                
                # 检查请求状态
                if response.status_code != 200:
                    print(f"请求失败，状态码: {response.status_code}，重试第{i+1}次")
                    time.sleep(2)
                    continue
                
                # 解析JSON数据
                raw_data = response.json()
                if raw_data.get("code") != 0:
                    print(f"API返回错误: {raw_data.get('msg', '未知错误')}")
                    time.sleep(2)
                    continue
                
                # 解析并返回结构化数据
                return self._parse_product_data(raw_data)
            
            except requests.exceptions.RequestException as e:
                print(f"请求异常: {str(e)}，重试第{i+1}次")
                time.sleep(2)
        
        # 多次重试失败后返回None
        print(f"获取商品{product_id}数据失败，已达最大重试次数")
        return None

if __name__ == "__main__":
    # 示例：采集商品ID为100012345678的详情数据
    spider = JDProductSpider()
    product_data = spider.get_product_detail("100012345678")
    
    if product_data:
        print("商品详情数据:")
        print(json.dumps(product_data, ensure_ascii=False, indent=2))
    else:
        print("采集失败")