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

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