商品详情页实时数据采集技术方案：京东 API 接口开发实战

import hashlib
import time
from urllib.parse import urlencode


class JDAPISignUtils:
    @staticmethod
    def generate_sign(params, app_secret):
        """生成签名：参数排序后拼接+app_secret，MD5加密"""
        # 按参数名ASCII升序排序
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        # 拼接为key=value&key=value格式
        sign_str = urlencode(sorted_params) + app_secret
        # MD5加密并转为大写
        sign = hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
        return sign

    @staticmethod
    def get_timestamp():
        """生成符合格式的时间戳（yyyy-MM-dd HH:mm:ss）"""
        return time.strftime('%Y-%m-%d %H:%M:%S', time.localtime())

import requests
import json
from typing import Dict, Optional


class JDGoodsDetailAPI:
    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"  # 京东API网关地址
        self.sign_utils = JDAPISignUtils()

    def get_goods_detail(self, ware_id: str) -> Optional[Dict]:
        """获取商品详情页数据"""
        # 1. 组装基础参数
        params = {
            "method": "jingdong.ware.detail.get",  # 接口方法名
            "app_key": self.app_key,
            "timestamp": self.sign_utils.get_timestamp(),
            "format": "json",  # 返回格式
            "v": "2.0",  # 接口版本
            "ware_id": ware_id  # 商品ID
        }

        # 2. 生成签名
        params["sign"] = self.sign_utils.generate_sign(params, self.app_secret)

        # 3. 发送POST请求
        try:
            response = requests.post(self.api_url, data=params, timeout=10)
            response.raise_for_status()  # 抛出HTTP错误状态码
        except requests.exceptions.RequestException as e:
            print(f"API请求失败：{e}")
            return None

        # 4. 解析响应数据
        try:
            result = json.loads(response.text)
            # 京东API返回格式：{"jingdong_ware_detail_get_response": {"result": {...}}}
            if "jingdong_ware_detail_get_response" in result:
                return result["jingdong_ware_detail_get_response"].get("result", {})
            else:
                print(f"API返回格式异常：{result}")
                return None
        except json.JSONDecodeError as e:
            print(f"JSON解析失败：{e}")
            return None

import pymysql
import redis
from pymysql.cursors import DictCursor


class DataStorage:
    def __init__(self, mysql_config: Dict, redis_config: Dict):
        # 初始化MySQL连接
        self.mysql_conn = pymysql.connect(
            host=mysql_config["host"],
            user=mysql_config["user"],
            password=mysql_config["password"],
            database=mysql_config["db"],
            cursorclass=DictCursor
        )
        # 初始化Redis连接
        self.redis_client = redis.Redis(
            host=redis_config["host"],
            port=redis_config["port"],
            password=redis_config.get("password", ""),
            db=redis_config.get("db", 0)
        )

    def save_goods_detail(self, goods_data: Dict):
        """保存商品详情到MySQL"""
        if not goods_data.get("ware_id"):
            print("商品ID不存在，跳过存储")
            return

        # 提取核心字段（根据实际返回数据调整）
        sql = """
        INSERT INTO jd_goods_detail (
            ware_id, goods_name, price, stock, main_image, brand, create_time, update_time
        ) VALUES (%s, %s, %s, %s, %s, %s, NOW(), NOW())
        ON DUPLICATE KEY UPDATE
            goods_name = VALUES(goods_name),
            price = VALUES(price),
            stock = VALUES(stock),
            main_image = VALUES(main_image),
            brand = VALUES(brand),
            update_time = NOW()
        """
        try:
            with self.mysql_conn.cursor() as cursor:
                cursor.execute(sql, (
                    goods_data["ware_id"],
                    goods_data.get("goods_name", ""),
                    goods_data.get("price", 0),
                    goods_data.get("stock_num", 0),
                    goods_data.get("main_image", ""),
                    goods_data.get("brand_name", "")
                ))
                self.mysql_conn.commit()
            print(f"商品ID {goods_data['ware_id']} 数据存储成功")
        except Exception as e:
            self.mysql_conn.rollback()
            print(f"MySQL存储失败：{e}")

    def cache_goods_detail(self, ware_id: str, goods_data: Dict, expire=3600):
        """缓存商品数据到Redis（过期时间1小时）"""
        try:
            self.redis_client.set(
                f"jd_goods:{ware_id}",
                json.dumps(goods_data, ensure_ascii=False),
                ex=expire
            )
            print(f"商品ID {ware_id} 缓存成功")
        except Exception as e:
            print(f"Redis缓存失败：{e}")

    def close(self):
        """关闭数据库连接"""
        self.mysql_conn.close()
        self.redis_client.close()

def main():
    # 配置参数（替换为实际的app_key、app_secret和数据库信息）
    APP_KEY = "your_app_key"
    APP_SECRET = "your_app_secret"
    WARE_ID = "100012345678"  # 测试商品ID

    mysql_config = {
        "host": "localhost",
        "user": "root",
        "password": "123456",
        "db": "jd_goods_db"
    }

    redis_config = {
        "host": "localhost",
        "port": 6379
    }

    # 1. 初始化API客户端
    jd_api = JDGoodsDetailAPI(APP_KEY, APP_SECRET)

    # 2. 调用API获取商品详情
    goods_data = jd_api.get_goods_detail(WARE_ID)
    if not goods_data:
        print("未获取到商品数据，程序退出")
        return

    # 3. 初始化数据存储
    storage = DataStorage(mysql_config, redis_config)

    # 4. 存储与缓存数据
    storage.save_goods_detail(goods_data)
    storage.cache_goods_detail(WARE_ID, goods_data)

    # 5. 关闭连接
    storage.close()
    print("流程执行完成")


if __name__ == "__main__":
    main()