超越基础调用：淘宝商品详情 API 的高级用法与业务集成

import requests
import time
import hashlib
import json
from typing import Dict, Optional

class TaobaoAPI:
    """淘宝API高级调用封装类"""
    def __init__(self, app_key: str, app_secret: str):
        self.app_key = app_key
        self.app_secret = app_secret
        self.gateway = "https://eco.taobao.com/router/rest"  # 正式环境网关
        self.timeout = 10  # 请求超时时间
        self.retry_times = 3  # 失败重试次数

    def _generate_sign(self, params: Dict) -> str:
        """
        生成API请求签名（淘宝MD5签名机制）
        :param params: 待签名的请求参数
        :return: 签名字符串
        """
        # 1. 按参数名ASCII升序排序
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        # 2. 拼接成key=value格式的字符串
        sign_str = self.app_secret
        for key, value in sorted_params:
            if value:  # 跳过空值
                sign_str += f"{key}{value}"
        sign_str += self.app_secret
        # 3. MD5加密并转大写
        sign = hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
        return sign

    def request(self, method: str, params: Dict) -> Optional[Dict]:
        """
        通用API请求方法，包含重试、签名、异常处理
        :param method: API方法名（如taobao.item.detail.get）
        :param params: 业务参数（如item_id、fields）
        :return: API返回的业务数据（已解析JSON）
        """
        # 基础公共参数
        common_params = {
            "app_key": self.app_key,
            "method": method,
            "format": "json",
            "v": "2.0",
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
            "sign_method": "md5"
        }
        # 合并公共参数与业务参数
        all_params = {**common_params, **params}
        # 生成签名
        all_params["sign"] = self._generate_sign(all_params)

        # 带重试的请求逻辑
        for retry in range(self.retry_times):
            try:
                response = requests.get(
                    url=self.gateway,
                    params=all_params,
                    timeout=self.timeout
                )
                response.raise_for_status()  # 抛出HTTP状态码异常
                result = response.json()

                # 处理API业务异常
                if "error_response" in result:
                    error_msg = result["error_response"]["msg"]
                    error_code = result["error_response"]["code"]
                    print(f"API调用失败（重试{retry+1}/{self.retry_times}）：{error_code} - {error_msg}")
                    continue

                return result["item_detail_get_response"]["item_detail"]
            except requests.exceptions.RequestException as e:
                print(f"请求异常（重试{retry+1}/{self.retry_times}）：{str(e)}")
                time.sleep(0.5)  # 重试间隔
        print("多次重试后仍调用失败")
        return None

import asyncio
import aiohttp
from typing import List

class TaobaoBatchAPI(TaobaoAPI):
    """基于异步的批量API调用类，含限流控制"""
    def __init__(self, app_key: str, app_secret: str, qps: int = 20):
        super().__init__(app_key, app_secret)
        self.qps = qps  # 设定QPS上限
        self.semaphore = asyncio.Semaphore(qps)  # 信号量控制并发

    async def _async_request(self, session: aiohttp.ClientSession, item_id: str, fields: str) -> Optional[Dict]:
        """异步单条商品查询"""
        async with self.semaphore:
            params = {"item_id": item_id, "fields": fields}
            # 复用父类的参数拼接和签名逻辑
            common_params = {
                "app_key": self.app_key,
                "method": "taobao.item.detail.get",
                "format": "json",
                "v": "2.0",
                "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
                "sign_method": "md5"
            }
            all_params = {**common_params, **params}
            all_params["sign"] = self._generate_sign(all_params)

            try:
                async with session.get(self.gateway, params=all_params, timeout=self.timeout) as response:
                    response.raise_for_status()
                    result = await response.json()
                    if "error_response" in result:
                        print(f"商品{item_id}查询失败：{result['error_response']['msg']}")
                        return None
                    return {
                        "item_id": item_id,
                        "data": result["item_detail_get_response"]["item_detail"]
                    }
            except Exception as e:
                print(f"商品{item_id}异步请求异常：{str(e)}")
                return None

    async def batch_get_items(self, item_ids: List[str], fields: str) -> List[Dict]:
        """
        批量查询商品详情，自动限流
        :param item_ids: 商品ID列表
        :param fields: 需要返回的字段
        :return: 批量查询结果
        """
        async with aiohttp.ClientSession() as session:
            tasks = [self._async_request(session, item_id, fields) for item_id in item_ids]
            results = await asyncio.gather(*tasks)
            # 过滤掉查询失败的结果
            return [res for res in results if res is not None]

# 批量调用示例
if __name__ == "__main__":
    # 替换为自己的AppKey和AppSecret
    APP_KEY = "your_app_key"
    APP_SECRET = "your_app_secret"
    
    # 1. 基础单商品查询
    api = TaobaoAPI(APP_KEY, APP_SECRET)
    item_detail = api.request(
        method="taobao.item.detail.get",
        params={
            "item_id": "123456789",  # 替换为真实商品ID
            "fields": "title,price,stock,sku,category_id,desc"
        }
    )
    if item_detail:
        print(f"商品标题：{item_detail['title']}")
        print(f"商品价格：{item_detail['price']}")
    
    # 2. 批量商品查询
    batch_api = TaobaoBatchAPI(APP_KEY, APP_SECRET, qps=15)  # 保守设置QPS，避免限流
    item_ids = ["123456789", "987654321", "112233445"]  # 替换为真实商品ID列表
    loop = asyncio.get_event_loop()
    batch_results = loop.run_until_complete(batch_api.batch_get_items(item_ids, "title,price,stock"))
    for res in batch_results:
        print(f"商品{res['item_id']}：{res['data']['title']} - {res['data']['price']}元")

import sqlite3
from datetime import datetime

def save_item_to_db(item_data: Dict, db_path: str = "taobao_items.db"):
    """
    将商品详情数据存入SQLite数据库（实际业务可替换为MySQL/PostgreSQL）
    :param item_data: API返回的商品数据
    :param db_path: 数据库路径
    """
    # 初始化数据库表
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS taobao_item_details (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        item_id TEXT UNIQUE NOT NULL,
        title TEXT,
        price DECIMAL(10,2),
        stock INTEGER,
        category_id TEXT,
        create_time DATETIME,
        update_time DATETIME
    )
    """)
    # 插入/更新数据
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    cursor.execute("""
    INSERT OR REPLACE INTO taobao_item_details 
    (item_id, title, price, stock, category_id, create_time, update_time)
    VALUES (?, ?, ?, ?, ?, COALESCE((SELECT create_time FROM taobao_item_details WHERE item_id=?), ?), ?)
    """, (
        item_data.get("item_id"),
        item_data.get("title"),
        item_data.get("price"),
        item_data.get("stock"),
        item_data.get("category_id"),
        item_data.get("item_id"),
        now,
        now
    ))
    conn.commit()
    conn.close()

# 集成示例：查询商品并存入数据库
if __name__ == "__main__":
    APP_KEY = "your_app_key"
    APP_SECRET = "your_app_secret"
    api = TaobaoAPI(APP_KEY, APP_SECRET)
    item_detail = api.request(
        method="taobao.item.detail.get",
        params={"item_id": "123456789", "fields": "item_id,title,price,stock,category_id"}
    )
    if item_detail:
        save_item_to_db(item_detail)
        print("商品数据已存入数据库，可用于选品分析")

import time

def price_monitor(item_id: str, threshold: float, check_interval: int = 3600):
    """
    商品价格监控：定时查询价格，波动超过阈值则告警
    :param item_id: 商品ID
    :param threshold: 价格波动阈值（元）
    :param check_interval: 检查间隔（秒）
    """
    APP_KEY = "your_app_key"
    APP_SECRET = "your_app_secret"
    api = TaobaoAPI(APP_KEY, APP_SECRET)
    # 获取初始价格
    initial_detail = api.request(
        method="taobao.item.detail.get",
        params={"item_id": item_id, "fields": "price"}
    )
    if not initial_detail:
        print("初始价格获取失败，监控启动失败")
        return
    initial_price = float(initial_detail["price"])
    print(f"价格监控启动：商品{item_id}初始价格{initial_price}元，阈值{threshold}元")

    # 循环监控
    while True:
        time.sleep(check_interval)
        current_detail = api.request(
            method="taobao.item.detail.get",
            params={"item_id": item_id, "fields": "price"}
        )
        if not current_detail:
            continue
        current_price = float(current_detail["price"])
        price_diff = abs(current_price - initial_price)
        if price_diff > threshold:
            # 此处可替换为邮件/钉钉/企业微信告警
            print(f"【价格告警】商品{item_id}价格波动超过阈值！当前价格{current_price}元，与初始价格相差{price_diff}元")
            initial_price = current_price  # 更新初始价格

# 启动价格监控
if __name__ == "__main__":
    price_monitor(item_id="123456789", threshold=5.0, check_interval=3600)