#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 事件综合分析技能测试脚本 验证核心模块功能 """ import sys import os # 添加技能目录到Python路径 sys.path.insert(0, os.path.dirname(os.path.dirname(__file__))) from core.collector import SourceCollector, SourceItem from core.analyzer import EventAnalyzer from core.narrative import NarrativeAnalyzer from core.timeline import TimelineAnalyzer from core.report_generator import ReportGenerator from datetime import datetime def test_collector(): """测试信源采集器""" print("\n=== 测试1: 信源采集器 ===") collector = SourceCollector() collector.load_sources_config() # 创建模拟数据 mock_items = [ SourceItem( id='test1', url='http://www.mod.gov.cn/test', source_type='official', source_name='中国国防部', content='2026-03-28 中国国防部发表声明，强调维护国家主权。', title='国防部声明', timestamp='2026-03-28 08:00' ), SourceItem( id='test2', url='http://www.xinhuanet.com/test', source_type='media', source_name='新华社', content='2026-03-28 新华社报道，相关军事演习是例行性的。', title='军事演习报道', timestamp='2026-03-28 09:30' ) ] collector.collected_items = mock_items # 测试验证 verification_results = collector.verify_sources(mock_items) print(f"✓ 采集器初始化成功") print(f"✓ 模拟数据加载成功: {len(mock_items)} 条") print(f"✓ 多源验证成功: {len(verification_results)} 个结果") for result in verification_results: print(f" - {result.item_id}: 一致性 {result.consistency_score:.2f}, 置信度 {result.confidence}") return True def test_narrative_analyzer(): """测试叙事分析器""" print("\n=== 测试2: 叙事分析器 ===") analyzer = NarrativeAnalyzer() # 测试文本 test_texts = [ "中国国防部发表声明，强调维护国家主权和领土完整的决心。", "相关军事演习是例行性的，不针对任何特定国家。", "美方对此表示关切，称此举可能加剧地区紧张局势。", "台湾地区当局称将密切关注事态发展。", "专家分析认为，各方应保持克制，避免误判。" ] # 执行分析 results = analyzer.analyze(test_texts) print(f"✓ 叙事分析器初始化成功") print(f"✓ 情感分析完成") print(f" - 情感极性: {results['emotional_analysis'].sentiment_polarity:.3f}") print(f" - 主导情感: {results['emotional_analysis'].dominant_emotion}") print(f" - 情感强度: {results['emotional_analysis'].sentiment_intensity:.3f}") print(f"✓ 叙事框架识别完成: {len(results['narrative_frames'])} 个框架") for frame in results['narrative_frames']: print(f" - {frame.frame_name} (强度: {frame.strength:.3f})") print(f"✓ 议题分析完成") print(f" - 主要议题: {results['topic_analysis'].main_topics[:5]}") return True def test_timeline_analyzer(): """测试时间线分析器""" print("\n=== 测试3: 时间线分析器 ===") analyzer = TimelineAnalyzer() # 模拟源数据 mock_sources = [ { 'source_type': 'official', 'source_name': '中国国防部', 'content': '2026-03-28 08:00 中国国防部发表声明，强调维护国家主权。', 'title': '国防部声明', 'url': 'http://www.mod.gov.cn/xxx', 'timestamp': '2026-03-28 08:00' }, { 'source_type': 'media', 'source_name': '新华社', 'content': '2026-03-28 09:30 新华社报道，相关军事演习是例行性的。', 'title': '军事演习报道', 'url': 'http://www.xinhuanet.com/xxx', 'timestamp': '2026-03-28 09:30' } ] # 构建时间线 timeline = analyzer.build_timeline( sources_data=mock_sources, time_range='2026-03-28 - 2026-03-31' ) print(f"✓ 时间线构建成功") print(f" - 总事件数: {len(timeline.events)}") print(f" - 关键事件数: {len(timeline.key_events)}") print(f" - 时间范围: {timeline.start_time} 至 {timeline.end_time}") return True def test_report_generator(): """测试报告生成器""" print("\n=== 测试4: 报告生成器 ===") generator = ReportGenerator() # 模拟报告数据 mock_report_data = { 'title': '测试事件综合分析报告', 'event_overview': '这是一个测试事件，用于验证报告生成功能。', 'timeline_events': [ { 'timestamp': '2026-03-28 08:00', 'source': '中国国防部', 'description': '发表声明' } ], 'key_evidence': [ { 'title': '国防部声明', 'source_name': '中国国防部', 'source_type': 'official', 'url': 'http://www.mod.gov.cn/test', 'content': '国防部声明内容...' } ], 'china_mainland_behavior': { 'action_type': '外交', 'tactical_intent': '政策宣示', 'strategic_goal': '维护主权' }, 'taiwan_region_behavior': { 'action_type': '未识别' }, 'external_behavior': { 'action_type': '未识别' }, 'influence_assessment': { 'political_diplomatic': '存在政治外交影响', 'security_military': '存在安全军事影响', 'economic_social': '影响有限', 'opinion_psychological': '舆论关注度中等' }, 'deep_analysis': { 'root_cause': '政治日程与历史因素', 'scenarios': [ { 'name': '基线情景', 'probability': 0.5, 'description': '维持现状', 'potential_outcomes': ['局势稳定'] } ], 'conflict_risks': ['误判风险'], 'misjudgment_factors': ['信息不对称'], 'recommendations': ['保持克制'] } } # 生成报告 md_report = generator.generate_markdown(mock_report_data) print(f"✓ Markdown报告生成成功") print(f" - 报告长度: {len(md_report)} 字符") # 保存测试报告 generator.save_report(md_report, 'reports/test_report.md') return True def main(): """主测试函数""" print("=" * 60) print("事件综合分析技能功能测试") print("=" * 60) test_results = [] try: test_results.append(("信源采集器", test_collector())) except Exception as e: print(f"✗ 信源采集器测试失败: {str(e)}") test_results.append(("信源采集器", False)) try: test_results.append(("叙事分析器", test_narrative_analyzer())) except Exception as e: print(f"✗ 叙事分析器测试失败: {str(e)}") test_results.append(("叙事分析器", False)) try: test_results.append(("时间线分析器", test_timeline_analyzer())) except Exception as e: print(f"✗ 时间线分析器测试失败: {str(e)}") test_results.append(("时间线分析器", False)) try: test_results.append(("报告生成器", test_report_generator())) except Exception as e: print(f"✗ 报告生成器测试失败: {str(e)}") test_results.append(("报告生成器", False)) # 总结 print("\n" + "=" * 60) print("测试总结") print("=" * 60) success_count = sum(1 for name, success in test_results if success) total_count = len(test_results) for name, success in test_results: status = "✓ 成功" if success else "✗ 失败" print(f"{status}: {name}") print(f"\n总计: {success_count}/{total_count} 测试通过") if success_count == total_count: print("\n🎉 所有测试通过！技能已就绪。") return 0 else: print(f"\n⚠️ {total_count - success_count} 个测试失败，请检查错误信息。") return 1 if __name__ == '__main__': exit(main())