电商评论自动化监控 + 文本分析落地教程，一套脚本搞定竞品舆情实时追踪

pip install requests pandas jieba wordcloud

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

# =====================配置区（自行修改）=====================
# OpenClaw个人凭证
ACCESS_TOKEN = "你的专属凭证"
# 需要监控的商品ID列表
ITEM_ID_LIST = ["723456123456", "723456123789"]
# 监控轮询间隔，单位秒，建议3600（1小时）
LOOP_INTERVAL = 3600
# 历史数据存储文件
SAVE_FILE = "shop_comment_data.csv"
# 已采集评论ID缓存，去重使用
record_id_set = set()
# ==========================================================

def get_item_comments(item_id):
    """
    获取单款商品全部评价，自动分页
    """
    all_comments = []
    page = 1
    while True:
        # 构造请求参数
        params = {
            "token": ACCESS_TOKEN,
            "item_id": item_id,
            "page": page,
            "page_size": 50
        }
        # 请求数据
        resp = requests.get("https://data-claw-service.com/comment/query", params=params, timeout=30)
        res_json = resp.json()
        # 判断返回状态
        if res_json.get("code") != 200:
            print(f"商品{item_id} 第{page}页获取失败：{res_json.get('msg')}")
            break
        comment_list = res_json.get("data", {}).get("list", [])
        if not comment_list:
            print(f"商品{item_id} 评价数据全部拉取完成，共{len(all_comments)}条")
            break
        # 过滤已采集旧数据，只保留新增评论
        new_comment = []
        for item in comment_list:
            cid = item.get("comment_id")
            if cid not in record_id_set:
                record_id_set.add(cid)
                new_comment.append(item)
        all_comments.extend(new_comment)
        page += 1
        time.sleep(1)
    return all_comments

# 模块2：数据清洗与本地存储
def save_comment_data(comment_list):
    if not comment_list:
        return
    df_new = pd.DataFrame(comment_list)
    # 判断本地文件是否存在，追加写入
    try:
        df_old = pd.read_csv(SAVE_FILE, encoding="utf-8-sig")
        df_total = pd.concat([df_old, df_new], ignore_index=True)
    except FileNotFoundError:
        df_total = df_new
    # 去重保存
    df_total.drop_duplicates(subset=["comment_id"], keep="last", inplace=True)
    df_total.to_csv(SAVE_FILE, index=False, encoding="utf-8-sig")
    print(f"新增{len(df_new)}条评价，当前总数据量：{len(df_total)}")

# 模块3：评论数据分析（情感、高频词统计）
def comment_analysis():
    df = pd.read_csv(SAVE_FILE, encoding="utf-8-sig")
    # 1. 基础评分统计
    score_stats = df["score"].value_counts().sort_index()
    print("=====商品评分分布统计=====")
    print(score_stats)
    # 区分好评差评（4-5星好评，1-2星差评，3星中性）
    good = len(df[df["score"] >= 4])
    bad = len(df[df["score"] <= 2])
    mid = len(df[df["score"] == 3])
    print(f"好评数量：{good} 差评数量：{bad} 中性评价：{mid}")
    # 2. 差评文本分词，提取高频吐槽词
    bad_text = "".join(df[df["score"] <= 2]["content"].astype(str))
    stop_words = {"的", "了", "很", "有点", "还", "就是", "但是", "一般", "感觉"}
    words = jieba.lcut(bad_text)
    filter_words = [w for w in words if len(w) > 1 and w not in stop_words]
    word_count = Counter(filter_words)
    print("\n=====差评高频关键词Top20=====")
    for word, cnt in word_count.most_common(20):
        print(f"{word}：{cnt}次")
    # 导出分析报表
    analysis_result = pd.DataFrame(word_count.most_common(30), columns=["关键词", "出现次数"])
    analysis_result.to_csv("comment_word_analysis.csv", index=False, encoding="utf-8-sig")

# 定时循环监控主程序
def main():
    global record_id_set
    # 加载历史评论ID，用于增量判断
    try:
        df_history = pd.read_csv(SAVE_FILE, encoding="utf-8-sig")
        record_id_set = set(df_history["comment_id"].tolist())
        print(f"加载历史采集记录，已有{len(record_id_set)}条历史评价")
    except FileNotFoundError:
        print("暂无历史数据，首次全量采集")
    # 循环监控
    while True:
        total_new = []
        for goods_id in ITEM_ID_LIST:
            print(f"\n开始采集商品：{goods_id}")
            new_data = get_item_comments(goods_id)
            total_new.extend(new_data)
        # 保存新增数据并分析
        if total_new:
            save_comment_data(total_new)
            comment_analysis()
        else:
            print("本轮无新增评价，等待下一轮监控...")
        print(f"等待{LOOP_INTERVAL/60}分钟后再次采集")
        time.sleep(LOOP_INTERVAL)

if __name__ == "__main__":
    main()

# 批量读取商品ID示例
with open("goods_id.txt", "r", encoding="utf-8") as f:
    ITEM_ID_LIST = [line.strip() for line in f if line.strip()]

def send_warning(text):
    robot_url = "你的机器人webhook地址"
    data = {"msgtype": "text", "text": {"content": text}}
    requests.post(robot_url, json=data)
# 在save_comment_data函数内新增差评判断
bad_new = [i for i in new_comment if i.get("score") <= 2]
if bad_new:
    send_warning(f"检测到{len(bad_new)}条新增差评，请及时查看！")