万邦api博客

Open Claw 落地京东商品自动化监控与数据分析，零基础搭建长效运营数据系统（完整可运行 Python 教程）

Sun, 02 Aug 2026 11:00:30 +0800

前言

做京东电商运营、竞品调研、无货源铺货、品牌数据复盘的从业者，长期会面临几个棘手痛点：

人工盯守几十上百款竞品，价格变动、库存缺货、活动调价发现严重滞后；
手动整理商品销量、评价、规格参数、历史价格，统计效率极低，数据误差大；
普通简易采集工具字段残缺，拿不到多 SKU 明细、图文详情、活动优惠、类目参数等深度数据；
爬虫采集容易风控封禁，频繁失效，维护成本居高不下。

近期一直在使用 Open Claw 做电商数据自动化落地，借助一款高阶商品详情查询能力，仅用几十行代码，就搭建出全天候竞品监控 + 结构化数据分析 + 数据存档一体化工具，无需频繁维护采集规则，稳定性强、字段全面。本文从原理讲解、环境搭建、基础单商品拉取、定时监控、批量数据分析、数据持久化全流程落地，所有代码复制即可运行，适配个人开发者、中小型电商团队日常使用，内容适合 CSDN、知乎技术社区发布。

一、方案整体思路介绍

1. Open Claw 定位

Open Claw 是轻量化自动化调度框架，核心作用是统一规范化数据拉取、任务调度、异常重试、数据清洗、规则判断，不用手动编写网络请求容错、签名校验、接口轮询逻辑，专注业务逻辑开发。搭配高阶商品详情查询能力，可以一次性获取京东商品全维度信息：基础标题、实时售价、划线原价、多规格 SKU、库存、店铺信息、商品图文、详情参数、累计销量、评价标签、优惠活动、历史价格区间等全部内容。

2. 整套系统三大核心模块

基础数据拉取模块：单商品精准采集，自定义筛选返回字段，精简冗余数据；
定时监控预警模块：循环巡检商品，价格下跌、库存清零、上新活动自动告警；
数据分析存档模块：批量采集数据清洗，计算折扣率、销量增速，落地 CSV 表格长期复盘。

3. 前置准备

运行环境：Python3.8 及以上，安装依赖库requests；
身份凭证：提前申请专属密钥对（key、secret）；
商品 ID：京东商品链接末尾数字串即为商品唯一 ID；安装依赖命令：

pip install requests

二、第一阶段：基础开发 —— 单商品全量数据稳定拉取

这是整个系统的底层基础，先完成单次商品详情稳定获取，Open Claw 会自动处理网络超时、请求重试、缓存加速、格式转换，我们只需要配置参数、解析返回数据。

完整基础代码

import requests
import json
from datetime import datetime

# ============ 基础配置区，自行修改 ============
KEY = "填写你的专属key"
SECRET = "填写你的专属secret"
# 目标京东商品ID，自行替换
TARGET_ITEM_ID = "10335871600"
# 可选参数配置：json返回格式，关闭缓存实时最新数据
BASE_PARAMS = {
    "key": KEY,
    "secret": SECRET,
    "cache": "no",
    "result_type": "json"
}

class JdGoodsClient:
    def __init__(self):
        self.base_url = "https://api-gw.xxx.cn/jd/item_get_pro"

    def get_single_goods_detail(self, item_id, need_fields=None):
        """
        获取单品高阶详情
        :param item_id: 商品ID
        :param need_fields: 自定义需要返回的字段，None则返回全量字段
        :return: 结构化商品字典 / None
        """
        params = BASE_PARAMS.copy()
        params["num_iid"] = item_id
        if need_fields is not None:
            params["fields"] = need_fields
        try:
            # Open Claw统一调度发起网络请求，内置3次超时重试
            resp = requests.get(url=self.base_url, params=params, timeout=12)
            resp.raise_for_status()
            raw_data = resp.json()
            # 请求状态校验
            if raw_data.get("code") != 200:
                print(f"数据拉取失败，原因：{raw_data.get('msg', '未知异常')}")
                return None
            return raw_data.get("item", {})
        except Exception as e:
            print(f"网络请求异常：{str(e)}")
            return None

# 基础调用测试
if __name__ == "__main__":
    client = JdGoodsClient()
    # 示例：仅获取标题、价格、销量、SKU精简字段
    simple_data = client.get_single_goods_detail(TARGET_ITEM_ID, need_fields="title,price,original_price,sales,sku_list")
    if simple_data:
        print("=====精简商品信息=====")
        print(json.dumps(simple_data, ensure_ascii=False, indent=2))

    # 获取商品全量详情
    full_data = client.get_single_goods_detail(TARGET_ITEM_ID)
    if full_data:
        print("\n=====商品全量详情提取关键信息=====")
        info = {
            "采集时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "商品标题": full_data.get("title"),
            "当前售价": full_data.get("price"),
            "原价": full_data.get("original_price"),
            "总销量": full_data.get("sales"),
            "店铺名称": full_data.get("seller_info", {}).get("shop_name"),
            "SKU规格数量": len(full_data.get("sku_list", []))
        }
        print(json.dumps(info, ensure_ascii=False, indent=2))

代码讲解与踩坑说明

核心入参说明

num_iid：必填，商品唯一编号，精准定位采集目标；
cache：可选，yes 为读取缓存（速度快、非实时），no 强制实时刷新，监控场景务必设置为 no；
fields：可选，按需填写字段，大幅减少数据体积，加快解析速度。

返回数据核心可用字段 基础信息：标题、售价、原价、品牌、类目、主图图集；商业数据：总销量、评价数量、好评率、优惠活动；深度明细：多 SKU 价格、库存、商品详情图文、规格参数、配送时效。
常见报错排查密钥填写错误、商品 ID 失效、网络波动，框架自带重试，无需额外编写复杂容错逻辑。

三、第二阶段：核心实战 —— 全天候自动化商品监控系统

基础拉取稳定后，搭建定时巡检监控脚本，实现价格下跌预警、库存清零提醒，适配竞品盯价、备货预警、活动蹲点场景。

监控完整代码

import time
import requests
import json
from datetime import datetime

KEY = "填写你的专属key"
SECRET = "填写你的专属secret"
BASE_URL = "https://api-gw.xxx.cn/jd/item_get_pro"
BASE_PARAMS = {
    "key": KEY,
    "secret": SECRET,
    "cache": "no",
    "result_type": "json"
}

class JdMonitor:
    def get_goods_data(self, item_id):
        params = BASE_PARAMS.copy()
        params["num_iid"] = item_id
        try:
            resp = requests.get(BASE_URL, params=params, timeout=10)
            res = resp.json()
            if res.get("code") == 200:
                return res.get("item", {})
            return None
        except:
            return None

    def single_monitor(self, item_id, warn_price, check_interval=600):
        """
        单品循环监控
        :param item_id: 商品ID
        :param warn_price: 预警价格，现价低于该值触发提醒
        :param check_interval: 巡检间隔，单位秒，默认10分钟
        """
        print(f"【监控启动】商品ID:{item_id}，降价预警阈值：{warn_price}元，巡检间隔：{check_interval}秒")
        while True:
            goods = self.get_goods_data(item_id)
            if not goods:
                print(f"{datetime.now()} 数据获取失败，等待下一轮巡检...")
                time.sleep(check_interval)
                continue

            title = goods.get("title")
            now_price = float(goods.get("price", 0))
            stock = goods.get("quantity", "未知")
            shop = goods.get("seller_info", {}).get("shop_name")

            print(f"\n[{datetime.now()}]")
            print(f"商品：{title}")
            print(f"店铺：{shop}  实时价格：{now_price}  库存：{stock}")

            # 价格预警判断
            if now_price <= warn_price:
                print("⚠️⚠️⚠️【降价预警】当前价格低于预设阈值！")
            # 库存预警
            if stock == 0 or stock == "0":
                print("⚠️⚠️⚠️【库存预警】商品已缺货！")

            time.sleep(check_interval)

    def batch_monitor(self, monitor_list, check_interval=900):
        """批量多商品监控
        monitor_list格式：[{"id":"xxx","warn_price":xxx},...]
        """
        print(f"【批量监控启动】共监控{len(monitor_list)}款商品，巡检间隔{check_interval}秒")
        while True:
            for item in monitor_list:
                iid = item["id"]
                wp = item["warn_price"]
                goods = self.get_goods_data(iid)
                if not goods:
                    print(f"{datetime.now()} 商品{iid}采集失败")
                    continue
                price = float(goods.get("price",0))
                title = goods.get("title")
                if price <= wp:
                    print(f"[{datetime.now()}]【批量预警】{title} 价格{price}低于阈值{wp}")
            time.sleep(check_interval)

# 运行监控
if __name__ == "__main__":
    monitor = JdMonitor()
    # 1.单品监控示例
    # monitor.single_monitor(item_id="10335871600", warn_price=199, check_interval=300)

    # 2.批量监控示例，自行新增商品
    batch_list = [
        {"id": "10335871600", "warn_price": 195},
        {"id": "10224567891", "warn_price": 269}
    ]
    monitor.batch_monitor(batch_list, check_interval=600)

落地使用说明

巡检间隔建议设置 300~1800 秒，高频采集无需担心风控；
可拓展改造：对接企业微信、钉钉机器人，预警消息推送至工作群；
批量监控适合店铺竞品全链路盯守，单品监控适合爆款蹲活动降价。

四、第三阶段：进阶落地 —— 批量数据分析 + 数据存档复盘

单纯监控只能看实时状态，长期运营需要历史数据做销量、价格走势分析。接下来实现批量采集、数据清洗、指标计算、自动写入 CSV 表格存档，后续可以用 Excel、Pandas 做可视化分析。

数据分析 + 持久化完整代码

import requests
import json
import csv
from datetime import datetime

KEY = "填写你的专属key"
SECRET = "填写你的专属secret"
BASE_URL = "https://api-gw.xxx.cn/jd/item_get_pro"
BASE_PARAMS = {
    "key": KEY,
    "secret": SECRET,
    "cache": "no"
}

class GoodsAnalysis:
    def get_data(self, item_id):
        params = BASE_PARAMS.copy()
        params["num_iid"] = item_id
        try:
            res = requests.get(BASE_URL, params=params, timeout=10).json()
            if res.get("code") == 200:
                return res["item"]
            return None
        except:
            return None

    def analysis_single(self, raw_data):
        """数据清洗+指标计算"""
        if not raw_data:
            return {}
        now_price = float(raw_data.get("price", 0))
        origin_price = float(raw_data.get("original_price", now_price))
        sales = int(raw_data.get("sales", 0))
        discount_rate = round(now_price / origin_price, 3) if origin_price != 0 else 1
        info = {
            "采集时间": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            "商品ID": raw_data.get("num_iid"),
            "标题": raw_data.get("title"),
            "店铺": raw_data.get("seller_info", {}).get("shop_name"),
            "现价": now_price,
            "原价": origin_price,
            "折扣系数": discount_rate,
            "总销量": sales,
            "SKU数量": len(raw_data.get("sku_list", []))
        }
        return info

    def batch_save_csv(self, id_list, file_name="商品数据存档.csv"):
        """批量采集，写入CSV表格存档"""
        headers = ["采集时间", "商品ID", "标题", "店铺", "现价", "原价", "折扣系数", "总销量", "SKU数量"]
        # 首次写入创建表头
        try:
            with open(file_name, "r", encoding="utf-8-sig") as f:
                pass
        except FileNotFoundError:
            with open(file_name, "w", newline="", encoding="utf-8-sig") as f:
                writer = csv.writer(f)
                writer.writerow(headers)

        # 追加写入数据
        save_list = []
        for iid in id_list:
            raw = self.get_data(iid)
            ana_data = self.analysis_single(raw)
            if ana_data:
                save_list.append(list(ana_data.values()))
                print(f"✅ 商品{iid}解析完成")
            else:
                print(f"❌ 商品{iid}数据获取失败")

        with open(file_name, "a", newline="", encoding="utf-8-sig") as f:
            writer = csv.writer(f)
            writer.writerows(save_list)
        print(f"\n批量采集完成，数据已保存至：{file_name}")

# 数据分析调用
if __name__ == "__main__":
    ana = GoodsAnalysis()
    # 待采集商品ID列表
    goods_ids = [
        "10335871600",
        "10224567891",
        "10112345678"
    ]
    ana.batch_save_csv(goods_ids)

数据分析拓展用法

折扣系数：数值越小优惠力度越大，筛选低价引流商品；
多次定时存档后，对比不同日期销量，判断商品增长潜力；
结合 Pandas 绘制价格走势图、销量柱状图，做成可视化运营看板。

五、落地优化与拓展方案

1. 稳定性优化

批量采集加入随机休眠time.sleep(1~3)，平稳请求频次；
异常商品单独记录日志，避免单次失败中断整批任务；
长期监控可使用 Linux 定时任务、Windows 计划任务，实现 7*24 小时离线运行。

2. 功能拓展方向

评价分析：提取返回评价文本，做关键词词云、差评统计；
选品筛选：新增销量、价格、店铺等级过滤规则，自动筛选蓝海潜力商品；
可视化：用 Matplotlib、Pyecharts 自动生成价格走势图片；
多平台兼容：可迁移适配淘宝、拼多多同类商品查询能力，搭建跨平台统一数据系统。

六、总结

依托 Open Claw 的调度能力 + 高阶商品详情查询，我们用三段轻量化 Python 代码，快速搭建了数据采集 - 实时监控 - 结构化分析 - 长期存档完整电商数据工具。对比传统爬虫，这套方案无需持续适配页面改版，开箱即用；对比付费商业监控工具，可高度自定义规则、无功能阉割、长期使用成本极低。

对于技术新手，直接复制代码修改密钥与商品 ID 即可快速落地；对于开发人员，可以在此基础上拓展数据库存储、前端可视化、多机器人告警等企业级功能。不管是个人卖家选品调研，还是团队竞品长效监控，这套落地思路都具备很强实用性。

Open Claw 实战教程：5 分钟搭建京东商品自动化监控与数据分析系统

Sun, 02 Aug 2026 09:50:47 +0800

前言

做电商运营、竞品调研、货源选品、铺货运维的朋友，几乎都会遇到同一个痛点：手动盯商品价格、库存、促销活动、规格变动效率极低；批量整理竞品数据、做价格走势复盘、销量口碑分析时，复制粘贴整理数据耗时耗力，人工统计还容易出错。

很多开发者尝试爬虫抓取页面数据，却频繁遭遇反爬封禁、页面改版失效、数据残缺等问题，调试与维护成本居高不下。最近一直在使用 Open Claw 工具完成电商数据自动化采集，依托其内置的高级商品详情采集能力，无需复杂逆向、不用频繁维护爬虫规则，就能稳定拿到京东商品全维度结构化数据，快速落地实时监控 + 长效数据分析两套常用业务，本文结合落地代码，完整分享从工具接入、实时监控脚本、数据分析整理全流程实操。

一、工具核心能力介绍

本次用到的高级详情采集能力，相比普通基础采集，最大优势是数据覆盖面更广、实时性更强、结构化程度更高：

基础信息全覆盖：商品标题、品牌、类目、主图、轮播图、长详情图文、商品参数、发货配送、店铺资质、自营 / 第三方店铺类型一键提取；
动态变动数据：实时售价、原价、优惠券、满减活动、多规格 SKU 单价与库存、近 30 天销量、库存余量；
口碑评价维度：好评率、总评论数、追评数量、差评标签，可用于竞品口碑打分；
灵活配置：支持关闭缓存获取毫秒级实时数据，可自由切换 JSON、格式化文本等数据格式，适配数据库存储、表格导出、可视化图表等场景。

调用逻辑十分简单，仅需要 2 个身份凭证 + 目标商品 ID，即可发起数据拉取，GET/POST 两种请求方式全部兼容，Python、Java、JS 等主流编程语言都能快速接入。

核心入参说明

表格

参数	是否必填	作用说明
身份凭证 1	是	个人专属调用密钥，平台后台注册后直接获取
身份凭证 2	是	安全校验密钥，和凭证 1 搭配使用，保障调用安全
商品 ID	是	京东商品详情页末尾数字编号，定位目标商品
cache	可选	yes 默认缓存加速；no 关闭缓存，用于价格、库存实时监控
result_type	可选	默认 JSON，日常数据分析可选择格式化阅读文本

返回核心可用字段（监控 & 分析高频使用）

基础：title 标题、shop_name 店铺名、pic_url 主图、goods_attr 商品参数
价格促销：price 现价、org_price 原价、coupon 优惠券列表
规格库存：sku 多规格集合、sku_price 单品价格、sku_stock 单品库存
销量口碑：sales 累计销量、goodRateShow 好评率、commentCount 总评论量

二、环境准备

本次全程使用 Python 实现，代码轻量化、无复杂依赖，只需要安装网络请求库：

pip install requests

运行环境：Python3.7 及以上，Windows、Mac、Linux 服务器均可部署，服务器部署可实现 7×24 小时不间断监控。

三、基础调用：单商品结构化数据获取

先写基础封装函数，完成单次商品数据拉取、异常重试、数据解析，为后续监控、数据分析打下基础。

import requests
import time
import json

class JDGoodsCollector:
    def __init__(self, key, secret):
        # 初始化身份凭证，替换为自己后台获取的凭证
        self.key = key
        self.secret = secret
        self.base_url = "https://api-gw.onebound.cn/jd/item_get_pro"
        self.session = requests.Session()  # 长连接优化请求速度

    def get_goods_detail(self, item_id, real_time=False):
        """
        拉取商品详情
        :param item_id: 京东商品ID
        :param real_time: 是否实时数据，True关闭缓存，用于监控价格库存变动
        :return: 结构化商品字典/None
        """
        params = {
            "key": self.key,
            "secret": self.secret,
            "api_name": "jd.item_get_pro",
            "num_iid": item_id,
            "cache": "no" if real_time else "yes",
            "result_type": "json"
        }
        retry_times = 3  # 3次失败重试
        for i in range(retry_times):
            try:
                resp = self.session.get(self.base_url, params=params, timeout=12)
                resp.raise_for_status()
                data = resp.json()
                if data.get("code") == 200:
                    return data.get("item", {})
                else:
                    print(f"请求异常：{data.get('msg', '未知错误')}")
            except Exception as e:
                print(f"第{i+1}次请求失败：{str(e)}，1秒后重试")
                time.sleep(1)
        print("多次请求失败，放弃本次采集")
        return None

# 测试基础调用
if __name__ == "__main__":
    # 替换为你的个人凭证
    KEY = "你的专属key"
    SECRET = "你的专属secret"
    collector = JDGoodsCollector(KEY, SECRET)
    target_item = "100012345678"  # 替换目标京东商品ID
    goods_info = collector.get_goods_detail(target_item, real_time=False)
    if goods_info:
        print("====商品基础信息====")
        print(f"商品标题：{goods_info.get('title')}")
        print(f"当前售价：{goods_info.get('price')} 元")
        print(f"原价：{goods_info.get('org_price')} 元")
        print(f"店铺名称：{goods_info.get('shop_name')}")
        print(f"累计销量：{goods_info.get('sales')}")
        print(f"好评率：{goods_info.get('goodRateShow')}")

运行后即可一键打印商品核心信息，代码做了超时、网络异常重试，稳定性远高于爬虫脚本。

四、实战 1：搭建商品实时价格 & 库存监控脚本

需求场景

竞品降价预警、自有商品库存告警、大促期间价格变动捕捉，一旦价格下调、库存大幅减少，立刻打印提醒，可拓展对接微信、钉钉推送消息。实现思路：循环定时采集实时数据，对比上一轮采集数据，出现变动触发告警。

import requests
import time
import json

class JDMonitor:
    def __init__(self, key, secret):
        self.collector = JDGoodsCollector(key, secret)
        # 历史数据缓存，存储上一轮采集结果
        self.history_data = {}

    def monitor_single_goods(self, item_id, interval=300):
        """
        单商品循环监控
        :param item_id: 商品ID
        :param interval: 采集间隔，单位秒，默认5分钟采集一次
        """
        print(f"====开始监控商品{item_id}，采集间隔{interval}秒====")
        while True:
            now_data = self.collector.get_goods_detail(item_id, real_time=True)
            if not now_data:
                time.sleep(interval)
                continue
            # 首次采集，仅缓存数据不对比
            if item_id not in self.history_data:
                self.history_data[item_id] = now_data
                print(f"初始化缓存完成，当前价格：{now_data.get('price')}元")
                time.sleep(interval)
                continue

            old_data = self.history_data[item_id]
            alert_list = []
            # 价格变动检测
            old_price = float(old_data.get("price", 0))
            new_price = float(now_data.get("price", 0))
            if new_price < old_price:
                alert_list.append(f"【降价提醒】原价{old_price} → 现价{new_price}")
            elif new_price > old_price:
                alert_list.append(f"【涨价提醒】原价{old_price} → 现价{new_price}")

            # 库存变动检测
            old_stock = old_data.get("sku", [])
            new_stock = now_data.get("sku", [])
            if len(new_stock) > 0 and str(new_stock[0].get("skuStock")) != str(old_stock[0].get("skuStock")):
                alert_list.append(f"【库存变动】当前库存：{new_stock[0].get('skuStock')}")

            # 输出告警
            if alert_list:
                print("\n=====触发告警=====")
                print(f"商品：{now_data.get('title')}")
                for alert in alert_list:
                    print(alert)
                print("==================\n")

            # 更新历史缓存
            self.history_data[item_id] = now_data
            time.sleep(interval)

# 监控运行入口
if __name__ == "__main__":
    KEY = "你的专属key"
    SECRET = "你的专属secret"
    monitor = JDMonitor(KEY, SECRET)
    # 监控目标商品，间隔300秒（5分钟）采集一次
    monitor.monitor_single_goods("100012345678", interval=300)

拓展优化方向

多商品批量监控：将商品 ID 存入列表，循环遍历采集；
消息推送：对接钉钉机器人、企业微信接口，变动推送至工作群；
阈值告警：设定降价幅度阈值，仅降价超过 5% 才提醒，过滤小幅波动。

五、实战 2：批量采集 + 长效数据分析整理脚本

日常竞品分析、选品调研，需要批量抓取多款商品数据，整理为表格结构化数据，用于价格对比、口碑排序、热度筛选，下面代码实现批量采集 + 数据整理，可直接导出 CSV 表格。

import requests
import time
import csv

class JDAnalysisTool:
    def __init__(self, key, secret):
        self.collector = JDGoodsCollector(key, secret)

    def batch_collect(self, id_list):
        """批量采集多款商品数据"""
        result_list = []
        print(f"开始批量采集{len(id_list)}个商品数据")
        for idx, item_id in enumerate(id_list):
            print(f"正在采集第{idx+1}/{len(id_list)}：{item_id}")
            info = self.collector.get_goods_detail(item_id, real_time=False)
            if info:
                item_dict = {
                    "商品ID": item_id,
                    "商品标题": info.get("title"),
                    "现价": info.get("price"),
                    "原价": info.get("org_price"),
                    "店铺": info.get("shop_name"),
                    "销量": info.get("sales"),
                    "好评率": info.get("goodRateShow"),
                    "总评论": info.get("commentCount")
                }
                result_list.append(item_dict)
            time.sleep(1)  # 防频繁请求限流
        return result_list

    def export_csv(self, data_list, save_name="jd_goods_analysis.csv"):
        """导出csv表格，Excel可直接打开"""
        if not data_list:
            print("无数据可导出")
            return
        headers = list(data_list[0].keys())
        with open(save_name, "w", encoding="utf-8-sig", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=headers)
            writer.writeheader()
            writer.writerows(data_list)
        print(f"数据已导出至：{save_name}")

    def sort_hot(self, data_list):
        """数据分析：按销量降序排序，筛选热门商品"""
        return sorted(data_list, key=lambda x: int(x.get("销量", 0)) if x.get("销量") else 0, reverse=True)

# 数据分析运行入口
if __name__ == "__main__":
    KEY = "你的专属key"
    SECRET = "你的专属secret"
    analysis = JDAnalysisTool(KEY, SECRET)
    # 批量竞品ID列表
    goods_ids = [
        "100012345678",
        "100023456789",
        "100034567890"
    ]
    data = analysis.batch_collect(goods_ids)
    analysis.export_csv(data)
    hot_data = analysis.sort_hot(data)
    print("\n====热门商品排序（销量从高到低）====")
    for item in hot_data:
        print(f"{item['商品标题']} | 销量：{item['销量']} | 价格：{item['现价']}")

数据分析拓展用法

价格区间筛选：增加函数过滤低价、高价竞品，锁定对标价位商品；
口碑筛选：过滤好评率低于 90% 的劣质竞品；
入库存储：将数据存入 MySQL、Sqlite，长期存储后绘制价格走势折线图，复盘大促价格规律。

六、落地常见问题与优化技巧

调用限流：个人日常使用间隔 1~3 秒，企业批量采集用异步队列错峰请求，不要短时间高频循环；
监控实时性：日常监控 5~10 分钟采集一次足够，618、双 11 大促可缩短至 2~3 分钟；
轻量化部署：脚本放到云服务器后台常驻运行，无需本地电脑一直开机；
冗余过滤：不需要详情图片、长参数时，可自行裁剪解析字段，减少内存占用。

七、总结

借助 Open Claw 的高级商品详情采集能力，不用钻研复杂网页逆向、不用反复维护爬虫规则，几十行代码就能快速落地稳定可靠的商品监控系统与数据分析工具。

个人从业者：用来盯竞品、选货源、薅促销活动；
电商团队：搭建自动化数据中台，批量竞品监控、价格调研、铺货素材整理；整套代码轻量化易修改，新手复制替换凭证与商品 ID 即可直接运行，长期使用可以根据业务叠加消息推送、可视化图表、数据库存储等拓展功能。

OpenClaw 实战：搭建京东商品监控与简易数据分析脚本

Fri, 31 Jul 2026 11:08:18 +0800

前言

做电商运营、竞品分析、选品调研的时候，我们经常需要持续盯一批商品：价格波动、库存变化、规格变动、销量走势。手动打开网页复制粘贴，效率低还容易遗漏关键变化。

最近折腾 OpenClaw 这套工具，发现可以快速把商品基础信息定时拉取到本地，做成轻量化监控脚本。不需要复杂爬虫框架，几行代码就可以完成定时采集、数据落盘、简单分析，适合个人小团队做竞品跟踪、选品复盘。

注意：本文仅用于技术学习，使用时请遵守平台规则，控制访问频率，请勿高频批量抓取，避免对网站造成压力。

工具准备

环境依赖：Python3，requests，pandas，sqlite3（内置）

pip install requests pandas

我们会实现这些功能：

根据商品编号获取完整商品元数据：标题、原价、售价、库存、SKU 规格、店铺信息、发货地等
定时采集，把历史价格、库存存入本地 SQLite 数据库，留存变化记录
简单数据分析：价格变动统计、库存波动、导出 CSV 报表
简易告警逻辑：价格大跌 / 库存清零打印提示

完整实现代码

# -*- coding:utf-8 -*-
import requests
import time
import sqlite3
import pandas as pd
from datetime import datetime

# --------------------------配置区域---------------------------
KEY = "your_key_here"
SECRET = "your_secret_here"
# 需要监控的商品ID列表
MONITOR_ITEM_IDS = ["10335871600"]
# 采集间隔，单位秒，测试不要设置过小，建议至少300秒以上
FETCH_INTERVAL = 600
DB_PATH = "./jd_monitor.db"
# -------------------------------------------------------------

def init_db():
    """初始化数据库，存储每次采集快照"""
    conn = sqlite3.connect(DB_PATH)
    cur = conn.cursor()
    cur.execute('''
    CREATE TABLE IF NOT EXISTS item_history (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        num_iid TEXT,
        title TEXT,
        price REAL,
        original_price REAL,
        stock INTEGER,
        sales INTEGER,
        shop_name TEXT,
        location TEXT,
        fetch_time TIMESTAMP
    )
    ''')
    conn.commit()
    conn.close()


def fetch_item_detail(num_iid):
    """获取单款商品完整详情数据"""
    headers = {
        "Accept-Encoding": "gzip",
        "Connection": "close"
    }
    base_url = "https://api-gw.onebound.cn/jd/item_get_pro/"
    params = {
        "key": KEY,
        "secret": SECRET,
        "num_iid": num_iid,
        "result_type": "json",
        "cache": "no"
    }
    try:
        resp = requests.get(base_url, params=params, headers=headers, timeout=20)
        resp.raise_for_status()
        return resp.json()
    except Exception as e:
        print(f"请求异常 num_iid:{num_iid}, err:{e}")
        return None


def save_snapshot(raw_data):
    """把单次采集结果写入数据库快照"""
    if not raw_data or "item" not in raw_data:
        return
    item = raw_data.get("item", {})
    conn = sqlite3.connect(DB_PATH)
    cur = conn.cursor()
    cur.execute('''
    INSERT INTO item_history
    (num_iid,title,price,original_price,stock,sales,shop_name,location,fetch_time)
    VALUES (?,?,?,?,?,?,?,?,?)
    ''', (
        item.get("num_iid"),
        item.get("title"),
        float(item.get("price", 0)),
        float(item.get("orginal_price", 0)),
        int(item.get("num", 0)),
        int(item.get("sales", 0)),
        item.get("nick", ""),
        item.get("location", ""),
        datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    ))
    conn.commit()
    conn.close()

    # 简单监控告警逻辑
    price = float(item.get("price", 0))
    stock = int(item.get("num", 0))
    print(f"【快照】{item.get('title')} | 当前价:{price} |库存:{stock} | {datetime.now()}")
    if stock <= 0:
        print(f"⚠️警告：商品 {item.get('num_iid')} 库存归零！")


def export_analysis_csv():
    """导出历史数据，做数据分析，输出csv"""
    conn = sqlite3.connect(DB_PATH)
    df = pd.read_sql("SELECT * FROM item_history ORDER BY fetch_time DESC", conn)
    conn.close()
    csv_name = f"./item_analysis_{datetime.now().strftime('%Y%m%d_%H%M')}.csv"
    df.to_csv(csv_name, index=False, encoding="utf_8_sig")
    print(f"✅分析报表导出完成：{csv_name}")
    return df


def simple_analysis(df):
    """简易数据分析：统计价格最大最小，价格变动次数，库存变化"""
    if df.empty:
        print("暂无采集数据")
        return
    print("\n========简易分析报告========")
    for item_id in df["num_iid"].unique():
        sub = df[df["num_iid"] == item_id]
        min_p = sub["price"].min()
        max_p = sub["price"].max()
        latest_price = sub.iloc[0]["price"]
        latest_stock = sub.iloc[0]["stock"]
        record_cnt = len(sub)
        print(f"商品ID:{item_id}")
        print(f"记录条数:{record_cnt} |历史最低:{min_p} |历史最高:{max_p} |最新价格:{latest_price} |最新库存:{latest_stock}")
        print("-"*40)


def monitor_loop():
    init_db()
    print("开始商品监控循环，按Ctrl+C停止程序")
    try:
        while True:
            for item in MONITOR_ITEM_IDS:
                res = fetch_item_detail(item)
                if res:
                    save_snapshot(res)
            # 每采集三轮导出一次报表，可自行修改
            export_analysis_csv()
            time.sleep(FETCH_INTERVAL)
    except KeyboardInterrupt:
        print("\n收到停止信号，准备导出最终报表...")
        df = export_analysis_csv()
        simple_analysis(df)
        print("程序退出")


if __name__ == "__main__":
    monitor_loop()

代码解读

数据库初始化：使用轻量 SQLite 保存每一轮的快照，每一次拉取都作为一条历史记录，后续用来回溯价格、库存变化。不需要部署 MySQL，开箱即用。
数据拉取函数：填入你的凭证和商品编号，就可以拿到商品标题、售价、原价、库存、店铺名称、发货地、SKU 规格全套信息。
快照存储：每次成功获取信息，就写入一条记录；同时内置简单告警，如果库存变为 0 控制台输出提醒。
报表导出与分析：定时把全部历史快照导出 CSV，直接可以用 Excel 打开做图表分析。脚本内置简单统计：最高最低价格、记录数量、最新状态。
循环监控：设置间隔时间，持续轮询。测试环境建议不要把间隔调太小，避免触发访问限制。

运行之后控制台输出示例：

开始商品监控循环，按Ctrl+C停止程序
【快照】京东鲜花爱丽丝泡泡鲜切花云南鲜花家庭插花爱丽丝泡泡10枝源头直发 | 当前价:30.9 |库存:99 | 2026‑07‑31 15:22:10
✅分析报表导出完成：./item_analysis_20260731_1522.csv

按下Ctrl+C终止脚本，程序会自动导出最后一次完整分析报告，打印各商品的价格区间统计。

拓展玩法

1. SKU 多规格监控

返回结果里面自带 skus 数组，可以扩展脚本把每一个子规格的价格、库存分开存入子表，监控不同规格的调价、缺货。

# 示例，解析sku片段
skus_data = raw_data["item"].get("skus",{})
sku_list = skus_data.get("sku",[])
for sku in sku_list:
    print(sku["properties_name"],sku["price"],sku["quantity"])

2. 邮件 / 企业微信告警

当检测到价格暴跌、库存售罄，可以增加推送逻辑，把告警消息推送到企业微信机器人、邮箱，不用盯着控制台。

3. Pandas 做可视化

导出 DataFrame 后，直接用 matplotlib 绘制价格走势图，把一段时间的价格波动生成图片保存，适合做竞品周报素材。

import matplotlib.pyplot as plt
plt.rcParams["font.sans-serif"]=["SimHei"]
df["fetch_time"] = pd.to_datetime(df["fetch_time"])
plt.plot(df["fetch_time"],df["price"])
plt.title("商品价格监控曲线")
plt.savefig("./price_trend.png")

踩坑和注意事项

凭证信息不要硬编码上传公开代码仓库，实际项目建议放到环境变量。
平台会提供缓存开关，调试阶段把 cache 设置为 no 拿到实时数据；日常监控打开缓存，可以减少消耗。
返回中的价格字段，部分场景下会和页面展示存在偏差，属于原始数据本身限制，做分析时要留意。
不要超大批量同时监控几百上千商品，控制轮询间隔，合理使用工具。

小结

借助 OpenClaw，不用手写复杂解析逻辑，只需要专注业务逻辑：存快照、写告警、做数据分析。对于个人做竞品跟踪、选品复盘，这套脚本足够轻便好用。

Open‑Claw 实战｜低成本搭建电商评论监控与口碑数据分析系统

Fri, 31 Jul 2026 10:08:49 +0800

前言

做电商运营、竞品调研、产品迭代的同学，几乎都会遇到一个痛点：想要持续跟踪商品用户反馈，人工翻看评价页面效率极低。当需要同时盯多款竞品，还要定期统计好评、差评、规格反馈、晒图情况，单纯手动复制粘贴几乎不可行。

很多同学会尝试自己写页面解析脚本，但平台防护策略更新快，经常没过几天就无法正常拿到内容，调试、维护要耗费大量时间。

最近在用 Open‑Claw 做项目，发现它可以省去页面解析、对抗防护的繁琐工作，直接拿到规整后的评价原始内容。我们可以基于这套能力，快速搭建一套简易的监控脚本，实现定时拉取商品评价、增量识别新增反馈、简单文本统计分析，输出本地报表，完成基础的竞品口碑监控。

提示：本文所有内容仅用于个人学习研究，使用相关能力请遵守对应平台规则，请勿用于违规批量爬取。

整体思路

整体分为 4 个步骤：

准备身份凭证，获取商品编号；
编写拉取逻辑，分页获取评价原始内容；
做增量判断，只处理新增评价，避免重复分析；
简单统计分析：好评差评占比、高频关键词提取，导出 csv 文件；
增加定时循环，实现简易持续监控。

拿到的数据字段包含：评价正文、评价时间、购买规格、买家脱敏昵称、晒图地址、追评内容等，足够做日常口碑调研分析万邦开放平...。

环境准备

本地需要安装依赖库

pip install requests pandas collections

完整可运行代码

import requests
import time
import csv
import pandas as pd
from collections import Counter

# ===================== 配置区域，请自行修改 =====================
TOKEN = "your_access_key"
SECRET = "your_secret_code"
TARGET_ITEM_ID = "723456789123"   # 目标商品id，从商品链接提取
MAX_PAGE = 5                      # 最多读取多少页评价
SAVE_CSV_PATH = "./goods_review.csv"
# 用于内存记录已经处理过的评价id，防止重复统计
processed_review_ids = set()
# ==============================================================


def fetch_review_page(item_id, page_no, sort_type=1, rate_filter=0):
    """
    获取单页评价
    sort_type:1最新,2最热
    rate_filter:0全部,1好评,2中评,3差评
    """
    params = {
        "key": TOKEN,
        "secret": SECRET,
        "num_iid": item_id,
        "page_no": page_no,
        "page_size": 40,
        "sort": sort_type,
        "rate_type": rate_filter,
        "result_type": "json"
    }
    try:
        resp = requests.get(
            "https://collect.openclaw.cloud/pull",
            params=params,
            timeout=15
        )
        json_data = resp.json()
        if json_data.get("code") != 0:
            print(f"第{page_no}页请求异常：{json_data.get('msg')}")
            return []
        review_list = json_data.get("items", {}).get("item", [])
        return review_list
    except Exception as e:
        print(f"网络异常 {str(e)}")
        return []


def parse_raw_review(raw_item):
    """清洗单条原始数据，提取需要字段"""
    rid = raw_item.get("rate_id", "")
    if not rid:
        return None
    info = {
        "review_id": rid,
        "nick": raw_item.get("display_user_nick", ""),
        "content": raw_item.get("rate_content", "").strip(),
        "sku_spec": raw_item.get("auction_sku", ""),
        "rate_date": raw_item.get("rate_date", ""),
        "star": raw_item.get("rate_star", 0),
        "pic_list": "|".join(raw_item.get("pics", [])) if raw_item.get("pics") else "",
        "add_review": raw_item.get("add_feedback", "").strip()
    }
    return info


def save_to_csv(review_data_list, file_path):
    """追加写入csv，保存评价记录"""
    headers = ["review_id","nick","content","sku_spec","rate_date","star","pic_list","add_review"]
    file_exist = False
    try:
        with open(file_path, "r", encoding="utf‑8‑sig") as f:
            file_exist = True
    except FileNotFoundError:
        pass

    with open(file_path, "a", newline="", encoding="utf‑8‑sig") as f:
        writer = csv.DictWriter(f, fieldnames=headers)
        if not file_exist:
            writer.writeheader()
        for row in review_data_list:
            writer.writerow(row)


def simple_text_analysis(df_new):
    """简单数据分析，打印统计结果"""
    if df_new.empty:
        print("本轮没有新评价")
        return
    total = len(df_new)
    good = len(df_new[df_new["star"] >=4])
    mid = len(df_new[df_new["star"] ==3])
    bad = len(df_new[df_new["star"] <=2])

    print("\n===========本轮新增评价统计===========")
    print(f"本轮新增条数：{total}")
    print(f"好评：{good}，占比 {good/total:.2%}")
    print(f"中评：{mid}，占比 {mid/total:.2%}")
    print(f"差评：{bad}，占比 {bad/total:.2%}")

    # 简单分词提取高频词，这里做简单切分，生产环境建议替换专业分词库
    all_words = []
    for text in df_new["content"]:
        words = [w for w in str(text).replace(" ","") if len(w)>=2]
        all_words.extend(list(words))
    top_word = Counter(all_words).most_common(12)
    print("高频词Top12：",top_word)
    print("======================================\n")


def run_one_cycle():
    """执行一轮采集任务"""
    new_reviews = []
    for page in range(1, MAX_PAGE+1):
        print(f"正在读取第 {page} 页")
        raw_items = fetch_review_page(TARGET_ITEM_ID, page)
        if not raw_items:
            break
        for raw in raw_items:
            parsed = parse_raw_review(raw)
            if not parsed:
                continue
            if parsed["review_id"] in processed_review_ids:
                continue
            processed_review_ids.add(parsed["review_id"])
            new_reviews.append(parsed)
        time.sleep(2)

    if len(new_reviews) > 0:
        save_to_csv(new_reviews, SAVE_CSV_PATH)
        df = pd.DataFrame(new_reviews)
        simple_text_analysis(df)
    else:
        print("本轮未发现新增评价")


if __name__ == "__main__":
    # 单次运行 run_one_cycle()
    # 如果要做持续监控，打开下面循环，注意间隔不要过短
    run_one_cycle()
    """
    while True:
        run_one_cycle()
        print("休眠30分钟，等待下一轮监控...")
        time.sleep(60 * 30)
    """

代码说明

fetch_review_page：负责分页获取内容，可以筛选好评、中评、差评，也可以选择最新或者最热排序。
parse_raw_review：把杂乱的原始结果做规整，把评价 id、规格、晒图、追评提取出来，评价 id 作为唯一标识，用来做增量过滤。
processed_review_ids内存集合，记录已经处理过的评价 id，每一轮只分析新出来的反馈，不会重复统计历史老评价。
simple_text_analysis简易分析，统计星级分布，简单做高频词统计。正式业务场景可以替换 jieba 分词，接入情绪分析模型做情感判断。
csv 持久化把每一轮数据追加写入本地文件，方便后续 Excel 打开复盘。
如果开启 while 循环，就变成定时监控脚本，注意休眠时间不能太短，避免短时间高频访问。

落地场景拓展

拿到结构化评价数据之后，还可以延伸做很多事情：

竞品监控：同时配置多个商品编号，循环遍历，定时输出竞品差评变化，快速感知竞品质量问题、用户集中吐槽点；
自家商品质检：持续盯自己店铺商品，一旦出现新增差评，脚本打印告警，运营可以及时跟进；
规格问题统计：统计哪个 SKU 规格对应的差评最多，辅助供应链、产品做优化；
晒图素材沉淀：把晒图链接保存，后续做用户研究素材。

踩坑总结

不要设置每页数量过大，单次返回数据量太高容易报错，建议控制在 40‑50 以内；
一定要做延时，循环中间 time.sleep，否则容易触发限流；
评价 id 务必保存，用来做增量识别，不然每次全量拉取，会重复分析历史内容；
部分评价会为空、或者只有表情符号，业务代码可以增加文本长度过滤，剔除无效内容；
正式部署不要硬编码凭证，可以放到环境变量中。

小结

借助 Open‑Claw 的能力，我们不需要花费大量精力维护页面解析规则，就可以快速完成评价获取、增量监控、简单数据分析，十几行核心代码就能搭建起个人调研使用的口碑监控脚本。

电商口碑自动化监控实战：基于 OpenClaw 搭建商品评价实时分析系统（附完整可运行 Python 代码）

Tue, 28 Jul 2026 17:38:04 +0800

一、前言：电商人做竞品口碑的痛点

做淘宝、天猫运营、产品调研、竞品分析的从业者，几乎每天都要面对商品评价数据：

手动采集效率极低：竞品多、商品量大，人工翻页复制评论，一天只能处理 2-3 款产品，数据严重滞后；
平台风控限制爬虫：自研 Selenium、异步爬取极易被限制访问，还要维护代理池、Cookie，维护成本极高；
数据无结构化：复制下来的评论杂乱无章，无法快速区分好评、差评、追评、图文评价，难以做批量统计；
缺少实时监控能力：新品差评爆发、批量负面反馈出现后，商家往往滞后数天才发现，错失售后补救、产品优化时机。

本文给大家一套轻量化落地方案，借助 OpenClaw 数据采集能力，零复杂爬虫开发，仅通过几十行 Python 代码，实现商品评价全量自动拉取、定时增量监控、文本情感分析、痛点关键词提取完整流程，数据可落地存储 Excel/MySQL，适合个人运营、中小商家、数据分析师直接复用。

二、方案核心优势

无需自研爬虫，规避平台反拦截 底层封装成熟数据通道，不用处理页面动态渲染、滑块验证、访问限流，输入商品 ID 即可稳定拉取全维度评价数据，包含评价正文、星级、购买 SKU、晒图、追评、评价时间、买家昵称等完整字段。
支持增量实时监控 可自定义定时轮询周期（小时 / 天），只抓取新增评价，避免重复拉取历史数据，7×24 小时监控竞品口碑异动。
一站式数据清洗 + 智能分析 配套文本处理逻辑，自动过滤刷单无效评价、空白内容，结合文本情感库自动区分正向 / 负向评价，统计差评高频问题，输出可视化口碑报表。
低开发门槛 Python 基础语法即可运行，配置专属访问密钥后直接调用，支持批量商品同时监控，适配选品调研、店铺舆情、竞品对标等多场景。

三、前期准备工作

1. 工具账号开通

注册完成实名后获取专属访问密钥（后续代码中替换KEY参数），支持海外手机号注册登录，无国内号码也可正常使用。

2. 环境依赖安装

本地 Python3.8 及以上环境，执行命令安装所需第三方库：

pip install requests snownlp pandas openpyxl apscheduler

requests：网络请求拉取评价数据
snownlp：中文评论情感分析
pandas：数据清洗、导出 Excel 报表
apscheduler：定时任务，实现自动化监控
openpyxl：Excel 文件读写依赖

3. 获取目标商品 ID

打开淘宝商品详情页，链接中id=后一串纯数字即为商品编号，例如item.taobao.com/item.htm?id=723456123456，商品 ID 为723456123456。

四、完整实战代码实现

4.1 全局配置与基础请求封装

新建review_monitor.py文件，封装基础数据拉取函数，统一处理分页、异常重试、数据格式化：

import requests
import json
import time
import pandas as pd
from snownlp import SnowNLP
from apscheduler.schedulers.background import BackgroundScheduler

# ===================== 配置区（自行修改） =====================
# OpenClaw后台获取的专属密钥
ACCESS_KEY = "你的专属密钥"
# 需要监控的商品ID列表，可批量添加多个商品
MONITOR_ITEM_IDS = ["723456123456", "698765432100"]
# 单次请求每页数据条数
PAGE_SIZE = 20
# 定时监控间隔，单位：小时（每2小时抓取一次新增评价）
MONITOR_INTERVAL = 2
# 本地存储文件路径
SAVE_EXCEL_PATH = "./商品评价监控报表.xlsx"
# ===========================================================

# 全局请求会话，自动重试
session = requests.Session()
retry_strategy = requests.adapters.Retry(
    total=3, backoff_factor=1, status_forcelist=[500, 502, 503, 429]
)
session.mount("https://", requests.adapters.HTTPAdapter(max_retries=retry_strategy))

def fetch_item_review(item_id: str, page: int = 1):
    """
    拉取单页商品评价数据
    :param item_id: 商品数字ID
    :param page: 分页页码
    :return: 结构化评价列表、总评价数量
    """
    params = {
        "key": ACCESS_KEY,
        "item_id": item_id,
        "page": page,
        "page_size": PAGE_SIZE,
        "sort": "create_time_desc"  # 按评价时间倒序，优先获取最新评价
    }
    # 底层数据请求地址
    base_url = "对应数据服务地址"
    try:
        resp = session.get(base_url, params=params, timeout=15)
        resp_data = resp.json()
        if resp_data.get("code") != 200:
            print(f"商品{item_id}第{page}页请求失败：{resp_data.get('msg')}")
            return [], 0
        review_list = resp_data.get("data", {}).get("review_list", [])
        total_count = resp_data.get("data", {}).get("total", 0)
        return review_list, total_count
    except Exception as e:
        print(f"网络请求异常：{str(e)}")
        return [], 0

4.2 数据清洗 + 情感分析函数

对原始评价文本做清洗、情感打分，自动标记好评 / 中评 / 差评：

def text_sentiment_analysis(content: str):
    """评论情感分析，返回情感标签、情感分数"""
    if not content or len(content.strip()) < 2:
        return "无效评价", 0.5
    s = SnowNLP(content)
    score = s.sentiments
    if score >= 0.7:
        label = "好评"
    elif score >= 0.4:
        label = "中评"
    else:
        label = "差评"
    return label, round(score, 2)

def parse_raw_review(raw_list: list, item_id: str):
    """原始数据结构化清洗"""
    clean_data = []
    for item in raw_list:
        content = item.get("content", "")
        sentiment_label, sentiment_score = text_sentiment_analysis(content)
        row = {
            "商品ID": item_id,
            "评价ID": item.get("review_id", ""),
            "评价时间": item.get("create_time", ""),
            "买家昵称": item.get("nick", ""),
            "购买规格SKU": item.get("sku_name", ""),
            "评价正文": content,
            "星级评分": item.get("rate", 5),
            "是否晒图": 1 if item.get("pic_urls") else 0,
            "是否追评": 1 if item.get("add_review") else 0,
            "情感标签": sentiment_label,
            "情感分数": sentiment_score
        }
        clean_data.append(row)
    return clean_data

4.3 全量拉取 + 本地存储核心逻辑

循环分页抓取所有评价，合并历史数据去重，导出 Excel 报表：

def crawl_all_reviews(item_id: str):
    """单商品全量评价抓取"""
    all_clean_reviews = []
    page = 1
    while True:
        raw_data, total = fetch_item_review(item_id, page)
        if not raw_data:
            break
        clean_rows = parse_raw_review(raw_data, item_id)
        all_clean_reviews.extend(clean_rows)
        # 当前页数据不足分页大小，说明已抓取完毕
        if len(raw_data) < PAGE_SIZE:
            break
        page += 1
        time.sleep(1)  # 分页间隔防限流
    print(f"商品{item_id}抓取完成，本次获取{len(all_clean_reviews)}条评价")
    return all_clean_reviews

def save_to_excel(new_data: list):
    """合并新旧数据，去重后保存Excel"""
    new_df = pd.DataFrame(new_data)
    try:
        # 读取历史报表
        old_df = pd.read_excel(SAVE_EXCEL_PATH)
        merge_df = pd.concat([old_df, new_df], ignore_index=True)
        # 根据评价ID去重，保留最新数据
        merge_df = merge_df.drop_duplicates(subset=["评价ID"], keep="last")
    except FileNotFoundError:
        # 文件不存在则新建
        merge_df = new_df
    # 按评价时间倒序排序
    merge_df = merge_df.sort_values(by="评价时间", ascending=False)
    merge_df.to_excel(SAVE_EXCEL_PATH, index=False)
    print(f"数据已更新至本地报表，当前总数据量：{len(merge_df)}条")

4.4 定时监控任务入口

开启后台定时任务，自动循环监控所有目标商品：

def monitor_task():
    """定时执行的监控任务"""
    total_new_data = []
    for goods_id in MONITOR_ITEM_IDS:
        review_data = crawl_all_reviews(goods_id)
        total_new_data.extend(review_data)
    if total_new_data:
        save_to_excel(total_new_data)
        # 简易预警：检测新增差评，打印提醒
        bad_comments = [i for i in total_new_data if i["情感标签"] == "差评"]
        if bad_comments:
            print(f"⚠️ 检测到新增{len(bad_comments)}条差评，请及时查看！")
            for bad in bad_comments[:3]:
                print(f"【差评预警】{bad['评价时间']} | {bad['评价正文'][:60]}")

if __name__ == "__main__":
    # 首次运行立即抓取一次全量数据
    monitor_task()
    # 启动定时调度器
    scheduler = BackgroundScheduler()
    scheduler.add_job(monitor_task, "interval", hours=MONITOR_INTERVAL)
    scheduler.start()
    print(f"自动监控已启动，每{MONITOR_INTERVAL}小时自动抓取一次评价数据...")
    try:
        while True:
            time.sleep(3600)
    except KeyboardInterrupt:
        scheduler.shutdown()
        print("监控任务已停止")

五、代码功能拆解与业务落地说明

1. 增量监控与去重机制

每次抓取完成后，脚本会读取本地历史 Excel 文件，通过评价ID唯一标识自动剔除重复数据，仅保留最新新增评价，大幅减少重复存储、重复分析的资源消耗。

2. 差评自动预警逻辑

每次抓取结束自动筛选新增差评并打印控制台提醒，可二次扩展：接入钉钉 / 企业微信机器人，差评实时推送工作群，实现舆情快速响应。

3. 多维度数据分析扩展思路

基于导出的 Excel 结构化数据，可快速完成以下分析工作：

痛点词统计：使用 jieba 分词拆分差评文本，统计 “发货慢”“材质差”“尺寸不符” 等高频负面词汇，指导产品优化；
竞品口碑对标：多商品同时监控，对比同款产品好评率、差评类型，找到自身差异化优势；
时间趋势分析：按日期分组统计每日评价数量、差评占比，判断活动、上新后的口碑波动；
图文评价专项分析：筛选带晒图评价，分析买家实拍反馈，优化商品主图、详情页。

4. 数据库存储改造（进阶）

如果需要长期海量存储，可替换save_to_excel函数，接入 MySQL 存储数据，示例核心写入逻辑：

import pymysql
def save_to_mysql(data_list):
    conn = pymysql.connect(host="127.0.0.1", user="root", password="数据库密码", database="taobao_review")
    cursor = conn.cursor()
    insert_sql = """
    INSERT INTO goods_review (item_id, review_id, create_time, nick, sku, content, rate, has_pic, is_add, sentiment, score)
    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
    ON DUPLICATE KEY UPDATE content=VALUES(content)
    """
    for row in data_list:
        cursor.execute(insert_sql, (
            row["商品ID"], row["评价ID"], row["评价时间"], row["买家昵称"],
            row["购买规格SKU"], row["评价正文"], row["星级评分"],
            row["是否晒图"], row["是否追评"], row["情感标签"], row["情感分数"]
        ))
    conn.commit()
    cursor.close()
    conn.close()

六、实战避坑指南

密钥权限问题 密钥仅绑定个人账号，不要泄露，若请求返回权限不足提示，重新登录后台核对密钥，海外注册账号同样支持正常调取数据。
抓取频率控制 代码内置分页 1 秒延迟，定时任务建议最低 1 小时抓取一次，短时间高频请求会触发访问限制，导致数据拉取失败。
情感分析误差优化 SnowNLP 基础分词对电商口语、网络梗识别存在少量偏差，高精度需求可替换哈工大 LTP、百度 AI 文本情感接口，提升正负向判断准确率。
大批量商品监控优化 同时监控 10 款以上商品时，可改用异步httpx并发请求，缩短整体抓取耗时。

七、总结

传统自研爬虫做商品评价监控，要解决反爬、代理、数据结构化、定时调度等多重难题，开发和维护成本极高。借助 OpenClaw 标准化数据采集能力，仅用百行 Python 代码即可搭建一套完整口碑监控系统，实现自动采集 - 清洗 - 分析 - 存储 - 预警全流程自动化。

无论是店铺运营实时盯自己商品差评，还是竞品调研批量对标用户反馈，这套方案都能大幅降低人工成本，把零散的评价转化为可量化、可落地的产品优化、运营决策数据，新手复制代码修改配置即可直接投入使用。

低成本搭建电商口碑监控体系：7×24 小时商品评论采集 + 情感分析完整实战（附 Python 源码）

Sun, 26 Jul 2026 16:14:05 +0800

前言

做电商运营、产品调研、竞品分析的同学基本都有同一个痛点：想要持续追踪商品真实用户反馈，靠手动复制评论、导出后台表格效率极低；自己写页面抓取脚本又频繁遇到平台拦截、页面改版失效、代理池维护成本高等问题。

传统方案要么受平台后台导出条数限制，要么需要投入大量服务器资源维持稳定采集，中小团队、个人开发者很难落地长效监控。本文分享一套基于 OpenClaw 轻量化数据能力的完整落地方案，不用逆向解析页面、不用搭建代理集群，仅通过简单代码即可实现商品全量评论拉取、增量实时监控、口碑情感统计，从数据同步、清洗存储到痛点挖掘一站式完成，复制代码就能直接运行，适配淘宝 / 天猫全品类商品调研场景。

一、方案整体能力拆解

整套流程分为四大核心模块，普通笔记本或低配云服务器就能稳定运行，无复杂分布式部署门槛：

全维度数据同步 输入商品唯一标识即可批量拉取完整评论信息，覆盖用户评分、评价正文、晒图链接、追评内容、购买规格、评价时间、互动点赞数等全维度信息，支持按最新评价分页循环采集，一次性获取商品全部历史反馈。
增量实时监控 依靠评论唯一标识自动完成数据去重，设置定时循环任务后仅抓取新增评价，大幅减少重复请求资源消耗，实现 7×24 小时不间断口碑监测，一旦出现新增差评可快速识别预警。
标准化数据清洗存储 自动过滤空白无效评价、统一时间与图文链接格式，规整尺码 / 颜色规格字段，小型项目可存入 CSV 表格，长期大批量监控可对接 SQLite 本地数据库，方便后续反复调取分析。
智能口碑分析挖掘 通过分词与情感判别区分好评、中评、差评，统计负面高频痛点词汇，输出商品评分分布、口碑变化趋势，快速定位产品品控、包装、使用体验等核心问题，同时对比不同规格商品用户反馈差异。

二、前期基础配置步骤

2.1 平台账号权限开通

登录数据服务控制台完成实名认证，创建个人项目获取专属访问凭证两组字符串（密钥标识、安全密钥）；
在项目权限面板开启商品评价数据读取权限，无需额外申请平台类目权限，全品类商品均可调取；
本地安装 Python 运行环境，提前安装依赖工具包，执行安装命令：

pip install requests pandas jieba wordcloud matplotlib

2.2 核心采集参数说明

程序运行仅需两类必填信息：目标商品 ID、单次读取条数，同时支持可选配置：读取页码、排序规则（最新评价优先 / 最热评价优先）、是否同步拉取晒图地址。返回结构化数据包含核心字段：

基础信息：评价发布时间、脱敏买家昵称、星级评分、评价唯一编号
内容信息：主评文字、追评文字、晒图链接数组、对应购买 SKU 规格
互动数据：评价点赞数、商家回复内容、是否带图标识

三、完整可运行 Python 代码实战

3.1 基础评论批量采集代码（分页全量拉取）

import requests
import pandas as pd
import time

# 替换为自己控制台获取的凭证
ACCESS_KEY = "你的密钥标识"
ACCESS_SECRET = "你的安全密钥"
TARGET_ITEM_ID = "目标商品ID"
PAGE_SIZE = 20  # 单页读取条数

def get_item_review(page):
    """获取单页商品评价数据"""
    params = {
        "key": ACCESS_KEY,
        "secret": ACCESS_SECRET,
        "item_id": TARGET_ITEM_ID,
        "page": page,
        "page_size": PAGE_SIZE,
        "sort": "new"  # new=最新评价 hot=热门评价
    }
    # 统一数据请求通道
    resp = requests.get("https://data-claw-gateway.com/tmall/review", params=params)
    res_data = resp.json()
    return res_data

def batch_crawl_all_reviews():
    """循环分页采集全部评价"""
    all_review_list = []
    page_num = 1
    while True:
        print(f"正在采集第{page_num}页评价...")
        result = get_item_review(page_num)
        # 判断数据是否返回正常
        if result.get("code") != 0:
            print("数据读取异常：", result.get("msg"))
            break
        page_data = result.get("data", {})
        review_items = page_data.get("item", [])
        if not review_items:
            print("所有评价采集完成")
            break
        # 结构化提取需要的字段
        for item in review_items:
            temp_dict = {
                "评价ID": item.get("rate_id"),
                "评价时间": item.get("rate_date"),
                "买家昵称": item.get("display_user_nick"),
                "商品规格": item.get("auction_sku"),
                "星级评分": item.get("user_star"),
                "主评内容": item.get("rate_content"),
                "追评内容": item.get("add_feedback"),
                "晒图链接": item.get("pics"),
                "点赞数": item.get("like_count")
            }
            all_review_list.append(temp_dict)
        page_num += 1
        time.sleep(1.2)  # 间隔延迟，避免高频请求限制
    # 保存至本地表格
    df = pd.DataFrame(all_review_list)
    df.to_csv(f"商品评价_{TARGET_ITEM_ID}.csv", index=False, encoding="utf-8-sig")
    print(f"共采集{len(all_review_list)}条评价，文件已保存")
    return df

# 执行全量采集
if __name__ == "__main__":
    review_df = batch_crawl_all_reviews()

3.2 增量监控 + 差评预警代码（7×24 小时定时轮询）

基于已采集历史评价 ID 做去重，仅捕获新增反馈，新增差评自动打印提醒：

import requests
import pandas as pd
import time

ACCESS_KEY = "你的密钥标识"
ACCESS_SECRET = "你的安全密钥"
TARGET_ITEM_ID = "目标商品ID"
# 存储已采集评价ID，用于增量判断
history_id_set = set()

def get_new_review():
    params = {
        "key": ACCESS_KEY,
        "secret": ACCESS_SECRET,
        "item_id": TARGET_ITEM_ID,
        "page": 1,
        "page_size": 50,
        "sort": "new"
    }
    resp = requests.get("https://data-claw-gateway.com/tmall/review", params=params)
    res = resp.json()
    if res.get("code") != 0:
        return []
    items = res.get("data", {}).get("item", [])
    new_reviews = []
    for item in items:
        rid = item.get("rate_id")
        if rid not in history_id_set:
            history_id_set.add(rid)
            new_reviews.append(item)
    return new_reviews

def monitor_loop():
    print("启动实时口碑监控，每10分钟轮询一次...")
    while True:
        new_data = get_new_review()
        if new_data:
            print(f"====检测到{len(new_data)}条新增评价====")
            for review in new_data:
                score = review.get("user_star")
                content = review.get("rate_content")
                sku = review.get("auction_sku")
                # 低分判定差评预警
                if int(score) <= 3:
                    print(f"【差评预警】规格：{sku} | 评分：{score}分 | 内容：{content}")
            # 追加写入本地表格
            df_new = pd.DataFrame([{
                "评价ID": i.get("rate_id"),
                "评价时间": i.get("rate_date"),
                "商品规格": i.get("auction_sku"),
                "星级评分": i.get("user_star"),
                "主评内容": i.get("rate_content"),
                "追评内容": i.get("add_feedback")
            } for i in new_data])
            df_new.to_csv(f"监控增量评价_{TARGET_ITEM_ID}.csv", mode="a", header=False, index=False, encoding="utf-8-sig")
        time.sleep(600)  # 10分钟轮询间隔

if __name__ == "__main__":
    monitor_loop()

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

# 读取采集好的评价表格
df = pd.read_csv(f"商品评价_{TARGET_ITEM_ID}.csv", encoding="utf-8-sig")

# 1. 星级评分分布统计
score_count = df["星级评分"].value_counts().sort_index()
plt.figure(figsize=(8,4))
score_count.plot(kind="bar", color="#409EFF")
plt.title("商品星级评分分布")
plt.xlabel("星级")
plt.ylabel("评价数量")
plt.savefig("评分分布图表.png")

# 2. 差评高频关键词提取
bad_comment = df[df["星级评分"] <= 3]["主评内容"].dropna().str.cat(sep=" ")
stop_words = {"的", "了", "很", "有点", "还", "但是", "就是", "感觉", "这个", "买", "收到"}
words = jieba.lcut(bad_comment)
clean_words = [w for w in words if len(w)>=2 and w not in stop_words]
word_counter = Counter(clean_words)
top10_word = word_counter.most_common(10)
print("差评Top10高频词汇：")
for word, num in top10_word:
    print(f"{word}：{num}次")

# 3. 不同规格商品平均分对比
sku_score = df.groupby("商品规格")["星级评分"].mean().sort_values()
print("\n各规格平均评分：")
print(sku_score)

四、落地使用场景与业务价值

场景 1：自有店铺口碑长效监控

搭配定时轮询代码实现全天监测，一旦出现批量差评可第一时间介入售后，降低负面评价扩散对转化的影响；同时按月统计差评高频词汇，针对性优化产品做工、包装、发货速度等短板。

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

批量采集同赛道头部竞品全量评价，通过关键词对比找出竞品差异化优势与固有缺陷，用于新品定位、详情页卖点策划、营销文案打磨。例如竞品大量反馈 “续航不足”，即可在自家产品宣传重点突出续航优势。

场景 3：产品迭代需求挖掘

从上万条真实用户反馈中筛选高频需求，比如服饰类 “尺码偏小”“面料起球”、小家电 “噪音大”，直接作为产品改版、品控升级的数据依据，替代传统问卷调研，数据真实度更高。

场景 4：运营活动效果复盘

大促前后分别采集评价数据，对比活动期用户反馈变化，判断活动赠品、优惠机制、发货时效是否带来负面口碑，优化后续大促运营策略。

五、避坑优化小贴士

采集间隔控制：分页采集代码中设置 1 秒以上延迟，短时间批量读取大量商品时适当拉长休眠时间，避免触发访问限制；
数据存储优化：单商品评价超 5000 条建议改用 SQLite 存储，CSV 文件打开加载速度会大幅下降；
关键词优化：可自定义停用词列表，剔除无意义虚词，提升痛点词汇提取精准度；
多商品批量监控：将商品 ID 存入列表循环执行监控任务，同时监控多款竞品，统一汇总分析报表；
数据脱敏合规：输出报表时可隐藏完整买家昵称，仅保留脱敏后缀，符合用户信息使用规范。

总结

这套依托 OpenClaw 构建的电商评论监控分析方案，完美解决传统采集方式稳定性差、维护成本高的痛点，无需复杂逆向开发，少量代码即可完成从数据获取、实时监测到智能分析的全链路自动化流程。不管是电商运营、产品经理还是数据从业者，都能快速落地使用，依托真实用户评价数据驱动产品与运营决策，大幅降低数据调研人力成本。

低成本搭建电商商品口碑实时监控系统｜从数据同步到情感分析完整落地（附 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 可视化后台、接入大模型做评价深度语义解读、多平台商品数据统一汇总分析。