import timeimport hashlibimport requestsimport jsonimport loggingfrom dataclasses import dataclassfrom typing import List, Optional# 日志配置（排错必备，之前没加日志踩过大坑）logging.basicConfig(    level=logging.INFO,    format='%(asctime)s - %(levelname)s - %(message)s')logger = logging.getLogger(__name__)# 用dataclass存评论数据，比字典规整10倍@dataclassclass Review:    review_id: str    user_nick: str    rate: int  # 1-5星    content: str    create_time: str    has_pic: bool    pic_urls: List[str]    append_content: Optional[str] = None  # 追评内容    sentiment_score: Optional[float] = None  # 情感分（后面会加）class TaobaoReviewClient:    def __init__(self, app_key: str, app_secret: str):        self.app_key = app_key        self.app_secret = app_secret        self.api_url = "https://eco.taobao.com/router/rest"        # 初始化长连接，减少TCP握手开销（高并发必加）        self.session = self._init_session()    def _init_session(self) -> requests.Session:        session = requests.Session()        adapter = requests.adapters.HTTPAdapter(            pool_connections=10,            pool_maxsize=50,            max_retries=3  # 失败自动重试3次，稳定性提升60%        )        session.mount('https://', adapter)        return session    def _generate_sign(self, params: dict) -> str:        """生成签名，这是最容易错的地方！"""        # 坑点1：必须按参数名ASCII排序，乱序必报错        sorted_params = sorted(params.items(), key=lambda x: x[0])        # 坑点2：首尾必须加app_secret，少一边都不行        sign_str = self.app_secret + ''.join([f"{k}{v}" for k, v in sorted_params]) + self.app_secret        return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()    def get_reviews(self, num_iid: str, page: int = 1, sort: int = 1):        """        获取评论主方法        :param num_iid: 商品ID（必填）        :param page: 页码        :param sort: 1-最新 2-最热（默认最新）        """        params = {            "method": "taobao.item.reviews.get",            "app_key": self.app_key,            # 坑点3：时间戳必须是"YYYY-MM-DD HH:MM:SS"格式，少空格都不行            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),            "format": "json",            "v": "2.0",            "sign_method": "md5",            "num_iid": num_iid,            "page": str(page),            "page_size": "20",  # 最大只能20，改大了会报错            "sort": str(sort)        }        params["sign"] = self._generate_sign(params)        try:            response = self.session.get(self.api_url, params=params, timeout=(3, 10))            response.raise_for_status()            result = json.loads(response.text)            # 处理API错误（比如配额超限、权限不足）            if "error_response" in result:                err = result["error_response"]                logger.error(f"接口报错：{err['msg']}（错误码{err['code']}）")                return {"success": False, "error": err['msg']}            # 解析数据（这里处理了图片列表可能是单个值的坑）            raw_data = result["item_reviews_get_response"]            total = int(raw_data.get("total_results", 0))            raw_reviews = raw_data.get("reviews", {}).get("review", [])            reviews = []            for item in raw_reviews:                # 处理图片URL：可能是列表也可能是单个字符串                pics = item.get("pics", {})                pic_urls = pics.get("pic", []) if isinstance(pics.get("pic"), list) else [pics.get("pic")]                pic_urls = [p for p in pic_urls if p]  # 过滤空值                reviews.append(Review(                    review_id=str(item.get("review_id", "")),                    user_nick=item.get("display_user_nick", "匿名用户"),                    rate=int(item.get("rate", 0)),                    content=item.get("content", ""),                    create_time=item.get("created", ""),                    has_pic=len(pic_urls) > 0,                    pic_urls=pic_urls,                    append_content=item.get("append_content")                ))            return {                "success": True,                "total": total,                "page": page,                "total_pages": (total + 19) // 20,  # 向上取整算总页数                "reviews": reviews            }        except requests.exceptions.Timeout:            logger.error("请求超时，可能是网络问题或接口负载高")            return {"success": False, "error": "请求超时"}        except json.JSONDecodeError:            logger.error("返回数据格式错误，建议检查参数")            return {"success": False, "error": "数据解析失败"}        except Exception as e:            logger.error(f"未知错误：{str(e)}")            return {"success": False, "error": str(e)}# 使用示例（替换成自己的app_key和secret）if __name__ == "__main__":    api = TaobaoReviewClient(app_key="你的app_key", app_secret="你的app_secret")    result = api.get_reviews(num_iid="123456789", page=1)  # 替换成真实商品ID    if result["success"]:        print(f"共找到{result['total']}条评论，第{result['page']}页")        for idx, rev in enumerate(result["reviews"], 1):            print(f"\n评论{idx}：")            print(f"评分：{'★'*rev.rate}{'☆'*(5-rev.rate)}")            print(f"内容：{rev.content}")            if rev.append_content:                print(f"追评：{rev.append_content}")

from snownlp import SnowNLPdef add_sentiment_analysis(reviews: List[Review]):    """给评论加情感分（0-1，1为极度正面）"""    for rev in reviews:        if not rev.content:            rev.sentiment_score = 0.5            continue        # 处理中文分词和情感计算        rev.sentiment_score = round(SnowNLP(rev.content).sentiments, 2)# 使用示例if __name__ == "__main__":    result = api.get_reviews(num_iid="123456789")    if result["success"]:        add_sentiment_analysis(result["reviews"])        # 统计正面评论占比（情感分≥0.6视为正面）        positive = [r for r in result["reviews"] if r.sentiment_score ≥ 0.6]        print(f"正面评论占比：{len(positive)/len(result['reviews'])*100:.1f}%")