Python 实战：快速接入淘宝 API 批量获取商品详情

pip install requests python-dotenv urllib3

# .env文件内容
APP_KEY=你的AppKey
APP_SECRET=你的AppSecret
ACCESS_TOKEN=你的AccessToken
API_URL=http://gw.api.taobao.com/router/rest

import os
import time
import hmac
import hashlib
import base64
import requests
from urllib.parse import urlencode
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

class TaobaoAPIClient:
    def __init__(self):
        self.app_key = os.getenv('APP_KEY')
        self.app_secret = os.getenv('APP_SECRET')
        self.access_token = os.getenv('ACCESS_TOKEN')
        self.api_url = os.getenv('API_URL')
        
    def _generate_sign(self, params):
        """生成API签名"""
        # 按参数名ASCII排序
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        # 拼接参数
        query_string = urlencode(sorted_params)
        # 生成签名字符串
        sign_str = self.app_secret + query_string + self.app_secret
        # HMAC-SHA1加密
        sign = hmac.new(
            sign_str.encode('utf-8'),
            digestmod=hashlib.sha1
        ).digest()
        # Base64编码
        return base64.b64encode(sign).decode('utf-8')
        
    def call(self, method, params=None):
        """调用淘宝API"""
        if params is None:
            params = {}
            
        # 公共参数
        public_params = {
            'app_key': self.app_key,
            'format': 'json',
            'method': method,
            'timestamp': time.strftime('%Y-%m-%d %H:%M:%S'),
            'v': '2.0',
            'sign_method': 'hmac-sha1',
            'session': self.access_token
        }
        
        # 合并参数
        all_params = {**public_params,** params}
        # 生成签名
        all_params['sign'] = self._generate_sign(all_params)
        
        try:
            # 发送请求
            response = requests.get(
                self.api_url,
                params=all_params,
                timeout=10
            )
            response.raise_for_status()
            result = response.json()
            
            # 处理错误响应
            if 'error_response' in result:
                error = result['error_response']
                raise Exception(f"API错误: {error['msg']} (错误码: {error['code']})")
                
            return result
        except Exception as e:
            print(f"调用失败: {str(e)}")
            return None

import time
import json
from typing import List, Dict
from taobao_api_client import TaobaoAPIClient

class ProductDetailFetcher:
    def __init__(self, max_retries=3, interval=2):
        self.client = TaobaoAPIClient()
        self.max_retries = max_retries  # 最大重试次数
        self.interval = interval  # API调用间隔（秒）
        
    def fetch_single_product(self, num_iid: str) -> Dict:
        """获取单个商品详情"""
        # 需要返回的字段列表
        fields = (
            "num_iid,title,price,stock,sale_count,shop_name,nick,"
            "cid,category_name,pic_url,detail_url,skus,params"
        )
        
        for retry in range(self.max_retries):
            try:
                result = self.client.call(
                    "taobao.item.get",
                    {
                        "num_iid": num_iid,
                        "fields": fields
                    }
                )
                
                if result and "item_get_response" in result:
                    return result["item_get_response"]["item"]
                    
            except Exception as e:
                print(f"获取商品 {num_iid} 失败（第{retry+1}次重试）: {str(e)}")
                time.sleep(self.interval)
                
        print(f"商品 {num_iid} 达到最大重试次数，获取失败")
        return None
        
    def fetch_batch_products(self, num_iids: List[str]) -> List[Dict]:
        """批量获取商品详情"""
        results = []
        total = len(num_iids)
        
        for index, num_iid in enumerate(num_iids, 1):
            print(f"正在获取商品 {index}/{total}: {num_iid}")
            product = self.fetch_single_product(num_iid)
            
            if product:
                results.append(product)
                # 保存中间结果
                self._save_temp_result(product)
                
            # 控制调用频率，避免触发限流
            if index < total:
                time.sleep(self.interval)
                
        return results
        
    def _save_temp_result(self, product: Dict, filename="temp_products.json"):
        """保存临时结果到文件"""
        try:
            with open(filename, 'a', encoding='utf-8') as f:
                json.dump(product, f, ensure_ascii=False)
                f.write('\n')
        except Exception as e:
            print(f"保存临时结果失败: {str(e)}")

    @staticmethod
    def save_results(products: List[Dict], filename="products_detail.json"):
        """保存最终结果到JSON文件"""
        try:
            with open(filename, 'w', encoding='utf-8') as f:
                json.dump(products, f, ensure_ascii=False, indent=2)
            print(f"成功保存 {len(products)} 个商品详情到 {filename}")
        except Exception as e:
            print(f"保存结果失败: {str(e)}")

def main():
    # 要查询的商品ID列表（示例）
    product_ids = [
        "654879215632",
        "648752136547",
        "632547896521",
        "612547896325",
        "687542136589"
        # 可添加更多商品ID
    ]
    
    # 创建获取器实例
    fetcher = ProductDetailFetcher(
        max_retries=3,  # 失败重试3次
        interval=2      # 每2秒调用一次API
    )
    
    # 批量获取商品详情
    print(f"开始批量获取 {len(product_ids)} 个商品详情...")
    products = fetcher.fetch_batch_products(product_ids)
    
    # 保存结果
    if products:
        fetcher.save_results(products)
        
        # 打印统计信息
        print("\n获取统计:")
        print(f"总商品数: {len(product_ids)}")
        print(f"成功获取: {len(products)}")
        print(f"失败数量: {len(product_ids) - len(products)}")

if __name__ == "__main__":
    main()