淘宝评论比价接口实战指南：商品评论 + 价格数据双链路采集与分析

review_id,display_user_nick,is_anonymous,content,score,created,auction_sku,
pic_urls,useful_count,is_append,append_content,append_time,seller_reply,
seller_reply_time,comment_labels

{
    "item_reviews_get_response": {
        "total_results": 1258,
        "reviews": {
            "review": [
                {
                    "review_id": "7295689452365896235",
                    "display_user_nick": "小***柚",
                    "is_anonymous": false,
                    "content": "面料柔软，尺码标准，做工精细，性价比很高",
                    "score": 5,
                    "created": "2026-04-12 09:22:36",
                    "auction_sku": "颜色:黑色;尺码:XL",
                    "useful_count": 36,
                    "pic_urls": ["https://img.alicdn.com/..."],
                    "is_append": true,
                    "append_content": "穿洗三次没缩水，版型不变形，推荐购买",
                    "append_time": "2026-04-25 11:15:22",
                    "seller_reply": {
                        "reply_content": "感谢您细致的评价...",
                        "reply_created": "2026-04-13 15:02:11"
                    },
                    "comment_labels": ["质量好", "尺码合适"]
                }
            ]
        }
    }
}

num_iid,title,nick,price,original_price,pic_url,sell_count,item_location

import hashlib
import time
def generate_top_sign(params: dict, app_secret: str) -> str:
    """
    淘宝开放平台 TOP 签名生成（评论 + 比价接口通用）
    """
    # 1. 过滤空值
    filtered = {k: v for k, v in params.items() 
                if v is not None and v != "" and k != "sign"}
    
    # 2. 按 key ASCII 升序排序
    sorted_items = sorted(filtered.items(), key=lambda x: x[0])
    
    # 3. 拼接签名字符串
    sign_str = app_secret
    for k, v in sorted_items:
        sign_str += f"{k}{v}"
    sign_str += app_secret
    
    # 4. MD5 加密转大写
    return hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()

import requests
import time
import pandas as pd
class TaobaoReviewAPI:
    def __init__(self, app_key, app_secret):
        self.app_key = app_key
        self.app_secret = app_secret
        self.gateway = "https://eco.taobao.com/router/rest"
    
    def _sign(self, params):
        """TOP 签名"""
        filtered = {k: v for k, v in params.items() 
                    if v is not None and v != "" and k != "sign"}
        sorted_kv = sorted(filtered.items(), key=lambda x: x[0])
        raw = self.app_secret + "".join([f"{k}{v}" for k, v in sorted_kv]) + self.app_secret
        return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()
    
    def get_reviews_page(self, num_iid, page_no=1, page_size=20, rate_type="all"):
        """单页评论采集"""
        params = {
            "method": "taobao.item.reviews.get",
            "app_key": self.app_key,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "v": "2.0",
            "format": "json",
            "sign_method": "md5",
            "num_iid": num_iid,
            "page_no": page_no,
            "page_size": page_size,
            "sort": 1,  # 最新在前
            "rate_type": rate_type,
            "has_content": "true",
            "fields": "review_id,display_user_nick,content,score,created,auction_sku,pic_urls,useful_count,is_append,append_content,seller_reply,comment_labels"
        }
        params["sign"] = self._sign(params)
        
        try:
            resp = requests.get(self.gateway, params=params, timeout=15)
            data = resp.json()
            if "error_response" in data:
                print(f"接口错误：{data['error_response']}")
                return []
            reviews = data["item_reviews_get_response"]["reviews"].get("review", [])
            return reviews
        except Exception as e:
            print(f"请求异常：{e}")
            return []
    
    def get_all_reviews(self, num_iid, rate_type="all", max_pages=100):
        """全量分页拉取评论（带流控）"""
        all_reviews = []
        for page in range(1, max_pages + 1):
            reviews = self.get_reviews_page(num_iid, page, 20, rate_type)
            if not reviews:
                break
            all_reviews.extend(reviews)
            print(f"第 {page} 页采集完成，本页 {len(reviews)} 条，累计 {len(all_reviews)} 条")
            time.sleep(0.6)  # QPS 流控，避免被封
        return all_reviews

class TaobaoPriceAPI:
    def __init__(self, app_key, app_secret):
        self.app_key = app_key
        self.app_secret = app_secret
        self.gateway = "https://eco.taobao.com/router/rest"
        self.signer = TaobaoReviewAPI(app_key, app_secret)  # 复用签名逻辑
    
    def get_item_detail(self, num_iid):
        """获取单商品详情（含价格、销量、SKU）"""
        params = {
            "method": "taobao.item.get",
            "app_key": self.app_key,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "v": "2.0",
            "format": "json",
            "sign_method": "md5",
            "num_iid": num_iid,
            "fields": "num_iid,title,price,original_price,num,sold_quantity,pic_url,sku,item_imgs,props_name"
        }
        params["sign"] = self.signer._sign(params)
        
        try:
            resp = requests.get(self.gateway, params=params, timeout=15)
            data = resp.json()
            if "error_response" in data:
                return None
            return data["item_get_response"]["item"]
        except Exception as e:
            print(f"商品详情获取失败：{e}")
            return None
    
    def search_items(self, keyword, page_no=1, page_size=20):
        """关键词搜索商品列表"""
        params = {
            "method": "taobao.items.search",
            "app_key": self.app_key,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "v": "2.0",
            "format": "json",
            "sign_method": "md5",
            "q": keyword,
            "page_no": page_no,
            "page_size": page_size,
            "fields": "num_iid,title,nick,price,original_price,pic_url,sell_count,item_location"
        }
        params["sign"] = self.signer._sign(params)
        
        try:
            resp = requests.get(self.gateway, params=params, timeout=15)
            data = resp.json()
            if "error_response" in data:
                return []
            items = data["items_search_response"]["items"]["item"]
            return items
        except Exception as e:
            print(f"搜索失败：{e}")
            return []

class TaobaoReviewPriceAnalyzer:
    """评论 + 比价双链路分析器"""
    
    def __init__(self, app_key, app_secret):
        self.review_api = TaobaoReviewAPI(app_key, app_secret)
        self.price_api = TaobaoPriceAPI(app_key, app_secret)
    
    def analyze_item(self, num_iid):
        """综合分析单个商品：价格 + 评论口碑"""
        print(f"正在分析商品 {num_iid}...")
        
        # 1. 获取商品详情（价格、销量）
        item = self.price_api.get_item_detail(num_iid)
        if not item:
            return None
        
        price = float(item.get("price", 0))
        original_price = float(item.get("original_price", price))
        sold = int(item.get("sold_quantity", 0))
        title = item.get("title", "")
        
        print(f"商品：{title}")
        print(f"售价：¥{price} | 原价：¥{original_price} | 销量：{sold}")
        
        # 2. 获取评论
        reviews = self.review_api.get_all_reviews(num_iid, max_pages=50)
        if not reviews:
            return {
                "num_iid": num_iid,
                "title": title,
                "price": price,
                "sold": sold,
                "review_count": 0,
                "good_rate": 0
            }
        
        # 3. 评论结构化分析
        df = pd.DataFrame(reviews)
        df["score"] = df["score"].astype(int)
        
        total = len(df)
        good = len(df[df["score"] >= 4])
        neutral = len(df[df["score"] == 3])
        bad = len(df[df["score"] <= 2])
        good_rate = good / total * 100 if total > 0 else 0
        
        # 差评关键词提取
        bad_reviews = df[df["score"] <= 2]["content"].tolist()
        bad_text = " ".join(bad_reviews)
        
        print(f"评论总数：{total} | 好评率：{good_rate:.1f}%")
        print(f"好评：{good} | 中评：{neutral} | 差评：{bad}")
        
        return {
            "num_iid": num_iid,
            "title": title,
            "price": price,
            "original_price": original_price,
            "discount": round((1 - price/original_price)*100, 1) if original_price > 0 else 0,
            "sold": sold,
            "review_count": total,
            "good_rate": round(good_rate, 1),
            "good_count": good,
            "neutral_count": neutral,
            "bad_count": bad,
            "bad_keywords": self._extract_keywords(bad_text, top=10),
            "reviews": reviews
        }
    
    def _extract_keywords(self, text, top=10):
        """简单关键词提取（实际生产环境建议用 jieba）"""
        import re
        from collections import Counter
        
        # 去除标点，分词
        words = re.findall(r'[\u4e00-\u9fa5]{2,}', text)
        stopwords = {"这个", "那个", "还是", "但是", "觉得", "非常", "有点"}
        words = [w for w in words if w not in stopwords and len(w) >= 2]
        return Counter(words).most_common(top)
    
    def compare_items(self, num_iid_list):
        """多商品对比分析"""
        results = []
        for num_iid in num_iid_list:
            result = self.analyze_item(num_iid)
            if result:
                results.append(result)
            time.sleep(1)  # 接口流控
        
        # 输出对比表格
        print("\n" + "="*80)
        print("多商品评论-价格对比分析")
        print("="*80)
        print(f"{'商品ID':<15}{'售价':<10}{'销量':<10}{'评论数':<10}{'好评率':<10}{'性价比评级'}")
        print("-"*80)
        
        for r in results:
            # 简单性价比评级：好评率 / 价格 * 1000
            value_score = r["good_rate"] / r["price"] * 1000 if r["price"] > 0 else 0
            rating = "⭐⭐⭐" if value_score > 50 else ("⭐⭐" if value_score > 30 else "⭐")
            print(f"{r['num_iid']:<15}¥{r['price']:<9}{r['sold']:<10}{r['review_count']:<10}{r['good_rate']:<9.1f}%{rating}")
        
        return results
# ===================== 使用示例 =====================
if __name__ == "__main__":
    APP_KEY = "你的AppKey"
    APP_SECRET = "你的AppSecret"
    
    analyzer = TaobaoReviewPriceAnalyzer(APP_KEY, APP_SECRET)
    
    # 分析单个商品
    result = analyzer.analyze_item("1234567890123")
    
    # 多商品对比（同款不同店）
    # analyzer.compare_items(["1234567890123", "1234567890124", "1234567890125"])