低成本搭建电商商品口碑实时监控系统｜从数据同步到情感分析完整落地（附 Python 可运行代码）

pip install requests pandas jieba wordcloud matplotlib

import requests
import time
import pandas as pd
import jieba
from collections import Counter
import matplotlib.pyplot as plt

# ===================== 基础配置区（自行修改） =====================
# 工具后台获取的凭证
ACCESS_KEY = "你的专属key"
ACCESS_SECRET = "你的专属密钥"
# 目标商品ID
TARGET_ITEM_ID = "1234567890123"
# 单次拉取条数
PAGE_SIZE = 20
# 监控轮询间隔（单位：秒，10分钟=600秒）
MONITOR_INTERVAL = 600
# 存储文件路径
SAVE_PATH = "goods_comments.csv"
# 负面关键词库（用于差评识别预警）
NEG_WORDS = ["质量差", "破损", "掉色", "发货慢", "售后差", "尺寸不符", "假货", "异味"]
# =================================================================

def get_review_batch(item_id, page=1):
    """
    批量获取商品评价数据
    :param item_id: 商品唯一ID
    :param page: 分页页码
    :return: 结构化评价列表
    """
    params = {
        "key": ACCESS_KEY,
        "secret": ACCESS_SECRET,
        "item_id": item_id,
        "page": page,
        "page_size": PAGE_SIZE,
        "result_type": "json"
    }
    # 工具统一访问通道
    resp = requests.get("https://gate.openclaw.cloud/data", params=params, timeout=30)
    res_json = resp.json()
    review_list = []
    if res_json.get("data") and res_json["data"].get("item"):
        for item in res_json["data"]["item"]:
            review_info = {
                "review_id": item.get("rate_id", ""),
                "content": item.get("rate_content", ""),
                "review_time": item.get("rate_date", ""),
                "user_name": item.get("display_user_nick", ""),
                "sku_info": item.get("auction_sku", ""),
                "img_urls": item.get("pics", []),
                "seller_reply": item.get("reply_content", "")
            }
            review_list.append(review_info)
    return review_list

def save_comments_to_csv(new_data):
    """清洗并保存评价数据，自动去重"""
    if len(new_data) == 0:
        return pd.DataFrame([])
    df_new = pd.DataFrame(new_data)
    # 过滤无内容评价
    df_new = df_new[df_new["content"].str.strip() != ""]
    # 处理文件存在/不存在两种场景
    try:
        df_all = pd.read_csv(SAVE_PATH, encoding="utf-8-sig")
        # 去重：根据评价唯一ID
        df_merge = pd.concat([df_all, df_new]).drop_duplicates(subset=["review_id"], keep="last")
        df_merge.to_csv(SAVE_PATH, index=False, encoding="utf-8-sig")
        print(f"本次新增有效评价：{len(df_merge)-len(df_all)} 条，累计数据：{len(df_merge)} 条")
        return df_merge
    except FileNotFoundError:
        df_new.to_csv(SAVE_PATH, index=False, encoding="utf-8-sig")
        print(f"首次创建数据文件，共写入 {len(df_new)} 条评价")
        return df_new

def analyze_comment_insight(file_path):
    """评论数据分析：正负向统计、负面痛点词提取"""
    df = pd.read_csv(file_path, encoding="utf-8-sig")
    if len(df) == 0:
        print("暂无评价数据，无法分析")
        return
    print("===== 商品口碑整体分析报告 =====")
    print(f"总有效评价数量：{len(df)}")
    # 标记负面评价
    df["is_negative"] = df["content"].apply(lambda x: 1 if any(w in x for w in NEG_WORDS) else 0)
    neg_count = df["is_negative"].sum()
    pos_count = len(df) - neg_count
    print(f"正向评价：{pos_count} 条，负面评价：{neg_count} 条")
    print(f"负面占比：{round(neg_count/len(df)*100,2)}%")
    # 提取差评文本分词统计高频痛点
    neg_text = "".join(df[df["is_negative"] == 1]["content"].tolist())
    words = jieba.lcut(neg_text)
    # 过滤无意义单字、助词
    filter_words = ["的", "了", "很", "有点", "一点", "还是", "但是", "觉得"]
    word_clean = [w for w in words if len(w)>=2 and w not in filter_words]
    word_count = Counter(word_clean)
    print("\n差评高频痛点TOP15：")
    for word, cnt in word_count.most_common(15):
        print(f"{word}：{cnt}次")
    # 简单绘制正负评价分布饼图
    plt.rcParams["font.sans-serif"] = ["SimHei"]
    plt.rcParams["axes.unicode_minus"] = False
    plt.pie([pos_count, neg_count], labels=["好评/中性", "差评"], autopct="%1.2f%%", colors=["#66cc66", "#ff6666"])
    plt.title("商品评价情感分布")
    plt.show()
    return df

def monitor_goods_review():
    """持续监控商品新增评价，定时轮询执行"""
    print(f"已启动商品口碑监控，商品ID：{TARGET_ITEM_ID}，每{MONITOR_INTERVAL/60}分钟同步一次数据")
    while True:
        all_new_reviews = []
        page = 1
        # 分批拉取最新评价（仅抓取第一页最新数据，如需全量同步可放开循环）
        batch = get_review_batch(TARGET_ITEM_ID, page)
        all_new_reviews.extend(batch)
        # 保存并清洗数据
        df_data = save_comments_to_csv(all_new_reviews)
        # 差评预警提示
        new_neg = df_data[df_data["is_negative"] == 1] if "is_negative" in df_data.columns else pd.DataFrame([])
        if len(new_neg) > 0:
            print("\n⚠️ 检测到新增差评，请及时处理：")
            for idx, row in new_neg.iterrows():
                print(f"【{row['review_time']}】用户{row['user_name']}：{row['content']}")
        print(f"\n等待{MONITOR_INTERVAL/60}分钟后执行下一轮同步...\n")
        time.sleep(MONITOR_INTERVAL)

if __name__ == "__main__":
    # 首次执行：拉取全量历史评价（循环分页）
    print("开始同步商品全部历史评价...")
    full_review_data = []
    current_page = 1
    while True:
        page_data = get_review_batch(TARGET_ITEM_ID, current_page)
        if len(page_data) == 0:
            break
        full_review_data.extend(page_data)
        print(f"已同步第{current_page}页，当前累计{len(full_review_data)}条")
        current_page += 1
        time.sleep(2)
    save_comments_to_csv(full_review_data)
    # 执行一次完整数据分析
    analyze_comment_insight(SAVE_PATH)
    # 启动持续实时监控
    monitor_goods_review()