功能介绍

这是一个深度分析微信对话数据的复杂系统，具备以下高级功能：

• 微信聊天记录多维度解析
• 社交关系网络图谱构建
• 对话情感波动分析
• 话题演化追踪
• 关键人物识别
• 异常对话检测
• 多账号数据聚合分析

系统架构

graph TD
    A[微信聊天记录] --> B[数据预处理]
    B --> C[自然语言处理]
    C --> D[关系网络构建]
    C --> E[情感分析]
    C --> F[话题建模]
    D --> G[社交图谱可视化]
    E --> H[情感时间线]
    F --> I[话题演化图]
    G --> J[综合仪表盘]
    H --> J
    I --> J

核心代码实现

import sqlite3
import json
import re
from datetime import datetime
from collections import defaultdict
import jieba
import jieba.analyse
import numpy as np
import pandas as pd
import networkx as nx
import matplotlib.pyplot as plt
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.decomposition import LatentDirichletAllocation
from transformers import pipeline
from pyecharts import options as opts
from pyecharts.charts import Graph, Line, WordCloud, Pie
from flask import Flask, render_template
from tqdm import tqdm
import spacy
from spacy.lang.zh import Chinese
from sentence_transformers import SentenceTransformer

nlp = spacy.load("zh_core_web_sm")
sentiment_analyzer = pipeline("sentiment-analysis", 
                            model="finiteautomata/bertweet-base-sentiment-analysis")
sbert_model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')

app = Flask(__name__)

class WeChatAnalyzer:
    """微信聊天记录分析核心类"""
    
    def __init__(self, db_path):
        self.db_path = db_path
        self.conn = sqlite3.connect(db_path)
        self.stopwords = self.load_stopwords()
        self.nlp = Chinese()
        self.nlp.add_pipe("sentencizer")
        
    def load_stopwords(self):
        """加载中文停用词表"""
        with open('chinese_stopwords.txt', 'r', encoding='utf-8') as f:
            return set([line.strip() for line in f])
    
    def preprocess_text(self, text):
        """文本预处理"""
        # 去除特殊字符
        text = re.sub(r'[^\w\s]', '', text)
        # 分词
        words = jieba.cut(text)
        # 去除停用词
        words = [word for word in words if word not in self.stopwords and len(word) > 1]
        return ' '.join(words)
    
    def load_conversations(self):
        """加载聊天记录"""
        query = """
        SELECT 
            msgId, talker, content, createTime, isSend 
        FROM message 
        WHERE type = 1 
        ORDER BY createTime
        """
        df = pd.read_sql(query, self.conn)
        df['createTime'] = pd.to_datetime(df['createTime'], unit='ms')
        df['processed_content'] = df['content'].apply(self.preprocess_text)
        return df
    
    def build_social_graph(self, df):
        """构建社交关系图"""
        G = nx.Graph()
        talkers = df['talker'].unique()
        
        # 添加节点
        for talker in talkers:
            G.add_node(talker, size=10)
        
        # 添加边（基于共同话题）
        vectorizer = TfidfVectorizer(max_features=1000)
        tfidf = vectorizer.fit_transform(df['processed_content'])
        
        # 计算用户话题分布
        user_topics = defaultdict(list)
        for _, row in df.iterrows():
            user_topics[row['talker']].append(row['processed_content'])
        
        # 计算用户相似度
        for i, talker1 in enumerate(talkers):
            for j, talker2 in enumerate(talkers):
                if i >= j:
                    continue
                
                # 使用SBERT计算语义相似度
                emb1 = sbert_model.encode(' '.join(user_topics[talker1]))
                emb2 = sbert_model.encode(' '.join(user_topics[talker2]))
                similarity = np.dot(emb1, emb2) / (np.linalg.norm(emb1) * np.linalg.norm(emb2))
                
                if similarity > 0.3:  # 相似度阈值
                    G.add_edge(talker1, talker2, weight=similarity)
        
        return G
    
    def analyze_sentiment_trend(self, df):
        """分析情感趋势"""
        df['sentiment'] = df['content'].apply(
            lambda x: sentiment_analyzer(x[:512])[0]['label']
        )
        
        # 情感分数映射
        sentiment_map = {'POSITIVE': 1, 'NEUTRAL': 0, 'NEGATIVE': -1}
        df['sentiment_score'] = df['sentiment'].map(sentiment_map)
        
        # 按时间窗口聚合
        df.set_index('createTime', inplace=True)
        daily_sentiment = df['sentiment_score'].resample('D').mean()
        
        return daily_sentiment
    
    def detect_key_topics(self, df, n_topics=5):
        """检测关键话题"""
        vectorizer = TfidfVectorizer(max_df=0.95, min_df=2)
        tfidf = vectorizer.fit_transform(df['processed_content'])
        
        # LDA主题建模
        lda = LatentDirichletAllocation(
            n_components=n_topics,
            max_iter=5,
            learning_method='online',
            learning_offset=50.,
            random_state=0
        )
        lda.fit(tfidf)
        
        # 提取关键词
        feature_names = vectorizer.get_feature_names_out()
        topics = []
        for topic_idx, topic in enumerate(lda.components_):
            top_features = [feature_names[i] for i in topic.argsort()[:-6:-1]]
            topics.append({
                'topic_id': topic_idx,
                'keywords': top_features,
                'weight': topic.sum()
            })
        
        return topics
    
    def identify_key_persons(self, df, G):
        """识别关键人物"""
        # 计算中心性指标
        degree_centrality = nx.degree_centrality(G)
        betweenness_centrality = nx.betweenness_centrality(G)
        pagerank = nx.pagerank(G)
        
        # 计算活跃度
        activity = df['talker'].value_counts().to_dict()
        
        # 综合评分
        persons = []
        for person in G.nodes():
            score = 0.3 * degree_centrality.get(person, 0) + \
                    0.3 * betweenness_centrality.get(person, 0) + \
                    0.2 * pagerank.get(person, 0) + \
                    0.2 * activity.get(person, 0) / len(df)
            
            persons.append({
                'name': person,
                'score': score,
                'degree': degree_centrality.get(person, 0),
                'betweenness': betweenness_centrality.get(person, 0),
                'activity': activity.get(person, 0)
            })
        
        # 按分数排序
        return sorted(persons, key=lambda x: x['score'], reverse=True)[:10]
    
    def detect_anomalies(self, df):
        """检测异常对话"""
        # 计算消息长度
        df['msg_length'] = df['content'].apply(len)
        
        # 计算时间间隔
        df['time_diff'] = df['createTime'].diff().dt.total_seconds().fillna(0)
        
        # 标记异常
        df['is_anomaly'] = 0
        df.loc[
            (df['msg_length'] > 500) | 
            (df['time_diff'] < 1), 
            'is_anomaly'
        ] = 1
        
        return df[df['is_anomaly'] == 1]
    
    def generate_visualizations(self, df, G, daily_sentiment, topics, key_persons):
        """生成可视化图表"""
        # 1. 社交关系图
        nodes = [{
            'name': node,
            'symbolSize': 10 + G.degree(node) * 2,
            'category': '普通成员' if G.degree(node) < 5 else '核心成员'
        } for node in G.nodes()]
        
        links = [{
            'source': edge[0],
            'target': edge[1],
            'value': G.edges[edge]['weight']
        } for edge in G.edges()]
        
        graph = (
            Graph()
            .add("", nodes, links,
                 repulsion=50,
                 linestyle_opts=opts.LineStyleOpts(curve=0.2),
                 label_opts=opts.LabelOpts(is_show=False))
            .set_global_opts(title_opts=opts.TitleOpts(title="社交关系网络"))
        )
        
        # 2. 情感趋势图
        sentiment_line = (
            Line()
            .add_xaxis(daily_sentiment.index.strftime('%Y-%m-%d').tolist())
            .add_yaxis("情感分数", daily_sentiment.values.round(2).tolist())
            .set_global_opts(
                title_opts=opts.TitleOpts(title="情感趋势"),
                tooltip_opts=opts.TooltipOpts(trigger="axis"),
                datazoom_opts=[opts.DataZoomOpts()]
            )
        )
        
        # 3. 话题词云
        wordcloud = (
            WordCloud()
            .add("", [(kw, int(weight*100)) for topic in topics 
                     for kw, weight in zip(topic['keywords'], 
                                         np.linspace(1, 0.2, len(topic['keywords'])))],
                 word_size_range=[12, 55])
            .set_global_opts(title_opts=opts.TitleOpts(title="话题关键词云"))
        )
        
        # 4. 关键人物饼图
        person_pie = (
            Pie()
            .add("", [(p['name'], p['score']) for p in key_persons],
                 radius=["30%", "75%"],
                 rosetype="radius")
            .set_global_opts(
                title_opts=opts.TitleOpts(title="关键人物影响力"),
                legend_opts=opts.LegendOpts(orient="vertical", pos_top="15%", pos_left="2%")
            )
        )
        
        return {
            'graph': graph.render_embed(),
            'sentiment': sentiment_line.render_embed(),
            'wordcloud': wordcloud.render_embed(),
            'key_persons': person_pie.render_embed()
        }

@app.route('/')
def dashboard():
    """主仪表盘"""
    analyzer = WeChatAnalyzer('wechat.db')
    df = analyzer.load_conversations()
    G = analyzer.build_social_graph(df)
    daily_sentiment = analyzer.analyze_sentiment_trend(df)
    topics = analyzer.detect_key_topics(df)
    key_persons = analyzer.identify_key_persons(df, G)
    anomalies = analyzer.detect_anomalies(df)
    
    visuals = analyzer.generate_visualizations(
        df, G, daily_sentiment, topics, key_persons
    )
    
    return render_template(
        'dashboard.html',
        total_messages=len(df),
        unique_talkers=len(G.nodes()),
        avg_sentiment=daily_sentiment.mean(),
        top_topics=topics[:3],
        key_persons=key_persons[:5],
        anomaly_count=len(anomalies),
        **visuals
    )

if __name__ == "__main__":
    app.run(host='0.0.0.0', port=5000, debug=True)

使用说明

1. 环境准备

# 安装核心依赖
pip install jieba pandas networkx matplotlib scikit-learn transformers sentence-transformers spacy flask pyecharts tqdm

# 下载中文语言模型
python -m spacy download zh_core_web_sm

# 下载预训练模型
python -c "from sentence_transformers import SentenceTransformer; SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')"

2. 数据准备

1. 使用微信备份工具导出聊天记录为SQLite数据库（命名为wechat.db）
2. 准备中文停用词表（chinese_stopwords.txt）

3. 启动系统

python wechat_analyzer.py

访问 http://localhost:5000 查看分析仪表盘

功能扩展建议

1. ​​多语言支持​​：扩展支持英文等其他语言分析
2. ​​实时分析​​：集成WebSocket实现实时更新
3. ​​多账号聚合​​：支持分析多个微信账号的数据
4. ​​语音消息分析​​：集成语音识别处理语音消息
5. ​​情感原因分析​​：识别导致情感变化的具体事件

适用场景

✅ 社交关系研究与分析
✅ 团队协作效率评估
✅ 客户服务对话质量监控
✅ 社群运营效果分析
✅ 个人社交网络管理

这个系统比之前的解决方案更专注于对话数据分析，整合了：

• 自然语言处理与文本挖掘
• 复杂网络分析
• 情感计算
• 话题建模
• 异常检测
• 交互式可视化

适合需要深度分析微信对话数据的个人或企业应用场景。