feedback_analyzer.py

"""
生成停用词和桑基图数据
"""

import pandas as pd
import jieba
import re
from collections import Counter
import networkx as nx
import matplotlib.pyplot as plt
import numpy as np
import warnings
from matplotlib.font_manager import FontProperties
import os
warnings.filterwarnings('ignore')
from collections import defaultdict
from sklearn.feature_extraction.text import CountVectorizer

# domain_vocab.py
DOMAIN_VOCAB = [
    # 中文通用领域词（去重 + 合并）
    "空间连接", "字段计算器", "建筑面积", "城市规划", "叠加分析", "空间连接功能",
    "数据表", "建筑层数", "地理处理", "相交功能", "现状地块", "相交叠加",
    "地块属性", "分地块", "容积率统计", "计算方法", "参数设置", "软件设置",
    "核密度分析", "热点分析", "带宽", "密度场", "焦点", "焦点统计",
    "地图代数", "条件分析", "差运算", "最大值", "交通", "像元大小",
    "参数", "凸包", "餐饮", "住宿", "搜索半径", "栅格计算器", "重分类", "Con函数",

    # 英文通用领域词（去重）
    "ArcGIS", "spatial join", "ArcMap", "Map algebra", "Kernel Density",
    "Con", "Getis - Ord Gi*", "NDVI", "Raster Calculator", "dwg", "catalog",
    "spatial join", "data manager", "POI",
]


def generate_domain_stopwords(df, text_columns, domain_keywords):
    """结合领域关键词生成停用词表"""
    # 1. 基础停用词（通用功能词）
    common_stopwords = {"的", "了", "在", "是", "是否", "我", "有", "和", "就", "不", "人", "都", "一", "一个", "上", "也", "很", "到", "说", "要", "去", "你", "会", "着", "没有", "看", "好", "自己", "这", "那", "这个", "那个", "什么", "怎么",
                        "哪里", "时候", "然后", "可能", "应该", "可以", "就是", "还是", "但是", "不过", "如果", "因为", "所以", "而且", "或者", "其实", "觉得", "认为", "希望", "能够", "需要", "知道", "表示", "这样", "那样", "这些", "那些", "有点", "一点",
                        "一些","进一步", "具体", "问题", "疑惑", "讲解", "需求", "难点", "操作", "应用", "场景", "进行", "对于", "实际", "情况", "结合",  "对于",
                        "学生", "老师", "实验", "报告", "作业", "课程", "课堂", "学习", "理解", "掌握", "明白", "清楚",
                        "建议", "希望", "请问", "想问", "不懂", "不会", "不知道", "不太会", "不太懂", "不太清楚",
                        }
    # 2. 领域关键词（如果有）
    domain_words = set(domain_keywords)
    # if domain_keywords_file and os.path.exists(domain_keywords_file):
    #     with open(domain_keywords_file, "r", encoding="utf-8") as f:
    #         domain_words = set([line.strip() for line in f if line.strip()])
    # 3. 合并文本并预处理
    all_text = ""
    for col in text_columns:
        all_text += " ".join(df[col].fillna("").astype(str))

    all_text = re.sub(r"[^\w\s]", "", all_text)
    all_text = re.sub(r"\d+", "", all_text)
    all_text = all_text.lower()  # 统一英文为小写（新增）
    words = jieba.lcut(all_text)

    # 4. 统计词频
    word_freq = Counter(words)
    word_freq = {word: freq for word, freq in word_freq.items() if len(word) > 1}

    # 5. 生成候选停用词
    # - 通用功能词
    # - 高频但非领域关键词的词汇（出现频率>50%且不在领域词表中）
    stopwords = common_stopwords.copy()

    for word, freq in word_freq.items():
        if freq > len(df) * 0.5 and word not in domain_words:
            stopwords.add(word)
    # 6. 保存停用词表
    # with open(output_path, "w", encoding="utf-8") as f:
    #     for word in stopwords:
    #         f.write(word + "\n")
    # print(f"领域定制化停用词表已生成，共{len(stopwords)}个词，保存至{output_path}")

    return stopwords

def load_and_preprocess_data(df, stopwords, domain_words):
    """加载数据并进行预处理（优化专业词汇识别）"""
    stopwords = set(stopwords)
    # with open(stopwords_path, "r", encoding="utf-8") as f:
    #     for line in f:
    #         stopwords.add(line.strip())

    question_types = {
        "s1": "难点",
        "s2": "讲解需求",
        "s3": "操作疑惑",
        "s4": "应用场景"
    }

    # 定义问题因果顺序（索引越小越靠前）
    question_hierarchy = ["s1", "s2", "s3", "s4"]

    # 创建专业词汇词典（提前加载所有领域词）
    professional_dict = {}
    # 添加域词词汇
    for word in domain_words:
        jieba.add_word(word, freq=10000)  # 高频率确保优先识别
        professional_dict[word] = 1


    # 创建专业词汇的正则表达式模式
    # 按长度降序排序，确保长词优先匹配
    sorted_domain = sorted(domain_words, key=len, reverse=True)
    pattern_str = "|".join(re.escape(word) for word in sorted_domain)
    professional_pattern = re.compile(f"({pattern_str})")

    def clean_text(text):
        if not isinstance(text, str):
            return []
        # 第一步：基础清洗
        text_cleaned = re.sub(r"[^\w\s]", "", text)
        text_cleaned = re.sub(r"\d+", "", text_cleaned)
        text_cleaned = text_cleaned.lower()
        # 第二步：专业词汇识别和标记
        # 找到所有专业词汇的位置
        matches = []
        for match in professional_pattern.finditer(text_cleaned):
            start, end = match.span()
            matches.append((start, end, text_cleaned[start:end]))
        # 第三步：分割文本（保护专业词汇）
        segments = []
        last_end = 0
        # 按匹配位置分割文本
        for start, end, word in matches:
            # 添加前面的普通文本
            if start > last_end:
                segments.append(text_cleaned[last_end:start])

            # 添加专业词汇（作为整体）
            segments.append(word)
            last_end = end
        # 添加最后一段文本
        if last_end < len(text_cleaned):
            segments.append(text_cleaned[last_end:])
        # 第四步：对非专业词汇部分进行分词
        final_words = []
        for segment in segments:
            if segment in professional_dict:
                # 专业词汇直接添加
                final_words.append(segment)
            else:
                # 普通文本进行分词
                words = jieba.lcut(segment)
                words = [w for w in words if w not in stopwords and len(w) > 1]
                final_words.extend(words)
        return final_words

    for col in ["s1", "s2", "s3", "s4"]:
        df[col + "_words"] = df[col].apply(clean_text)

    return df, question_types, question_hierarchy

def build_sankey_data(df, question_columns, top_n=30):
    """
    构建桑基图用的数据（DataFrame: source, target, value）
    - 仅保留前 top_n 个全局高频关键词
    """
    question_labels = {
        "s1": "S1_难点",
        "s2": "S2_讲解需求",
        "s3": "S3_操作疑惑",
        "s4": "S4_应用场景"
    }

    # 1. 统计全局关键词频率
    all_keywords = []
    for col in question_columns:
        all_keywords.extend([kw for kws in df[col + "_words"] for kw in kws])
    keyword_freq = Counter(all_keywords)
    core_keywords = set([kw for kw, _ in keyword_freq.most_common(top_n)])

    # 2. 构建桑基图原始数据
    link_counter = Counter()
    for _, row in df.iterrows():
        for q in question_columns:
            q_label = question_labels[q]
            keywords = row[q + "_words"]
            for kw in keywords:
                if kw in core_keywords:
                    link_counter[(q_label, kw)] += 1

    # 3. 转为 DataFrame
    sankey_data = pd.DataFrame([
        {"source": src, "target": tgt, "value": val}
        for (src, tgt), val in link_counter.items()
    ])
    # sankey_data.to_csv("E:\\data\\20250621Edu\\sankey_d\\sankey_data.csv", index=False, encoding='utf-8-sig')
    # sankey_data.to_csv("E:\\data\\20250621Edu\\sankey_d\\sankey_data2.csv", index=False, encoding='utf-8-sig')
    print("桑基图保存成功!")
    return sankey_data

def generate_sankey_df(file_path, text_columns, domain_words, top_n=30):
    df = pd.read_excel(file_path)

    stopwords = generate_domain_stopwords(df, text_columns, domain_words)
    df, question_types, question_hierarchy = load_and_preprocess_data(df, stopwords, domain_words)
    sankey_data = build_sankey_data(df, text_columns, top_n)
    return sankey_data

if __name__ == '__main__':
    """主函数：执行完整的分析流程"""
    file_path = "E:\\data\\20250621Edu\\ex02.xlsx"
    stopwords_path = "E:\\data\\20250621Edu\\stop\\stop2.txt"
    text_columns = ["s1", "s2", "s3", "s4"]
    top_n = 30 # 设置频数排在前30的关键词
    sankey_data = generate_sankey_df(file_path, text_columns, DOMAIN_VOCAB, top_n)