电商口碑自动化监控方案：搭建商品评论实时采集 + 情感分析系统

┌─────────────────────────────────────────────────────────────┐
│                    电商口碑自动化监控系统                      │
├─────────────┬─────────────┬─────────────┬───────────────────┤
│ 数据采集层   │ 数据处理层   │ 智能分析层   │ 应用展示层         │
├─────────────┼─────────────┼─────────────┼───────────────────┤
│ • 评论API   │ • 数据清洗   │ • 情感判定   │ • 实时监控看板     │
│   批量拉取   │ • 去重过滤   │ • 属性提取   │ • 负面预警推送     │
│ • 增量同步   │ • 分词向量化 │ • 风险分级   │ • 趋势分析报告     │
│ • 多平台适配 │ • 异常检测   │ • 痛点聚类   │ • 竞品对比分析     │
└─────────────┴─────────────┴─────────────┴───────────────────┘

import requests
import time
import pandas as pd
from datetime import datetime
# ===================== 基础配置区 =====================
ACCESS_KEY = "你的专属key"
ACCESS_SECRET = "你的专属密钥"
TARGET_ITEM_ID = "1234567890123"      # 目标商品ID
PAGE_SIZE = 20                        # 单次拉取条数
MONITOR_INTERVAL = 600                # 监控轮询间隔（秒）
SAVE_PATH = "goods_comments.csv"
# ===================================================
def get_review_batch(item_id, page=1):
    """
    批量获取商品评价数据
    """
    params = {
        "key": ACCESS_KEY,
        "secret": ACCESS_SECRET,
        "item_id": item_id,
        "page": page,
        "page_size": PAGE_SIZE,
        "result_type": "json"
    }
    resp = requests.get(
        "https://gate.openclaw.cloud/data",  # 标准化数据通道
        params=params, timeout=30
    )
    res_json = resp.json()
    review_list = []
    if res_json.get("data") and res_json["data"].get("item"):
        for item in res_json["data"]["item"]:
            review_info = {
                "review_id": item.get("rate_id", ""),
                "content": item.get("rate_content", ""),
                "review_time": item.get("rate_date", ""),
                "user_name": item.get("display_user_nick", ""),
                "sku_info": item.get("auction_sku", ""),
                "img_urls": item.get("pics", []),
                "seller_reply": item.get("reply_content", ""),
                "star_rating": item.get("rate", 5),  # 评分
            }
            review_list.append(review_info)
    return review_list

def save_comments_to_csv(new_data):
    """清洗并保存评价数据，自动去重"""
    if len(new_data) == 0:
        return pd.DataFrame([])
    
    df_new = pd.DataFrame(new_data)
    # 过滤无内容评价
    df_new = df_new[df_new["content"].str.strip() != ""]
    # 过滤过短评价（少于3个字）
    df_new = df_new[df_new["content"].str.len() >= 3]
    
    try:
        df_all = pd.read_csv(SAVE_PATH, encoding="utf-8-sig")
        # 根据评价唯一ID去重
        df_merge = pd.concat([df_all, df_new]).drop_duplicates(
            subset=["review_id"], keep="last"
        )
        df_merge.to_csv(SAVE_PATH, index=False, encoding="utf-8-sig")
        print(f"本次新增有效评价：{len(df_merge)-len(df_all)} 条，"
              f"累计数据：{len(df_merge)} 条")
        return df_merge
    except FileNotFoundError:
        df_new.to_csv(SAVE_PATH, index=False, encoding="utf-8-sig")
        print(f"首次创建数据文件，共写入 {len(df_new)} 条评价")
        return df_new

import re
import jieba
import jieba.posseg as pseg
# 加载自定义词典（电商领域词）
jieba.load_userdict("ecommerce_dict.txt")
# 停用词表
STOPWORDS = set(["的", "了", "很", "有点", "还是", "但是", "觉得", 
                 "这个", "那个", "一个", "非常", "真的"])
def preprocess_text(text):
    """
    评论文本预处理流水线
    """
    # 1. 去除HTML标签和特殊符号
    text = re.sub(r'<[^>]+>', '', text)
    text = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9]', ' ', text)
    
    # 2. 去除重复字符（如"好好好好" → "好"）
    text = re.sub(r'(.)\\1{3,}', r'\\1', text)
    
    # 3. 分词 + 词性标注
    words = pseg.cut(text)
    
    # 4. 过滤停用词、保留名词/动词/形容词
    filtered = []
    for word, flag in words:
        if word in STOPWORDS or len(word) < 2:
            continue
        if flag.startswith(('n', 'v', 'a')):  # 名词、动词、形容词
            filtered.append(word)
    
    return " ".join(filtered)

def detect_fake_review(review):
    """
    水军评论检测规则
    """
    red_flags = 0
    
    # 1. 内容重复度检测
    if len(set(review["content"])) / len(review["content"]) < 0.3:
        red_flags += 1
    
    # 2. 图片数量异常（大量无图或全图无文）
    if review["content"] == "" and len(review.get("img_urls", [])) > 5:
        red_flags += 1
    
    # 3. 时间集中度（同一时段大量相似评论）
    # 需结合历史数据统计
    
    # 4. 用户名模式（随机字母数字组合）
    if re.match(r'^[a-z0-9]{8,12}$', review["user_name"]):
        red_flags += 1
    
    return red_flags >= 2  # 命中2条以上标记为疑似水军

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import SGDClassifier
import joblib
import numpy as np
# 负面关键词库（用于快速预警）
NEG_WORDS = ["质量差", "破损", "掉色", "发货慢", "售后差", 
             "尺寸不符", "假货", "异味", "退货", "差评", 
             "失望", "愤怒", "被骗", "投诉"]
class SentimentAnalyzer:
    def __init__(self):
        self.vectorizer = TfidfVectorizer(max_features=5000, ngram_range=(1,2))
        self.model = SGDClassifier(loss='log_loss')
        self.is_trained = False
        
    def train(self, texts, labels):
        """训练模型"""
        X = self.vectorizer.fit_transform(texts)
        self.model.fit(X, labels)
        self.is_trained = True
        joblib.dump((self.vectorizer, self.model), "sentiment_model.pkl")
        
    def predict(self, texts):
        """预测情感"""
        if not self.is_trained:
            try:
                self.vectorizer, self.model = joblib.load("sentiment_model.pkl")
                self.is_trained = True
            except:
                # 未训练时使用规则引擎兜底
                return self._rule_based_predict(texts)
        
        X = self.vectorizer.transform(texts)
        probs = self.model.predict_proba(X)
        # 返回 [负向概率, 正向概率]
        return probs[:, 1]  # 正向概率
    
    def _rule_based_predict(self, texts):
        """规则兜底：基于负面词典"""
        results = []
        for text in texts:
            neg_score = sum(1 for w in NEG_WORDS if w in text)
            # 简单映射：负面词越多，正向概率越低
            pos_prob = max(0, 1 - neg_score * 0.3)
            results.append(pos_prob)
        return np.array(results)
    
    def analyze_batch(self, df):
        """批量分析评论情感"""
        texts = df["content"].apply(preprocess_text).tolist()
        sentiments = self.predict(texts)
        
        df["sentiment_score"] = sentiments
        df["sentiment_label"] = df["sentiment_score"].apply(
            lambda x: "正向" if x > 0.6 else ("负向" if x < 0.4 else "中性")
        )
        return df
# 实例化分析器
analyzer = SentimentAnalyzer()

# 商品属性词典
ATTR_DICT = {
    "质量": ["质量", "做工", "材质", "手感", "耐用"],
    "物流": ["物流", "快递", "发货", "配送", "速度"],
    "服务": ["客服", "售后", "服务", "态度", "回复"],
    "价格": ["价格", "贵", "便宜", "性价比", "划算"],
    "外观": ["外观", "颜色", "颜值", "设计", "款式"],
}
def extract_aspect_sentiment(text, sentiment_score):
    """
    提取属性级情感
    返回: {属性: 情感倾向}
    """
    aspects = {}
    for aspect, keywords in ATTR_DICT.items():
        for kw in keywords:
            if kw in text:
                # 简单规则：根据整体情感分数和关键词上下文判断
                aspects[aspect] = "正向" if sentiment_score > 0.5 else "负向"
                break
    return aspects

import json
# 钉钉/企业微信机器人配置
WEBHOOK_URL = "https://oapi.dingtalk.com/robot/send?access_token=xxx"
def send_alert(reviews, level="P1"):
    """发送预警通知"""
    contents = []
    for _, row in reviews.iterrows():
        contents.append(
            f"【{level}预警】\\n"
            f"时间：{row['review_time']}\\n"
            f"用户：{row['user_name']}\\n"
            f"内容：{row['content'][:100]}...\\n"
            f"规格：{row['sku_info']}"
        )
    
    message = {
        "msgtype": "markdown",
        "markdown": {
            "title": f"商品口碑{level}预警",
            "text": "\\n\\n---\\n\\n".join(contents[:5])  # 最多推送5条
        }
    }
    
    requests.post(WEBHOOK_URL, json=message, timeout=10)
def monitor_goods_review():
    """持续监控商品新增评价"""
    print(f"已启动商品口碑监控，商品ID：{TARGET_ITEM_ID}，"
          f"每{MONITOR_INTERVAL/60}分钟同步一次数据")
    
    while True:
        all_new_reviews = []
        page = 1
        batch = get_review_batch(TARGET_ITEM_ID, page)
        all_new_reviews.extend(batch)
        
        # 保存并清洗数据
        df_data = save_comments_to_csv(all_new_reviews)
        
        if len(df_data) == 0:
            time.sleep(MONITOR_INTERVAL)
            continue
        
        # 情感分析
        df_data = analyzer.analyze_batch(df_data)
        
        # 识别新增差评
        new_reviews = df_data[df_data["review_time"] >= 
                              datetime.now().strftime("%Y-%m-%d")]
        neg_reviews = new_reviews[new_reviews["sentiment_label"] == "负向"]
        
        if len(neg_reviews) > 0:
            print(f"\\n⚠️ 检测到 {len(neg_reviews)} 条新增差评：")
            for _, row in neg_reviews.iterrows():
                print(f"【{row['review_time']}】{row['user_name']}：{row['content'][:80]}")
            
            # P0 检测：敏感词
            p0_mask = neg_reviews["content"].str.contains(
                "|".join(["假货", "欺诈", "诈骗", "骗子"]), na=False
            )
            if p0_mask.any():
                send_alert(neg_reviews[p0_mask], level="P0")
            
            # P1 检测：高频属性负面
            for attr, kws in ATTR_DICT.items():
                attr_neg = neg_reviews[neg_reviews["content"].str.contains(
                    "|".join(kws), na=False
                )]
                if len(attr_neg) >= 3:
                    send_alert(attr_neg, level="P1")
        
        print(f"\\n等待{MONITOR_INTERVAL/60}分钟后执行下一轮同步...\\n")
        time.sleep(MONITOR_INTERVAL)

import matplotlib.pyplot as plt
from collections import Counter
from wordcloud import WordCloud
def generate_report(file_path):
    """生成口碑分析报告"""
    df = pd.read_csv(file_path, encoding="utf-8-sig")
    if len(df) == 0:
        return
    
    print("=" * 50)
    print("📊 商品口碑整体分析报告")
    print("=" * 50)
    print(f"总有效评价数量：{len(df)}")
    
    # 情感分布
    pos = len(df[df["sentiment_label"] == "正向"])
    neg = len(df[df["sentiment_label"] == "负向"])
    neu = len(df) - pos - neg
    print(f"正向：{pos} ({pos/len(df)*100:.1f}%) | "
          f"中性：{neu} ({neu/len(df)*100:.1f}%) | "
          f"负向：{neg} ({neg/len(df)*100:.1f}%)")
    
    # 差评高频痛点
    neg_text = "".join(df[df["sentiment_label"] == "负向"]["content"].tolist())
    words = jieba.lcut(neg_text)
    filter_words = ["的", "了", "很", "还是", "但是"]
    word_clean = [w for w in words if len(w) >= 2 and w not in filter_words]
    word_count = Counter(word_clean)
    
    print("\\n🔴 差评高频痛点 TOP10：")
    for word, cnt in word_count.most_common(10):
        print(f"  {word}：{cnt} 次")
    
    # 属性级分析
    print("\\n📦 各属性满意度：")
    for attr in ATTR_DICT.keys():
        attr_reviews = df[df["content"].str.contains("|".join(ATTR_DICT[attr]), na=False)]
        if len(attr_reviews) > 0:
            pos_rate = len(attr_reviews[attr_reviews["sentiment_label"] == "正向"]) / len(attr_reviews)
            print(f"  {attr}：正向率 {pos_rate*100:.1f}%（{len(attr_reviews)} 条）")
    
    # 绘制情感分布饼图
    plt.rcParams["font.sans-serif"] = ["SimHei"]
    plt.rcParams["axes.unicode_minus"] = False
    plt.pie([pos, neu, neg], labels=["正向", "中性", "负向"], 
            autopct="%1.1f%%", colors=["#52c41a", "#faad14", "#f5222d"])
    plt.title("商品评价情感分布")
    plt.savefig("sentiment_pie.png", dpi=150)
    plt.show()
    
    # 生成词云
    wc = WordCloud(font_path="simhei.ttf", width=800, height=400, 
                   background_color="white")
    wc.generate(" ".join(word_clean[:200]))
    wc.to_file("wordcloud.png")

if __name__ == "__main__":
    # 阶段1：首次全量同步历史评价
    print("开始同步商品全部历史评价...")
    full_review_data = []
    current_page = 1
    while True:
        page_data = get_review_batch(TARGET_ITEM_ID, current_page)
        if len(page_data) == 0:
            break
        full_review_data.extend(page_data)
        print(f"已同步第{current_page}页，当前累计{len(full_review_data)}条")
        current_page += 1
        time.sleep(2)  # 礼貌请求间隔
    
    save_comments_to_csv(full_review_data)
    
    # 阶段2：执行一次完整分析
    generate_report(SAVE_PATH)
    
    # 阶段3：启动持续实时监控
    print("\\n🚀 启动 7×24 小时口碑监控...")
    monitor_goods_review()