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

pip install requests pandas jieba wordcloud matplotlib

# 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)  # 每小时自动采集一次，按需调整间隔

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