API 数据采集实战：以京东商品详情接口为例的接入与调试全过程

pip install requests pandas

import requests
import json
import time
import hashlib
import pandas as pd

class JDProductAPI:
    def __init__(self, app_key, app_secret):
        self.app_key = app_key
        self.app_secret = app_secret
        self.api_url = "https://api.jd.com/routerjson"  # 京东API通用网关地址

    def generate_sign(self, params):
        """生成签名"""
        # 1. 按参数名ASCII升序排序
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        # 2. 拼接参数字符串
        param_str = "&".join([f"{k}={v}" for k, v in sorted_params])
        # 3. 拼接AppSecret
        sign_str = f"{param_str}{self.app_secret}"
        # 4. MD5加密
        sign = hashlib.md5(sign_str.encode("utf-8")).hexdigest()
        return sign

    def get_product_detail(self, product_id, fields=None):
        """获取商品详情"""
        # 构造基础参数
        params = {
            "app_key": self.app_key,
            "method": "jingdong.product.read.get",  # 商品详情查询接口方法名
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
            "format": "json",
            "v": "1.0",
            "product_id": product_id
        }
        # 可选参数：字段筛选（如需指定返回字段，可传入逗号分隔的字段名字符串）
        if fields:
            params["fields"] = fields
        # 生成签名
        params["sign"] = self.generate_sign(params)
        try:
            # 发送GET请求（京东API支持GET/POST，此处用GET）
            response = requests.get(self.api_url, params=params, timeout=10)
            # 解析响应数据
            result = json.loads(response.text)
            # 处理返回结果
            if result.get("code") == 200:
                # 提取商品核心信息（根据业务需求筛选字段）
                product_data = {
                    "商品ID": result["result"]["product_id"],
                    "商品名称": result["result"]["name"],
                    "品牌名称": result["result"]["brand_name"],
                    "售价": result["result"]["price"],
                    "库存": result["result"]["stock"],
                    "商品链接": result["result"]["url"],
                    "创建时间": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
                }
                return product_data
            else:
                print(f"接口调用失败：错误码{result['code']}，错误信息{result['msg']}")
                return None
        except Exception as e:
            print(f"请求异常：{str(e)}")
            return None

    def save_to_csv(self, product_data, filename="jd_products.csv"):
        """将商品数据保存到CSV文件"""
        # 检查文件是否存在，不存在则创建并写入表头
        try:
            pd.read_csv(filename)
            # 追加数据
            df = pd.DataFrame([product_data])
            df.to_csv(filename, mode="a", header=False, index=False, encoding="utf-8-sig")
        except FileNotFoundError:
            # 创建新文件并写入表头
            df = pd.DataFrame([product_data])
            df.to_csv(filename, index=False, encoding="utf-8-sig")
        print(f"数据已成功保存到{filename}")

if __name__ == "__main__":
    # 替换为你的AppKey和AppSecret
    APP_KEY = "你的AppKey"
    APP_SECRET = "你的AppSecret"
    # 初始化API实例
    jd_api = JDProductAPI(APP_KEY, APP_SECRET)
    # 要查询的商品ID（可从京东商品详情页URL中获取，如https://item.jd.com/100012345678.html中的100012345678）
    PRODUCT_ID = "100012345678"
    # 调用接口获取商品详情（如需获取所有字段，可去掉fields参数）
    product_detail = jd_api.get_product_detail(PRODUCT_ID, fields="product_id,name,brand_name,price,stock,url")
    if product_detail:
        print("商品详情：")
        print(json.dumps(product_detail, ensure_ascii=False, indent=2))
        # 保存数据到CSV
        jd_api.save_to_csv(product_detail)
    else:
        print("未获取到商品详情")

# 批量查询商品详情
def batch_get_product_details(self, product_ids, fields=None):
    all_products = []
    for product_id in product_ids:
        print(f"正在查询商品ID：{product_id}")
        product_data = self.get_product_detail(product_id, fields)
        if product_data:
            all_products.append(product_data)
        time.sleep(1)  # 控制请求频率
    return all_products

# 使用示例
product_ids = ["100012345678", "100012345679", "100012345680"]  # 多个商品ID
batch_products = jd_api.batch_get_product_details(product_ids)
if batch_products:
    # 保存批量数据到CSV
    df = pd.DataFrame(batch_products)
    df.to_csv("jd_batch_products.csv", index=False, encoding="utf-8-sig")
    print(f"批量查询完成，共获取{len(batch_products)}个商品数据")

from tenacity import retry, stop_after_attempt, wait_exponential

# 添加重试装饰器
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def get_product_detail(self, product_id, fields=None):
    # 原接口调用逻辑不变
    pass

import pymysql

def save_to_mysql(self, product_data, db_config):
    """保存数据到MySQL"""
    # 连接数据库
    db = pymysql.connect(
        host=db_config["host"],
        port=db_config["port"],
        user=db_config["user"],
        password=db_config["password"],
        database=db_config["database"],
        charset="utf8"
    )
    cursor = db.cursor()
    # 插入数据SQL语句
    sql = """
    INSERT INTO jd_products (product_id, product_name, brand_name, price, stock, product_url, create_time)
    VALUES (%s, %s, %s, %s, %s, %s, %s)
    """
    try:
        cursor.execute(sql, (
            product_data["商品ID"],
            product_data["商品名称"],
            product_data["品牌名称"],
            product_data["售价"],
            product_data["库存"],
            product_data["商品链接"],
            product_data["创建时间"]
        ))
        db.commit()
        print("数据已保存到MySQL")
    except Exception as e:
        db.rollback()
        print(f"MySQL保存失败：{str(e)}")
    finally:
        db.close()

# 数据库配置（替换为你的数据库信息）
db_config = {
    "host": "localhost",
    "port": 3306,
    "user": "root",
    "password": "your_password",
    "database": "jd_data"
}
# 保存数据到MySQL
if product_detail:
    jd_api.save_to_mysql(product_detail, db_config)

import logging

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    filename="jd_api_logs.log",
    filemode="a"
)

# 在接口调用方法中添加日志
def get_product_detail(self, product_id, fields=None):
    logging.info(f"开始查询商品ID：{product_id}")
    try:
        # 接口调用逻辑
        # ...
        if result.get("code") == 200:
            logging.info(f"商品ID：{product_id} 查询成功")
            return product_data
        else:
            logging.error(f"商品ID：{product_id} 查询失败，错误码{result['code']}，错误信息{result['msg']}")
            return None
    except Exception as e:
        logging.error(f"商品ID：{product_id} 请求异常：{str(e)}")
        return None