1688 的 item_review 接口（官方标准方法名：alibaba.item.review.get）是阿里巴巴开放平台提供的核心接口之一，用于获取指定商品的买家评论数据，包括文本评价、评分、晒图、采购属性（如采购数量、用途）等信息。该接口是分析商品质量、供应商信誉、买家需求的关键数据来源，广泛应用于供应链风控、竞品分析、商品优化等场景。
一、接口核心特性深度分析
- 接口定位与核心价值
功能本质：从 1688 商品详情页的 “买家评价” 模块提取结构化数据，支持按时间、评分等维度筛选，解决手动爬取评论的效率与合规问题；
核心价值：
供应链风控：通过差评分析商品质量缺陷（如 “材质不符”“尺寸偏差”）；
买家需求挖掘：从评论中提取高频关键词（如 “需要定制 LOGO”“批量包装”）；
供应商信誉评估：统计好评率、重复采购评论占比，判断供应商服务稳定性；
数据范围：仅返回 1688 平台公开的买家评论（匿名评论会隐藏部分信息），不包含未公开的私下沟通记录。 - 认证与请求机制
遵循 1688 开放平台统一的 “appkey + 签名” 认证体系，无需用户授权（session非必填），核心流程：
参数准备：公共参数（app_key、method、timestamp等）+ 业务参数（商品 ID、评论筛选条件）；
签名生成：
所有参数按参数名 ASCII 升序排序；
拼接为key=value&key=value格式字符串；
末尾拼接appsecret，通过 MD5 加密生成 32 位大写签名（sign）；
请求方式：仅支持GET（1688 开放平台多数基础数据接口均为GET）；
网关地址：固定为 https://gw.open.1688.com/openapi/http/1/system.oauth2（1688 开放平台统一网关）。 - 核心参数与响应结构
（1）关键请求参数
参数名 类型 是否必填 说明
method String 是 接口方法名，固定为 alibaba.item.review.get
app_key String 是 1688 开放平台应用 ID（需申请 “商品评论读取” 权限）
timestamp String 是 时间戳，格式 yyyy-MM-dd HH:mm:ss（误差≤5 分钟）
format String 否 响应格式，默认 json，可选 xml
v String 是 接口版本，固定为 2.0
sign_method String 是 签名方式，固定为 md5
product_id String 是 商品 ID（1688 商品详情页 URL 中 id= 后的数值，如 694521898765）
page Int 否 页码，默认 1，最大支持 100 页（超过返回空数据）
page_size Int 否 每页评论数，默认 20，最大支持 50
review_type String 否 评论类型：all（全部，默认）、positive（好评）、negative（差评）
sort String 否 排序方式：create_time_desc（时间倒序，默认）、create_time_asc（时间正序）
（2）响应核心字段
响应以 alibaba_item_review_get_response 为根节点，核心数据分为 3 层：
分页元数据：total_results（总评论数）、page（当前页）、page_size（每页数量）；
评论列表（reviews 数组）：每个评论包含：
基础信息：review_id（评论 ID）、buyer_nick（买家昵称，匿名会脱敏）、create_time（评论时间）；
评分信息：star_level（星级，1-5 星）、star_desc（评分描述，如 “质量很好”）；
内容信息：content（评论文本）、images（晒图 URL 列表，无晒图为空数组）；
采购属性：purchase_quantity（采购数量）、purchase_unit（单位，如 “件”）、usage（用途，如 “生产使用”）；
统计信息：positive_rate（好评率，如 0.92 代表 92%）、positive_count（好评数）、negative_count（差评数）。
二、Python 脚本实现（含评论分析功能）
以下实现 item_review 接口的完整调用逻辑，支持评论获取、数据格式化、基础分析（好评率计算、关键词提取），需先在 1688 开放平台申请 appkey 和 appsecret 并开通接口权限。
import requests import hashlib import time import json import logging import re from collections import Counter from typing import Dict, Optional, List, Tuple from requests.exceptions import RequestException from wordcloud import WordCloud # 需额外安装：pip install wordcloud
配置日志
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s"
)
停用词列表（过滤无意义词汇）
STOP_WORDS = set ([
"的", "了", "是", "我", "在", "有", "和", "就", "不", "人", "都", "一", "一个", "上", "也",
"很", "到", "说", "要", "去", "你", "会", "着", "没有", "看", "好", "自己", "这", "件", "个"
])
class AlibabaItemReviewAPI:
def init(self, appkey: str, appsecret: str):
"""
初始化 1688 item_review 接口客户端
:param appkey: 1688 开放平台 appkey
:param appsecret: 1688 开放平台 appsecret
"""
self.appkey = appkey
self.appsecret = appsecret
self.base_url = "https://gw.open.1688.com/openapi/http/1/system.oauth2"
self.session = requests.Session()
def _generate_sign (self, params: Dict) -> str:
"""生成 1688 API 签名（MD5 大写）""" - 按参数名 ASCII 升序排序
sorted_params = sorted(params.items(), key=lambda x: x[0]) - 拼接为 key=value&key=value 格式
sign_str = "&".join([f"{k}={v}" for k, v in sorted_params]) - 末尾拼接 appsecret 并 MD5 加密
sign_str += self.appsecret
return hashlib.md5(sign_str.encode("utf-8")).hexdigest().upper()
def _get_timestamp (self) -> str:
"""生成符合 1688 规范的时间戳（yyyy-MM-dd HH:mm:ss）"""
return time.strftime ("% Y-% m-% d % H:% M:% S", time.localtime ())
def get_item_reviews (self,
product_id: str,
page: int = 1,
page_size: int = 20,
review_type: str = "all",
sort: str = "create_time_desc") -> Optional [Dict]:
"""
获取商品评论数据
:param product_id: 1688 商品 ID
:param page: 页码（1~100）
:param page_size: 每页评论数（1~50）
:param review_type: 评论类型：all/positive/negative
:param sort: 排序方式：create_time_desc/create_time_asc
:return: 格式化后的评论数据（含分页、统计、评论列表）
""" - 参数校验
if review_type not in ["all", "positive", "negative"]:
logging.error (f"无效评论类型：{review_type}，仅支持 all/positive/negative")
return None
if page < 1 or page > 100:
logging.error (f"页码超出范围：{page}，仅支持 1~100")
return None
if page_size < 1 or page_size > 50:
logging.error (f"每页数量超出范围：{page_size}，仅支持 1~50")
return None - 构造请求参数
params = {
"method": "alibaba.item.review.get",
"app_key": self.appkey,
"timestamp": self._get_timestamp(),
"format": "json",
"v": "2.0",
"sign_method": "md5",
"product_id": product_id,
"page": str(page),
"page_size": str(page_size),
"review_type": review_type,
"sort": sort
} - 生成签名
params["sign"] = self._generate_sign(params)
try: - 发送 GET 请求
response = self.session.get (
self.base_url,
params=params,
timeout=15
)
response.raise_for_status () # 抛出 HTTP 错误（如 401、429）
result = response.json () - 处理错误响应
if "error_response" in result:
error = result ["error_response"]
logging.error (
f"接口错误：{error.get ('msg', ' 未知错误 ')}（错误码：{error.get ('code', ' 未知 ')}，子码：{error.get ('sub_code', ' 无 ')}）"
)
return None - 提取并格式化数据
raw_response = result.get ("alibaba_item_review_get_response", {})
raw_data = raw_response.get ("result", {})
if not raw_data:
logging.warning ("未获取到评论数据")
return None
return self._format_review_data(raw_data)
except RequestException as e:
logging.error (f"请求异常：{str (e)}")
return None
except json.JSONDecodeError:
logging.error (f"响应解析失败：{response.text [:200]}...")
return None
def _format_review_data (self, raw_data: Dict) -> Dict:
"""格式化评论数据，提取核心字段并清洗""" - 格式化分页信息
pagination = {
"total_reviews": int(raw_data.get("total_results", 0)),
"current_page": int(raw_data.get("page", 1)),
"page_size": int(raw_data.get("page_size", 20)),
"total_pages": (int(raw_data.get("total_results", 0)) + int(raw_data.get("page_size", 20)) - 1) // int(raw_data.get("page_size", 20))
} - 格式化统计信息
stats = {
"positive_rate": round (float (raw_data.get ("positive_rate", 0)), 4), # 好评率（保留 4 位小数）
"positive_count": int (raw_data.get ("positive_count", 0)), # 好评数
"negative_count": int (raw_data.get ("negative_count", 0)), # 差评数
"has_image_ratio": round (
int (raw_data.get ("has_image_count", 0)) /max (int (raw_data.get ("total_results", 1)), 1),
4
) if raw_data.get ("total_results") else 0 # 带图评论占比
} - 格式化评论列表（清洗文本、提取关键信息）
formatted_reviews = []
for raw_review in raw_data.get("reviews", {}).get("review", []):
清洗评论文本（去除 HTML 标签、多余空格）
content = re.sub(r"<[^>]+>", "", raw_review.get("content", "")).strip()
处理晒图（提取 URL 列表）
images = raw_review.get ("images", {}).get ("image_url", [])
if isinstance (images, str): # 兼容单张图片的情况（返回字符串而非数组）
images = [images] if images else []
formatted_reviews.append ({
"review_id": raw_review.get ("review_id"),
"buyer_nick": raw_review.get ("buyer_nick", "匿名买家"), # 脱敏昵称
"create_time": raw_review.get ("create_time"),
"star_level": int (raw_review.get ("star_level", 0)), # 1-5 星
"star_desc": raw_review.get ("star_desc", ""),
"content": content,
"images": images,
"purchase_info": {
"quantity": int (raw_review.get ("purchase_quantity", 0)),
"unit": raw_review.get ("purchase_unit", "件"),
"usage": raw_review.get ("usage", "未填写") # 用途（如 “生产使用”“批发零售”）
}
})
return {
"product_id": raw_data.get("product_id"),
"pagination": pagination,
"stats": stats,
"reviews": formatted_reviews
}
def get_all_reviews (self, product_id: str, review_type: str = "all", max_pages: int = 10) -> Optional [Dict]:
"""
获取商品所有评论（自动分页）
:param product_id: 商品 ID
:param review_type: 评论类型
:param max_pages: 最大页数（防止过度请求，建议≤20）
:return: 合并后的所有评论数据
"""
all_reviews = []
current_page = 1
total_reviews = 0
total_pages = 1
while current_page <= max_pages and current_page <= total_pages:
logging.info(f"获取商品 {product_id} 第 {current_page} 页评论（类型：{review_type}）")
page_result = self.get_item_reviews (
product_id=product_id,
page=current_page,
page_size=50, # 用最大页大小减少请求次数
review_type=review_type,
sort="create_time_desc"
)
if not page_result:
break
初始化总页数（第一页后更新）
if current_page == 1:
total_reviews = page_result["pagination"]["total_reviews"]
total_pages = page_result["pagination"]["total_pages"]
logging.info(f"商品 {product_id} 共 {total_reviews} 条评论，共 {total_pages} 页")
追加当前页评论
all_reviews.extend(page_result["reviews"])
current_page += 1
控制请求频率（避免触发 QPS 限制）
time.sleep(1.5)
重新计算统计信息（合并后）
if all_reviews:
positive_count = sum (1 for r in all_reviews if r ["star_level"] >= 4)
negative_count = sum (1 for r in all_reviews if r ["star_level"] <= 2)
has_image_count = sum (1 for r in all_reviews if len (r ["images"]) > 0)
return {
"product_id": product_id,
"pagination": {
"total_reviews": len (all_reviews),
"total_pages": current_page - 1,
"page_size": 50,
"current_page": current_page - 1
},
"stats": {
"positive_rate": round (positive_count /len (all_reviews), 4),
"positive_count": positive_count,
"negative_count": negative_count,
"has_image_ratio": round (has_image_count /len (all_reviews), 4)
},
"reviews": all_reviews
}
else:
logging.warning (f"未获取到商品 {product_id} 的任何评论")
return None
def analyze_reviews (self, reviews: List [Dict]) -> Dict:
"""
评论基础分析（关键词提取、评分分布、采购用途统计）
:param reviews: 格式化后的评论列表
:return: 分析结果
"""
if not reviews:
return {"
