#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 时间线分析模块 构建详细时间线，识别事件序列和关键时间点 """ from typing import Dict, List, Any, Optional from dataclasses import dataclass, field from datetime import datetime import re @dataclass class TimelineEvent: """时间线事件""" id: str = "" timestamp: str = "" date: str = "" time: str = "" event_type: str = "" actor: str = "" action: str = "" description: str = "" source: str = "" source_type: str = "" evidence: List[str] = field(default_factory=list) def to_dict(self) -> Dict[str, Any]: return { 'id': self.id, 'timestamp': self.timestamp, 'date': self.date, 'time': self.time, 'event_type': self.event_type, 'actor': self.actor, 'action': self.action, 'description': self.description, 'source': self.source, 'source_type': self.source_type, 'evidence': self.evidence } @dataclass class Timeline: """时间线""" events: List[TimelineEvent] = field(default_factory=list) start_time: str = "" end_time: str = "" total_duration: str = "" key_events: List[TimelineEvent] = field(default_factory=list) def sort_events(self): """按时间排序事件""" self.events.sort(key=lambda x: x.timestamp) def add_event(self, event: TimelineEvent): """添加事件""" self.events.append(event) self.sort_events() def to_dict(self) -> Dict[str, Any]: return { 'start_time': self.start_time, 'end_time': self.end_time, 'total_duration': self.total_duration, 'total_events': len(self.events), 'key_events': [e.to_dict() for e.e in self.key_events], 'all_events': [e.to_dict() for e.e in self.events] } class TimelineAnalyzer: """时间线分析器""" def __init__(self): self.timeline = Timeline() def build_timeline(self, sources_data: List[Dict], time_range: str = None) -> Timeline: """ 构建时间线 Args: sources_data: 多源数据列表 time_range: 时间范围（如 '2026-03-01 - 2026-03-31'） Returns: Timeline: 完整时间线 """ print("正在构建时间线...") # 解析时间范围 if time_range: self.timeline.start_time, self.timeline.end_time = self._parse_time_range(time_range) self.timeline.total_duration = self._calculate_duration( self.timeline.start_time, self.timeline.end_time ) # 从源数据中提取事件 for source in sources_data: events = self._extract_events_from_source(source) for event in events: self.timeline.add_event(event) # 识别关键事件 self._identify_key_events() print(f"时间线构建完成，共 {len(self.timeline.events)} 个事件") return self.timeline def _parse_time_range(self, time_range: str) -> tuple: """解析时间范围""" if '-' in time_range: parts = time_range.split('-') return parts[0].strip(), parts[1].strip() return "", time_range def _calculate_duration(self, start: str, end: str) -> str: """计算持续时间""" if start and end: return f"{start} 至 {end}" return "未指定" def _extract_events_from_source(self, source: Dict) -> List[TimelineEvent]: """从源数据提取事件""" events = [] source_type = source.get('source_type', 'unknown') source_name = source.get('source_name', 'unknown') content = source.get('content', '') title = source.get('title', '') url = source.get('url', '') timestamp = source.get('timestamp', '') # 尝试从内容中提取时间信息 extracted_times = self._extract_timestamps(content) # 如果没有明确的时间戳，使用源数据中的时间戳 if not extracted_times and timestamp: event = TimelineEvent( id=f"{source_name}_{timestamp}", timestamp=timestamp, date=timestamp[:10] if len(timestamp) >= 10 else timestamp, time=timestamp[11:19] if len(timestamp) >= 19 else '', event_type='generic', actor=source_name, action='声明/报道', description=title, source=source_name, source_type=source_type, evidence=[url] ) events.append(event) # 如果提取到多个时间点，创建多个事件 for i, (date_str, time_str, context) in enumerate(extracted_times): event = TimelineEvent( id=f"{source_name}_{i}", timestamp=f"{date_str} {time_str}", date=date_str, time=time_str, event_type=self._infer_event_type(context), actor=source_name, action=self._infer_action(context), description=context, source=source_name, source_type=source_type, evidence=[url] ) events.append(event) return events def _extract_timestamps(self, text: str) -> List[tuple]: """从文本中提取时间戳""" timestamps = [] # 匹配日期时间格式 patterns = [ r'(\d{4}-\d{2}-\d{2})\s+(\d{2}:\d{2})', r'(\d{4}年\d{1,2}月\d{1,2}日)\s+(\d{1,2}:\d{2})', r'(\d{4}-\d{2}-\d{2})' ] for pattern in patterns: matches = re.finditer(pattern, text) for match in matches: groups = match.groups() if len(groups) >= 2: date_str = groups[0] time_str = groups[1] start_pos = max(0, match.start() - 50) end_pos = min(len(text), match.end() + 50) context = text[start_pos:end_pos] timestamps.append((date_str, time_str, context)) elif len(groups) == 1: date_str = groups[0] time_str = "00:00" start_pos = max(0, match.start() - 50) end_pos = min(len(text), match.end() + 50) context = text[start_pos:end_pos] timestamps.append((date_str, time_str, context)) return timestamps def _infer_event_type(self, context: str) -> str: """推断事件类型""" context_lower = context.lower() if '军事' in context_lower or '演习' in context_lower or '部署' in context_lower: return 'military_action' elif '外交' in context_lower or '声明' in context_lower: return 'official_statement' elif '报道' in context_lower: return 'media_report' elif '分析' in context_lower: return 'expert_analysis' else: return 'generic' def _infer_action(self, context: str) -> str: """推断行动类型""" context_lower = context.lower() if '发表' in context_lower or '发布' in context_lower: return '发表声明' elif '回应' in context_lower or '回应' in context_lower: return '回应' elif '举行' in context_lower or '进行' in context_lower: return '举行活动' elif '报道' in context_lower: return '报道' else: return '行动' def _identify_key_events(self): """识别关键事件""" if len(self.timeline.events) < 2: self.timeline.key_events = self.timeline.events return # 简化实现：选择开头、中间、结尾的事件作为关键事件 total = len(self.timeline.events) # 第一个事件 self.timeline.key_events.append(self.timeline.events[0]) # 中间事件（每3个事件选1个） for i in range(2, total - 2, 3): self.timeline.key_events.append(self.timeline.events[i]) # 最后一个事件 if total > 1: self.timeline.key_events.append(self.timeline.events[-1]) print(f"识别到 {len(self.timeline.key_events)} 个关键事件") def main(): """测试主函数""" 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' }, { 'source_type': 'official', 'source_name': '台湾地区防务部门', 'content': '2026-03-28 10:00 台湾地区防务部门称将密切关注事态发展。', 'title': '防务部门回应', 'url': 'http://example.xxx', 'timestamp': '2026-03-28 10:00' } ] # 构建时间线 timeline = analyzer.build_timeline( sources_data=mock_sources, time_range='2026-03-28 - 2026-03-31' ) print(f"\n时间线信息:") print(f" 开始时间: {timeline.start_time}") print(f" 结束时间: {timeline.end_time}") print(f" 持续时间: {timeline.total_duration}") print(f" 总事件数: {len(timeline.events)}") print(f" 关键事件数: {len(timeline.key_events)}") print(f"\n关键事件:") for event in timeline.key_events: print(f" [{event.date} {event.time}] {event.actor}: {event.action.action}") if __name__ == '__main__': main()