Python 调用淘宝 API 实战：高效采集商品详情数据

pip install requests

import requests
import hashlib
import time
import urllib.parse
import csv

# -------------------------- 1. 配置API核心参数 --------------------------
APP_KEY = "你的App Key"  # 替换为你的App Key
APP_SECRET = "你的App Secret"  # 替换为你的App Secret
ACCESS_TOKEN = "你的Access Token"  # 替换为你的Access Token
API_URL = "https://eco.taobao.com/router/rest"  # 淘宝API网关地址


# -------------------------- 2. 生成淘宝API签名 --------------------------
def generate_sign(params, app_secret):
    """
    按淘宝API签名规则生成签名：
    1. 参数按key字母升序排序；
    2. 拼接为"key=value"格式，并用&连接；
    3. 拼接app_secret，用MD5加密（大写）。
    """
    # 1. 按key升序排序
    sorted_params = sorted(params.items(), key=lambda x: x[0])
    # 2. 拼接参数字符串
    param_str = "&".join([f"{key}={value}" for key, value in sorted_params])
    # 3. 拼接app_secret并MD5加密
    sign_str = param_str + app_secret
    sign = hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
    return sign


# -------------------------- 3. 调用taobao.item.get接口获取商品详情 --------------------------
def get_taobao_item_detail(num_iid):
    """
    通过商品ID（num_iid）获取商品详情
    :param num_iid: 商品ID（可从淘宝商品详情页URL中获取，如https://item.taobao.com/item.htm?id=123456789中的123456789）
    :return: 商品详情字典（失败返回None）
    """
    # 公共参数（所有淘宝API均需携带）
    public_params = {
        "app_key": APP_KEY,
        "method": "taobao.item.get",  # 接口名称
        "format": "json",  # 返回格式（支持json/xml）
        "v": "2.0",  # API版本
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),  # 时间戳（格式固定）
        "sign_method": "md5",  # 签名方式
        "session": ACCESS_TOKEN  # Access Token（部分接口需携带）
    }

    # 业务参数（接口特定参数，参考taobao.item.get文档）
    business_params = {
        "num_iid": num_iid,
        "fields": "num_iid,title,price,stock,cid,seller_id,nick,item_desc"  # 需要获取的字段，用逗号分隔
    }

    # 合并所有参数
    all_params = {**public_params, **business_params}
    # 生成签名
    all_params["sign"] = generate_sign(all_params, APP_SECRET)

    # 发送HTTP GET请求
    try:
        response = requests.get(API_URL, params=all_params, timeout=10)
        response.raise_for_status()  # 若状态码非200，抛出异常
        result = response.json()

        # 检查返回结果是否正常（淘宝API错误会放在error_response中）
        if "error_response" in result:
            error = result["error_response"]
            print(f"API调用失败：{error['msg']}（错误码：{error['code']}）")
            return None
        else:
            item_detail = result["item_get_response"]["item"]
            print(f"成功获取商品详情：{item_detail['title']}")
            return item_detail

    except Exception as e:
        print(f"请求异常：{str(e)}")
        return None


# -------------------------- 4. 数据存储：将商品详情保存到CSV文件 --------------------------
def save_item_to_csv(item_detail, file_path="taobao_item_detail.csv"):
    """
    将商品详情保存到CSV文件
    :param item_detail: 商品详情字典
    :param file_path: 保存路径
    """
    if not item_detail:
        print("无有效商品数据，无需保存")
        return

    # 定义CSV表头（与获取的fields对应）
    headers = ["num_iid", "title", "price", "stock", "cid", "seller_id", "nick", "item_desc"]
    # 提取数据（确保顺序与表头一致）
    row_data = [
        item_detail.get("num_iid", ""),
        item_detail.get("title", ""),
        item_detail.get("price", ""),
        item_detail.get("stock", ""),
        item_detail.get("cid", ""),
        item_detail.get("seller_id", ""),
        item_detail.get("nick", ""),
        item_detail.get("item_desc", "")  # 商品描述可能是HTML，需后续处理
    ]

    # 写入CSV（若文件不存在则创建，存在则追加）
    with open(file_path, "a", encoding="utf-8-sig", newline="") as f:
        writer = csv.writer(f)
        # 首次写入时添加表头
        import os
        if os.path.getsize(file_path) == 0:
            writer.writerow(headers)
        writer.writerow(row_data)
    print(f"商品数据已保存到：{file_path}")


# -------------------------- 5. 批量采集示例 --------------------------
if __name__ == "__main__":
    # 待采集的商品ID列表（可从Excel/数据库中读取，此处为示例）
    item_id_list = ["123456789", "987654321", "112233445"]  # 替换为实际需要采集的商品ID

    # 循环采集每个商品
    for num_iid in item_id_list:
        item_detail = get_taobao_item_detail(num_iid)
        if item_detail:
            save_item_to_csv(item_detail)
        time.sleep(2)  # 避免请求频率过高被限制（淘宝API有QPS限制，通常个人应用为10次/秒）

from bs4 import BeautifulSoup
soup = BeautifulSoup(item_detail["item_desc"], "html.parser")
pure_desc = soup.get_text().strip()

import pandas as pd
df = pd.read_excel("item_ids.xlsx")
item_id_list = df["num_iid"].tolist()  # 假设Excel中列名为num_iid

import java.util.*;
import java.security.MessageDigest;

public class TaobaoApiSign {
    public static String generateSign(Map<String, String> params, String appSecret) {
        // 按key升序排序
        Map<String, String> sortedMap = new TreeMap<>(params);
        // 拼接参数
        StringBuilder sb = new StringBuilder();
        for (Map.Entry<String, String> entry : sortedMap.entrySet()) {
            sb.append(entry.getKey()).append("=").append(entry.getValue()).append("&");
        }
        // 移除最后一个&，拼接appSecret
        String signStr = sb.substring(0, sb.length() - 1) + appSecret;
        // MD5加密
        try {
            MessageDigest md = MessageDigest.getInstance("MD5");
            byte[] bytes = md.digest(signStr.getBytes("UTF-8"));
            // 转为大写十六进制
            StringBuilder signSb = new StringBuilder();
            for (byte b : bytes) {
                signSb.append(String.format("%02X", b));
            }
            return signSb.toString();
        } catch (Exception e) {
            throw new RuntimeException("生成签名失败", e);
        }
    }
}