京东商品评论接口深度开发实战：从情感分析到商业决策的全链路技术落地

import hashlib
import time
import requests
from aliyunsdkcore.client import AcsClient
from aliyunsdkkms.request.v20160120 import GetSecretValueRequest
import json
from typing import Dict, List, Tuple
# 阿里云KMS客户端初始化（用于安全获取京东接口密钥，避免明文存储）
def init_aliyun_kms_client(region_id: str, access_key_id: str, access_key_secret: str) -> AcsClient:
    """
    初始化阿里云KMS客户端（适配阿里云服务器环境）
    :param region_id: 阿里云地域ID（如「cn-hangzhou」）
    :return: 阿里云KMS客户端实例
    """
    return AcsClient(access_key_id, access_key_secret, region_id)
def get_jd_secret_from_kms(kms_client: AcsClient, secret_name: str) -> Tuple[str, str]:
    """
    从阿里云KMS获取京东接口密钥（合规存储方案，避免明文泄露）
    :param secret_name: KMS中存储的密钥名称（如「jd-union-api-secret」）
    :return: appkey, app_secret
    """
    request = GetSecretValueRequest.GetSecretValueRequest()
    request.set_SecretName(secret_name)
    response = kms_client.do_action_with_exception(request)
    secret_data = json.loads(response.decode('utf-8'))
    secret_json = json.loads(secret_data['SecretData'])
    return secret_json['appkey'], secret_json['app_secret']
def generate_jd_sign(params: Dict, app_secret: str) -> str:
    """
    京东官方签名算法实现（严格遵循京东开放平台文档，避免签名错误）
    原理：1. 参数按key升序排序；2. 拼接「app_secret+key1value1key2value2+app_secret」；3. MD5加密并转大写
    """
    # 1. 按参数名ASCII升序排序（官方强制要求，否则签名失败）
    sorted_params = sorted(params.items(), key=lambda x: x[0])
    # 2. 拼接签名字符串（忽略空值参数）
    sign_str = app_secret
    for key, value in sorted_params:
        if value is not None and str(value).strip() != "":
            sign_str += f"{key}{value}"
    sign_str += app_secret
    # 3. MD5加密并转为大写（官方指定加密方式）
    return hashlib.md5(sign_str.encode('utf-8')).hexdigest().upper()
def get_jd_comment_params(appkey: str, access_token: str, sku_id: str, page: int, page_size: int) -> Dict:
    """
    构造京东评论接口请求参数（严格按官方字段规范，无冗余参数）
    :param sku_id: 京东商品SKU ID（如「100012345678」）
    :param page: 页码（官方限制≤100）
    :param page_size: 每页条数（官方限制1-100）
    """
    # 官方必填参数（参考京东开放平台文档v3.0）
    params = {
        "method": "jd.union.open.comment.query",
        "app_key": appkey,
        "access_token": access_token,
        "timestamp": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
        "format": "json",
        "v": "1.0",
        "sign_method": "md5",
        # 接口业务参数（需转为JSON字符串，官方要求）
        "360buy_param_json": json.dumps({
            "skuId": sku_id,
            "pageIndex": page,
            "pageSize": page_size,
            "score": 0,  # 0=全部评分，1-5=指定评分（官方字段）
            "sortType": 2  # 2=按有用数排序（官方推荐，数据价值更高）
        })
    }
    return params

def call_jd_comment_api(kms_client: AcsClient, kms_secret_name: str, access_token: str, 
                        sku_id: str, page: int, page_size: int) -> Tuple[List[Dict], int]:
    """
    京东评论接口合规调用（含官方错误码处理+阿里云环境适配）
    :return: 评论数据列表, 总评论数
    """
    # 1. 从阿里云KMS获取密钥（安全合规）
    appkey, app_secret = get_jd_secret_from_kms(kms_client, kms_secret_name)
    # 2. 构造请求参数并生成签名
    params = get_jd_comment_params(appkey, access_token, sku_id, page, page_size)
    params["sign"] = generate_jd_sign(params, app_secret)
    # 3. 阿里云环境适配：使用requests会话池（减少连接开销，适配云服务器网络）
    session = requests.Session()
    session.mount('https://', requests.adapters.HTTPAdapter(pool_connections=5, pool_maxsize=10))
    try:
        # 官方接口网关（唯一合法地址，避免使用非官方域名）
        response = session.get(
            url="https://api.jd.com/routerjson",
            params=params,
            timeout=(10, 30),  # 阿里云网络环境适配：长超时应对跨地域请求
            headers={"User-Agent": "Aliyun-Server-Java/1.0"}  # 标识阿里云服务器环境
        )
        response.raise_for_status()  # 触发HTTP错误（如403、500）
        result = response.json()
        # 4. 官方错误码处理（参考京东开放平台《错误码规范》）
        if "error_response" in result:
            error = result["error_response"]
            error_code = error.get("code")
            error_msg = error.get("msg")
            # 常见错误码的官方解决方案
            error_solution = {
                "1001": "密钥无效：检查appkey/app_secret是否正确，IP白名单是否配置",
                "2003": "QPS超限：降低调用频率（基础版≤5次/秒，进阶版≤10次/秒）",
                "3002": "权限不足：申请「jd.union.open.comment.query」进阶版权限",
                "4001": "参数错误：检查360buy_param_json格式是否正确，skuId是否存在"
            }
            raise Exception(f"京东接口错误[{error_code}]：{error_msg} → 解决方案：{error_solution.get(str(error_code), '参考京东官方文档')}")
        # 5. 解析官方返回数据（严格按官方字段结构，避免字段错误）
        data = result.get("jd_union_open_comment_query_response", {}).get("result", {})
        return data.get("comments", []), data.get("totalCount", 0)
    except requests.exceptions.RequestException as e:
        raise Exception(f"网络请求异常（阿里云环境）：{str(e)} → 检查服务器安全组是否开放443端口")
    except Exception as e:
        raise Exception(f"接口调用失败：{str(e)}")
    finally:
        session.close()

from concurrent.futures import ThreadPoolExecutor, as_completed
import redis
def init_aliyun_redis(redis_addr: str, redis_password: str, db: int) -> redis.Redis:
    """
    初始化阿里云Redis（用于缓存高频请求数据，减少接口调用量）
    :param redis_addr: 阿里云Redis实例地址（如「r-xxx.redis.rds.aliyuncs.com:6379」）
    """
    return redis.Redis(host=redis_addr.split(':')[0], port=int(redis_addr.split(':')[1]),
                       password=redis_password, db=db, decode_responses=True)
def batch_get_jd_comments(kms_client: AcsClient, kms_secret_name: str, access_token: str,
                          sku_ids: List[str], max_workers: int = 3, redis_client: redis.Redis = None) -> Dict:
    """
    批量获取多SKU评论（阿里云ECS并发优化：基于线程池+Redis缓存）
    :param max_workers: 线程数（阿里云2核4G ECS建议≤5，避免CPU过载）
    :param redis_client: 阿里云Redis实例（缓存7天内的评论数据，减少重复调用）
    """
    result = {}
    # 线程池适配阿里云ECS性能：2核4G实例建议max_workers=3，4核8G建议≤5
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {}
        for sku_id in sku_ids:
            # 先查Redis缓存（命中则直接返回，未命中则调用接口）
            if redis_client and redis_client.exists(f"jd_comment:{sku_id}"):
                result[sku_id] = json.loads(redis_client.get(f"jd_comment:{sku_id}"))
                continue
            # 提交线程任务（适配阿里云ECS的CPU调度）
            futures[executor.submit(
                call_jd_comment_api, kms_client, kms_secret_name, access_token, sku_id, 1, 100
            )] = sku_id
        # 处理线程结果
        for future in as_completed(futures):
            sku_id = futures[future]
            try:
                comments, total_count = future.result()
                result[sku_id] = {"comments": comments, "total_count": total_count}
                # 写入Redis缓存（设置7天过期，适配阿里云Redis的内存管理）
                if redis_client:
                    redis_client.setex(f"jd_comment:{sku_id}", 60*60*24*7, json.dumps(result[sku_id]))
            except Exception as e:
                result[sku_id] = {"error": str(e)}
    return result

if __name__ == "__main__":
    # 1. 阿里云资源配置（需替换为实际环境参数）
    ALIYUN_REGION = "cn-hangzhou"  # 阿里云地域
    ALIYUN_ACCESS_KEY = "LTAIxxxxxx"  # 阿里云AccessKey（建议使用RAM子账号，最小权限）
    ALIYUN_ACCESS_SECRET = "xxxxxx"
    KMS_SECRET_NAME = "jd-union-api-secret"  # 阿里云KMS中存储的京东密钥名称
    REDIS_ADDR = "r-xxxx.redis.rds.aliyuncs.com:6379"  # 阿里云Redis地址
    REDIS_PASSWORD = "xxxxxx"
    JD_ACCESS_TOKEN = "xxxxxx"  # 京东开放平台获取的access_token（有效期30天）
    TARGET_SKUS = ["100012345678", "100012345679"]  # 目标SKU列表
    try:
        # 2. 初始化阿里云资源
        kms_client = init_aliyun_kms_client(ALIYUN_REGION, ALIYUN_ACCESS_KEY, ALIYUN_ACCESS_SECRET)
        redis_client = init_aliyun_redis(REDIS_ADDR, REDIS_PASSWORD, db=0)
        # 3. 批量获取评论数据（阿里云环境适配）
        comment_result = batch_get_jd_comments(
            kms_client=kms_client,
            kms_secret_name=KMS_SECRET_NAME,
            access_token=JD_ACCESS_TOKEN,
            sku_ids=TARGET_SKUS,
            max_workers=3,  # 2核4G ECS适配
            redis_client=redis_client
        )
        # 4. 输出结果（阿里云日志服务适配：建议使用logging而非print）
        for sku_id, data in comment_result.items():
            if "error" in data:
                print(f"SKU {sku_id} 处理失败：{data['error']}")
            else:
                print(f"SKU {sku_id} 总评论数：{data['total_count']}，获取评论数：{len(data['comments'])}")
    except Exception as e:
        print(f"实战调用失败：{str(e)}")