京东商品评论接口（jingdong.ware.comment.get）技术解析：数据拉取与情感分析优化

import time
import hashlib
import requests
import json
import logging
from dataclasses import dataclass
from typing import List, Optional, Dict
# 日志配置（技术排错必备，记录请求参数与错误信息）
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# 评论数据结构化存储类（避免字典key混乱）
@dataclass
class JdReview:
    review_id: str
    user_nick: str
    rate: int
    content: str
    create_time: str
    properties: Dict[str, str]
    has_pic: bool
    pic_urls: List[str]
    reply_content: Optional[str] = None
    append_content: Optional[str] = None
    sentiment_score: Optional[float] = None
class JdReviewClient:
    def __init__(self, app_key: str, app_secret: str):
        self.app_key = app_key
        self.app_secret = app_secret
        self.api_url = "https://api.jd.com/routerjson"
        self.session = self._init_session()  # 长连接优化（减少TCP握手开销）
    def _init_session(self) -> requests.Session:
        """初始化请求会话，配置重试策略（应对网络波动）"""
        session = requests.Session()
        adapter = requests.adapters.HTTPAdapter(
            pool_connections=10,
            pool_maxsize=50,
            max_retries=3  # 临时网络错误自动重试（3次上限）
        )
        session.mount('https://', adapter)
        return session
    def _generate_sign(self, params: dict) -> str:
        """
        生成MD5签名（京东接口核心技术卡点）
        避坑点1：参数按ASCII码升序排序（而非字母顺序）
        避坑点2：首尾拼接app_secret（缺一不可）
        """
        # 按参数名ASCII升序排序
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        # 拼接参数字符串（无分隔符）
        param_str = ''.join([f"{k}{v}" for k, v in sorted_params])
        # 首尾加app_secret后MD5加密（转大写）
        sign_str = f"{self.app_secret}{param_str}{self.app_secret}"
        return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
    def get_reviews(self, sku_id: str, page: int = 1, page_size: int = 10, sort_type: int = 5) -> dict:
        """
        评论数据拉取主方法
        避坑点3：请求方式为POST（GET将返回405 Method Not Allowed）
        避坑点4：pageSize≤20（超过将触发参数错误）
        避坑点5：timestamp格式严格匹配（少空格/秒数错误均导致签名失败）
        """
        # 基础参数配置
        params = {
            "method": "jingdong.ware.comment.get",
            "app_key": self.app_key,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),  # 严格格式
            "format": "json",
            "v": "2.0",
            "sign_method": "md5",
            "skuId": sku_id,
            "page": str(page),
            "pageSize": str(min(page_size, 20)),  # 强制限制pageSize≤20
            "sortType": str(sort_type)
        }
        # 生成签名
        params["sign"] = self._generate_sign(params)
        try:
            # 发送POST请求（参数放data而非params）
            response = self.session.post(
                url=self.api_url,
                data=params,
                timeout=(3, 10)  # 连接超时3秒，读取超时10秒
            )
            response.raise_for_status()  # 捕获4xx/5xx HTTP错误
            result = json.loads(response.text)
            # 处理接口业务错误
            if result.get("code") != 0:
                err_msg = f"{result.get('message', '未知错误')}（错误码：{result.get('code')}）"
                logger.error(f"接口业务错误：{err_msg}")
                return {"success": False, "error": err_msg}
            # 解析返回数据
            return self._parse_review_data(result.get("data", {}))
        except requests.exceptions.Timeout:
            logger.error("请求超时（可能为接口高峰期负载过高）")
            return {"success": False, "error": "请求超时"}
        except json.JSONDecodeError:
            logger.error("返回数据非JSON格式（检查参数是否正确）")
            return {"success": False, "error": "数据解析失败"}
        except Exception as e:
            logger.error(f"未知错误：{str(e)}")
            return {"success": False, "error": str(e)}
    def _parse_review_data(self, raw_data: dict) -> dict:
        """
        评论数据解析（核心技术点：字段兼容性处理）
        避坑点6：properties可能为null/非列表（需类型判断）
        避坑点7：pictures含空URL（需过滤）
        避坑点8：replies可能为空列表（需判断长度）
        """
        total = int(raw_data.get("total", 0))
        raw_reviews = raw_data.get("comments", [])
        parsed_reviews = []
        for item in raw_reviews:
            # 处理结构化标签（兼容null/非列表格式）
            props = item.get("properties", [])
            properties = {}
            if isinstance(props, list):
                properties = {p["name"]: p["value"] for p in props if "name" in p and "value" in p}
            # 处理图片URL（过滤空值）
            pic_urls = item.get("pictures", [])
            pic_urls = [url for url in pic_urls if isinstance(url, str) and url.strip()]
            has_pic = len(pic_urls) > 0
            # 处理商家回复（取第一条有效回复）
            reply_content = None
            replies = item.get("replies", [])
            if isinstance(replies, list) and len(replies) > 0:
                reply_content = replies[0].get("content")
            # 处理追评内容
            append_content = item.get("afterSaleReview", {}).get("content") if "afterSaleReview" in item else None
            # 构造结构化评论对象
            parsed_reviews.append(JdReview(
                review_id=str(item.get("id", "")),
                user_nick=item.get("userNick", "匿名用户"),
                rate=int(item.get("rating", 0)),
                content=item.get("content", ""),
                create_time=item.get("creationTime", ""),
                properties=properties,
                has_pic=has_pic,
                pic_urls=pic_urls,
                reply_content=reply_content,
                append_content=append_content
            ))
        return {
            "success": True,
            "total_count": total,
            "current_page": int(raw_data.get("page", 1)),
            "total_pages": (total + int(raw_data.get("pageSize", 10)) - 1) // int(raw_data.get("pageSize", 10)),
            "reviews": parsed_reviews
        }

from snownlp import SnowNLP
def optimize_sentiment_analysis(reviews: List[JdReview]) -> None:
    """
    情感分析优化（基于结构化标签提升准确率）
    技术逻辑：利用properties字段的星级标签，调整基础情感分
    避坑点9：标签星级提取需处理“5星”“四星”等不同表述（统一提取数字）
    """
    for review in reviews:
        if not review.content.strip():
            review.sentiment_score = 0.5
            continue
        # 基础情感分（基于评论正文）
        base_score = SnowNLP(review.content).sentiments
        # 结合结构化标签调整分数
        for tag_name, tag_value in review.properties.items():
            # 提取星级数字（兼容“5星”“四星”“3个星”等格式）
            star_match = re.search(r'(\d+)', tag_value)
            if star_match:
                star = int(star_match.group(1))
                # 星级≥4加权重，≤2减权重（避免极端值影响）
                if star >= 4:
                    base_score = min(1.0, base_score + 0.1)
                elif star <= 2:
                    base_score = max(0.0, base_score - 0.15)
        # 结果保留2位小数
        review.sentiment_score = round(base_score, 2)