低成本搭建电商商品口碑监控系统｜从实时数据采集到情感分析完整实战（附可运行 Python 代码）

pip install requests pandas jieba snownlp matplotlib sqlite3

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)

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

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)