30 分钟搭建电商商品口碑实时监控系统，自动化采集 + 情感分析全流程实战（附完整可运行 Python 代码）

前言

做电商运营、产品调研、竞品分析的同学，几乎都有同一个痛点：想要持续掌握商品真实用户反馈，却只能手动刷新页面复制评论，效率极低。大批量竞品、多店铺商品同时跟踪时，人工统计完全跟不上节奏；想及时捕捉新增差评、用户吐槽痛点，往往滞后几天，错失产品优化、舆情处理最佳时机。

传统页面抓取方案存在大量短板：频繁触发平台访问限制、需要长期维护登录凭证、页面改版后代码直接失效、批量采集稳定性差。今天分享一套轻量化成熟方案，借助 OpenClaw 一站式数据工具，无需逆向页面、不用处理复杂反爬规则，快速搭建全量评论自动采集、定时增量监控、文本情感挖掘、痛点可视化完整体系，新手零基础也能落地。

整套方案覆盖三大核心能力：

批量拉取商品历史全部评价，包含主评、追评、晒图、买家规格、评分、发布时间等完整信息；
定时轮询增量抓取新增评论，自动去重，新增差评支持消息告警；
本地完成文本清洗、情感正负向分类、高频关键词统计，输出可视化口碑报表。

一、整体搭建流程概述

整套系统分为 4 个环节，普通电脑 / 轻量云服务器即可稳定运行，无需分布式部署：

工具后台初始化：创建数据监控项目，生成专属访问凭证，配置采集频次、批量商品清单；
数据拉取脚本：通过 Python 程序读取商品评价数据，自动清洗无效内容、存储本地文件；
长效监控定时任务：循环轮询抓取当日新增评价，实时捕捉负面反馈；
智能数据分析模块：分词、情感打分、词云生成、评分分布统计，输出分析结果。

二、工具后台基础配置步骤

登录管理后台，新建电商数据采集项目，系统自动生成唯一访问令牌与项目标识；
批量录入需要监控的商品 ID（自有店铺 + 竞品商品分开分组管理，方便后续分类统计）；
自定义采集规则：设置单次最大抓取条数、轮询间隔时长，内置平台访问限流策略，无需手动调整延时；
开启增量同步模式：系统自动记录已抓取评论唯一标识，后续仅同步新增评价，大幅减少重复数据与资源消耗。

配置完成后，后台会生成专属数据访问地址，下文代码中仅需替换凭证参数即可直接运行，不用额外调试页面解析逻辑。

三、完整 Python 实操代码（分三大模块）

模块 1：商品评论批量采集与数据存储脚本

实现一次性拉取商品全量历史评价，过滤系统默认无意义好评，自动去重保存至本地 CSV 文件。

import requests
import pandas as pd
import time
import re

# ===================== 基础配置区（替换为自己后台生成的凭证） =====================
TOKEN = "你的专属访问令牌"
PROJECT_ID = "项目编号"
TARGET_GOODS_ID = "待监控商品ID"
SAVE_FILE = "goods_comment_data.csv"
# ==============================================================================

def get_all_comments(goods_id, page=1):
    """拉取单页商品评价数据"""
    params = {
        "token": TOKEN,
        "project": PROJECT_ID,
        "goods_id": goods_id,
        "page": page,
        "sort": 1  # 1=按最新评价排序，0=综合排序
    }
    try:
        resp = requests.get("https://task-data.clawopen.com/query", params=params, timeout=20)
        result = resp.json()
        if result.get("code") != 200:
            print(f"数据请求异常：{result.get('msg')}")
            return None
        return result.get("data", {})
    except Exception as e:
        print(f"页面{page}采集失败，错误信息：{str(e)}")
        time.sleep(3)
        return None

def parse_comment_data(raw_data):
    """结构化解析原始评价数据，提取核心字段"""
    comment_list = []
    item_arr = raw_data.get("item", [])
    for item in item_arr:
        # 合并主评+追评完整文本
        main_text = item.get("rate_content", "")
        append_text = item.get("add_feedback", "")
        full_content = f"{main_text} {append_text}".strip()
        # 过滤无内容评价
        if len(full_content) < 2:
            continue
        info = {
            "comment_id": item.get("rate_id"),
            "goods_id": TARGET_GOODS_ID,
            "user_name": item.get("display_user_nick"),
            "comment_text": full_content,
            "spec": item.get("auction_sku", ""),
            "publish_time": item.get("rate_date"),
            "has_img": 1 if item.get("pics") else 0,
            "has_video": 1 if item.get("video") else 0,
            "useful_num": item.get("useful_count", 0)
        }
        comment_list.append(info)
    return comment_list, raw_data.get("total_results", 0)

def batch_fetch_all_comments():
    """分页循环抓取全部历史评价"""
    all_data = []
    page = 1
    while True:
        raw = get_all_comments(TARGET_GOODS_ID, page)
        if not raw:
            break
        page_data, total = parse_comment_data(raw)
        if not page_data:
            break
        all_data.extend(page_data)
        print(f"已抓取第{page}页，当前累计{len(all_data)}条评价，商品总评价数：{total}")
        # 到达最后一页终止循环
        if len(all_data) >= int(total):
            break
        page += 1
        time.sleep(2)
    # 数据去重，基于评论唯一ID过滤重复内容
    df = pd.DataFrame(all_data)
    df.drop_duplicates(subset=["comment_id"], keep="first", inplace=True)
    # 写入本地文件，不存在则新建，存在则追加
    try:
        old_df = pd.read_csv(SAVE_FILE, encoding="utf-8-sig")
        df = pd.concat([old_df, df], ignore_index=True)
        df.drop_duplicates(subset=["comment_id"], keep="first", inplace=True)
    except FileNotFoundError:
        pass
    df.to_csv(SAVE_FILE, index=False, encoding="utf-8-sig")
    print(f"全量采集完成，本地文件共存储{len(df)}条有效评价")

if __name__ == "__main__":
    batch_fetch_all_comments()

模块 2：定时增量监控脚本（实时捕捉新增差评）

程序后台常驻运行，每小时自动抓取当日新增评价，识别负面内容后打印告警，可自行扩展企业微信 / 邮件推送功能。

import time
import datetime
import pandas as pd

# 复用模块1配置
TOKEN = "你的专属访问令牌"
PROJECT_ID = "项目编号"
TARGET_GOODS_ID = "待监控商品ID"
SAVE_FILE = "goods_comment_data.csv"
MONITOR_INTERVAL = 3600  # 监控轮询间隔，单位秒，此处1小时

def check_new_comments():
    """单次增量抓取新增评价并检测负面内容"""
    from module1 import get_all_comments, parse_comment_data
    raw = get_all_comments(TARGET_GOODS_ID, page=1)
    if not raw:
        return
    new_data, _ = parse_comment_data(raw)
    if not new_data:
        return
    # 读取历史数据，对比筛选全新评价
    try:
        history_df = pd.read_csv(SAVE_FILE, encoding="utf-8-sig")
        history_ids = set(history_df["comment_id"].tolist())
    except FileNotFoundError:
        history_ids = set()
    new_valid = [row for row in new_data if row["comment_id"] not in history_ids]
    if len(new_valid) == 0:
        print(f"{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')} 暂无新增评价")
        return
    # 新增评价入库
    new_df = pd.DataFrame(new_valid)
    try:
        old_df = pd.read_csv(SAVE_FILE, encoding="utf-8-sig")
        total_df = pd.concat([old_df, new_df], ignore_index=True)
    except FileNotFoundError:
        total_df = new_df
    total_df.to_csv(SAVE_FILE, index=False, encoding="utf-8-sig")
    print(f"检测到{len(new_valid)}条新增评价，已更新本地数据")
    # 简易负面关键词预警（可自行扩充词库）
    negative_words = {"差", "不好", "瑕疵", "掉色", "破损", "卡顿", "异味", "不值", "漏水", "变形"}
    for item in new_valid:
        text = item["comment_text"]
        if any(word in text for word in negative_words):
            print("==================== 负面评价告警 ====================")
            print(f"评价时间：{item['publishing_time']}")
            print(f"买家规格：{item['spec']}")
            print(f"评价内容：{text}")
            print("======================================================")

def start_long_term_monitor():
    """启动长效后台监控循环"""
    print("商品口碑实时监控程序已启动，每小时自动检测新增评价")
    while True:
        check_new_comments()
        time.sleep(MONITOR_INTERVAL)

if __name__ == "__main__":
    start_long_term_monitor()

模块 3：评论情感分析 + 可视化报表生成

基于分词与情感模型自动区分好评 / 中评 / 差评，提取高频痛点词汇，生成词云与评分分布图表，快速输出产品优化方向。

import pandas as pd
import jieba
from snownlp import SnowNLP
import matplotlib.pyplot as plt
from wordcloud import WordCloud
from collections import Counter

# 图表中文显示配置
plt.rcParams["font.sans-serif"] = ["SimHei"]
plt.rcParams["axes.unicode_minus"] = False
SAVE_FILE = "goods_comment_data.csv"
# 无意义停用词过滤库
STOP_WORDS = {"的", "了", "很", "比较", "就是", "还", "有点", "但是", "非常", "感觉", "一般", "这个", "东西", "买来"}

def sentiment_classify():
    """情感打分，划分好评、中评、差评"""
    df = pd.read_csv(SAVE_FILE, encoding="utf-8-sig")
    sentiment_res = []
    all_words = []
    negative_words_pool = []
    for _, row in df.iterrows():
        text = row["comment_text"]
        s = SnowNLP(text)
        score = s.sentiments
        # 情感阈值划分
        if score >= 0.6:
            label = "好评"
        elif score >= 0.3:
            label = "中评"
        else:
            label = "差评"
        sentiment_res.append(label)
        # 分词处理
        cut_words = jieba.lcut(text)
        valid_words = [w for w in cut_words if w not in STOP_WORDS and len(w) > 1]
        all_words.extend(valid_words)
        if label == "差评":
            negative_words_pool.extend(valid_words)
    df["sentiment_label"] = sentiment_res
    # 保存带情感标签的完整数据
    df.to_csv("comment_with_sentiment.csv", index=False, encoding="utf-8-sig")
    return df, all_words, negative_words_pool

def draw_analysis_chart(df, all_words, negative_words):
    """绘制评分分布图表、全量评价词云、差评痛点词云"""
    # 1. 情感分布柱状图
    sentiment_count = df["sentiment_label"].value_counts()
    fig, axes = plt.subplots(1, 3, figsize=(18, 6))
    axes[0].bar(sentiment_count.index, sentiment_count.values, color=["#66cc66", "#ffcc66", "#ff6666"])
    axes[0].set_title("商品评价情感分布统计", fontsize=14)
    axes[0].set_ylabel("评价数量")

    # 2. 全量评价词云
    full_text = " ".join(all_words)
    wc1 = WordCloud(width=600, height=400, background_color="white", font_path="simhei.ttf").generate(full_text)
    axes[1].imshow(wc1)
    axes[1].axis("off")
    axes[1].set_title("全部评价高频词云", fontsize=14)

    # 3. 差评痛点词云
    neg_text = " ".join(negative_words)
    wc2 = WordCloud(width=600, height=400, background_color="white", font_path="simhei.ttf").generate(neg_text)
    axes[2].imshow(wc2)
    axes[2].axis("off")
    axes[2].set_title("差评痛点关键词云", fontsize=14)

    plt.tight_layout()
    plt.savefig("comment_analysis_report.png", dpi=300)
    plt.show()

def stat_hot_keywords(word_list, top_num=15):
    """统计TOP高频关键词"""
    count = Counter(word_list)
    top_words = count.most_common(top_num)
    print(f"\nTOP{top_num}高频用户词汇：")
    for word, num in top_words:
        print(f"{word}：{num}次")
    return top_words

if __name__ == "__main__":
    data_df, all_word_list, neg_word_list = sentiment_classify()
    draw_analysis_chart(data_df, all_word_list, neg_word_list)
    stat_hot_keywords(all_word_list, 15)
    print("\n差评高频痛点词汇：")
    stat_hot_keywords(neg_word_list, 10)
    print("分析报表已生成：comment_with_sentiment.csv、comment_analysis_report.png")

四、落地使用实操指南

环境依赖安装 新建 Python 虚拟环境，执行以下命令安装所需第三方库：

pip install requests pandas jieba snownlp matplotlib wordcloud

参数替换 将代码中TOKEN、PROJECT_ID、TARGET_GOODS_ID全部替换为 OpenClaw 后台生成的专属信息，多商品监控可封装循环批量处理。
分步执行流程

第一步：运行模块 1，一次性抓取商品全部历史评价，建立本地基础数据库；
第二步：后台启动模块 2 监控脚本，长期实时跟踪新增评价，负面反馈即时预警；
第三步：每日 / 每周运行模块 3，生成完整情感分析报表，提炼产品优化痛点。

拓展优化方向

消息推送：对接企业微信机器人、邮件接口，新增差评自动推送运营人员；
数据库存储：将 CSV 替换为 MySQL/SQLite，支持上万条商品长期数据存储；
关键词自定义：扩充负面词库，区分物流、材质、售后、尺寸等多维度痛点；
多商品批量监控：循环遍历商品 ID 列表，同时跟踪数十款竞品口碑变化。

五、方案优势对比（对比传统爬虫）

稳定性更强：底层已适配平台访问规则，无需手动维护 Cookie、UA、延时策略，不会频繁被限制访问；
开发成本极低：不用分析页面动态渲染逻辑，无需反复调试页面改版适配，开箱即用；
功能覆盖完整：原生支持分页、增量同步、晒图 / 视频 / 追评全字段获取，省去大量解析代码；
轻量化部署：单台普通电脑即可运行，无需服务器集群，中小商家、个人调研完全够用；
长期监控省心：自动去重、增量抓取，避免重复采集浪费资源，负面舆情实时预警。

六、合规与使用提醒

本套方案仅适用于个人学习、自有店铺运营优化、竞品市场调研等非大规模商用场景；采集数据仅用于内部产品分析，禁止未经授权批量分发、售卖采集到的用户评价信息，遵守电商平台用户协议与网络数据相关法律法规。

总结

借助 OpenClaw 数据工具搭配 Python 轻量脚本，仅需 30 分钟就能搭建一套完整的电商商品口碑监控分析体系，告别手动复制整理评论的低效工作。从历史评价批量采集、长效实时监控，到文本情感挖掘、可视化痛点报表，全链路自动化落地，不管是产品经理挖掘用户需求、运营实时处理差评舆情，还是竞品调研分析市场反馈，都能大幅提升工作效率。

整套代码可直接复制运行，仅替换后台凭证即可快速投入使用，后续可根据自身业务需求扩展消息推送、数据库存储、多维度关键词分类等进阶功能。

万邦api博客

Nice to meet you, too!

api 数据挖掘