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

pip install requests snownlp pandas openpyxl apscheduler

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

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

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)}条")

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("监控任务已停止")

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