淘宝券后价详情API接口完全指南

券后价 = 原价(zk_final_price或price) - 优惠券面额(coupon_amount)

{
  "tbk_item_info_get_response": {
    "results": {
      "n_tbk_item": [
        {
          "num_iid": "567890123",
          "title": "时尚纯棉男士短袖T恤",
          "price": "99.90",              // 原价（划线价）
          "zk_final_price": "39.90",      // 券后价（优惠后价格）⭐核心字段
          "coupon_info": "满50减10",      // 优惠券信息描述
          "coupon_amount": "10.00",       // 优惠券面额
          "coupon_start_fee": "50.00",   // 优惠券使用门槛
          "coupon_start_time": "2025-03-01 00:00:00",
          "coupon_end_time": "2025-03-31 23:59:59",
          "commission_rate": "20.00",    // 佣金比例
          "commission": "7.98",          // 佣金金额
          "volume": 500,                 // 30天销量
          "pict_url": "https://example.com/tshirt.jpg"
        }
      ]
    }
  }}

import hashlibimport urllib.parseimport timefrom typing import Dictdef generate_taobao_sign(params: Dict[str, str], app_secret: str) -> str:
    """
    生成淘宝API签名（MD5）
    """
    # 1. 过滤sign参数，按ASCII排序
    sorted_params = sorted(
        [(k, v) for k, v in params.items() if k != "sign" and v is not None],
        key=lambda x: x[0]
    )
    
    # 2. 拼接为 key1value1key2value2 格式（值需URL编码）
    sign_str = "".join([
        f"{k}{urllib.parse.quote_plus(str(v))}" 
        for k, v in sorted_params    ])
    
    # 3. 首尾拼接app_secret
    sign_str = f"{app_secret}{sign_str}{app_secret}"
    
    # 4. MD5加密并转大写
    return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()# 使用示例params = {
    "app_key": "your_app_key",
    "method": "taobao.tbk.item.info.get",
    "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
    "format": "json",
    "v": "2.0",
    "sign_method": "md5",
    "num_iids": "123456789",
    "fields": "num_iid,title,price,zk_final_price,coupon_info"}sign = generate_taobao_sign(params, "your_app_secret")print(f"生成的签名: {sign}")

import requestsimport timeimport jsonimport hashlibimport urllib.parseimport loggingfrom typing import Dict, List, Optional, Unionfrom dataclasses import dataclassfrom requests.adapters import HTTPAdapterfrom urllib3.util.retry import Retry

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')logger = logging.getLogger(__name__)@dataclassclass CouponPriceInfo:
    """券后价信息数据类"""
    num_iid: str
    title: str
    original_price: float      # 原价
    final_price: float         # 券后价（zk_final_price）
    coupon_amount: float       # 优惠券面额
    coupon_start_fee: float    # 使用门槛
    coupon_info: str           # 优惠券描述
    commission_rate: float     # 佣金比例
    commission_amount: float   # 佣金金额
    coupon_start_time: Optional[str] = None
    coupon_end_time: Optional[str] = None
    is_coupon_valid: bool = True  # 优惠券是否有效
    volume: int = 0            # 销量
    pict_url: str = ""
    
    def calculate_real_savings(self) -> Dict:
        """计算真实优惠力度"""
        # 判断优惠券是否可用（当前价是否满足门槛）
        can_use_coupon = self.original_price >= self.coupon_start_fee        
        # 实际支付价（如果优惠券不可用，则按原价）
        real_pay_price = self.final_price if can_use_coupon else self.original_price        
        # 实际优惠金额
        real_savings = self.original_price - real_pay_price        
        # 折扣率
        discount_rate = (real_pay_price / self.original_price) * 10 if self.original_price > 0 else 0
        
        return {
            "can_use_coupon": can_use_coupon,
            "real_pay_price": round(real_pay_price, 2),
            "real_savings": round(real_savings, 2),
            "discount_rate": round(discount_rate, 1),  # 如8.5折
            "savings_percentage": round((real_savings / self.original_price) * 100, 1) if self.original_price > 0 else 0
        }class TaobaoCouponAPI:
    """
    淘宝券后价API客户端
    支持淘宝联盟官方API和第三方聚合API
    """
    
    def __init__(self, 
                 app_key: Optional[str] = None,
                 app_secret: Optional[str] = None,
                 session: Optional[str] = None,
                 third_party_key: Optional[str] = None,
                 use_official: bool = True):
        self.app_key = app_key
        self.app_secret = app_secret
        self.session = session
        self.third_party_key = third_party_key
        self.use_official = use_official        
        # 官方API端点
        self.official_url = "https://eco.taobao.com/router/rest"
        # 备用官方域名
        self.backup_url = "https://gw.api.taobao.com/router/rest"
        
        # 第三方API端点（示例）
        self.third_party_urls = {
            "aliprice": "https://www.aliprice.com/items/itemDetail",
            "o0b": "https://o0b.cn/ibrad/taobao/item_get/",
            "vv_tool": "http://api.vv-tool.com/tool/erps/taobaoget"
        }
        
        # 配置会话和重试
        self.session_req = requests.Session()
        retry = Retry(total=3, backoff_factor=0.5, status_forcelist=[429, 500, 502, 503])
        self.session_req.mount('https://', HTTPAdapter(max_retries=retry))
        
        # 请求频率控制（官方API默认QPS=10）
        self.last_request_time = 0
        self.min_interval = 0.1  # 100ms间隔
    
    def _rate_limit(self):
        """请求频率限制"""
        current_time = time.time()
        elapsed = current_time - self.last_request_time        if elapsed < self.min_interval:
            time.sleep(self.min_interval - elapsed)
        self.last_request_time = time.time()
    
    def _generate_sign(self, params: Dict) -> str:
        """生成淘宝官方API签名"""
        sorted_params = sorted(
            [(k, v) for k, v in params.items() if k != "sign" and v is not None],
            key=lambda x: x[0]
        )
        
        sign_str = "".join([
            f"{k}{urllib.parse.quote_plus(str(v))}" 
            for k, v in sorted_params        ])
        sign_str = f"{self.app_secret}{sign_str}{self.app_secret}"
        
        return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
    
    def _call_official_api(self, method: str, biz_params: Dict) -> Optional[Dict]:
        """调用淘宝官方API"""
        self._rate_limit()
        
        # 构建公共参数
        params = {
            "app_key": self.app_key,
            "method": method,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "format": "json",
            "v": "2.0",
            "sign_method": "md5"
        }
        
        # 添加业务参数
        params.update(biz_params)
        
        # 添加session（如需要）
        if self.session:
            params["session"] = self.session        
        # 生成签名
        params["sign"] = self._generate_sign(params)
        
        try:
            response = self.session_req.get(
                self.official_url, 
                params=params, 
                timeout=15
            )
            response.raise_for_status()
            result = response.json()
            
            # 处理错误
            if "error_response" in result:
                error = result["error_response"]
                logger.error(f"API错误: {error.get('msg')} (代码: {error.get('code')})")
                return None
            
            # 返回业务数据
            response_key = f"{method.replace('.', '_')}_response"
            return result.get(response_key)
            
        except requests.exceptions.RequestException as e:
            logger.error(f"请求异常: {e}")
            return None
        except json.JSONDecodeError:
            logger.error("JSON解析失败")
            return None
    
    def get_coupon_price_single(self, num_iid: str) -> Optional[CouponPriceInfo]:
        """
        获取单个商品的券后价详情
        """
        if self.use_official and self.app_key:
            # 使用官方API
            method = "taobao.tbk.item.info.get"
            fields = (
                "num_iid,title,pict_url,price,zk_final_price,"
                "coupon_info,coupon_amount,coupon_start_fee,"
                "coupon_start_time,coupon_end_time,"
                "commission_rate,commission,volume"
            )
            
            biz_params = {
                "num_iids": num_iid,
                "fields": fields            }
            
            result = self._call_official_api(method, biz_params)
            if not result:
                return None
            
            items = result.get("results", {}).get("n_tbk_item", [])
            if not items:
                logger.warning(f"商品 {num_iid} 无返回数据")
                return None
            
            item = items[0]
            
            # 解析价格字段
            original_price = float(item.get("price", 0))
            final_price = float(item.get("zk_final_price", 0))
            coupon_amount = float(item.get("coupon_amount", 0) or 0)
            coupon_start_fee = float(item.get("coupon_start_fee", 0) or 0)
            
            # 判断优惠券是否有效（当前时间是否在有效期内）
            # 简化处理，实际应解析时间字符串对比
            is_valid = True
            
            return CouponPriceInfo(
                num_iid=str(item.get("num_iid")),
                title=item.get("title", ""),
                original_price=original_price,
                final_price=final_price,
                coupon_amount=coupon_amount,
                coupon_start_fee=coupon_start_fee,
                coupon_info=item.get("coupon_info", ""),
                commission_rate=float(item.get("commission_rate", 0) or 0),
                commission_amount=float(item.get("commission", 0) or 0),
                coupon_start_time=item.get("coupon_start_time"),
                coupon_end_time=item.get("coupon_end_time"),
                is_coupon_valid=is_valid,
                volume=int(item.get("volume", 0)),
                pict_url=item.get("pict_url", "")
            )
        
        else:
            # 使用第三方API（简化示例）
            return self._get_third_party_price(num_iid)
    
    def get_coupon_price_batch(self, num_iids: List[str]) -> List[CouponPriceInfo]:
        """
        批量获取券后价（官方API支持一次查询40个商品）
        """
        if not num_iids:
            return []
        
        # 分批处理（每批最多40个）
        batch_size = 40
        results = []
        
        for i in range(0, len(num_iids), batch_size):
            batch = num_iids[i:i+batch_size]
            batch_str = ",".join(batch)
            
            method = "taobao.tbk.item.info.get"
            fields = (
                "num_iid,title,pict_url,price,zk_final_price,"
                "coupon_info,coupon_amount,coupon_start_fee,"
                "commission_rate,volume"
            )
            
            biz_params = {
                "num_iids": batch_str,
                "fields": fields            }
            
            result = self._call_official_api(method, biz_params)
            if result:
                items = result.get("results", {}).get("n_tbk_item", [])
                for item in items:
                    try:
                        info = CouponPriceInfo(
                            num_iid=str(item.get("num_iid")),
                            title=item.get("title", ""),
                            original_price=float(item.get("price", 0)),
                            final_price=float(item.get("zk_final_price", 0)),
                            coupon_amount=float(item.get("coupon_amount", 0) or 0),
                            coupon_start_fee=float(item.get("coupon_start_fee", 0) or 0),
                            coupon_info=item.get("coupon_info", ""),
                            commission_rate=float(item.get("commission_rate", 0) or 0),
                            commission_amount=float(item.get("commission", 0) or 0),
                            volume=int(item.get("volume", 0)),
                            pict_url=item.get("pict_url", "")
                        )
                        results.append(info)
                    except Exception as e:
                        logger.error(f"解析商品数据失败: {e}")
        
        return results    
    def _get_third_party_price(self, num_iid: str) -> Optional[CouponPriceInfo]:
        """第三方API调用示例（需根据实际平台调整）"""
        try:
            # 示例：使用订单侠等第三方API
            url = f"https://o0b.cn/ibrad/taobao/item_get/"
            params = {
                "key": self.third_party_key,
                "num_iid": num_iid            }
            
            response = self.session_req.get(url, params=params, timeout=10)
            data = response.json()
            
            item = data.get("item", {})
            # 解析第三方返回结构...
            # 实际字段需根据第三方文档调整
            
            return CouponPriceInfo(
                num_iid=num_iid,
                title=item.get("title", ""),
                original_price=float(item.get("original_price", 0)),
                final_price=float(item.get("final_price", 0)),
                coupon_amount=float(item.get("coupon_amount", 0)),
                coupon_start_fee=float(item.get("coupon_start_fee", 0)),
                coupon_info=item.get("coupon_info", ""),
                commission_rate=float(item.get("commission_rate", 0)),
                volume=int(item.get("volume", 0)),
                pict_url=item.get("pic_url", "")
            )
            
        except Exception as e:
            logger.error(f"第三方API调用失败: {e}")
            return None
    
    def parse_taobao_link(self, url: str) -> Optional[Dict]:
        """
        解析淘宝链接/淘口令，获取商品ID和券后价
        需要 taobao.tbk.link.parse 权限
        """
        if not self.use_official:
            logger.warning("解析功能仅支持官方API")
            return None
        
        method = "taobao.tbk.link.parse"
        biz_params = {
            "url": url,
            "fields": "num_iid,title,zk_final_price,coupon_info"
        }
        
        return self._call_official_api(method, biz_params)# ==================== 使用示例 ====================if __name__ == "__main__":
    # 初始化（使用官方API）
    api = TaobaoCouponAPI(
        app_key="your_app_key",
        app_secret="your_app_secret",
        session="your_session",  # 可选
        use_official=True
    )
    
    # 示例1：获取单个商品券后价
    print("=" * 50)
    print("示例1：单个商品查询")
    print("=" * 50)
    
    result = api.get_coupon_price_single("6543217890")
    if result:
        print(f"商品标题: {result.title}")
        print(f"原价: ¥{result.original_price}")
        print(f"券后价: ¥{result.final_price}")
        print(f"优惠券: {result.coupon_info} (面额¥{result.coupon_amount})")
        print(f"佣金比例: {result.commission_rate}%")
        print(f"30天销量: {result.volume}")
        
        # 计算真实优惠
        savings = result.calculate_real_savings()
        print(f"\n💰 优惠分析:")
        print(f"  是否可用券: {'是' if savings['can_use_coupon'] else '否'}")
        print(f"  实际支付: ¥{savings['real_pay_price']}")
        print(f"  实际优惠: ¥{savings['real_savings']}")
        print(f"  折扣力度: {savings['discount_rate']}折")
        print(f"  节省比例: {savings['savings_percentage']}%")
    else:
        print("获取失败")
    
    # 示例2：批量查询
    print("\n" + "=" * 50)
    print("示例2：批量商品查询")
    print("=" * 50)
    
    item_ids = ["123456", "789012", "345678"]
    batch_results = api.get_coupon_price_batch(item_ids)
    
    for item in batch_results:
        print(f"\n{item.title[:20]}...")
        print(f"  券后价: ¥{item.final_price} (省¥{item.coupon_amount})")

{
  "item": {
    "num_iid": "123456",
    "title": "商品标题",
    "original_price": "199.00",
    "final_price": "89.00",        // 券后价
    "coupon_amount": "110.00",     // 优惠券面额
    "coupon_start_fee": "100.00",  // 使用门槛
    "commission_rate": "30.00",    // 佣金比例
    "volume": 10000,               // 销量
    "pic_url": "https://..."
  }}

def calculate_final_price(
    original_price: float,
    zk_final_price: float,
    coupon_amount: float,
    coupon_start_fee: float,
    shop_discount: float = 0,      // 店铺满减
    platform_discount: float = 0,   // 平台券
    is_tmall: bool = False) -> Dict:
    """
    计算最终到手价（考虑多重优惠叠加）
    """
    # 1. 判断优惠券是否可用
    can_use_coupon = original_price >= coupon_start_fee    
    # 2. 计算优惠顺序（通常：店铺券 > 平台券 > 跨店满减）
    price_after_shop = original_price - shop_discount
    price_after_coupon = price_after_shop - (coupon_amount if can_use_coupon else 0)
    final_price = price_after_coupon - platform_discount    
    # 3. 确保价格不为负（淘宝不允许负价）
    final_price = max(final_price, 0.01)
    
    # 4. 校验与zk_final_price的一致性（误差应<0.1元）
    is_consistent = abs(final_price - zk_final_price) < 0.1
    
    return {
        "original_price": original_price,
        "shop_discount": shop_discount,
        "coupon_discount": coupon_amount if can_use_coupon else 0,
        "platform_discount": platform_discount,
        "final_price": round(final_price, 2),
        "total_savings": round(original_price - final_price, 2),
        "savings_rate": round((original_price - final_price) / original_price * 100, 1),
        "is_coupon_used": can_use_coupon,
        "is_price_consistent": is_consistent    }