1688 平台的 item_review 接口是获取商品采购商评价数据的核心接口，专注于 B2B 场景下的交易评价信息。与面向消费者的电商平台不同，1688 的评论更侧重于产品质量、供应商服务、物流速度等批发采购相关维度，对供应商评估和采购决策具有重要参考价值。
一、接口核心特性分析
1. 接口功能与定位
2. 认证机制
3. 核心参数与响应结构
请求参数
响应核心字段
二、Python 脚本实现
import requests
import time
import json
import logging
import re
from typing import Dict, Optional, List
from requests.exceptions import RequestException
from snownlp import SnowNLP # 用于情感分析，需安装：pip install snownlp
# 配置日志
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s"
)
class Alibaba1688ItemReviewAPI:
def __init__(self, appkey: str, appsecret: str):
"""
初始化1688评论API客户端
:param appkey: 1688开放平台appkey
:param appsecret: 1688开放平台appsecret
"""
self.appkey = appkey
self.appsecret = appsecret
self.base_url = "https://gw.open.1688.com/openapi"
self.access_token = None
self.token_expires_at = 0 # token过期时间戳
self.session = requests.Session()
self.session.headers.update({
"Content-Type": "application/x-www-form-urlencoded",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36"
})
def _get_access_token(self) -> Optional[str]:
"""获取访问令牌"""
# 检查token是否有效
if self.access_token and self.token_expires_at > time.time() + 60:
return self.access_token
logging.info("获取新的access_token")
params = {
"method": "alibaba.oauth2.getToken",
"client_id": self.appkey,
"client_secret": self.appsecret,
"grant_type": "client_credentials",
"format": "json"
}
try:
response = self.session.get(f"{self.base_url}/gateway.do", params=params, timeout=10)
response.raise_for_status()
result = response.json()
if "error_response" in result:
logging.error(f"获取access_token失败: {result['error_response']['msg']} (错误码: {result['error_response']['code']})")
return None
self.access_token = result["access_token"]
self.token_expires_at = time.time() + result.get("expires_in", 86400) # 默认为24小时
return self.access_token
except RequestException as e:
logging.error(f"获取access_token请求异常: {str(e)}")
return None
def get_item_reviews(self,
offer_id: str,
page: int = 1,
page_size: int = 20,
sort: str = "newest",
is_retail: bool = False,
has_image: bool = False) -> Optional[Dict]:
"""
获取商品评论
:param offer_id: 商品ID（1688中称为offer_id）
:param page: 页码
:param page_size: 每页条数
:param sort: 排序方式
:param is_retail: 是否只看零售单评价
:param has_image: 是否只看有图评价
:return: 评论数据
"""
# 验证参数
valid_sorts = ["newest", "helpful", "high_rating"]
if sort not in valid_sorts:
logging.error(f"无效的排序方式: {sort}，支持: {valid_sorts}")
return None
if page_size < 1 or page_size > 50:
logging.error(f"每页条数必须在1-50之间，当前为: {page_size}")
return None
# 获取有效的access_token
if not self._get_access_token():
return None
params = {
"method": "alibaba.item.review.get",
"client_id": self.appkey,
"access_token": self.access_token,
"offer_id": offer_id,
"page": page,
"page_size": page_size,
"sort": sort,
"is_retail": "true" if is_retail else "false",
"has_image": "true" if has_image else "false",
"format": "json",
"v": "1.0",
"timestamp": time.strftime("%Y-%m-%d %H:%M:%S")
}
try:
response = self.session.get(f"{self.base_url}/gateway.do", params=params, timeout=15)
response.raise_for_status()
result = response.json()
if "error_response" in result:
logging.error(f"获取评论失败: {result['error_response']['msg']} (错误码: {result['error_response']['code']})")
return None
review_response = result.get("alibaba_item_review_get_response", {})
reviews_data = review_response.get("result", {})
if not reviews_data:
logging.warning("未获取到评论数据")
return None
# 格式化评论数据
return self._format_review_data(reviews_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, review_data: Dict) -> Dict:
"""格式化评论数据"""
# 分页信息
pagination = {
"total_reviews": int(review_data.get("total_count", 0)),
"total_pages": (int(review_data.get("total_count", 0)) + int(review_data.get("page_size", 20)) - 1) // int(review_data.get("page_size", 20)),
"current_page": int(review_data.get("current_page", 1)),
"page_size": int(review_data.get("page_size", 20))
}
# 格式化评论列表
reviews = []
for review in review_data.get("reviews", []):
# 处理评价内容（去除HTML标签）
content = self._clean_text(review.get("content", ""))
# 情感分析（0-1之间，越接近1越积极）
sentiment_score = self._analyze_sentiment(content)
sentiment = "positive" if sentiment_score > 0.6 else "negative" if sentiment_score < 0.4 else "neutral"
# 处理评价图片
images = []
if review.get("images"):
images = [img.get("url") for img in review.get("images") if img.get("url")]
# 处理追评（1688中可能包含长期使用反馈）
append_comment = None
if review.get("append_comment"):
append_comment = {
"content": self._clean_text(review["append_comment"].get("content", "")),
"created": review["append_comment"].get("gmt_create"),
"days_after_purchase": review["append_comment"].get("days_after_purchase") # 购买后多少天追评
}
# 处理交易信息（B2B特有的采购信息）
trade_info = {
"purchase_quantity": int(review.get("purchase_quantity", 0)), # 采购数量
"purchase_amount": float(review.get("purchase_amount", 0)), # 采购金额
"is_repeat_buyer": review.get("is_repeat_buyer", False), # 是否回头客
"cooperation_count": int(review.get("cooperation_count", 1)), # 合作次数
"purchase_time": review.get("purchase_time") # 采购时间
}
reviews.append({
"review_id": review.get("review_id"),
"buyer": {
"user_id": review.get("buyer_user_id"),
"nickname": review.get("buyer_nick"),
"level": review.get("buyer_level"), # 采购商等级
"region": review.get("buyer_region") # 采购商地区
},
"rating": {
"product_quality": int(review.get("product_quality_rating", 0)), # 产品质量评分
"seller_service": int(review.get("seller_service_rating", 0)), # 卖家服务评分
"logistics_speed": int(review.get("logistics_speed_rating", 0)), # 物流速度评分
"match_description": int(review.get("match_description_rating", 0)) # 与描述相符度
},
"content": content,
"created_time": review.get("gmt_create"),
"images": images,
"append_comment": append_comment,
"useful_count": int(review.get("useful_count", 0)), # 有用数
"tags": review.get("tags", "").split(","), # 评价标签
"trade_info": trade_info,
"is_retail": review.get("is_retail", False), # 是否零售单
"sentiment": {
"score": round(sentiment_score, 4),
"label": sentiment
}
})
return {
"pagination": pagination,
"reviews": reviews,
"raw_data": review_data # 保留原始数据
}
def _clean_text(self, text: str) -> str:
"""清理文本，去除HTML标签和特殊字符"""
if not text:
return ""
# 去除HTML标签
clean = re.sub(r'<.*?>', '', text)
# 去除多余空格和换行
clean = re.sub(r'\s+', ' ', clean).strip()
# 去除特殊字符
clean = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9,.?!，。？！]', ' ', clean)
return clean
def _analyze_sentiment(self, text: str) -> float:
"""使用SnowNLP进行情感分析"""
if not text:
return 0.5 # 中性
try:
return SnowNLP(text).sentiments
except:
return 0.5 # 分析失败时返回中性
def get_all_reviews(self, offer_id: str, max_pages: int = 10, has_image: bool = False) -> List[Dict]:
"""
获取多页评论数据
:param offer_id: 商品ID
:param max_pages: 最大页数限制
:param has_image: 是否只看有图评价
:return: 所有评论列表
"""
all_reviews = []
page = 1
while page <= max_pages:
logging.info(f"获取第 {page} 页评论")
result = self.get_item_reviews(
offer_id=offer_id,
page=page,
page_size=50, # 使用最大页大小减少请求次数
sort="newest",
has_image=has_image
)
if not result or not result["reviews"]:
break
all_reviews.extend(result["reviews"])
# 检查是否已到最后一页
if page >= result["pagination"]["total_pages"]:
break
page += 1
# 控制请求频率，遵守1688 API的QPS限制
time.sleep(2)
return all_reviews
def analyze_b2b_reviews(self, reviews: List[Dict]) -> Dict:
"""分析B2B评论数据，生成针对批发采购场景的统计报告"""
if not reviews:
return {}
total = len(reviews)
sentiment_counts = {"positive": 0, "neutral": 0, "negative": 0}
rating_stats = {
"product_quality": [],
"seller_service": [],
"logistics_speed": [],
"match_description": []
}
tag_counts = {}
has_image_count = 0
repeat_buyer_count = 0
retail_order_count = 0
total_purchase_quantity = 0
total_purchase_amount = 0
# B2B场景关键词（与批发采购相关）
b2b_keywords = {
"quality": ["质量", "材质", "做工", "品质", "用料"],
"price": ["价格", "性价比", "便宜", "贵", "优惠"],
"service": ["服务", "态度", "响应", "沟通", "售后"],
"logistics": ["物流", "快递", "运输", "速度", "包装"],
"moq": ["起订量", "批量", "数量", "MOQ"],
"delivery": ["发货", "交期", "工期", "准时"]
}
keyword_counts = {k:0 for k in b2b_keywords}
# 统计基础数据
for review in reviews:
# 情感统计
sentiment = review["sentiment"]["label"]
sentiment_counts[sentiment] += 1
# 评分统计
for key in rating_stats:
if key in review["rating"]:
rating_stats[key].append(review["rating"][key])
# 标签统计
for tag in review["tags"]:
if tag:
tag_counts[tag] = tag_counts.get(tag, 0) + 1
# 有图评价统计
if review["images"]:
has_image_count += 1
# 回头客统计（B2B重要指标）
if review["trade_info"]["is_repeat_buyer"]:
repeat_buyer_count += 1
# 零售单统计
if review["is_retail"]:
retail_order_count += 1
# 采购量和采购金额统计
total_purchase_quantity += review["trade_info"]["purchase_quantity"]
total_purchase_amount += review["trade_info"]["purchase_amount"]
# B2B关键词统计
content = review["content"].lower()
for category, kws in b2b_keywords.items():
for kw in kws:
if kw in content:
keyword_counts[category] += 1
break # 每个类别只计数一次
# 计算平均评分
avg_ratings = {}
for key, values in rating_stats.items():
if values:
avg_ratings[key] = round(sum(values) / len(values), 1)
else:
avg_ratings[key] = 0
# 获取热门标签（前10）
top_tags = sorted(tag_counts.items(), key=lambda x: x[1], reverse=True)[:10]
# 计算平均采购量和采购金额
avg_purchase = {
"quantity": round(total_purchase_quantity / total, 1) if total > 0 else 0,
"amount": round(total_purchase_amount / total, 2) if total > 0 else 0
}
return {
"total_reviews": total,
"sentiment_distribution": {
"count": sentiment_counts,
"percentage": {
k: round(v / total * 100, 1) for k, v in sentiment_counts.items()
}
},
"average_rating": avg_ratings,
"image_review_ratio": round(has_image_count / total * 100, 1) if total > 0 else 0,
"repeat_buyer_ratio": round(repeat_buyer_count / total * 100, 1) if total > 0 else 0, # 回头客比例（B2B重要指标）
"retail_order_ratio": round(retail_order_count / total * 100, 1) if total > 0 else 0, # 零售单比例
"avg_purchase": avg_purchase, # 平均采购量和金额
"total_purchase": {
"quantity": total_purchase_quantity,
"amount": total_purchase_amount
},
"top_tags": top_tags,
"b2b_keyword_distribution": keyword_counts
}
# 示例调用
if __name__ == "__main__":
# 替换为实际的appkey和appsecret（从1688开放平台获取）
APPKEY = "your_appkey"
APPSECRET = "your_appsecret"
# 替换为目标商品offer_id
OFFER_ID = "61234567890"
# 初始化API客户端
api = Alibaba1688ItemReviewAPI(APPKEY, APPSECRET)
# 方式1：获取单页评论
# review_result = api.get_item_reviews(
# offer_id=OFFER_ID,
# page=1,
# page_size=20,
# sort="newest",
# has_image=False
# )
# 方式2：获取多页评论
review_result = api.get_all_reviews(
offer_id=OFFER_ID,
max_pages=3,
has_image=False
)
if isinstance(review_result, dict) and "reviews" in review_result:
print(f"共获取到 {review_result['pagination']['total_reviews']} 条评论")
print(f"当前第 {review_result['pagination']['current_page']}/{review_result['pagination']['total_pages']} 页\n")
# 打印前3条评论
for i, review in enumerate(review_result["reviews"][:3], 1):
print(f"{i}. 采购商: {review['buyer']['nickname']} ({review['buyer']['region'] or '未知地区'})")
print(f" 采购信息: {review['trade_info']['purchase_quantity']}件, {review['trade_info']['purchase_amount']}元, {'回头客' if review['trade_info']['is_repeat_buyer'] else '新客户'}")
print(f" 评分: 质量 {review['rating']['product_quality']}, 服务 {review['rating']['seller_service']}, 物流 {review['rating']['logistics_speed']}")
print(f" 时间: {review['created_time']}")
print(f" 内容: {review['content'][:100]}{'...' if len(review['content'])>100 else ''}")
print(f" 情感: {review['sentiment']['label']} (得分: {review['sentiment']['score']})")
if review['images']:
print(f" 图片数: {len(review['images'])}")
if review['tags']:
print(f" 标签: {', '.join(review['tags'])}")
if review['append_comment']:
print(f" 追评({review['append_comment']['days_after_purchase']}天后): {review['append_comment']['content'][:50]}{'...' if len(review['append_comment']['content'])>50 else ''}")
print("-" * 100)
# 分析评论
analysis = api.analyze_b2b_reviews(review_result["reviews"])
print("\n=== B2B评论分析报告 ===")
print(f"总评论数: {analysis['total_reviews']}")
print(f"情感分布: 正面 {analysis['sentiment_distribution']['percentage']['positive']}%, 中性 {analysis['sentiment_distribution']['percentage']['neutral']}%, 负面 {analysis['sentiment_distribution']['percentage']['negative']}%")
print(f"平均评分: 质量 {analysis['average_rating']['product_quality']}, 服务 {analysis['average_rating']['seller_service']}, 物流 {analysis['average_rating']['logistics_speed']}")
print(f"回头客比例: {analysis['repeat_buyer_ratio']}%")
print(f"平均采购量: {analysis['avg_purchase']['quantity']}件, 平均采购金额: {analysis['avg_purchase']['amount']}元")
print("热门标签:")
for tag, count in analysis['top_tags']:
print(f" {tag}: {count}次")
print("B2B关键维度提及次数:")
for category, count in analysis['b2b_keyword_distribution'].items():
print(f" {category}: {count}次")
elif isinstance(review_result, list):
# 处理多页评论结果
print(f"共获取到 {len(review_result)} 条评论")
# 分析评论
analysis = api.analyze_b2b_reviews(review_result)
print("\n=== B2B评论分析报告 ===")
print(f"总评论数: {analysis['total_reviews']}")
print(f"情感分布: 正面 {analysis['sentiment_distribution']['percentage']['positive']}%, 中性 {analysis['sentiment_distribution']['percentage']['neutral']}%, 负面 {analysis['sentiment_distribution']['percentage']['negative']}%")
print(f"回头客比例: {analysis['repeat_buyer_ratio']}%")