实战解析：阿里巴巴国际站 alibaba.item_get 商品详情数据API接口

App Key = ALI_2025_XXXXXXXXApp Secret = XXXXXXXXXXXXXXXXXXXXXXXXAccess Token = (通过OAuth2.0获取，有效期2小时)

pip install requests>=2.31.0
pip install cryptography>=41.0.0  # 用于HMAC-SHA256签名pip install redis>=4.5.0          # 可选，用于数据缓存

import requestsimport timeimport jsonfrom typing import Optionalclass AlibabaAuth:
    """OAuth2.0认证管理器"""
    
    def __init__(self, app_key: str, app_secret: str):
        self.app_key = app_key
        self.app_secret = app_secret
        self.token_url = "https://auth.alibaba.com/oauth/token"
        self.token_cache = {}
    
    def get_access_token(self) -> Optional[str]:
        """
        获取Access Token（带缓存机制）
        有效期2小时，提前10分钟刷新
        """
        cache_key = f"alibaba_token_{self.app_key}"
        
        # 检查缓存
        if cache_key in self.token_cache:
            token_info = self.token_cache[cache_key]
            if time.time() < token_info['expire_time']:
                return token_info['token']
        
        # 请求新Token
        try:
            response = requests.post(
                self.token_url,
                data={
                    "grant_type": "client_credentials",
                    "client_id": self.app_key,
                    "client_secret": self.app_secret,
                    "scope": "item_get,trade_data"  # 申请的权限范围
                },
                headers={"Content-Type": "application/x-www-form-urlencoded"},
                timeout=10
            )
            response.raise_for_status()
            
            result = response.json()
            token = result.get("access_token")
            expires_in = result.get("expires_in", 7200)  # 默认2小时
            
            # 缓存Token
            self.token_cache[cache_key] = {
                "token": token,
                "expire_time": time.time() + expires_in - 600  # 提前10分钟过期
            }
            
            print(f"✅ Token获取成功: {token[:20]}...")
            return token            
        except Exception as e:
            print(f"❌ Token获取失败: {e}")
            return None

import hmacimport hashlibfrom urllib.parse import urlencodeclass AlibabaSigner:
    """签名生成器"""
    
    def __init__(self, app_secret: str):
        self.app_secret = app_secret    
    def generate_sign(self, params: dict) -> str:
        """
        生成HMAC-SHA256签名
        规则：app_secret + 排序后的key1value1key2value2... + app_secret
        """
        # 过滤空值和sign字段
        filtered_params = {k: v for k, v in params.items() if v is not None and k != "sign"}
        
        # 按键名ASCII码升序排序
        sorted_params = sorted(filtered_params.items(), key=lambda x: x[0])
        
        # 拼接签名内容
        sign_content = self.app_secret        for key, value in sorted_params:
            sign_content += f"{key}{value}"
        sign_content += self.app_secret        
        # HMAC-SHA256加密并转大写
        sign = hmac.new(
            self.app_secret.encode('utf-8'),
            sign_content.encode('utf-8'),
            hashlib.sha256        ).hexdigest().upper()
        
        return sign

class AlibabaAPI:
    """阿里巴巴国际站API客户端"""
    
    def __init__(self, app_key: str, app_secret: str):
        self.app_key = app_key
        self.app_secret = app_secret
        self.base_url = "https://openapi.alibaba.com/api/v3/item/get"
        self.auth = AlibabaAuth(app_key, app_secret)
        self.signer = AlibabaSigner(app_secret)
    
    def get_item_detail(
        self, 
        item_id: str, 
        fields: str = None,
        language: str = "en",
        include_raw: bool = True
    ) -> Optional[dict]:
        """
        获取商品详情主方法
        :param item_id: 商品ID（Alibaba.com商品链接中的数字部分）
        :param fields: 指定返回字段（逗号分隔），None则返回全部
        :param language: 返回语言（en/zh/es/ru等）
        :param include_raw: 是否包含原始HTML描述
        :return: 标准化商品详情
        """
        # 获取Token
        access_token = self.auth.get_access_token()
        if not access_token:
            print("❌ 无法获取Access Token")
            return None
        
        # 构造请求参数
        params = {
            "app_key": self.app_key,
            "item_id": item_id,
            "language": language,
            "timestamp": str(int(time.time() * 1000)),
            "include_raw": str(include_raw).lower(),
            "sign_method": "hmac-sha256"
        }
        
        if fields:
            params["fields"] = fields        
        # 生成签名
        params["sign"] = self.signer.generate_sign(params)
        
        # 添加认证头
        headers = {
            "Authorization": f"Bearer {access_token}",
            "Content-Type": "application/json",
            "X-Api-Version": "3.0"
        }
        
        try:
            print(f"🚀 开始获取商品: {item_id}")
            response = requests.get(
                self.base_url,
                params=params,
                headers=headers,
                timeout=30
            )
            response.raise_for_status()
            
            result = response.json()
            
            # 检查业务错误码
            if result.get("code") != 0:
                print(f"❌ API业务错误: {result.get('msg')} (code: {result.get('code')})")
                return None
            
            # 解析并标准化数据
            raw_data = result.get("result", {})
            return self._parse_item_detail(raw_data, item_id)
            
        except requests.exceptions.RequestException as e:
            print(f"❌ 网络请求异常: {e}")
            return None
        except json.JSONDecodeError as e:
            print(f"❌ JSON解析失败: {e}")
            return None
    
    def _parse_item_detail(self, raw_data: dict, item_id: str) -> dict:
        """
        解析并标准化商品详情（核心方法）
        """
        if not raw_data:
            return None
        
        # 1. 基础信息
        base_info = {
            "platform": "alibaba",
            "product_id": item_id,
            "subject": raw_data.get("subject", ""),
            "product_url": f"https://www.alibaba.com/product-detail/{item_id}.html",
            "category_id": raw_data.get("category_id"),
            "main_image": raw_data.get("main_image", ""),
            "extra_images": raw_data.get("extra_images", []),
            "video_url": raw_data.get("video_url", ""),
            "html": raw_data.get("desc", "")  # 原始HTML描述
        }
        
        # 2. 价格体系（关键）
        price_info = raw_data.get("price_info", {})
        base_info["price_info"] = {
            "currency": price_info.get("currency", "USD"),
            "price_unit": price_info.get("price_unit"),
            "price_range": price_info.get("price_range"),  # 如 "5.0-10.0"
            "fob_price": price_info.get("fob_price"),
            "min_order": raw_data.get("min_order_quantity", 0),
            "min_order_unit": raw_data.get("min_order_unit", "piece")
        }
        
        # 3. 阶梯价格（Bulk Pricing）
        tier_prices = []
        for tier in price_info.get("tier_prices", []):
            tier_prices.append({
                "min_quantity": tier.get("min_quantity"),
                "price": tier.get("price"),
                "discount_rate": tier.get("discount")
            })
        base_info["tier_prices"] = tier_prices        
        # 4. SKU详情（最核心）
        sku_list = []
        for sku in raw_data.get("sku_list", []):
            sku_list.append({
                "sku_id": sku.get("sku_id"),
                "attributes": sku.get("attributes"),  # 颜色、尺寸等
                "price": sku.get("price"),
                "stock": sku.get("inventory", 0),
                "image": sku.get("image_url", ""),
                "can_customize": sku.get("support_customization", False),
                "custom_fields": sku.get("custom_fields", [])
            })
        base_info["sku_list"] = sku_list        
        # 5. 供应商信息
        supplier = raw_data.get("supplier_info", {})
        base_info["supplier"] = {
            "supplier_id": supplier.get("supplier_id"),
            "name": supplier.get("supplier_name"),
            "gold_supplier": supplier.get("is_gold_supplier", False),
            "verified": supplier.get("is_verified", False),
            "response_rate": supplier.get("response_rate", 0),
            "transaction_level": supplier.get("transaction_level"),
            "country": supplier.get("country", "CN"),
            "province": supplier.get("province")
        }
        
        # 6. 交易数据（需特殊权限）
        trade_info = raw_data.get("trade_info", {})
        base_info["trade_data"] = {
            "30day_transactions": trade_info.get("transactions", 0),
            "30day_buyers": trade_info.get("buyers", 0),
            "repeat_buyer_rate": trade_info.get("repeat_buyer_rate", 0),
            "avg_lead_time": trade_info.get("avg_lead_time", 0),  # 平均交期（天）
            "payment_methods": trade_info.get("payment_methods", [])
        }
        
        # 7. 物流与港口
        logistics = raw_data.get("shipping_info", {})
        base_info["shipping"] = {
            "ports": logistics.get("ports", []),  # 起运港
            "shipping_methods": logistics.get("methods", []),
            "lead_time": logistics.get("lead_time")  # 发货时间
        }
        
        # 8. 认证信息
        certifications = []
        for cert in raw_data.get("certifications", []):
            certifications.append({
                "type": cert.get("certification_type"),
                "authority": cert.get("authority"),
                "valid_until": cert.get("valid_date")
            })
        base_info["certifications"] = certifications        
        # 9. 关键词SEO
        seo = raw_data.get("seo_info", {})
        base_info["seo"] = {
            "keywords": seo.get("keywords", []),
            "search_volume": seo.get("search_volume", 0),
            "competition_index": seo.get("competition_index", 0)
        }
        
        print(f"✅ 解析成功: {base_info['subject'][:50]}...")
        print(f"   SKU数量: {len(sku_list)}")
        print(f"   供应商: {base_info['supplier']['name']}")
        print(f"   价格区间: {base_info['price_info']['price_range']}")
        
        return base_info

# 配置文件：config.pyALIBABA_CONFIG = {
    "app_key": "your_app_key_here",
    "app_secret": "your_app_secret_here"}# 主程序：main.pyimport sysfrom pathlib import Path
sys.path.append(str(Path(__file__).parent))from alibaba_api import AlibabaAPIimport jsonfrom datetime import datetimedef main():
    print("=" * 70)
    print("阿里巴巴国际站API实战演示")
    print("=" * 70)
    
    # 初始化API客户端
    api = AlibabaAPI(
        app_key=ALIBABA_CONFIG["app_key"],
        app_secret=ALIBABA_CONFIG["app_secret"]
    )
    
    # 商品ID列表（从商品详情页URL提取）
    item_ids = [
        "1600590399629",  # 示例：手机配件
        "6234567890123",  # 示例：服装
        "7894561230456"   # 示例：家居用品
    ]
    
    results = []
    for item_id in item_ids:
        try:
            # 获取完整详情（包含原始HTML）
            detail = api.get_item_detail(
                item_id=item_id,
                fields="subject,price_info,sku_list,supplier_info,trade_info,"
                       "shipping_info,certifications,seo_info,desc",
                language="en",
                include_raw=True
            )
            
            if detail:
                results.append({
                    "item_id": item_id,
                    "status": "success",
                    "data": detail,
                    "crawled_at": datetime.now().isoformat()
                })
            else:
                results.append({
                    "item_id": item_id,
                    "status": "failed",
                    "error": "API返回空数据"
                })
                
        except Exception as e:
            results.append({
                "item_id": item_id,
                "status": "failed",
                "error": str(e)
            })
        
        # 频率控制（建议QPS≤5）
        time.sleep(0.3)
    
    # 保存结果
    output_file = f"data/alibaba_products_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
    Path("data").mkdir(exist_ok=True)
    with open(output_file, "w", encoding="utf-8") as f:
        json.dump(results, f, ensure_ascii=False, indent=2)
    
    print("\n" + "=" * 70)
    print(f"✅ 数据采集完成！成功: {len([r for r in results if r['status']=='success'])}/{len(results)}")
    print(f"📄 结果已保存至: {output_file}")
    print("=" * 70)if __name__ == "__main__":
    main()

import redisimport hashlibclass CachedAlibabaAPI(AlibabaAPI):
    """带缓存的API客户端"""
    
    def __init__(self, *args, redis_host="localhost", **kwargs):
        super().__init__(*args, **kwargs)
        self.redis = redis.Redis(host=redis_host, port=6379, decode_responses=True)
    
    def get_item_detail(self, item_id: str, **kwargs):
        # 生成缓存键：hash(item_id + date)
        cache_key = f"alibaba:item:{hashlib.md5(f'{item_id}_{datetime.now().strftime(%Y%m%d)}'.encode()).hexdigest()}"
        
        # 尝试读取缓存
        cached = self.redis.get(cache_key)
        if cached:
            print(f"💾 命中缓存: {item_id}")
            return json.loads(cached)
        
        # 调用API
        data = super().get_item_detail(item_id, **kwargs)
        
        # 写入缓存（有效期24小时）
        if data:
            self.redis.setex(cache_key, 86400, json.dumps(data))
        
        return data

from concurrent.futures import ThreadPoolExecutor, as_completeddef batch_get_items(api: AlibabaAPI, item_ids: List[str], max_workers: int = 5):
    """
    多线程批量获取商品详情
    """
    results = []
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交任务
        future_to_item = {
            executor.submit(api.get_item_detail, item_id): item_id 
            for item_id in item_ids        }
        
        # 收集结果
        for future in as_completed(future_to_item):
            item_id = future_to_item[future]
            try:
                data = future.result()
                results.append({
                    "item_id": item_id,
                    "status": "success",
                    "data": data                })
            except Exception as e:
                results.append({
                    "item_id": item_id,
                    "status": "failed",
                    "error": str(e)
                })
    
    return results

def calculate_bulk_price(item_detail: dict, quantity: int) -> float:
    """
    根据采购数量计算实际价格（利用阶梯价格）
    """
    tier_prices = item_detail.get("tier_prices", [])
    if not tier_prices:
        return float(item_detail["price_info"]["price_range"].split("-")[0])
    
    # 按数量降序排序
    sorted_tiers = sorted(tier_prices, key=lambda x: x["min_quantity"], reverse=True)
    
    for tier in sorted_tiers:
        if quantity >= tier["min_quantity"]:
            return float(tier["price"])
    
    # 默认返回最低价
    return float(sorted_tiers[-1]["price"])

# 安全获取嵌套字段price = raw_data.get("price_info", {}).get("price_range", "N/A")stock = raw_data.get("stock", {}).get("total", 0)supplier_name = raw_data.get("supplier_info", {}).get("name", "Unknown")