#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 叙事分析模块 识别叙事框架、情感操纵、议题设置和传播路径 """ from typing import Dict, List, Any from dataclasses import dataclass, field from collections import Counter import re try: import jieba JIEBA_AVAILABLE = True except ImportError: JIEBA_AVAILABLE = False try: from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer VADER_AVAILABLE = True except ImportError: VADER_AVAILABLE = False @dataclass class NarrativeFrame: """叙事框架""" frame_type: str = "" # threat, hope, anger, fear, value frame_name: str = "" description: str = "" keywords: List[str] = field(default_factory=list) strength: float = 0.0 # 0-1 target_audience: str = "" @dataclass class EmotionalAnalysis: """情感分析结果""" sentiment_polarity: float = 0.0 # -1到1 sentiment_intensity: float = 0.0 # 0到1 dominant_emotion: str = "" # positive, negative, neutral emotion_keywords: Dict[str, int] = field(default_factory=dict) emotion_triggers: List[str] = field(default_factory=list) @dataclass class TopicAnalysis: """议题分析""" main_topics: List[str] = field(default_factory=list) topic_weights: Dict[str, float] = field(default_factory=dict) topic_evolution: List[Dict[str, Any]] = field(default_factory=list) @dataclass class PropagationAnalysis: """传播分析""" key_nodes: List[str] = field(default_factory=list) propagation_path: List[str] = field(default_factory=list) propagation_speed: str = "" propagation_scope: str = "" amplification_factors: List[str] = field(default_factory=list) class NarrativeAnalyzer: """叙事分析器""" def __init__(self): # 初始化情感分析器 if VADER_AVAILABLE: self.sentiment_analyzer = SentimentIntensityAnalyzer() else: self.sentiment_analyzer = None print("警告：VADER未安装，情感分析将受限") # 定义叙事框架库 self.narrative_frames = { 'threat': NarrativeFrame( frame_type='threat', frame_name='威胁框架', description='制造威胁感知，激起恐惧和不安', keywords=['威胁', '危险', '风险', '危机', '紧急', '严峻', '挑战'], strength=0.0 ), 'hope': NarrativeFrame( frame_type='hope', frame_name='希望框架', description='塑造愿景，激发积极情绪', keywords=['希望', '愿景', '未来', '前景', '机遇', '发展', '和平'], strength=0.0 ), 'anger': NarrativeFrame( frame_type='anger', frame_name='愤怒框架', description='激化对立，引发愤怒情绪', keywords=['愤怒', '抗议', '反对', '谴责', '抵制', '抗议'], strength=0.0 ), 'fear': NarrativeFrame( frame_type='fear', frame_name='恐惧框架', description='制造恐惧和恐慌', keywords=['恐惧', '恐慌', '害怕', '担心', '忧虑', '不安'], strength=0.0 ), 'value': NarrativeFrame( frame_type='value', frame_name='价值观框架', description='强调价值观，构建道德高地', keywords=['自由', '民主', '人权', '正义', '公平', '平等'], strength=0.0 ) } def analyze(self, texts: List[str]) -> Dict[str, Any]: """ 执行完整的叙事分析 Args: texts: 待分析的文本列表 Returns: Dict: 完整的叙事分析结果 """ print("正在进行叙事分析...") results = {} # 1. 情感分析 results['emotional_analysis'] = self._analyze_emotions(texts) # 2. 叙事框架识别 results['narrative_frames'] = self._identify_narrative_frames(texts) # 3. 议题分析 results['topic_analysis'] = self._analyze_topics(texts) # 4. 传播分析（简化） results['propagation_analysis'] = self._analyze_propagation(texts) return results def _analyze_emotions(self, texts: List[str]) -> EmotionalAnalysis: """情感分析""" analysis = EmotionalAnalysis() if not texts: return analysis # 合并所有文本 combined_text = ' '.join(texts) if self.sentiment_analyzer: # 使用VADER分析 sentiment = self.sentiment_analyzer.polarity_scores(combined_text) analysis.sentiment_polarity = sentiment['compound'] analysis.sentiment_intensity = abs(sentiment['compound']) # 确定主导情感 if sentiment['compound'] >= 0.05: analysis.dominant_emotion = 'positive' elif sentiment['compound'] <= -0.05: analysis.dominant_emotion = 'negative' else: analysis.dominant_emotion = 'neutral' else: # 简化情感分析 positive_words = ['好', '积极', '正面', '优秀', '成功', '希望'] negative_words = ['坏', '消极', '负面', '失败', '威胁', '愤怒'] positive_count = sum(1 for word in positive_words if word in combined_text) negative_count = sum(1 for word in negative_words if word in combined_text) total = positive_count + negative_count if total > 0: analysis.sentiment_polarity = (positive_count - negative_count) / total analysis.sentiment_intensity = total / len(combined_text) * 100 if positive_count > negative_count: analysis.dominant_emotion = 'positive' elif negative_count > positive_count: analysis.dominant_emotion = 'negative' else: analysis.dominant_emotion = 'neutral' # 提取情感关键词 emotion_keywords = {} for frame_name, frame in self.narrative_frames.items(): for keyword in frame.keywords: count = combined_text.count(keyword) if count > 0: emotion_keywords[keyword] = count analysis.emotion_keywords = emotion_keywords # 识别情感触发词 triggers = sorted(emotion_keywords.items(), key=lambda x: x[1], reverse=True)[:5] analysis.emotion_triggers = [word for word, count in triggers] return analysis def _identify_narrative_frames(self, texts: List[str]) -> List[NarrativeFrame]: """识别叙事框架""" combined_text = ' '.join(texts) detected_frames = [] for frame_name, frame in self.narrative_frames.items(): # 计算框架关键词出现频率 keyword_count = sum(combined_text.count(kw) for kw in frame.keywords) if keyword_count > 0: # 计算框架强度 frame_strength = min(1.0, keyword_count / len(combined_text) * 100) detected_frame = NarrativeFrame( frame_type=frame.frame_type, frame_name=frame.frame_name, description=frame.description, keywords=frame.keywords, strength=round(frame_strength, 3) ) detected_frames.append(detected_frame) # 按强度排序 detected_frames.sort(key=lambda x: x.strength, reverse=True) return detected_frames def _analyze_topics(self, texts: List[str]) -> TopicAnalysis: """议题分析""" analysis = TopicAnalysis() if not texts: return analysis combined_text = ' '.join(texts) # 使用jieba分词（如果可用） if JIEBA_AVAILABLE: words = jieba.cut(combined_text) words = [word.strip() for word in words if len(word.strip()) > 1] else: # 简单分词（按空格和标点） words = re.findall(r'[\w\u4e00-\u9fff]+', combined_text) # 词频统计 word_counter = Counter(words) # 提取高频主题词 top_topics = word_counter.most_common(10) analysis.main_topics = [word for word, count in top_topics] # 计算主题权重 total_count = sum(count for word, count in top_topics) for word, count in top_topics: analysis.topic_weights[word] = count / total_count return analysis def _analyze_propagation(self, texts: List[str]) -> PropagationAnalysis: """传播分析（简化）""" analysis = PropagationAnalysis() if not texts: return analysis combined_text = ' '.join(texts) # 识别传播节点（简化的关键词匹配） propagation_keywords = ['媒体报道', '官方声明', '舆论反应', '社交平台', '国际关注'] for keyword in propagation_keywords: if keyword in combined_text: analysis.key_nodes.append(keyword) # 判断传播速度（基于文本长度和关键词密度） if len(analysis.key_nodes) > 3: analysis.propagation_speed = '快速' elif len(analysis.key_nodes) > 1: analysis.propagation_speed = '中等' else: analysis.propagation_speed = '缓慢' # 判断传播范围 scope_keywords = ['国际', '全球', '地区', '国内'] for keyword in scope_keywords: if keyword in combined_text: analysis.propagation_scope = keyword break if not analysis.propagation_scope: analysis.propagation_scope = '未明确' # 识别放大因素 amplification_keywords = ['热点', '热搜', '转发', '评论', '关注'] for keyword in amplification_keywords: if keyword in combined_text: analysis.amplification_factors.append(keyword) return analysis def main(): """测试主函数""" analyzer = NarrativeAnalyzer() # 测试文本 test_texts = [ "中国国防部发表声明，强调维护国家主权和领土完整的决心。", "相关军事演习是例行性的，不针对任何特定国家。", "美方对此表示关切，称此举可能加剧地区紧张局势。", "台湾地区当局称将密切关注事态发展。", "专家分析认为，各方应保持克制，避免误判。" ] # 执行分析 results = analyzer.analyze(test_texts) print("\n=== 叙事分析结果 ===") print(f"情感极性: {results['emotional_analysis'].sentiment_polarity:.3f}") print(f"主导情感: {results['emotional_analysis'].dominant_emotion}") print(f"检测到的叙事框架: {len(results['narrative_frames'])}") for frame in results['narrative_frames']: print(f" - {frame.frame_name} (强度: {frame.strength:.3f})") print(f"主要议题: {results['topic_analysis'].main_topics[:5]}") if __name__ == '__main__': main()