万邦api博客

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

Wed, 22 Jul 2026 17:30:08 +0800

前言

做电商运营、产品调研、竞品分析的朋友应该都有同款痛点：想要持续跟踪商品用户真实反馈，手动翻评论、复制整理效率极低；传统页面抓取方式容易被限制、数据残缺，批量分析更是无从下手。

今天分享一套轻量化落地方案，借助 OpenClaw 稳定数据通道，不用复杂部署、不用大量代理，快速搭建7×24 小时商品评论增量监控 + 自动化数据分析体系，普通电脑、轻量云服务器均可稳定运行，完整流程附带可直接复制运行的 Python 代码，适合商家日常口碑巡检、竞品长期追踪、产品迭代痛点挖掘。

一、整套方案整体逻辑

整套系统分为四大核心模块，全程规避繁琐逆向、页面解析等高风险操作：

增量数据同步模块：输入商品唯一标识，批量拉取全维度评价内容，包含评价正文、评价时间、买家晒图、购买规格、脱敏用户昵称、卖家回复等完整信息，支持分页分批获取海量历史评价；
实时监控模块：基于评价唯一编号做数据去重，定时循环轮询，仅抓取新增评价，大幅降低资源消耗，实现口碑动态实时捕捉；
数据清洗存储模块：自动过滤空白评价、重复灌水好评、无效符号，统一结构化后保存至本地表格，长期监控可切换轻量数据库；
智能分析预警模块：文本分词区分好评 / 中评 / 差评，统计负面高频痛点词，新增低分评价自动识别，输出口碑趋势、评分分布统计结果，直接支撑产品优化与客服应急处理。

二、前期准备工作

注册获取专属密钥与访问凭证（工具后台直接复制，无需额外开发资质）；
提取目标商品 ID：打开商品详情页，链接中id=后方一串数字即为商品标识；
Python 环境依赖安装，执行下方命令：

pip install requests pandas jieba wordcloud matplotlib

三、完整实战代码实现

3.1 基础数据拉取工具函数（批量获取商品评价）

核心逻辑：传入商品标识、页码，工具通道自动返回结构化评价数据，无需手动解析页面，支持批量分页拉取全部历史评价。

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

3.2 数据清洗与持久化存储函数

自动去重、过滤无效空白评价，追加写入本地表格，长期监控数据不会重复堆积。

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

3.3 口碑情感分析 + 痛点统计函数

读取本地评价数据，区分正负向评价，提取差评高频关键词，输出统计结果，可直接用于产品迭代分析。

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

3.4 7×24 小时增量监控主循环

定时轮询抓取最新评价，自动识别新增差评并打印预警，实现实时口碑监控。

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()

四、代码使用说明与优化方案

1. 快速上手步骤

修改代码顶部ACCESS_KEY、ACCESS_SECRET为工具后台生成的专属凭证；
替换TARGET_ITEM_ID为需要监控的商品编号；
直接运行脚本，程序会先一次性同步全部历史评价，自动生成 CSV 数据表并输出分析报告；
同步完成后自动进入循环监控模式，每 10 分钟抓取一次最新评价，出现差评即时弹窗打印预警。

2. 场景化优化拓展

多商品批量监控：将商品 ID 存入列表，循环遍历执行同步逻辑，同时跟踪多款竞品口碑；
消息预警推送：新增差评识别后，对接企业微信 / 钉钉机器人，自动推送评价详情至运营群；
长期大数据存储：替换 CSV 存储为 SQLite、MySQL 数据库，支持百万级评价长期存储；
可视化看板：基于 Matplotlib 拓展时间趋势折线图，观察不同周期差评数量波动；
自定义维度分析：根据商品规格sku_info拆分分析不同尺码、颜色款式的用户反馈差异。

五、这套方案对比传统方式的核心优势

稳定无封禁风险：依托 OpenClaw 标准化数据通道，无需搭建大量代理池、无需模拟浏览器滑动翻页，不会触发平台访问限制，7×24 小时稳定运行；
数据完整结构化：一次性获取评价正文、晒图、规格、追评、卖家回复等全维度信息，不用自己写复杂页面解析逻辑；
轻量化低门槛：仅依赖基础 Python 第三方库，无需分布式、服务器集群，个人笔记本即可完成竞品长期监控；
自动化闭环：从数据抓取、清洗存储、情感分析、差评预警全流程自动化，不用人工导出、整理表格，大幅节省运营时间；
成本可控：按需调取数据，增量抓取机制减少重复数据消耗，中小商家、个人调研均可低成本落地。

六、落地场景总结

自有店铺运营：实时监控本店商品新增差评，第一时间介入客服安抚，降低负面口碑扩散；
竞品调研分析：长期跟踪同类爆款商品用户痛点，挖掘竞品短板，指导自家产品优化升级；
产品迭代调研：批量提取海量用户真实反馈，统计高频吐槽点，作为产品改版、工艺优化核心依据；
直播 / 活动复盘：活动期间开启高频监控，快速捕捉活动带来的集中负面反馈，及时调整活动策略。

结尾

很多做电商、产品的同学还在手动复制评论、用简易页面抓取工具，不仅耗时，还经常出现数据缺失、账号受限问题。这套基于 OpenClaw 搭建的自动化监控分析体系，把数据获取、清洗、分析、预警全部打通，一段代码就能实现长效口碑监测，适合长期复用。

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

Tue, 21 Jul 2026 11:52:27 +0800

前言

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

传统页面抓取方案存在大量短板：频繁触发平台访问限制、需要长期维护登录凭证、页面改版后代码直接失效、批量采集稳定性差。今天分享一套轻量化成熟方案，借助 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 分钟就能搭建一套完整的电商商品口碑监控分析体系，告别手动复制整理评论的低效工作。从历史评价批量采集、长效实时监控，到文本情感挖掘、可视化痛点报表，全链路自动化落地，不管是产品经理挖掘用户需求、运营实时处理差评舆情，还是竞品调研分析市场反馈，都能大幅提升工作效率。

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

电商竞品口碑自动化监控与深度数据分析实战（完整可运行 Python 源码）

Tue, 21 Jul 2026 10:07:56 +0800

前言

做电商运营、产品调研、竞品分析的朋友，几乎都有同一个痛点：想要持续追踪商品用户评价，手动翻页复制效率极低，批量对比多款竞品口碑、挖掘差评痛点、统计好评卖点更是耗时耗力。

很多人尝试页面批量抓取，又频繁遇到限制、页面结构变动失效、数据不全（晒图、追评、购买规格丢失）等问题。今天给大家分享一套基于 OpenClaw 稳定数据通道搭建的全自动评论监控分析方案，无需复杂环境部署，普通电脑即可 7×24 小时增量监控商品评价，自动完成数据清洗、情感分类、词频统计，输出可视化报表，新手半小时就能完整跑通整套流程。

本文全程无晦涩底层配置，附带完整可运行代码，适配淘宝 / 天猫全品类商品，支持全量历史评论采集 + 实时新增评论增量监控，适合商家做自有店铺口碑预警、运营做竞品用户需求挖掘、数据分析从业者做电商文本挖掘实验。

一、整体方案架构说明

整套工具分为四大核心模块，轻量化无分布式成本：

数据同步采集模块 输入商品唯一标识，即可拉取完整评价数据集，包含评价正文、评价时间、晒图素材、买家购买规格、追评内容、卖家回复等全部字段；支持两种排序模式：默认综合排序、最新评价优先排序，自动分页循环拉取，无需手动处理分页逻辑。
增量实时监控模块 内置数据去重逻辑，定时循环轮询目标商品，仅抓取新增评价，大幅降低资源消耗；可配置差评自动提醒，出现负面评价即时输出预警信息，实现全天候口碑监控。
数据清洗存储模块 自动过滤空白无效评价、脱敏冗余文本，统一规整时间、规格、图文链接字段；小规模使用直接保存 CSV 表格，长期大批量监控可无缝切换 SQLite 本地数据库存储。
智能文本分析模块 结合中文分词区分好评 / 中评 / 差评，统计负面高频痛点词汇、好评核心卖点；自动生成评分分布统计图、评价时间趋势图，直观展示商品口碑变化。

二、前期准备工作

2.1 工具权限配置

登录控制台，创建数据项目，获取两组凭证：访问密钥、安全密钥，用于稳定数据通道鉴权，所有数据请求均通过加密校验，合规稳定。

2.2 提取商品唯一标识

打开目标商品详情页，从商品链接中提取数字 ID（链接id=xxxx后的数字串），后续代码仅需填入该数字即可绑定对应商品。

2.3 Python 环境依赖安装

打开终端执行命令，安装项目所需第三方库：

pip install requests pandas jieba wordcloud matplotlib

三、完整实战代码（分模块讲解）

3.1 核心数据采集工具类（获取商品全量评价）

封装数据拉取、分页循环、数据解析逻辑，自动处理缓存、返回格式、语言适配，支持自定义采集页数、排序规则。

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

# 全局配置（替换为自己控制台获取的凭证）
ACCESS_KEY = "你的访问密钥"
SEC_KEY = "你的安全密钥"
# 全局存储评论列表
all_comment_data = []

class CommentCollector:
    def __init__(self):
        self.base_url = "稳定数据通道地址"
        self.cache_switch = "yes"  # 开启缓存提升访问速度
        self.data_format = "json" # 返回结构化json数据

    def get_single_page_comment(self, item_id, page=1, sort_type=0):
        """
        单页评价采集
        :param item_id: 商品数字ID
        :param page: 分页页码
        :param sort_type: 0综合排序 / 1最新评价排序
        :return: 单页结构化评价列表
        """
        # 组装请求参数
        params = {
            "key": ACCESS_KEY,
            "secret": SEC_KEY,
            "cache": self.cache_switch,
            "result_type": self.data_format,
            "num_iid": item_id,
            "page": page,
            "sort": sort_type
        }
        try:
            resp = requests.get(url=self.base_url, params=params, timeout=12)
            result = json.loads(resp.text)
            # 判断是否存在评价数据
            if "rateList" not in result.get("result", {}):
                return []
            rate_list = result["result"]["rateList"]
            parse_res = []
            for item in rate_list:
                info = {
                    "评价时间": item.get("rate_date", ""),
                    "评价内容": item.get("rate_content", "").strip(),
                    "购买规格": item.get("sku_properties_name", ""),
                    "是否晒图": 1 if item.get("pics") else 0,
                    "追评内容": item.get("append_comment", ""),
                    "卖家回复": item.get("reply_content", ""),
                    "星级": item.get("star", "")
                }
                parse_res.append(info)
            return parse_res
        except Exception as e:
            print(f"第{page}页采集异常：{str(e)}")
            return []

    def batch_collect_comments(self, item_id, max_page=10, sort_type=0):
        """批量分页全量采集"""
        global all_comment_data
        for page_num in range(1, max_page + 1):
            print(f"正在采集第{page_num}页评价数据...")
            page_data = self.get_single_page_comment(item_id, page_num, sort_type)
            if not page_data:
                print("无更多评价，采集终止")
                break
            all_comment_data.extend(page_data)
            time.sleep(1) # 间隔防限流
        print(f"全量采集完成，共获取{len(all_comment_data)}条用户评价")
        return all_comment_data

3.2 增量监控模块（7×24 小时实时监测新增评价）

基于历史评价文本做去重校验，定时轮询，新增差评自动打印预警，实现长效口碑监控。

class CommentMonitor:
    def __init__(self, collector: CommentCollector, item_id, interval=300):
        self.collector = collector
        self.item_id = item_id
        self.check_interval = interval # 监控轮询间隔，单位秒
        self.history_text_set = set()
        # 初始化历史评价文本库
        self.init_history()

    def init_history(self):
        """首次运行加载全部历史评价"""
        history = self.collector.batch_collect_comments(self.item_id, max_page=5)
        for item in history:
            self.history_text_set.add(item["评价内容"])
        print("历史评价初始化完成，开始实时监控新增评价")

    def monitor_loop(self):
        """循环监控主逻辑"""
        while True:
            # 每次仅拉取最新一页评价，减少请求量
            new_page = self.collector.get_single_page_comment(self.item_id, page=1, sort_type=1)
            for comment in new_page:
                text = comment["评价内容"]
                star = comment["星级"]
                if text not in self.history_text_set and text != "":
                    print("===== 捕获新增评价 =====")
                    print(f"时间：{comment['评价时间']}")
                    print(f"星级：{star}星 | 内容：{text}")
                    # 差评预警逻辑
                    if int(star) <= 2:
                        print("【重要预警】检测到新增差评，请及时跟进处理！")
                    self.history_text_set.add(text)
            print(f"等待{self.check_interval/60}分钟后再次检测...\n")
            time.sleep(self.check_interval)

3.3 数据分析与可视化模块（情感统计 + 词云生成）

对采集完成的评价数据集做清洗、星级分布统计、高频词提取，自动生成图表与词云文件。

class CommentAnalyzer:
    def __init__(self, data_list):
        self.df = pd.DataFrame(data_list)
        # 过滤空白评价
        self.df = self.df[self.df["评价内容"] != ""]
        self.stop_words = {"的", "了", "很", "还", "比较", "感觉", "就是", "这个", "挺", "一般"}

    def star_distribution_stat(self):
        """星级分布统计+绘图"""
        star_count = self.df["星级"].value_counts().sort_index()
        print("===== 商品星级分布统计 =====")
        print(star_count)
        plt.rcParams["font.sans-serif"] = ["SimHei"]
        plt.rcParams["axes.unicode_minus"] = False
        star_count.plot(kind="bar", color=["#ff4444", "#ffbb33", "#ffff33", "#bbee33", "#33bb33"])
        plt.title("商品评价星级分布")
        plt.xlabel("星级")
        plt.ylabel("评价数量")
        plt.savefig("星级分布统计图.png", dpi=300)
        plt.show()
        return star_count

    def extract_high_freq_words(self):
        """提取评价高频关键词，区分好评/差评"""
        good_text = "".join(self.df[self.df["星级"] >= 4]["评价内容"].tolist())
        bad_text = "".join(self.df[self.df["星级"] <= 2]["评价内容"].tolist())
        # 分词过滤停用词
        def cut_words(text):
            words = jieba.lcut(text)
            clean = [w for w in words if len(w) > 1 and w not in self.stop_words]
            return Counter(clean).most_common(20)
        good_words = cut_words(good_text)
        bad_words = cut_words(bad_text)
        print("\n===== 好评高频关键词TOP20 =====")
        for w, num in good_words:
            print(f"{w}：{num}次")
        print("\n===== 差评痛点关键词TOP20 =====")
        for w, num in bad_words:
            print(f"{w}：{num}次")
        # 生成差评词云
        wc = WordCloud(
            background_color="white",
            font_path="simhei.ttf",
            width=1000, height=600
        ).generate(bad_text)
        wc.to_file("差评痛点词云.png")
        print("差评词云图片已生成")
        return good_words, bad_words

    def export_csv(self):
        """导出完整评价数据表"""
        self.df.to_csv("商品完整评价数据.csv", encoding="utf-8-sig", index=False)
        print("评价数据表已导出为csv文件，可直接用Excel打开")

3.4 程序主运行入口

整合采集、监控、分析全流程，按需切换一次性全量采集分析或长效实时监控两种模式。

if __name__ == "__main__":
    # 配置目标商品ID，替换为自己调研商品数字ID
    TARGET_ITEM_ID = "600530677643"
    # 初始化采集工具
    collector = CommentCollector()
    # 模式1：一次性全量采集+数据分析（调研竞品使用）
    print("===== 模式1：全量采集并分析商品评价 =====")
    comment_data = collector.batch_collect_comments(TARGET_ITEM_ID, max_page=15, sort_type=0)
    if comment_data:
        analyzer = CommentAnalyzer(comment_data)
        analyzer.star_distribution_stat()
        analyzer.extract_high_freq_words()
        analyzer.export_csv()

    # 模式2：开启7×24小时增量监控（商家实时口碑预警，取消注释即可运行）
    # monitor = CommentMonitor(collector, TARGET_ITEM_ID, interval=300)
    # monitor.monitor_loop()

四、代码使用说明与场景落地

4.1 快速上手步骤

替换代码顶部ACCESS_KEY、SEC_KEY为控制台生成的专属凭证；
修改TARGET_ITEM_ID为目标商品 ID；
直接运行脚本，默认执行全量采集 + 数据分析流程，自动输出统计表格、统计图、差评词云；
需要长期监控时，注释模式 1 代码，取消模式 2 监控代码注释，脚本后台持续运行即可。

4.2 核心实用场景

自有店铺口碑监控 开启增量监控模式，新增差评实时预警，快速响应买家负面反馈，降低售后风险；定期导出评价数据，迭代产品优化方向。
竞品深度调研分析 批量采集多款竞品全量评价，通过词云提取竞品核心痛点，找到自身产品差异化优势；统计竞品好评高频卖点，优化商品详情页宣传话术。
电商文本挖掘学习 完整结构化评价数据集包含文本、时间、规格、晒图标识，可用于 NLP 情感分类、用户画像、销量关联分析等实验。

4.3 避坑优化小技巧

采集页数max_page建议控制在 20 页以内，如需大批量数据可拆分多次运行，避免单次请求过多；
监控轮询间隔interval建议设置 300 秒（5 分钟）以上，减少通道请求频次；
词云生成如提示字体缺失，将系统黑体字体文件路径填入font_path参数；
数据存储量巨大时，可扩展代码将 CSV 存储替换为 SQLite 数据库，方便长期数据检索。

五、方案优势总结

数据完整稳定 依托成熟稳定数据通道，无需处理页面渲染、签名校验、反爬限制，一次配置永久可用，完整抓取晒图、追评、卖家回复等普通页面抓取无法获取的字段。
轻量化低成本 不需要服务器集群、复杂中间件，本地电脑、轻量云主机均可部署，单人即可完成整套数据监控分析流程，大幅降低电商数据调研人力成本。
自动化闭环流程 从数据拉取、去重存储、情感分析到可视化报表全程自动化，无需人工复制整理评价，每天节省数小时手工统计时间。
灵活拓展性强 代码模块化拆分，可快速拓展功能：对接企业微信 / 钉钉消息推送差评预警、批量循环采集多款竞品、训练情感分类模型做精细化用户分层。

结语

电商行业竞争愈发激烈，用户真实评价是产品迭代、运营策略调整最核心的一手数据。这套基于 OpenClaw 搭建的自动化评价监控分析工具，解决了传统手动统计、页面抓取不稳定两大核心痛点，不管是商家、运营还是数据分析从业者，都能快速落地使用。

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

Mon, 20 Jul 2026 16:21:07 +0800

前言

做电商运营、产品、竞品分析的同学，几乎都绕不开用户评价数据。商品好评、差评、晒图、规格反馈、追评里藏着最真实的用户痛点：面料差、尺寸不准、物流慢、包装破损，或是产品爆火的核心卖点。

传统人工统计方式痛点极多：

手动翻页复制评论，单品上千条数据至少耗费半天；
无法实时监控新增差评，售后反馈滞后；
数据零散，很难批量统计负面关键词、评分分布；
自研网页抓取容易遭遇平台限制、频繁失效，维护成本高。

本文给大家分享一套成熟稳定的落地方案，依托 OpenClaw 数据通道，无需复杂逆向、不用处理页面加密，一行配置即可批量拉取全量评价，搭配 Python 实现7×24 小时增量监控、数据清洗、情感分析、可视化报表，小卖家、数据分析实习生、竞品调研团队都能直接上手。

一、方案整体架构

整套系统仅依赖轻量 Python 环境，普通笔记本 / 低配云服务器就能稳定运行，分为四大核心模块：

全量数据同步模块：输入商品唯一标识，批量拉取评价全字段，包含评论文本、星级、晒图链接、购买规格、评价时间、脱敏买家昵称、追评内容；支持分页循环采集，自动适配海量评论商品。
增量实时监控模块：基于评价唯一标识做本地去重，定时轮询拉取最新评价，仅新增未记录数据，大幅降低请求频次，实现差评实时预警。
数据清洗存储模块：过滤空白评价、无效系统默认话术，统一时间格式，结构化存入 CSV 文件（小体量项目）或 SQLite 本地数据库（长期监控多商品）。
智能分析可视化模块：中文分词 + 情感判断区分好评 / 中评 / 差评，统计负面高频痛点词汇，输出评分占比饼图、每日评价量趋势折线图，直观呈现口碑变化。

二、前期准备工作

2.1 环境依赖安装

新建 Python 项目，执行命令安装全部依赖库：

pip install requests pandas jieba snownlp matplotlib sqlite3

各库作用说明：

requests：发起网络请求拉取商品评价数据
pandas：批量数据清洗、表格存储
jieba：中文评论分词，提取关键词
snownlp：轻量中文情感打分，区分正负向评价
matplotlib：生成口碑可视化图表
sqlite3：本地轻量数据库，长期存储多商品历史评价

2.2 工具基础配置

登录创建项目获取两组身份凭证（密钥串），用于数据通道鉴权；复制商品 ID（商品链接中id=后的数字串）作为采集目标，无需解析复杂页面链接。

三、完整可运行代码实现

3.1 基础数据拉取工具类（核心采集逻辑）

import requests
import json
import time
import random
import pandas as pd
import sqlite3
from snownlp import SnowNLP
import jieba
import matplotlib.pyplot as plt

# 1. 基础鉴权配置（替换为自己后台凭证）
ACCESS_KEY = "你的密钥1"
SECRET_KEY = "你的密钥2"
GATEWAY_URL = "稳定数据通道地址"

class GoodsCommentCollector:
    def __init__(self, item_id):
        self.item_id = item_id
        self.all_comment_list = []
        self.db_conn = sqlite3.connect("goods_comment.db")
        self.init_db()

    # 初始化本地数据库，存储历史评价用于增量监控
    def init_db(self):
        create_sql = """
        CREATE TABLE IF NOT EXISTS comment_data (
            comment_id TEXT PRIMARY KEY,
            nick TEXT,
            star INTEGER,
            content TEXT,
            sku TEXT,
            rate_date TEXT,
            has_pic INTEGER,
            add_feedback TEXT,
            emotion_score REAL
        )
        """
        self.db_conn.execute(create_sql)
        self.db_conn.commit()

    # 单页拉取评价原始数据
    def fetch_single_page(self, page_num=1, page_size=20):
        params = {
            "key": ACCESS_KEY,
            "secret": SECRET_KEY,
            "item_id": self.item_id,
            "page": page_num,
            "page_size": page_size,
            "result_type": "json"
        }
        try:
            resp = requests.get(GATEWAY_URL, params=params, timeout=15)
            resp_data = json.loads(resp.text)
            return resp_data
        except Exception as e:
            print(f"第{page_num}页采集异常：{str(e)}")
            return None

    # 解析原始数据，清洗结构化字段
    def parse_raw_data(self, raw_json):
        if not raw_json or "items" not in raw_json:
            return False
        comment_items = raw_json["items"].get("item", [])
        if len(comment_items) == 0:
            return False
        for item in comment_items:
            comment_info = {
                "comment_id": item.get("rate_id", ""),
                "nick": item.get("display_user_nick", "匿名用户"),
                "star": int(item.get("user_star", 5)),
                "content": item.get("rate_content", "").strip(),
                "sku": item.get("auction_sku", "无规格"),
                "rate_date": item.get("rate_date", ""),
                "has_pic": 1 if item.get("pics", []) else 0,
                "add_feedback": item.get("add_feedback", "无追评")
            }
            # 过滤空白无意义评价
            if comment_info["content"] in ["此用户没有填写评价。", ""]:
                continue
            self.all_comment_list.append(comment_info)
        return True

    # 批量全量采集，自动分页循环
    def batch_crawl_all(self, max_page=50):
        for page in range(1, max_page + 1):
            print(f"正在采集第{page}页评价数据...")
            raw_data = self.fetch_single_page(page_num=page)
            parse_result = self.parse_raw_data(raw_data)
            if not parse_result:
                print("无更多评价，全量采集完成")
                break
            # 随机休眠，平稳请求频率
            time.sleep(random.uniform(1.2, 3.0))
        print(f"本次共采集有效评价{len(self.all_comment_list)}条")

    # 增量入库：仅写入本地不存在的新评价
    def save_to_local_db(self):
        insert_sql = """
        INSERT OR IGNORE INTO comment_data
        (comment_id, nick, star, content, sku, rate_date, has_pic, add_feedback, emotion_score)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
        """
        for comment in self.all_comment_list:
            # 情感打分0-1，大于0.5正向，小于0.3负向
            s = SnowNLP(comment["content"])
            score = s.sentiments
            self.db_conn.execute(insert_sql, (
                comment["comment_id"],
                comment["nick"],
                comment["star"],
                comment["content"],
                comment["sku"],
                comment["rate_date"],
                comment["has_pic"],
                comment["add_feedback"],
                score
            ))
        self.db_conn.commit()
        print("新评价数据已存入本地数据库")

    # 导出全部数据到Excel表格
    def export_to_csv(self):
        df = pd.DataFrame(self.all_comment_list)
        df.to_csv(f"商品{self.item_id}_评价数据.csv", index=False, encoding="utf-8-sig")
        print("评价表格导出完成")

    # 7×24小时定时监控入口，每30分钟轮询一次
    def start_real_time_monitor(self, interval=1800):
        print("启动实时口碑监控，每30分钟自动抓取新增评价...")
        while True:
            self.batch_crawl_all(max_page=2)
            self.save_to_local_db()
            self.all_comment_list.clear()
            print(f"等待{interval/60}分钟后执行下一轮采集\n")
            time.sleep(interval)

3.2 评论情感分析 + 可视化代码

def draw_emotion_chart(item_id):
    conn = sqlite3.connect("goods_comment.db")
    df = pd.read_sql(f"SELECT emotion_score FROM comment_data", conn)
    conn.close()

    # 划分情感标签
    def label_emotion(score):
        if score >= 0.5:
            return "好评"
        elif score >= 0.3:
            return "中评"
        else:
            return "差评"
    df["emotion_type"] = df["emotion_score"].apply(label_emotion)
    emo_count = df["emotion_type"].value_counts()

    # 绘制饼图
    plt.rcParams["font.sans-serif"] = ["SimHei"]
    plt.figure(figsize=(8, 6))
    plt.pie(emo_count.values, labels=emo_count.index, autopct="%1.2f%%",
            colors=["#46b946", "#ffcc5c", "#ff5959"])
    plt.title(f"商品{item_id}评价情感分布")
    plt.savefig(f"商品{item_id}_口碑分布图.png", dpi=300)
    plt.show()

# 提取差评高频关键词
def get_negative_keywords():
    conn = sqlite3.connect("goods_comment.db")
    bad_df = pd.read_sql("SELECT content FROM comment_data WHERE emotion_score < 0.3", conn)
    conn.close()
    all_words = []
    stop_words = {"的", "了", "很", "太", "就", "还", "有点", "但是", "感觉"}
    for text in bad_df["content"]:
        words = jieba.lcut(text)
        for w in words:
            if len(w) > 1 and w not in stop_words:
                all_words.append(w)
    word_series = pd.Series(all_words).value_counts().head(15)
    print("差评高频痛点词TOP15：")
    print(word_series)
    return word_series

3.3 程序主运行入口

if __name__ == "__main__":
    # 替换为目标商品ID
    TARGET_GOODS_ID = "1234567890123"
    collector = GoodsCommentCollector(TARGET_GOODS_ID)

    # 操作1：一次性全量采集所有评价
    collector.batch_crawl_all(max_page=30)
    collector.save_to_local_db()
    collector.export_to_csv()

    # 操作2：生成口碑情感分布图
    draw_emotion_chart(TARGET_GOODS_ID)

    # 操作3：提取差评核心痛点词
    get_negative_keywords()

    # 操作4：开启长期实时监控（取消注释即可7×24小时运行）
    # collector.start_real_time_monitor(interval=1800)

四、功能落地效果说明

4.1 数据采集优势

字段完整全覆盖：自动获取买家昵称、星级、正文、晒图标记、购买规格、评价时间、追评，完整还原用户消费反馈；
分页自动处理：无需手动控制翻页，循环采集至无新数据自动终止，万条评价一键拉取；
稳定无封锁：依托成熟数据通道，避开平台页面限制，不会出现传统网页抓取频繁失效、验证码拦截问题；
缓存加速：重复采集时可复用历史缓存，大幅缩短数据返回时间，降低资源消耗。

4.2 监控与分析能力

增量更新不重复：本地数据库存储每条评价唯一标识，定时轮询只抓取新增内容，避免重复存储浪费空间；
差评自动识别：通过情感模型自动区分负面评价，批量提取高频痛点，快速定位产品缺陷（掉色、尺寸偏小、发货慢等）；
可视化报表直观：一键生成好评 / 中评 / 差评占比图表，运营人员无需看懂代码也能看懂口碑整体情况；
长期数据沉淀：SQLite 本地数据库永久存储历史评价，可按月、季度对比口碑变化，用于产品迭代、竞品对比分析。

五、实际业务使用场景

店铺售后预警：开启实时监控后，新增差评即时入库，搭配企业微信 / 钉钉推送脚本可实现负面消息秒级提醒，及时联系用户处理；
产品迭代优化：批量提取差评高频词汇，快速锁定产品核心短板，优化面料、做工、包装、物流等环节；
竞品口碑调研：批量采集多款同类竞品评价，对比各方优缺点，制定差异化宣传卖点；
内容营销素材挖掘：筛选高分带图好评，提取用户真实夸赞话术，用于商品详情页、短视频文案；
运营数据复盘：按月导出评价表格，统计晒图率、追评占比、差评率，量化店铺服务与产品质量指标。

六、避坑实操小贴士

采集间隔不要低于 1 秒，定时监控建议设置 30 分钟以上，平稳请求频率保证通道稳定；
空白评价、系统默认无内容评价代码已自动过滤，无需手动清洗；
如需多商品同时监控，可多实例初始化采集类，分别建立独立数据表；
长期大批量监控建议部署在轻量云服务器，保持程序后台常驻运行；
情感分析仅作参考，部分中性吐槽语句会存在打分偏差，可自定义负面关键词词库提升准确度。

总结

过去想要搭建一套完整的电商评价监控分析体系，要么投入大量人力手动统计，要么投入开发成本自研抓取方案，维护难度极高。借助 OpenClaw 一站式数据能力，仅需几十行 Python 代码就能完成采集 - 存储 - 监控 - 分析 - 可视化全流程闭环，零基础开发、低成本落地，不管是个人电商从业者、数据实习生还是企业运营团队，都能快速落地用户口碑数字化管理。

后续可拓展优化方向：接入消息推送工具实现差评告警、搭建 Web 可视化后台、接入大模型做评价深度语义解读、多平台商品数据统一汇总分析。

低成本搭建电商商品评论监控分析系统｜从实时采集到口碑洞察完整实战（附可运行 Python 代码）

Fri, 17 Jul 2026 17:28:04 +0800

前言

做电商运营、产品调研、竞品分析的朋友应该都有同样痛点：想要持续追踪商品真实用户反馈，手动翻评论效率极低，批量多商品监控、历史评论沉淀、差评预警、痛点关键词提取全靠人工几乎不现实。

传统本地爬虫方案有三大硬伤：页面频繁改版导致脚本失效、频繁访问触发平台风控被限制、多商品定时轮询需要大量服务器资源维护。今天分享一套轻量化落地方案，借助 OpenClaw 一站式数据能力，无需复杂逆向、不用搭建代理池，几分钟完成商品全量评论采集、增量实时监控、文本情感数据分析，个人 / 中小团队零成本即可落地。

本文完整覆盖工具配置、数据拉取、持久化存储、定时增量监控、评论情感分析、差评告警全流程，代码开箱即用，适配淘宝 / 天猫全品类商品，适合 CSDN、知乎技术分享阅读。

一、方案整体架构

整套系统分为四大模块，普通电脑 / 轻量云服务器就能稳定运行，无分布式部署门槛：

数据同步模块：依托 OpenClaw 稳定数据通道，输入商品标识即可拉取评论全字段，包含评分、评论文本、晒图、追评、购买规格、评价时间、买家秀视频等完整信息，支持按最新评论排序分页拉取；
增量监控模块：基于评论唯一标识做数据去重，定时循环轮询，只抓取新增评论，减少资源消耗，实现 7×24 小时口碑实时监控；
数据清洗存储模块：过滤空评论、重复无效内容，统一结构化后存入 CSV 文件（小型项目）或 SQLite 数据库（长期大批量监控）；
智能分析预警模块：结合分词与情感判断区分好评 / 中评 / 差评，统计负面高频痛点词，新增差评自动推送提醒，输出评分分布、口碑变化趋势数据。

二、前期工具基础配置

登录管理后台，新建数据采集项目，系统自动生成专属身份校验令牌与项目编号；
后台自定义采集频率、单次拉取条数，内置平台访问防护策略，无需手动处理访问限制；
支持多商品分组管理，可分开监控自家店铺商品与竞品商品，数据隔离便于独立复盘；
提前提取目标商品 ID：商品详情页链接中id=后一串数字即为商品唯一标识，作为代码核心入参。

配置完成后，仅需将后台生成的凭证填入代码变量，即可直接启动采集流程。

三、完整 Python 实战代码

3.1 依赖安装

pip install requests pandas jieba snownlp schedule

3.2 核心采集 + 增量监控主脚本

import requests
import pandas as pd
import time
import schedule
from snownlp import SnowNLP
import os

# ===================== 配置区，替换为你后台生成的凭证 =====================
TOKEN = "你的后台TOKEN"
PROJECT_ID = "你的项目编号"
MONITOR_ITEM_IDS = ["600530677643", "698754321098"]  # 需要监控的商品ID列表
SAVE_FILE = "商品评论数据.csv"
MONITOR_INTERVAL = 10  # 监控轮询间隔，单位：分钟
# =======================================================================

# 全局存储已采集评论ID，用于增量去重
saved_review_ids = set()

def init_history_data():
    """初始化历史数据，读取已保存评论ID"""
    global saved_review_ids
    if os.path.exists(SAVE_FILE):
        df = pd.read_csv(SAVE_FILE, encoding="utf-8-sig")
        saved_review_ids = set(df["评论唯一ID"].dropna().tolist())
        print(f"加载历史数据，已有历史评论 {len(saved_review_ids)} 条")
    else:
        print("暂无历史数据，首次全量采集")

def get_goods_review(item_id, page=1, sort=1):
    """
    拉取商品评论数据
    :param item_id: 商品ID
    :param page: 分页页码
    :param sort: 1=最新评论排序，0=默认排序
    :return: 单页评论列表
    """
    headers = {
        "Content-Type": "application/json"
    }
    params = {
        "token": TOKEN,
        "project_id": PROJECT_ID,
        "goods_id": item_id,
        "page": page,
        "sort": sort
    }
    try:
        resp = requests.get("https://data-collect.openclaw.cloud/goods/review", headers=headers, params=params, timeout=20)
        res_json = resp.json()
        if res_json.get("code") != 200:
            print(f"商品{item_id}第{page}页请求失败：{res_json.get('msg')}")
            return []
        return res_json.get("data", [])
    except Exception as e:
        print(f"网络请求异常：{str(e)}")
        return []

def parse_review_raw(raw_list, item_id):
    """解析原始返回数据，结构化清洗"""
    review_list = []
    for item in raw_list:
        review_id = item.get("rate_id", "")
        if review_id in saved_review_ids:
            continue
        # 单条评论结构化字段
        review_info = {
            "商品ID": item_id,
            "评论唯一ID": review_id,
            "买家昵称": item.get("display_user_nick", ""),
            "评论星级": item.get("rate_star", ""),
            "评论正文": item.get("rate_content", ""),
            "购买规格": item.get("auction_sku", ""),
            "评论时间": item.get("rate_date", ""),
            "晒图链接": ",".join(item.get("pics", [])) if item.get("pics") else "",
            "追评内容": item.get("add_feedback", ""),
            "追评图片": ",".join(item.get("add_feedback_images", [])) if item.get("add_feedback_images") else "",
            "评价视频": item.get("video", "")
        }
        # 简单情感分析
        text_all = review_info["评论正文"] + review_info["追评内容"]
        if text_all.strip():
            s = SnowNLP(text_all)
            sentiment_score = s.sentiments
            if sentiment_score >= 0.6:
                review_info["情感标签"] = "好评"
            elif sentiment_score >= 0.3:
                review_info["情感标签"] = "中评"
            else:
                review_info["情感标签"] = "差评"
                # 新增差评触发告警逻辑，可扩展企业微信/邮件推送
                print(f"【差评预警】商品{item_id}新增差评：{review_info['评论正文']}")
        else:
            review_info["情感标签"] = "无内容"
        saved_review_ids.add(review_id)
        review_list.append(review_info)
    return review_list

def batch_crawl_item(item_id):
    """单商品全量/增量采集"""
    all_new_data = []
    page = 1
    while True:
        raw_data = get_goods_review(item_id, page=page, sort=1)
        if not raw_data:
            break
        page_data = parse_review_raw(raw_data, item_id)
        if not page_data:
            # 当前页无新增评论，停止分页
            break
        all_new_data.extend(page_data)
        page += 1
        time.sleep(1)
    return all_new_data

def save_to_csv(new_data):
    """数据持久化保存至CSV"""
    if not new_data:
        return
    df_new = pd.DataFrame(new_data)
    if os.path.exists(SAVE_FILE):
        df_old = pd.read_csv(SAVE_FILE, encoding="utf-8-sig")
        df_merge = pd.concat([df_old, df_new], ignore_index=True)
    else:
        df_merge = df_new
    df_merge.to_csv(SAVE_FILE, index=False, encoding="utf-8-sig")
    print(f"本次新增评论 {len(new_data)} 条，已完成存储")

def monitor_task():
    """定时监控主任务"""
    print("===== 启动一轮商品评论监控 =====")
    init_history_data()
    total_new = []
    for goods_id in MONITOR_ITEM_IDS:
        print(f"正在采集商品 {goods_id} 新增评论...")
        new_comments = batch_crawl_item(goods_id)
        total_new.extend(new_comments)
    save_to_csv(total_new)
    print("===== 本轮监控执行完成，等待下一轮 ====\n")

if __name__ == "__main__":
    # 初始化历史数据
    init_history_data()
    # 定时任务配置，每10分钟执行一次监控
    schedule.every(MONITOR_INTERVAL).minutes.do(monitor_task)
    print(f"监控系统已启动，每{MONITOR_INTERVAL}分钟自动拉取新增评论，Ctrl+C停止程序")
    # 常驻循环
    while True:
        schedule.run_pending()
        time.sleep(30)

3.3 评论数据分析脚本（痛点关键词统计 + 评分分布）

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

plt.rcParams["font.sans-serif"] = ["SimHei"]
plt.rcParams["axes.unicode_minus"] = False

def analyze_comment_data(file_path):
    df = pd.read_csv(file_path, encoding="utf-8-sig")
    print("=== 商品评论整体统计 ===")
    print(f"总评论条数：{len(df)}")
    print(df["情感标签"].value_counts())
    print("-" * 40)

    # 提取差评文本，分词统计高频痛点词
    bad_comments = df[df["情感标签"] == "差评"]["评论正文"].tolist()
    all_text = "".join(bad_comments)
    words = jieba.lcut(all_text)
    # 过滤无意义停用词
    stop_words = {"的", "了", "很", "有点", "还是", "就是", "感觉", "这个", "买", "收到"}
    word_filter = [w for w in words if len(w) > 1 and w not in stop_words]
    word_count = Counter(word_filter)
    print("差评高频痛点关键词Top15：")
    for word, cnt in word_count.most_common(15):
        print(f"{word}：{cnt}次")

    # 绘制情感分布饼图
    emo_count = df["情感标签"].value_counts()
    plt.figure(figsize=(8, 6))
    plt.pie(emo_count.values, labels=emo_count.index, autopct="%1.2f%%", colors=["#66cc66", "#ffcc66", "#ff6666", "#cccccc"])
    plt.title("商品评论情感分布")
    plt.savefig("评论情感分布.png", dpi=300)
    plt.show()

if __name__ == "__main__":
    analyze_comment_data("商品评论数据.csv")

四、代码核心能力说明

增量去重机制：启动时自动读取历史所有评论唯一标识，轮询时仅抓取全新评论，避免重复存储，大幅降低存储占用；
全字段完整覆盖：自动拉取用户昵称、星级、正文、购买规格、晒图、追评、视频等运营核心分析字段，无需额外二次处理；
定时长效监控：依靠 schedule 库实现后台常驻运行，自定义监控间隔，多商品批量同步追踪；
自动情感分类：SnowNLP 轻量化文本情感模型，自动区分好评 / 中评 / 差评，出现差评实时控制台预警，可自行扩展对接企业微信、钉钉消息推送；
可视化数据分析：一键统计差评高频痛点词，生成情感分布饼图，快速定位产品质量、物流、售后等核心问题；
低门槛部署：纯 Python 脚本，无复杂依赖，Windows、Mac、Linux、轻量云服务器均可直接运行。

五、落地业务场景

场景 1：自有店铺口碑实时监控

运营人员无需每天手动刷商品评论，程序自动定时抓取新增评价，一旦出现差评立刻预警，可第一时间联系用户处理售后，降低店铺负面口碑扩散风险；同时统计长期差评高频词，反向优化产品做工、包装、发货时效。

场景 2：竞品深度调研分析

批量添加多款竞品商品 ID，长期沉淀竞品全量用户反馈，对比自身与竞品的用户痛点差异，找到差异化产品优化方向，用于新品研发、详情页卖点调整。

场景 3：市场舆情复盘

按月导出历史评论数据，分析不同时间段口碑波动，结合活动大促节点，判断活动对用户体验、商品评价的影响，优化后续营销方案。

六、避坑优化小贴士

监控间隔设置：普通日常监控建议 10–15 分钟一轮；大促流量高峰期可缩短至 5 分钟，兼顾实时性与稳定性；
存储方案选择：商品数量少于 10 款、评论总量 10 万条以内使用 CSV 即可；长期大规模监控建议替换为 SQLite/MySQL 数据库；
风控优化：工具后台自带访问限流策略，代码内已增加分页休眠，无需额外搭建代理池，避免频繁访问限制；
情感模型升级：SnowNLP 适合轻量快速分析，若需要更高精准度可替换第三方大模型文本识别能力，优化评价分类准确率；
告警拓展：可在差评预警代码段增加 webhook 请求，将差评内容自动推送至企业微信群、钉钉群，实现团队协同处理。

七、总结

借助 OpenClaw 数据能力，我们避开了传统爬虫频繁维护、逆向失效、风控拦截等一系列难题，仅用几十行 Python 代码就能搭建一套完整的商品评论监控与数据分析体系。整套方案轻量化、低成本，个人开发者、电商运营、产品调研都能快速上手落地。

从定时抓取新增评论、自动存储沉淀、情感自动分类，到痛点关键词统计、口碑可视化分析，全流程自动化完成，把人工重复翻阅评论的时间释放出来，聚焦数据背后的业务优化决策，是电商用户反馈分析性价比极高的落地方案。

后续还可以基于这套代码拓展更多功能：评论图片自动下载归档、多维度评分趋势折线图、按商品规格拆分评价数据、月度口碑自动生成简报文档等。

电商口碑自动化监控方案：30 分钟搭建商品评论实时采集 + 情感分析系统（完整可运行代码）

Thu, 16 Jul 2026 15:04:16 +0800

前言

做电商运营、竞品调研、产品迭代的同学，几乎都有同一个痛点：人工刷商品评价效率极低。多店铺、多竞品同时跟踪时，每天花几小时翻评价、筛选差评、统计用户痛点，不仅耗时，还容易遗漏突发负面反馈，错过舆情处理黄金时间。

过去尝试过页面解析采集，但平台页面频繁改版、风控拦截、验证码、IP 限制等问题层出不穷，采集脚本隔三差五失效，维护成本极高。今天分享一套稳定成熟的商用数据采集方案，无需逆向页面、无需处理风控，开箱即可实现全量评论自动抓取、定时监控、文本情感分析、痛点可视化，适配淘宝 / 天猫全品类商品，新手也能快速落地。

一、方案核心优势

数据维度完整：一次性获取主评、追评、买家晒图、购买 SKU 规格、评价星级、发布时间、脱敏买家昵称等全部原生字段，无缺失信息万邦开放平...；
稳定无风控：底层封装好请求逻辑，无需搭建代理池、处理验证码、应对页面改版，7×24 小时稳定拉取数据；
自动化监控：搭配定时任务循环采集，自动对比历史数据，识别新增差评、新增追评，实现口碑预警；
一站式分析：采集后直接接入 NLP 情感分析，自动区分好评 / 中评 / 差评，提取高频痛点关键词，生成统计图表；
轻量化部署：Python 实现，无需复杂服务架构，本地电脑、云服务器均可运行，兼容 Windows、Linux、Mac。

二、前期准备工作

1. 环境依赖安装

执行命令安装所需工具库：

pip install requests pandas jieba snownlp matplotlib wordcloud schedule

requests：数据请求工具
pandas：结构化数据存储与处理
snownlp：中文情感打分
jieba：中文分词提取关键词
wordcloud/matplotlib：可视化词云、评分分布图
schedule：定时循环监控任务

2. 基础账号配置

前往数据服务控制台注册账号，获取专属访问凭证（下文代码中access_token替换为自己的凭证），无需额外申请平台权限，开通后即可调用全量商品评价数据。

3. 获取商品 ID

打开商品详情页，地址栏id=后面的数字即为商品唯一标识，例如https://item.taobao.com/item.htm?id=789654321098，商品 ID：789654321098。

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

模块 1：商品评论批量采集工具

实现分页拉取全量评价，自动清洗结构化数据，保存至本地 CSV 文件，支持多商品批量监控。

import requests
import time
import pandas as pd

# 基础配置
ACCESS_TOKEN = "你的专属访问凭证"
TARGET_ITEM_ID = "789654321098"  # 目标商品ID
SAVE_PATH = "goods_comment_data.csv"

def fetch_goods_comments(item_id, page=1, page_size=20):
    """拉取单页商品评价数据"""
    params = {
        "token": ACCESS_TOKEN,
        "goods_id": item_id,
        "page": page,
        "limit": page_size
    }
    # 数据请求地址
    res = requests.get("https://data-collect-service.com/shop_comment", params=params)
    result = res.json()
    if result.get("code") != 200:
        print(f"采集失败：{result.get('msg')}")
        return None
    return result.get("data", {})

def parse_comment_data(raw_data):
    """清洗原始数据，提取有效字段"""
    comment_list = []
    item_arr = raw_data.get("item", [])
    for item in item_arr:
        info = {
            "comment_id": item.get("rate_id", ""),
            "comment_time": item.get("rate_date", ""),
            "buyer_nick": item.get("display_user_nick", ""),
            "buy_sku": item.get("auction_sku", ""),
            "main_comment": item.get("rate_content", ""),
            "append_comment": item.get("add_feedback", ""),
            "has_img": 1 if item.get("pics") else 0,
            "has_append_img": 1 if item.get("add_feedback_images") else 0,
            "star_score": item.get("user_star", 0)
        }
        comment_list.append(info)
    return comment_list

def batch_collect_all_comments(item_id):
    """分页循环采集全部评价"""
    all_data = []
    page = 1
    while True:
        print(f"正在采集第{page}页评价...")
        raw = fetch_goods_comments(item_id, page=page)
        if not raw:
            break
        page_data = parse_comment_data(raw)
        if not page_data:
            print("无更多评价，采集结束")
            break
        all_data.extend(page_data)
        # 判断是否最后一页
        total_page = raw.get("totalpage", 1)
        if page >= int(total_page):
            break
        page += 1
        time.sleep(1)  # 限流休眠
    # 保存至本地CSV
    df = pd.DataFrame(all_data)
    # 追加写入，避免覆盖历史数据
    try:
        old_df = pd.read_csv(SAVE_PATH, encoding="utf-8-sig")
        df = pd.concat([old_df, df]).drop_duplicates(subset=["comment_id"])
    except FileNotFoundError:
        pass
    df.to_csv(SAVE_PATH, index=False, encoding="utf-8-sig")
    print(f"本次采集完成，累计有效评价：{len(df)}条")
    return all_data

# 单次采集测试
if __name__ == "__main__":
    batch_collect_all_comments(TARGET_ITEM_ID)

模块 2：评论情感分析 + 痛点可视化

读取采集后的评价文件，自动打分区分好评 / 中评 / 差评，提取高频负面关键词，生成词云与评分分布图表，快速定位产品问题。

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 comment_sentiment_analysis():
    df = pd.read_csv(SAVE_FILE, encoding="utf-8-sig")
    sentiment_label = []
    all_text = []
    negative_text = []
    for idx, row in df.iterrows():
        # 合并主评+追评完整文本
        text = str(row["main_comment"]) + str(row["append_comment"])
        if text.strip() == "":
            sentiment_label.append("无评价")
            continue
        # 情感打分 0~1，越高越正向
        s = SnowNLP(text)
        score = s.sentiments
        if score >= 0.6:
            label = "好评"
        elif score >= 0.3:
            label = "中评"
        else:
            label = "差评"
        sentiment_label.append(label)
        all_text.append(text)
        if label == "差评":
            negative_text.append(text)
    df["sentiment"] = sentiment_label
    # 保存带情感标签的数据
    df.to_csv("comment_with_sentiment.csv", index=False, encoding="utf-8-sig")

    # 1. 评分分布统计图表
    count_result = df["sentiment"].value_counts()
    count_result.plot(kind="bar", color=["#66cc66", "#ffcc66", "#ff6666"])
    plt.title("商品评价情感分布统计")
    plt.savefig("sentiment_dist.png", dpi=300, bbox_inches="tight")
    plt.close()

    # 2. 差评关键词词云
    words = []
    for text in negative_text:
        cut_words = jieba.lcut(text)
        for w in cut_words:
            if len(w) > 1 and w not in STOP_WORDS:
                words.append(w)
    word_count = Counter(words)
    cloud_text = " ".join(words)
    wc = WordCloud(
        font_path="C:/Windows/Fonts/simhei.ttf",
        width=1000, height=600, background_color="white"
    ).generate(cloud_text)
    wc.to_file("negative_wordcloud.png")
    print("情感分析完成，已生成分布图与差评词云")
    print("高频负面关键词TOP10：", word_count.most_common(10))
    return df

# 执行分析
if __name__ == "__main__":
    comment_sentiment_analysis()

模块 3：定时自动监控任务

设置定时循环采集，每小时同步一次最新评价，新增差评自动打印预警，实现 7×24 小时口碑监控。

import schedule
import time
from module1 import batch_collect_all_comments
from module2 import comment_sentiment_analysis

TARGET_GOODS_ID = "789654321098"

def monitor_task():
    """定时监控主任务"""
    print("===== 启动新一轮商品口碑监控 =====")
    # 1. 拉取最新评价
    batch_collect_all_comments(TARGET_GOODS_ID)
    # 2. 执行情感分析
    df = comment_sentiment_analysis()
    # 3. 筛选最新差评预警
    latest_df = df.tail(10)
    bad_comment = latest_df[latest_df["sentiment"] == "差评"]
    if len(bad_comment) > 0:
        print("⚠️ 监测到新增差评，请及时处理：")
        for idx, row in bad_comment.iterrows():
            print(f"时间：{row['comment_time']} 评价：{row['main_comment']}{row['append_comment']}")
    print("本轮监控任务结束，等待下一轮\n")

# 设置每60分钟执行一次监控
schedule.every(60).minutes.do(monitor_task)

if __name__ == "__main__":
    print("商品口碑监控系统已启动，持续运行中...")
    while True:
        schedule.run_pending()
        time.sleep(30)

四、落地使用场景详解

场景 1：自有店铺口碑维护

定时监控自家主推款商品，一旦出现新增差评、吐槽质量 / 物流 / 尺寸问题，脚本实时预警，运营可第一时间联系买家协商整改，降低负面评价对转化的影响；同时汇总月度差评关键词，同步供应链优化产品缺陷。

场景 2：竞品深度调研

批量添加同行竞品商品 ID，持续采集竞品全周期评价，通过词云对比双方用户痛点，找到竞品短板作为自身产品差异化卖点；统计不同 SKU 规格的好评占比，调整店铺备货、主推规格。

场景 3：新品市场反馈测试

新品上架后开启自动监控，7 天、15 天、30 天分别导出评价分析报告，快速判断用户接受度，若负面反馈集中则及时调整详情页、产品工艺。

五、方案对比：传统采集 VS 本方案

表格

对比维度	页面爬虫采集	本文稳定采集方案
维护成本	极高，页面改版、风控频繁失效	几乎零维护，底层持续更新适配
数据完整性	容易缺失追评、晒图、SKU 信息	原生全字段，无数据丢失
风控风险	频繁验证码、IP 封禁、账号限制	底层规避风控，稳定长期运行
开发难度	需要逆向 JS、代理池、解析适配	开箱即用，仅修改商品 ID 与凭证
自动化能力	定时采集易中断，无预警机制	自带定时任务 + 差评实时预警

六、拓展优化建议

多商品批量监控：将商品 ID 存入列表，循环遍历执行采集，一次性监控十款以上商品；
数据库存储：替换 CSV 文件，接入 SQLite/MySQL 存储海量历史评价，支持长期数据回溯；
消息推送预警：新增企业微信 / 钉钉机器人接口，检测到差评自动推送消息至运营群；
评分趋势统计：按日期分组统计每日好评率，绘制折线图，直观观察口碑波动；
图片素材抓取：基于返回的图片链接，批量下载买家晒图，用于详情页买家秀素材。

结语

电商数据驱动运营已经是行业标配，人工处理海量评价早已跟不上市场节奏。这套轻量化监控分析方案无需复杂开发能力，依托成熟稳定的数据获取能力，搭配 Python 自然语言分析工具，即可搭建完整的商品口碑监控体系。

无论是中小商家日常运营，还是数据分析师做竞品调研、产品迭代分析，都能大幅降低数据获取与处理成本，通过真实用户反馈反向优化商品、服务与运营策略。代码可直接复制运行，仅需替换凭证与商品 ID 即可快速落地。

电商口碑自动化监控 + 全量评论数据分析实战教程（Python 完整落地）

Wed, 15 Jul 2026 16:57:33 +0800

前言

做电商运营、产品、竞品分析的朋友应该都有同一个痛点：想实时掌握自家 / 竞品商品的用户真实反馈，人工翻评论效率极低，批量导出、长期追踪、差评预警、文本情感分析更是耗时耗力。

很多人尝试自研页面抓取方案，却频繁遇到访问限制、页面改版失效、图片 / 追评数据缺失、大量账号风控封禁等问题，维护成本极高。今天分享一套稳定轻量化方案，借助成熟的数据能力底座，实现商品评论全量自动抓取、长期数据监控、批量文本分析、口碑可视化一站式落地，新手半小时就能完整跑通，适配店铺日常运营、竞品调研、产品迭代需求。

一、方案核心优势（对比传统人工 / 自研抓取）

数据维度完整无缺失 自动同步主评、追评、买家晒图 / 视频、购买规格 SKU、评价时间、脱敏买家昵称、点赞互动数据，无需手动解析页面 DOM，不会因为平台页面改版导致采集失效。
长效监控自动更新 配置商品标识后，定时轮询增量数据，新增差评、新增追评会第一时间同步，可搭配消息推送做舆情预警，不用每天手动刷新页面。
合规稳定无风控风险 不用搭建代理池、不用维护账号 Cookie、不用应对平台反爬限制，支持批量商品并行监控，单日可承载上万条数据采集需求。
全链路数据分析闭环 采集完成后直接用 Python 做文本清洗、情感打分、高频词提取、口碑统计，输出表格与可视化图表，直接用于运营复盘与产品优化。

二、前期准备工作

1. 平台账号开通

登录数据服务控制台完成实名认证，创建专属应用，获取两组密钥凭证（后文代码中key、secret参数），开通商品评价数据查询权限，免费额度足够中小卖家日常使用，企业批量监控可升级扩容。

2. 开发环境依赖安装

本地 Python 环境执行以下命令安装所需工具包：

pip install requests pandas jieba wordcloud snowNLP matplotlib

requests：发起远程数据请求
pandas：数据存储、筛选、导出 Excel
jieba：中文评论分词
snowNLP：评论情感正负判定
wordcloud/matplotlib：生成好评差评词云可视化

三、完整采集代码实现（自动分页全量拉取评论）

3.1 基础数据采集工具类

将商品唯一 ID 填入item_id，自动分页循环拉取全部评论，统一整理成结构化数据，自动去重避免重复存储：

import requests
import time
import pandas as pd

class ShopCommentMonitor:
    def __init__(self, access_key, access_secret):
        # 平台密钥，替换成自己控制台获取的凭证
        self.key = access_key
        self.secret = access_secret
        self.base_url = "远程数据服务地址"
        self.all_comment_data = []

    def get_single_page_comment(self, item_id, page_num=1, page_size=20):
        """单页获取商品评价数据"""
        params = {
            "app_key": self.key,
            "app_secret": self.secret,
            "product_id": item_id,
            "page": page_num,
            "size": page_size
        }
        try:
            resp = requests.get(self.base_url, params=params, timeout=20)
            result = resp.json()
            if result.get("code") != 200:
                print(f"第{page_num}页数据获取失败：{result.get('msg')}")
                return None
            return result["data"]
        except Exception as e:
            print(f"请求异常：{str(e)}")
            return None

    def pull_all_comments(self, item_id):
        """循环分页拉取商品全部评论"""
        page = 1
        while True:
            page_data = self.get_single_page_comment(item_id, page)
            if not page_data or len(page_data["list"]) == 0:
                print("全部评论抓取完成")
                break
            # 解析单页评论列表
            for item in page_data["list"]:
                info = {
                    "评论ID": item.get("rate_id", ""),
                    "评价内容": item.get("rate_content", ""),
                    "追评内容": item.get("add_feedback", ""),
                    "评价时间": item.get("rate_date", ""),
                    "购买规格": item.get("auction_sku", ""),
                    "买家昵称": item.get("display_user_nick", ""),
                    "是否带图": 1 if item.get("pics") else 0,
                    "是否有追评": 1 if item.get("add_feedback") else 0
                }
                self.all_comment_data.append(info)
            print(f"已抓取第{page}页，当前累计{len(self.all_comment_data)}条评价")
            page += 1
            time.sleep(1)  # 请求间隔，避免高频访问限制
        return self.all_comment_data

# ------------------- 执行采集逻辑 -------------------
if __name__ == "__main__":
    # 替换为自己的密钥
    AK = "你的access_key"
    SK = "你的access_secret"
    # 目标商品ID，从商品详情页链接提取
    target_item = "1234567890123"
    monitor = ShopCommentMonitor(AK, SK)
    comment_list = monitor.pull_all_comments(target_item)
    # 导出原始评论到Excel本地存档
    df_raw = pd.DataFrame(comment_list)
    df_raw.to_excel("商品原始评论数据.xlsx", index=False)
    print("原始评论已导出至本地Excel")

3.2 定时增量监控脚本（长效舆情监控）

如果需要 7×24 小时追踪新增评价、及时捕捉差评，增加定时循环逻辑，对比本地历史数据实现增量更新：

import schedule

def monitor_task():
    """定时监控任务，每2小时执行一次增量采集"""
    AK = "你的access_key"
    SK = "你的access_secret"
    target_item = "1234567890123"
    monitor = ShopCommentMonitor(AK, SK)
    new_data = monitor.pull_all_comments(target_item)
    # 读取历史存档，合并去重
    try:
        history_df = pd.read_excel("商品原始评论数据.xlsx")
        new_df = pd.DataFrame(new_data)
        merge_df = pd.concat([history_df, new_df]).drop_duplicates(subset=["评论ID"])
        merge_df.to_excel("商品原始评论数据.xlsx", index=False)
        add_count = len(merge_df) - len(history_df)
        print(f"本次监控新增{add_count}条评价")
        # 筛选新增差评预警（可对接企业微信/钉钉推送）
        bad_comment = merge_df[merge_df["评价内容"].str.contains("差|不好|掉色|破损|不值", na=False)]
        if len(bad_comment) > 0:
            print("【预警】检测到负面评价：")
            print(bad_comment[["评价时间","评价内容","购买规格"]])
    except FileNotFoundError:
        print("无历史存档，首次全量采集")

# 设置每2小时自动执行一次监控
schedule.every(2).hours.do(monitor_task)
# 持续运行监控程序
while True:
    schedule.run_pending()
    time.sleep(60)

四、评论深度数据分析代码（情感 + 词云 + 统计）

采集完成后，基于本地 Excel 文件做用户反馈挖掘，区分好评 / 中评 / 差评，提取高频痛点词，直观展示用户核心诉求：

import jieba
from snowNLP import SnowNLP
from wordcloud import WordCloud
import matplotlib.pyplot as plt

# 读取采集好的评论数据
df = pd.read_excel("商品原始评论数据.xlsx")

# 1. 合并主评+追评全部文本
def merge_text(row):
    text = str(row["评价内容"]) + str(row["追评内容"])
    return text.replace("无评价", "").strip()

df["完整文本"] = df.apply(merge_text, axis=1)
df = df[df["完整文本"] != ""]

# 2. 情感打分，划分好评/中评/差评
def get_sentiment(text):
    s = SnowNLP(text)
    score = s.sentiments
    if score >= 0.6:
        return "好评", score
    elif score >= 0.3:
        return "中评", score
    else:
        return "差评", score

sent_result = df["完整文本"].apply(get_sentiment)
df["情感标签"] = [x[0] for x in sent_result]
df["情感分值"] = [x[1] for x in sent_result]

# 3. 基础口碑统计
stat_count = df["情感标签"].value_counts()
print("商品口碑分布统计：")
print(stat_count)
print(f"总评价数：{len(df)}")
print(f"好评占比：{round(stat_count['好评']/len(df)*100,2)}%")

# 4. 生成差评关键词词云（定位产品痛点）
bad_text = " ".join(df[df["情感标签"]=="差评"]["完整文本"].tolist())
cut_words = jieba.cut(bad_text)
word_list = [w for w in cut_words if len(w)>=2 and w not in ["物流","发货","客服"]]
cloud_text = " ".join(word_list)

wc = WordCloud(
    font_path="simhei.ttf",  # 黑体字体文件路径
    width=1000, height=600,
    background_color="white",
    max_words=100
)
wc.generate(cloud_text)
plt.figure(figsize=(12,7))
plt.imshow(wc)
plt.axis("off")
plt.title("差评高频痛点词云", fontsize=16)
plt.show()

# 5. 导出分析结果
df.to_excel("评论情感分析结果.xlsx", index=False)
print("带情感标签的分析文件已导出")

五、落地业务场景说明

场景 1：自有店铺舆情监控

定时脚本自动抓取新增评价，出现负面关键词实时预警，运营第一时间联系买家解决问题，降低差评对转化的影响，长期沉淀用户反馈优化详情页、售后话术。

场景 2：竞品对标分析

批量填入多款竞品商品 ID，一次性拉取全量评价，通过情感分值、高频痛点词对比竞品优缺点，找到差异化产品卖点，调整定价与推广文案。

场景 3：产品迭代调研

汇总全周期评价数据，提取差评高频词汇（如面料薄、尺寸偏小、续航短），同步给供应链 / 产品部优化材质、规格、功能，减少售后退货率。

场景 4：内容营销素材挖掘

筛选高赞好评、带图评价文本，提取用户自发夸赞的关键词，用于商品标题、短视频种草、直通车创意文案，提升转化说服力。

六、避坑实操小贴士

采集频率控制：批量监控建议间隔 1-2 小时，短时间高频请求会触发限流，导致部分数据返回为空；
文本清洗优化：无文字评价、纯晒图数据可在代码中过滤，减少情感分析干扰；
字体文件配置：生成词云必须配置中文字体路径，否则中文会显示方框乱码；
数据存储建议：长期监控建议搭配 MySQL 存储，Excel 仅做短期导出查看，百万级数据表格会卡顿；
多商品批量监控：封装商品 ID 列表循环执行采集函数，支持同时监控数十款商品口碑。

总结

这套方案避开了传统爬虫频繁失效、账号封禁、数据不全的痛点，一套代码完成自动采集 - 长效监控 - 情感分析 - 可视化输出全流程，无需复杂运维。不管是个人卖家、电商运营团队还是产品调研岗，都能低成本搭建属于自己的用户口碑数据体系，把零散的买家评价转化为可落地的运营决策依据。

电商商品口碑自动化监控与文本分析完整实操方案（附可运行 Python 代码）

Tue, 14 Jul 2026 17:27:54 +0800

前言

做电商运营、产品、市场调研的朋友基本都有同一个痛点：想要持续掌握商品真实用户口碑，只能每天手动打开商品页面翻评论，几十上百款商品同步监控时，人工整理、统计、筛选差评耗时巨大，错过负面反馈还会直接影响转化与店铺权重。

之前尝试过浏览器自动化工具、本地模拟浏览采集，要么频繁触发平台风控限制，要么字段缺失拿不到晒图、追评、视频、规格等完整信息，数据杂乱难以二次分析。偶然接触到一款轻量化数据采集工具 Open Claw，仅需简单配置就能稳定批量拉取全维度用户评价，搭配 Python 实现定时监控、数据清洗、情感词频分析，整套流程不用复杂部署，普通笔记本、轻量云服务器都能 7×24 小时自动运行，下面把完整落地流程、可直接复制运行的代码分享给大家。

一、整体方案架构

整套系统分为三层，全程无复杂分布式、无环境门槛：

定时数据同步层：借助 Open Claw 配置商品任务，输入商品编号即可批量拉取综合 / 最新排序评价，自动覆盖评论正文、评价时间、买家昵称、购买规格、晒图、追评、视频素材等全量字段，支持缓存加速，减少重复请求耗时；
数据清洗持久层：Python 脚本自动过滤系统默认好评、空白无效内容、重复评价，统一时间格式，将结构化数据存入本地 CSV，也可无缝对接 MySQL、飞书多维表；
口碑智能分析层：基于分词工具拆分评论文本，区分正向 / 负向反馈，统计差评高频痛点词，生成好评 / 差评词云、情感分布饼图，自动输出商品口碑简报。

工具优势对比

稳定性强：规避平台页面动态渲染、反爬拦截问题，无需维护代理池、Cookie；
字段完整：原生支持抓取追加评价、买家实拍图、商品短视频、SKU 规格信息，普通浏览器采集很难一次性获取；
灵活调度：支持分页循环采集、定时轮询，新增评论实时同步，实现口碑 7×24 小时监控；
多格式输出：原始数据可直接转为 JSON、表格文本，适配各类数据分析脚本。

二、环境前置准备

1. Python 依赖库安装

新建项目文件夹，执行命令安装所需第三方包：

pip install requests pandas jieba wordcloud matplotlib

2. Open Claw 基础配置

注册登录后台，创建数据采集任务，录入需要监控的商品编号；
生成专属访问凭证与任务访问地址（后文代码中DATA_SOURCE_URL替换为自己生成的链接）；
可自定义缓存开关、输出文本语言、分页采集数量，默认开启缓存提升重复采集速度。

三、完整可运行代码

模块 1：定时拉取商品评价 + 数据清洗存储脚本

功能：循环分页采集全量评价，过滤无效数据，增量保存本地文件，支持定时循环实现实时监控

# coding:utf-8
import requests
import pandas as pd
import time
import re
import os

# =====================基础配置区（自行修改）=====================
# Open Claw生成的任务访问链接
DATA_SOURCE_URL = "https://xxx.xxx/task/review?goods_id=600530677643"
# 本地数据存储文件
SAVE_CSV = "goods_review_data.csv"
# 每次采集间隔秒数，防止高频请求
SLEEP_SECONDS = 3
# 是否开启缓存加速，yes/no
CACHE_SWITCH = "yes"
# 采集排序：0综合排序，1最新排序（监控新增评论选1）
SORT_TYPE = 1
# 最大采集页数
MAX_PAGE = 10
# ==============================================================

# 清洗文本：去除换行、空格、特殊符号
def clean_text(text):
    if not text:
        return ""
    text = re.sub(r"\s+", "", str(text))
    text = re.sub(r"[^\u4e00-\u9fa5a-zA-Z0-9，。！？、]", "", text)
    return text

# 请求单页评价数据
def get_review_page(page_num):
    headers = {
        "Accept-Encoding": "gzip",
        "Connection": "close"
    }
    params = {
        "cache": CACHE_SWITCH,
        "page": page_num,
        "sort": SORT_TYPE
    }
    try:
        resp = requests.get(DATA_SOURCE_URL, params=params, headers=headers, timeout=20)
        resp.raise_for_status()
        res_json = resp.json()
        # 状态判断：无数据/任务异常直接返回空列表
        if res_json.get("error_code") != "0000":
            print(f"第{page_num}页数据异常：{res_json.get('error_msg')}")
            return []
        return res_json.get("review_list", [])
    except Exception as e:
        print(f"第{page_num}页请求失败：{str(e)}")
        return []

# 解析单条评价结构化数据
def parse_review_item(item):
    data_dict = {
        "评论ID": item.get("rate_id", ""),
        "买家昵称": item.get("display_user_nick", ""),
        "评价正文": clean_text(item.get("rate_content", "")),
        "追加评价": clean_text(item.get("add_feedback", "")),
        "评价时间": item.get("rate_date", ""),
        "购买规格": item.get("auction_sku", ""),
        "晒图链接": ",".join(item.get("pics", [])) if item.get("pics") else "",
        "追评图片": ",".join(item.get("add_feedback_images", [])) if item.get("add_feedback_images") else "",
        "评价视频": item.get("video", ""),
        "视频封面": item.get("videoCover", "")
    }
    # 过滤空白无意义评价
    if len(data_dict["评价正文"]) < 2 and len(data_dict["追加评价"]) < 2:
        return None
    return data_dict

# 批量采集全页数据
def batch_crawl_all_review():
    all_review_data = []
    for page in range(1, MAX_PAGE + 1):
        print(f"正在采集第{page}页评价...")
        raw_list = get_review_page(page)
        if not raw_list:
            print("无更多评价数据，停止分页采集")
            break
        for item in raw_list:
            parse_data = parse_review_item(item)
            if parse_data:
                all_review_data.append(parse_data)
        time.sleep(SLEEP_SECONDS)
    return all_review_data

# 增量写入CSV，避免重复存储
def save_to_csv(data_list):
    if not data_list:
        print("本次无新增有效评价")
        return
    df_new = pd.DataFrame(data_list)
    # 判断本地文件是否存在，增量合并去重
    if os.path.exists(SAVE_CSV):
        df_old = pd.read_csv(SAVE_CSV, dtype=str)
        df_merge = pd.concat([df_old, df_new], ignore_index=True)
        df_merge = df_merge.drop_duplicates(subset=["评论ID"], keep="last")
        df_merge.to_csv(SAVE_CSV, index=False, encoding="utf-8-sig")
    else:
        df_new.to_csv(SAVE_CSV, index=False, encoding="utf-8-sig")
    print(f"成功存储{len(data_list)}条有效评价，文件路径：{SAVE_CSV}")

# 定时监控主循环（取消注释while True开启7×24小时轮询）
if __name__ == "__main__":
    # while True:
    review_data = batch_crawl_all_review()
    save_to_csv(review_data)
    # time.sleep(3600)  # 每小时自动采集一次，按需调整间隔

模块 2：评论情感分析 + 词云可视化脚本

功能：读取本地评价表格，拆分好评 / 差评，提取高频关键词，生成可视化词云、情感分布饼图

# coding:utf-8
import pandas as pd
import jieba
from collections import Counter
from wordcloud import WordCloud
import matplotlib.pyplot as plt

# 停用词（过滤无意义词汇）
STOP_WORDS = {"很好", "不错", "满意", "还行", "一般", "收到", "发货", "物流", "包装", "客服", "推荐", "购买", "使用"}
CSV_FILE = "goods_review_data.csv"

# 合并所有评论文本
def get_all_comment_text():
    df = pd.read_csv(CSV_FILE, dtype=str)
    text_total = ""
    for idx, row in df.iterrows():
        text_total += row["评价正文"] + row["追加评价"]
    return text_total

# 分词+过滤停用词，统计高频词
def get_word_freq(top_n=20):
    full_text = get_all_comment_text()
    words = jieba.lcut(full_text)
    valid_words = [w for w in words if len(w) > 1 and w not in STOP_WORDS]
    word_count = Counter(valid_words)
    return word_count.most_common(top_n)

# 生成词云图
def generate_word_cloud():
    word_freq = get_word_freq(100)
    word_dict = dict(word_freq)
    wc = WordCloud(
        font_path="C:/Windows/Fonts/simhei.ttf",  # Windows黑体路径，Mac替换为对应字体
        width=1000, height=600,
        background_color="white",
        max_words=100
    )
    wc.generate_from_frequencies(word_dict)
    plt.figure(figsize=(12, 7))
    plt.imshow(wc, interpolation="bilinear")
    plt.axis("off")
    plt.title("商品评价关键词词云", fontsize=16)
    plt.savefig("review_wordcloud.png", dpi=300, bbox_inches="tight")
    plt.show()
    print("词云图片已保存为 review_wordcloud.png")

# 简易情感区分（根据负面关键词划分正负向）
def sentiment_statistics():
    negative_key = {"差", "不好", "掉色", "起球", "破损", "异味", "偏小", "偏大", "质量差", "退货", "瑕疵"}
    df = pd.read_csv(CSV_FILE, dtype=str)
    good_count = 0
    bad_count = 0
    total = len(df)
    for idx, row in df.iterrows():
        content = row["评价正文"] + row["追加评价"]
        if any(k in content for k in negative_key):
            bad_count += 1
        else:
            good_count += 1
    # 绘制情感分布饼图
    plt.figure(figsize=(8, 8))
    label = ["正向评价", "负面评价"]
    num = [good_count, bad_count]
    color = ["#66cc66", "#ff6666"]
    plt.pie(num, labels=label, colors=color, autopct="%1.2f%%", startangle=90)
    plt.title(f"商品口碑情感分布（总评价{total}条）")
    plt.savefig("sentiment_pie.png", dpi=300)
    plt.show()
    print(f"正向评价：{good_count}条，负面评价：{bad_count}条")
    print("TOP20高频用户关键词：", get_word_freq(20))

if __name__ == "__main__":
    sentiment_statistics()
    generate_word_cloud()

四、实操落地步骤详解

步骤 1：配置 Open Claw 采集任务

后台新建商品采集任务，填入需要监控的商品 ID；
调整采集参数：缓存开启yes、排序选择1（最新）用于监控新增差评；
复制生成的专属访问链接，替换代码中DATA_SOURCE_URL；
支持多商品批量创建任务，循环调用脚本实现多款商品同步监控。

步骤 2：数据采集与增量存储

运行第一段采集脚本，首次执行会自动创建goods_review_data.csv，所有评价完整入库；取消代码中while True注释、设置time.sleep(3600)即可实现每小时自动拉取新评价，无需人工重复操作。采集到的结构化字段包含：

基础文本：评价正文、追加追评、买家昵称
素材信息：晒图链接、追评图片、商品短视频地址
商品属性：下单 SKU 规格、评价时间
唯一标识：评论 ID（用于自动去重，避免重复统计）

步骤 3：口碑数据分析与可视化

运行第二段分析脚本，自动输出两项可视化文件：

review_wordcloud.png：用户高频关键词词云，快速抓取用户核心关注点（面料、尺寸、物流、做工等）；
sentiment_pie.png：正负评价占比饼图，直观判断商品整体口碑；控制台同步打印 TOP20 高频词汇、正负评价数量，可直接用于运营复盘、产品优化报告。

五、业务场景落地价值

差评实时预警：定时轮询抓取最新评价，脚本可拓展对接企业微信 / 飞书推送负面评价，2 小时内响应用户投诉，降低店铺纠纷；
竞品调研：批量采集同行同款商品评价，提取竞品痛点，优化自身产品文案、尺码、材质；
产品迭代依据：从高频关键词挖掘用户需求，比如大量评价提及 “尺码偏小”，可调整商品详情尺码表、修改版型；
投放素材优化：提取好评高频词作为直通车、短视频宣传关键词，提升转化点击率；
店铺数据沉淀：长期积累评价数据，按月统计口碑变化，直观看到产品优化后的用户反馈提升。

六、常见问题优化方案

采集速度慢：开启缓存参数cache=yes，减少重复请求平台数据源，采集效率提升 60% 以上；
文本乱码：CSV 文件写入统一使用encoding="utf-8-sig"，Windows、Mac 打开表格无中文乱码；
词云无文字：修改代码中font_path字体路径，Windows 使用 simhei.ttf，Mac 用 PingFang.ttc；
数据重复堆积：代码内置根据评论 ID 自动去重逻辑，每次采集仅新增未收录评价，文件不会无限膨胀；
大量商品监控：循环遍历多任务链接，配合轻量云服务器 7×24 小时挂机运行，一次性监控几十款商品无压力。

七、总结

传统人工整理评价的方式效率极低，而 Open Claw 搭配 Python 搭建的自动化口碑监控体系，零复杂开发成本，兼顾采集稳定性与数据完整性。整套流程从数据自动拉取、清洗存储到情感可视化分析一站式完成，运营、产品、市场人员不用学习复杂爬虫、分布式技术，复制代码修改少量配置即可落地，长期使用能大幅降低用户反馈调研成本，快速从海量评论中挖掘业务优化机会。

电商竞品全自动监控 + 数据分析落地教程，半小时搞定全品类行情追踪

Mon, 13 Jul 2026 17:50:54 +0800

前言

做电商运营、产品调研、市场分析的朋友应该都有同款痛点：想盯同行定价、活动优惠、销量波动、评价风向，手动复制粘贴商品页面、整理表格，每天耗费 2-3 小时；大促节点竞品调价、上新节奏完全跟不上，错过最佳调价、备货时机；多品类、几十上百款商品同步追踪时，人工统计极易漏数据、错数据，历史行情也没法留存复盘。

试过自建采集脚本、多浏览器轮询监控，要么频繁触发平台限制，要么代码维护成本极高，新增监控商品就要改一堆逻辑。后来接触到一套成熟稳定的数据抓取工具（Open Claw），无需复杂底层搭建，开箱实现商品全维度实时监控，搭配 Python 完成数据清洗、可视化分析，个人 / 中小企业不用投入服务器、代理池成本，单人就能完成全流程市场监测。

本文完整分享落地流程，附带可直接运行的 Python 代码，覆盖商品定时监控、数据存储、行情统计、价格波动预警四大核心场景，适配国内主流电商平台商品数据处理。

一、方案整体思路

整套流程分为三层，低门槛实现自动化监测：

数据获取层：依托成熟云端数据能力，输入商品标识即可一键拉取完整商品信息（售价、活动价、销量、库存、图文详情、评价、店铺资质等），内置访问调度机制，规避平台访问限制，支持批量商品批量同步。
自动化监控层：设置定时轮询周期（10 分钟 / 1 小时 / 每日），对比前后两次抓取的数据，识别价格变动、上新、下架、优惠活动、差评激增等异动，自动留存历史快照。
数据分析应用层：Python 读取本地存储的数据，完成数据清洗、价格区间分布统计、销量趋势绘图、竞品分层分析，生成 Excel 报表，支持配置消息提醒（企业微信 / 邮件）。

优势对比传统手动采集 / 自研爬虫：

无需搭建代理池、解决访问封禁、验证码拦截等问题
批量一次性导入上百款商品，统一管理监控清单
自动留存历史数据，支持长期行情回溯
代码轻量化，零基础运营也能直接复用

二、环境准备

本地仅需安装 Python 基础依赖，执行安装命令：

pip install requests pandas matplotlib openpyxl schedule

依赖说明：

requests：本地程序与数据工具交互
pandas：数据清洗、表格导出
matplotlib：行情可视化绘图
schedule：本地定时任务，循环监控
openpyxl：读写 Excel 文件

三、完整实操代码实现

3.1 基础配置与数据拉取核心函数

import requests
import json
import pandas as pd
import matplotlib.pyplot as plt
import schedule
import time
from datetime import datetime
import os

# 基础配置
# 工具后台获取的个人凭证（替换为自己账号生成的密钥）
TOKEN = "xxxxxxxxxxxxxxxxxxxx"
# 需要监控的商品ID列表，批量新增直接追加
MONITOR_GOODS = ["723456123987", "723456987654", "723456543210"]
# 数据存储文件
DATA_FILE = "goods_monitor_data.csv"
# 绘图字体适配中文
plt.rcParams["font.family"] = ["SimHei", "WenQuanYi Micro Hei"]
plt.rcParams["axes.unicode_minus"] = False

def get_goods_info(item_id):
    """
    调用云端数据能力，获取单款商品完整信息
    :param item_id: 商品唯一标识
    :return: 结构化商品字典数据
    """
    headers = {
        "Content-Type": "application/json"
    }
    params = {
        "token": TOKEN,
        "item_id": item_id
    }
    # 工具数据请求地址
    resp = requests.get("https://data-claw.collector.work/shop/goods", headers=headers, params=params)
    res_data = resp.json()
    if res_data.get("code") != 200:
        print(f"商品{item_id}数据获取失败：{res_data.get('msg')}")
        return None
    info = res_data.get("data")
    # 提取核心业务字段
    result = {
        "crawl_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "item_id": item_id,
        "title": info.get("title"),
        "price": float(info.get("price", 0)),
        "prom_price": float(info.get("prom_price", 0)),
        "sales": int(info.get("sales", 0)),
        "stock": int(info.get("stock", 0)),
        "shop_name": info.get("shop_name"),
        "score": float(info.get("score", 0)),
        "review_count": int(info.get("review_count", 0))
    }
    return result

3.2 数据存储与异动检测函数

def save_and_check_change(goods_data):
    """
    写入本地CSV历史文件，对比历史数据识别价格/销量异动
    """
    if not goods_data:
        return
    # 初始化数据文件
    if not os.path.exists(DATA_FILE):
        df_init = pd.DataFrame([goods_data])
        df_init.to_csv(DATA_FILE, index=False, encoding="utf-8-sig")
        print(f"首次录入商品{goods_data['item_id']}监控数据")
        return
    # 读取历史数据
    df_history = pd.read_csv(DATA_FILE, encoding="utf-8-sig")
    item_id = goods_data["item_id"]
    # 筛选该商品历史最新一条记录
    history_item = df_history[df_history["item_id"] == item_id]
    # 写入新数据
    df_new = pd.concat([df_history, pd.DataFrame([goods_data])], ignore_index=True)
    df_new.to_csv(DATA_FILE, index=False, encoding="utf-8-sig")

    # 异动判断逻辑
    if len(history_item) > 0:
        last_record = history_item.iloc[-1]
        price_old = last_record["prom_price"]
        price_new = goods_data["prom_price"]
        sales_old = last_record["sales"]
        sales_new = goods_data["sales"]
        # 价格变动预警（差价大于5元触发提醒）
        if abs(price_new - price_old) > 5:
            print(f"【价格异动提醒】商品{goods_data['title']}，原价{price_old}→现价{price_new}")
        # 销量暴涨提醒（单次增量超200）
        if sales_new - sales_old > 200:
            print(f"【销量暴涨提醒】商品{goods_data['title']}，新增销量{sales_new - sales_old}")

3.3 批量监控主任务

def monitor_task():
    """单次批量监控任务，循环所有商品"""
    print(f"\n===== 开始执行批量监控 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} =====")
    for good_id in MONITOR_GOODS:
        data = get_goods_info(good_id)
        if data:
            save_and_check_change(data)
    print("===== 本轮监控执行完成 =====")

3.4 数据分析与可视化报表函数

def data_analysis_export():
    """读取全部历史数据，统计分析并导出Excel+价格趋势图"""
    if not os.path.exists(DATA_FILE):
        print("暂无监控数据，无法分析")
        return
    df = pd.read_csv(DATA_FILE, encoding="utf-8-sig")
    # 1. 基础统计
    stat_df = df.groupby(["item_id", "title"]).agg(
        min_price=("prom_price", "min"),
        max_price=("prom_price", "max"),
        avg_price=("prom_price", "mean"),
        total_sales=("sales", "max"),
        latest_score=("score", "last")
    ).reset_index()
    # 导出统计报表
    stat_df.to_excel("竞品行情统计报表.xlsx", index=False)
    print("行情统计报表已生成：竞品行情统计报表.xlsx")

    # 2. 绘制价格趋势图（取第一个商品做示例）
    target_id = MONITOR_GOODS[0]
    target_data = df[df["item_id"] == target_id].sort_values("crawl_time")
    if len(target_data) > 1:
        fig, ax = plt.subplots(figsize=(12, 6))
        ax.plot(target_data["crawl_time"], target_data["prom_price"], marker="o", label="活动售价")
        ax.set_title("商品价格历史波动趋势")
        ax.set_xlabel("监控时间")
        ax.set_ylabel("价格（元）")
        plt.xticks(rotation=45)
        ax.legend()
        plt.tight_layout()
        plt.savefig("price_trend.png")
        print("价格趋势图表已保存：price_trend.png")

3.5 定时循环启动入口

if __name__ == "__main__":
    # 设置每1小时自动执行一轮监控，可修改为30/60/120分钟
    schedule.every(1).hours.do(monitor_task)
    # 程序启动先执行一次监控+数据分析
    monitor_task()
    data_analysis_export()
    print("\n监控程序后台运行中，每小时自动同步数据，按Ctrl+C终止")
    # 持续循环调度
    while True:
        schedule.run_pending()
        time.sleep(60)

四、代码使用说明

凭证替换：代码中TOKEN替换为 Open Claw 后台个人专属密钥，登录工具后台即可一键生成，权限默认开放商品全字段查询；
监控商品新增：修改MONITOR_GOODS列表，填入平台商品 ID，支持几十上百款批量监控；
监控周期调整：schedule.every(1).hours修改数字，0.5 代表 30 分钟，2 代表 2 小时；
预警阈值自定义：异动检测中价格差值、销量增量数值可按需修改，适配低价小商品 / 高客单价大件；
拓展消息推送：可自行接入企业微信机器人、邮件发送模块，异动时自动推送消息到工作群，不用反复查看程序日志。

五、落地效果与业务应用场景

场景 1：竞品定价监控

定时抓取同行日常售价、大促活动价，通过生成的价格趋势图，清晰看到竞品调价节点、优惠力度，结合自身成本调整定价区间，避免低价内卷或定价过高丢失流量。统计报表自动输出单品历史最低价、均价，快速锁定竞品价格带。

场景 2：爆款销量追踪

持续监测竞品销量变化，识别短时间销量暴涨的爆款，反向拆解标题、主图、活动玩法、评价关键词，复刻爆款运营逻辑；同时监控自身同款商品，及时发现销量下滑，排查价格、评价、流量问题。

场景 3：批量市场类目分析

导入整个类目几十上百款商品 ID，长期采集后用 Pandas 做整体价格区间分布、店铺评分分层、销量梯队划分，输出类目市场报告，作为选品、备货、上新的核心数据依据，替代人工逐页截图统计。

场景 4：库存与活动监测

同步抓取商品库存数据，监控竞品是否清仓、断货，抓住竞品缺货窗口期加大推广；识别满减、优惠券、第二件半价等活动，同步跟进营销活动节奏。

六、避坑优化小技巧

数据存储轻量化：长期监控会积累大量历史快照，可增加定时清理 30 天前旧数据的逻辑，避免 CSV 文件过大加载缓慢；
异常容错增强：代码可新增 try-except 捕获网络波动，监控失败自动重试，防止单次接口超时中断整轮批量任务；
多账号分区监控：若需要区分不同类目、不同竞品店铺，可复制多份监控列表，分开存储数据文件，报表互不干扰；
本地部署轻量化：程序支持 Windows、Mac、Linux 服务器后台挂起运行，服务器 24 小时不间断监控，本地电脑仅需运行时查看报表即可。

七、总结

以往做电商市场监测，要么靠人工低效统计，要么投入大量开发成本搭建采集体系，对中小商家、个人运营极不友好。借助 Open Claw 一站式数据能力，搭配几十行轻量化 Python 代码，零底层开发成本实现7×24 小时全自动商品监控，自动留存历史数据、识别异动、输出可视化分析报表。

整套方案落地门槛极低，运营不懂编程也能直接复制代码修改配置使用，长期积累的市场数据可以形成完整竞品数据库，为定价、选品、营销、备货提供客观数据支撑，大幅降低市场调研人力成本。

后续可在此基础上拓展进阶功能：评价文本情感分析、类目热销词提取、多平台商品比价、自动生成每日市场简报，完整搭建属于自己的轻量化电商数据监测体系。

电商竞品评论自动监控 + 深度文本分析完整实操方案（附可运行 Python 代码）

Fri, 10 Jul 2026 16:01:32 +0800

前言

做电商运营、产品、市场调研的朋友，几乎都绕不开竞品用户评价分析这件事。传统手动复制评论、Excel 整理的方式效率极低：单款商品上万条评价，逐条筛选差评、统计吐槽点、跟踪新增反馈，人工至少耗费半天；多商品长期监控更是分身乏术，等整理完数据，竞品已经完成一轮产品迭代。

本文分享一套轻量化落地方案，借助 OpenClaw 一站式数据能力，无需复杂环境搭建，半小时完成商品评论全量获取、实时增量监控、情感分词、关键词统计，附带完整可运行代码，个人 / 小团队均可直接复用，适配淘宝、天猫国内电商场景。

一、方案核心优势

全量数据快速拉取 支持指定商品批量读取全部买家评价，包含星级、评价正文、晒图、追评、评价时间、买家标签，无需逐页翻页加载，规避平台访问限制。
增量实时监控 配置定时轮询任务，自动抓取上新产生的评价，一旦出现大批量负面反馈、高频吐槽关键词，可触发本地日志提醒，第一时间捕捉竞品产品缺陷、服务漏洞。
轻量化文本分析 内置基础分词、情感打分功能，自动区分好评 / 中评 / 差评，统计高频吐槽词（如掉色、尺寸偏小、发货慢），输出结构化统计结果，直接用于产品优化、营销文案调整。
低门槛部署 仅需 Python 基础环境，无需搭建代理集群、解析复杂页面加密参数，屏蔽平台反爬拦截，新手复制代码即可运行。

二、前期准备工作

1. 环境依赖安装

本地安装 Python3.8 及以上版本，执行 pip 安装所需工具包：

pip install requests pandas jieba wordcloud

2. 工具基础配置

三、完整实操代码（分三大模块：数据拉取、增量监控、评论分析）

模块 1：批量获取商品全部评价数据

作用：一次性导出历史所有评价，存入本地 Excel，用于初期竞品大盘分析

import requests
import pandas as pd
import time

# 基础配置
TOKEN = "你的个人身份凭证字符串"
TARGET_ITEM_IDS = ["728945612345", "728945678901"]  # 待监控商品ID列表
SAVE_PATH = "电商评论原始数据.xlsx"

def get_all_reviews(item_id):
    review_list = []
    page = 1
    while True:
        # 构建请求参数
        params = {
            "token": TOKEN,
            "item_id": item_id,
            "page": page,
            "limit": 50
        }
        resp = requests.get("https://data-claw.fetchcontent.com/comment", params=params)
        res_data = resp.json()
        # 判断无更多数据则退出循环
        if len(res_data.get("data", [])) == 0:
            break
        # 清洗单条评价字段
        for item in res_data["data"]:
            temp_dict = {
                "商品ID": item_id,
                "评价时间": item.get("publish_time"),
                "星级评分": item.get("star"),
                "评价内容": item.get("content"),
                "是否追评": item.get("is_add_comment"),
                "晒图数量": item.get("img_num"),
                "买家标签": item.get("buyer_tag")
            }
            review_list.append(temp_dict)
        page += 1
        time.sleep(1.2)  # 限流防高频访问
    return review_list

if __name__ == "__main__":
    total_data = []
    for goods_id in TARGET_ITEM_IDS:
        print(f"正在采集商品{goods_id}评价数据...")
        goods_reviews = get_all_reviews(goods_id)
        total_data.extend(goods_reviews)
        print(f"商品{goods_id}共获取{len(goods_reviews)}条评价")
    # 保存至本地Excel
    df = pd.DataFrame(total_data)
    df.to_excel(SAVE_PATH, index=False)
    print(f"全部数据采集完成，已保存至{SAVE_PATH}")

模块 2：定时增量监控新增评价（核心监控功能）

作用：定时运行，对比历史数据，只抓取新增评价，实时预警负面反馈

import requests
import pandas as pd
import time
from datetime import datetime

TOKEN = "你的个人身份凭证字符串"
TARGET_ITEM_IDS = ["728945612345", "728945678901"]
HISTORY_FILE = "电商评论原始数据.xlsx"
NEW_REVIEW_LOG = "新增评价实时日志.txt"
# 负面关键词库，可自行扩充
BAD_WORDS = ["色差", "掉色", "破损", "发货慢", "质量差", "尺寸不符", "气味大"]

def get_new_reviews(item_id):
    new_data = []
    params = {
        "token": TOKEN,
        "item_id": item_id,
        "page": 1,
        "limit": 30
    }
    resp = requests.get("https://data-claw.fetchcontent.com/comment", params=params)
    res_data = resp.json()
    return res_data.get("data", [])

def monitor_task():
    # 读取历史已有评价
    try:
        history_df = pd.read_excel(HISTORY_FILE)
        history_content_set = set(history_df["评价内容"].tolist())
    except:
        history_content_set = set()
    
    today = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    log_text = f"\n===== {today} 监控轮次 =====\n"
    alert_flag = False

    for goods_id in TARGET_ITEM_IDS:
        fresh_comments = get_new_reviews(goods_id)
        for comment in fresh_comments:
            content = comment.get("content", "")
            # 判断是否为全新评价
            if content not in history_content_set and content.strip() != "":
                log_text += f"【商品{goods_id}】{comment['publish_time']} | {comment['star']}星 | {content}\n"
                history_content_set.add(content)
                # 负面关键词预警
                for word in BAD_WORDS:
                    if word in content:
                        log_text += f"⚠️ 负面预警：命中关键词【{word}】\n"
                        alert_flag = True
    # 写入日志文件
    with open(NEW_REVIEW_LOG, "a", encoding="utf-8") as f:
        f.write(log_text)
    if alert_flag:
        print("【重要提醒】本轮监控发现负面评价，请查看日志文件！")
    else:
        print("本轮监控完成，无新增负面反馈")

# 定时循环，每30分钟执行一次监控
if __name__ == "__main__":
    while True:
        monitor_task()
        time.sleep(1800)

模块 3：评论文本深度分析（情感统计 + 高频词提取）

作用：基于采集到的全量评价，自动统计好评差评占比、生成吐槽关键词词频表

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

plt.rcParams["font.sans-serif"] = ["SimHei"]
plt.rcParams["axes.unicode_minus"] = False
DATA_FILE = "电商评论原始数据.xlsx"
STOP_WORDS = ["很好", "不错", "还行", "满意", "物流快", "质量好", "收到", "非常"]

def analysis_comments():
    df = pd.read_excel(DATA_FILE)
    # 1. 星级分布统计
    star_count = df["星级评分"].value_counts().sort_index()
    print("===== 商品星级分布统计 =====")
    print(star_count)
    # 划分好评/中差评
    good = df[df["星级评分"] >= 4].shape[0]
    bad = df[df["星级评分"] <= 3].shape[0]
    total = df.shape[0]
    print(f"好评占比：{round(good/total*100,2)}%，中差评占比：{round(bad/total*100,2)}%")

    # 2. 提取中差评高频关键词
    bad_df = df[df["星级评分"] <= 3]
    all_text = "".join(bad_df["评价内容"].astype(str))
    words = jieba.lcut(all_text)
    word_filter = [w for w in words if len(w)>=2 and w not in STOP_WORDS]
    word_freq = Counter(word_filter)
    top20 = word_freq.most_common(20)
    print("\n===== 中差评Top20高频关键词 =====")
    for word, num in top20:
        print(f"{word}：{num}次")
    
    # 3. 绘制词频柱状图
    words_list = [i[0] for i in top20]
    count_list = [i[1] for i in top20]
    plt.figure(figsize=(12,6))
    plt.bar(words_list, count_list)
    plt.xticks(rotation=45)
    plt.title("竞品差评高频关键词统计")
    plt.tight_layout()
    plt.savefig("差评关键词分布图.png")
    print("\n词频图表已保存至本地！")

if __name__ == "__main__":
    analysis_comments()

四、落地使用流程详解

初始化全量采集 运行第一段代码，填入需要调研的竞品商品 ID，一次性拉取全部历史评价，生成基础 Excel 数据集，完成竞品口碑大盘摸底。
开启 7×24 小时监控 后台常驻运行第二段监控脚本，每 30 分钟自动扫描新增评价，一旦出现色差、破损、发货慢等负面词汇，日志文件会标记预警，不用人工每日反复查看商品页面。
周期性深度分析 每周执行第三段分析代码，自动计算好评差评比例、提取用户集中吐槽点，生成可视化图表，输出可直接用于产品优化、运营调整的分析结论。

五、实战落地场景分享

场景 1：新品上线竞品对标调研

新品筹备阶段，批量采集同赛道头部 3-5 款竞品全部评价，通过词频分析找到用户核心痛点：比如服装类目集中吐槽 “尺码偏小”，开发阶段直接调整版型；小家电频繁反馈 “噪音大”，优化内部电机结构，精准打造差异化卖点。

场景 2：店铺口碑风险预警

自有店铺 + 对标竞品同时监控，一旦竞品集中出现同类差评，预判行业共性问题，提前调整详情页说明、升级售后方案，规避同类投诉；自身商品新增大量负面评价时，快速定位问题批次，及时联系买家处理。

场景 3、营销内容优化

提取好评高频正向词汇（面料柔软、性价比高、包装精致），融入直通车标题、商品详情、短视频文案；规避差评高频负面关键词对应的宣传话术，降低用户心理预期落差，提升转化。

六、常见问题优化方案

采集速度慢、频繁中断 代码内已设置 1.2 秒间隔限流，若商品评价超 10 万条，可拆分商品 ID 分批次采集，避免单次请求数据量过大。
分词关键词杂乱无意义 可自行扩充STOP_WORDS停用词列表，过滤无分析价值词汇，聚焦产品、物流、服务相关核心词。
监控日志文件过大 新增定时清理脚本，按月归档历史日志，只保留近 30 天新增评价记录，节省本地存储空间。

七、总结

这套基于 OpenClaw 搭建的电商评论监控分析方案，完美解决传统人工整理数据低效、反馈滞后的痛点。无需复杂底层开发，三段代码覆盖数据采集 - 实时监控 - 智能分析全流程，个人运营、中小团队、市场调研人员均可快速上手。

长期持续监控竞品用户反馈，能持续挖掘产品迭代机会、规避经营风险，把零散的买家评价转化为可落地的数据决策依据，在同质化电商竞争中建立信息优势。