淘宝买家秀 API 深度开发：多模态内容解析与合规推荐技术拆解

import timeimport hashlibimport requestsfrom datetime import datetime, timedeltaclass TaobaoReviewAPI:    def __init__(self, app_key: str, app_secret: str, access_token: str):        self.app_key = app_key        self.app_secret = app_secret        self.access_token = access_token        self.api_url = "https://eco.taobao.com/router/rest"        self.last_call = datetime.min  # 控制调用频率        def _generate_sign(self, params: dict) -> str:        """生成合规签名，处理特殊字符编码"""        # 过滤空值并按ASCII排序        valid_params = {k: v for k, v in params.items() if v is not None}        sorted_params = sorted(valid_params.items(), key=lambda x: x[0])        # 拼接签名字符串（含app_secret首尾包裹）        sign_str = self.app_secret        for k, v in sorted_params:            value_str = str(v).encode('utf-8', errors='replace').decode('utf-8')            sign_str += f"{k}{value_str}"        sign_str += self.app_secret        # MD5加密并转大写        return hashlib.md5(sign_str.encode()).hexdigest().upper()        def get_reviews(self, num_iid: str, has_image: bool = False, page: int = 1) -> dict:        """合规获取买家秀数据，控制调用频率"""        # 限流控制：确保QPS≤1        interval = (datetime.now() - self.last_call).total_seconds()        if interval < 1.5:            time.sleep(1.5 - interval)                params = {            "method": "taobao.reviews.get",            "app_key": self.app_key,            "access_token": self.access_token,            "num_iid": num_iid,            "has_image": has_image,            "page": page,            "page_size": 20,            "fields": "review_id,content,images,rating,snick,created",            "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),            "format": "json",            "v": "2.0"        }        # 生成签名        params["sign"] = self._generate_sign(params)        resp = requests.get(self.api_url, params=params, timeout=15)        self.last_call = datetime.now()        return resp.json()

{  "reviews_get_response": {    "reviews": {      "review": [        {          "review_id": "R1234567890",          "content": "👍面料很舒服，颜色和图片一样，尺码正！😘",          "images": "https://img.alicdn.com/imgextra/i1/xxx.jpg,https://img.alicdn.com/imgextra/i2/xxx.jpg",          "rating": "5",          "snick": "tb12345678",          "created": "2025-09-30 14:32:15"        }      ]    },    "total_results": 120,    "request_id": "xxx"  }}

from dataclasses import dataclassfrom typing import List, Optional, Dictfrom datetime import date@dataclassclass StandardReview:    """买家秀标准化数据模型"""    review_id: str  # 唯一标识    product_id: str  # 关联商品ID    content: str  # 清洗后文本    content_length: int  # 文本长度（过滤短评）    images: List[str]  # 有效图片URL列表    image_count: int  # 图片数量    rating: int  # 评分（1-5）    sentiment_score: float  # 情感得分（0-1，1为正面）    keywords: List[str]  # 核心关键词（Top3）    user_nick: str  # 脱敏后昵称    create_date: date  # 发布日期    is_valid: bool  # 是否为有效评价    @classmethod    def from_raw(cls, raw_review: dict, product_id: str) -> "StandardReview":        """从原始数据转换为标准化模型"""        # 1. 文本清洗与情感分析        clean_content = cls._clean_text(raw_review["content"])        sentiment = cls._analyze_sentiment(clean_content)        keywords = cls._extract_keywords(clean_content)                # 2. 图片URL处理        image_list = raw_review.get("images", "").split(",") if raw_review.get("images") else []        valid_images = [img for img in image_list if cls._check_image_valid(img)]                # 3. 脱敏与格式统一       脱敏昵称 = cls._desensitize_nick(raw_review["snick"])        create_date = datetime.strptime(raw_review["created"], "%Y-%m-%d %H:%M:%S").date()                # 4. 有效性判断（文本≥10字且有意义）        is_valid = len(clean_content) >= 10 and sentiment != 0.5                return cls(            review_id=raw_review["review_id"],            product_id=product_id,            content=clean_content,            content_length=len(clean_content),            images=valid_images,            image_count=len(valid_images),            rating=int(raw_review["rating"]),            sentiment_score=round(sentiment, 2),            keywords=keywords,            user_nick=脱敏昵称,            create_date=create_date,            is_valid=is_valid        )        @staticmethod    def _clean_text(text: str) -> str:        """清洗文本：去除表情、URL、特殊符号"""        import re        # 移除表情符号（Unicode范围）        text = re.sub(r'[\U00010000-\U0010ffff]', '', text)        # 移除URL与特殊字符        text = re.sub(r'http[s]?://\S+', '', text)        text = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9,.，。！!？? ]', '', text)        return re.sub(r'\s+', ' ', text).strip()        @staticmethod    def _analyze_sentiment(text: str) -> float:        """情感分析：基于SnowNLP实现"""        from snownlp import SnowNLP        if not text:            return 0.5  # 中性        return SnowNLP(text).sentiments        @staticmethod    def _extract_keywords(text: str) -> List[str]:        """提取关键词：基于TF-IDF算法"""        import jieba.analyse        if len(text) < 10:            return []        # 提取Top3关键词（过滤停用词）        return jieba.analyse.extract_tags(text, topK=3, stop_words=True)        @staticmethod    def _check_image_valid(url: str) -> bool:        """校验图片URL有效性（轻量HEAD请求）"""        try:            resp = requests.head(url, timeout=3, allow_redirects=True)            return resp.status_code == 200 and "image" in resp.headers.get("Content-Type", "")        except:            return False        @staticmethod    def _desensitize_nick(nick: str) -> str:        """用户昵称脱敏：如tb12345678→tb12****78"""        if len(nick) <= 4:            return nick[:2] + "**"        return nick[:4] + "****" + nick[-2:]

def review_standard_pipeline(api: TaobaoReviewAPI, product_id: str) -> List[StandardReview]:    """买家秀数据标准化流水线"""    standard_reviews = []    page = 1    while True:        # 调用接口获取原始数据        raw_data = api.get_reviews(num_iid=product_id, has_image=True, page=page)        reviews = raw_data.get("reviews_get_response", {}).get("reviews", {}).get("review", [])        if not reviews:            break                # 转换为标准化数据        for raw_review in reviews:            std_review = StandardReview.from_raw(raw_review, product_id)            if std_review.is_valid:                standard_reviews.append(std_review)                # 超过5页停止（有效评价占比骤降）        if page >= 5:            break        page += 1    return standard_reviews

import numpy as npfrom sklearn.feature_extraction.text import TfidfVectorizerfrom sklearn.metrics.pairwise import cosine_similarityclass ReviewRecommendation:    def __init__(self, standard_reviews: List[StandardReview]):        self.reviews = {r.review_id: r for r in standard_reviews if r.is_valid}        self.tfidf = self._build_tfidf_vectorizer()        self.feature_matrix = self._build_feature_matrix()        def _build_tfidf_vectorizer(self) -> TfidfVectorizer:        """构建TF-IDF向量器（用于关键词匹配）"""        all_keywords = []        for review in self.reviews.values():            all_keywords.extend(review.keywords)        # 训练向量器        tfidf = TfidfVectorizer(vocabulary=list(set(all_keywords)))        tfidf.fit([" ".join(review.keywords) for review in self.reviews.values()])        return tfidf        def _build_feature_matrix(self) -> dict:        """构建买家秀特征矩阵"""        feature_matrix = {}        for review_id, review in self.reviews.items():            # 1. 文本特征：关键词TF-IDF向量（取均值）            keyword_vec = self.tfidf.transform([" ".join(review.keywords)]).toarray()[0]            keyword_mean = np.mean(keyword_vec) if len(keyword_vec) > 0 else 0                        # 2. 数值特征：标准化处理            rating_norm = review.rating / 5  # 评分归一化（0-1）            image_norm = min(review.image_count / 5, 1.0)  # 图片数上限5张            sentiment_norm = review.sentiment_score                        # 3. 时间特征：时效性权重（近30天为1，超过90天为0.3）            days_since_post = (date.today() - review.create_date).days            time_weight = max(1 - (days_since_post / 90), 0.3)                        # 合并特征向量            feature_matrix[review_id] = np.array([                keyword_mean, rating_norm, image_norm, sentiment_norm, time_weight            ])        return feature_matrix        def get_detail_page_recommendations(self, product_keywords: List[str], top_n: int = 6) -> List[dict]:        """详情页精选推荐：匹配商品核心关键词"""        # 生成商品关键词向量        product_vec = self.tfidf.transform([" ".join(product_keywords)]).toarray()[0]        product_mean = np.mean(product_vec) if len(product_vec) > 0 else 0                # 计算综合得分（关键词匹配+基础权重）        review_scores = []        for review_id, feat_vec in self.feature_matrix.items():            # 关键词匹配度（占比40%）            keyword_match = min(feat_vec[0] / (product_mean + 0.01), 1.0) * 0.4            # 基础权重（评分20%+图片20%+情感10%+时效10%）            base_score = (feat_vec[1]*0.2 + feat_vec[2]*0.2 + feat_vec[3]*0.1 + feat_vec[4]*0.1)            total_score = keyword_match + base_score            review_scores.append((review_id, total_score))                # 按得分排序取TopN        sorted_reviews = sorted(review_scores, key=lambda x: x[1], reverse=True)[:top_n]        return [            {                "review_id": rid,                "content": self.reviews[rid].content,                "images": self.reviews[rid].images,                "rating": self.reviews[rid].rating,                "score": round(score, 2)            } for rid, score in sorted_reviews        ]

def get_similar_product_reviews(self, target_product_attrs: dict, top_n: int = 4) -> List[dict]:    """相似商品买家秀推荐：基于商品属性相似度"""    # 目标商品属性（类目、价格带、核心关键词）    target_category = target_product_attrs["category"]    target_price_range = target_product_attrs["price_range"]    target_keywords = target_product_attrs["keywords"]        # 筛选同类目、同价格带的买家秀    candidate_reviews = [        r for r in self.reviews.values()        if r.product_attrs["category"] == target_category        and r.product_attrs["price_range"] == target_price_range    ]        # 计算关键词相似度    target_vec = self.tfidf.transform([" ".join(target_keywords)]).toarray()[0]    similar_scores = []    for review in candidate_reviews:        review_vec = self.tfidf.transform([" ".join(review.keywords)]).toarray()[0]        sim = cosine_similarity(target_vec.reshape(1, -1), review_vec.reshape(1, -1))[0][0]        similar_scores.append((review.review_id, sim))        # 取相似度TopN    sorted_similar = sorted(similar_scores, key=lambda x: x[1], reverse=True)[:top_n]    return [        {            "review_id": rid,            "product_id": self.reviews[rid].product_id,            "content": self.reviews[rid].content,            "similarity": round(sim, 2)        } for rid, sim in sorted_similar    ]

┌─────────────┐  热点商品买家秀  ┌─────────────┐│  本地缓存   │◄─────────────►│  Redis缓存  ││（内存，1h） │               │（24小时）   │└─────────────┘               └─────────────┘

import redisfrom functools import lru_cacheclass ReviewCache:    def __init__(self):        self.redis_client = redis.Redis(host="localhost", port=6379, db=1, decode_responses=True)        self.local_cache = lru_cache(maxsize=50)  # 本地缓存热点商品        def get_cached_reviews(self, product_id: str, is_hot: bool = False) -> Optional[List[dict]]:        """从缓存获取买家秀数据"""        # 热点商品查本地缓存        if is_hot:            try:                return self.local_cache[product_id]            except KeyError:                pass                # 非热点商品查Redis        cache_key = f"review:standard:{product_id}"        cached_data = self.redis_client.get(cache_key)        if not cached_data:            return None                # 反序列化（实际项目建议用JSON）        import json        return json.loads(cached_data)        def set_cached_reviews(self, product_id: str, reviews: List[StandardReview], is_hot: bool = False):        """写入缓存"""        # 转换为可序列化格式        review_dict_list = [            {                "review_id": r.review_id,                "content": r.content,                "images": r.images,                "rating": r.rating,                "sentiment_score": r.sentiment_score,                "keywords": r.keywords            } for r in reviews        ]        import json        serialized = json.dumps(review_dict_list)                # 热点商品写入本地缓存        if is_hot:            self.local_cache[product_id] = review_dict_list                # 写入Redis（24小时过期）        cache_key = f"review:standard:{product_id}"        self.redis_client.setex(cache_key, 86400, serialized)

def smart_retry(api_call_func, max_retries: int = 3) -> dict:    """智能重试：基于错误类型调整重试策略"""    retries = 0    while retries < max_retries:        try:            result = api_call_func()            # 正常返回或非限流错误，直接返回            if result.get("error_response") is None:                return result            err_code = result["error_response"]["code"]            if err_code != 27:  # 非限流错误（如权限问题）                return result            # 限流错误：指数退避重试            retry_interval = 2 ** retries  # 1s→2s→4s            time.sleep(retry_interval)            retries += 1        except Exception as e:            print(f"调用异常：{str(e)}")            retries += 1            time.sleep(1)    return {"error": "达到最大重试次数"}