#!/usr/bin/env python3 """ Cross 模式统计报告模块 提供事件记录和统计报告生成功能 """ import json import os import tempfile import fcntl from datetime import datetime from pathlib import Path from typing import Dict, List, Any, Optional from collections import defaultdict # 统计数据存储路径 STATS_DIR = Path(__file__).parent / "data" STATS_FILE = STATS_DIR / "stats_data.json" # 敏感字段列表（不记录这些字段） SENSITIVE_FIELDS = {"password", "token", "secret", "key", "credential", "auth"} def _ensure_stats_dir(): """确保统计目录存在""" STATS_DIR.mkdir(parents=True, exist_ok=True) # 锁文件路径 LOCK_FILE = STATS_DIR / "stats_data.lock" def _acquire_lock(lock_file_path: Path): """获取文件锁""" lock_file_path.parent.mkdir(parents=True, exist_ok=True) lock_fd = open(lock_file_path, 'w') fcntl.flock(lock_fd.fileno(), fcntl.LOCK_EX) return lock_fd def _release_lock(lock_fd): """释放文件锁""" fcntl.flock(lock_fd.fileno(), fcntl.LOCK_UN) lock_fd.close() def _load_stats_data() -> Dict[str, Any]: """加载统计数据（带锁保护）""" if not STATS_FILE.exists(): return {"events": [], "metadata": {"created_at": datetime.now().isoformat()}} lock_fd = None try: lock_fd = _acquire_lock(LOCK_FILE) with open(STATS_FILE, 'r', encoding='utf-8') as f: return json.load(f) except json.JSONDecodeError as e: # 备份损坏的文件 backup_path = STATS_FILE.with_suffix(f'.corrupted.{datetime.now().strftime("%Y%m%d_%H%M%S")}.bak') try: STATS_FILE.rename(backup_path) except OSError: pass # 备份失败时忽略 return {"events": [], "metadata": {"created_at": datetime.now().isoformat()}} except IOError: return {"events": [], "metadata": {"created_at": datetime.now().isoformat()}} finally: if lock_fd: _release_lock(lock_fd) def _atomic_write(data: Dict[str, Any]): """原子写入统计数据（带锁保护）""" _ensure_stats_dir() lock_fd = None temp_path = None try: lock_fd = _acquire_lock(LOCK_FILE) # 写入临时文件 with tempfile.NamedTemporaryFile( mode='w', encoding='utf-8', dir=STATS_DIR, suffix='.tmp', delete=False ) as temp_file: json.dump(data, temp_file, ensure_ascii=False, indent=2) temp_path = temp_file.name # 原子重命名 os.replace(temp_path, STATS_FILE) temp_path = None # 防止在 finally 中再次删除 finally: if lock_fd: _release_lock(lock_fd) # 清理临时文件 if temp_path and os.path.exists(temp_path): os.unlink(temp_path) def _filter_sensitive_data(data: Dict[str, Any]) -> Dict[str, Any]: """过滤敏感数据""" filtered = {} for key, value in data.items(): # 检查键名是否包含敏感字段 if any(sensitive in key.lower() for sensitive in SENSITIVE_FIELDS): filtered[key] = "[REDACTED]" elif isinstance(value, dict): filtered[key] = _filter_sensitive_data(value) else: filtered[key] = value return filtered def record_event( event_type: str, mode: str = "cross", metadata: Optional[Dict[str, Any]] = None ) -> bool: """ 记录事件 Args: event_type: 事件类型（如 'command_executed', 'task_completed' 等） mode: 工作模式（默认 'cross'） metadata: 事件元数据 Returns: bool: 记录是否成功 Raises: TypeError: metadata 不是 dict 类型时 """ try: # metadata 类型校验 if metadata is not None and not isinstance(metadata, dict): raise TypeError(f"metadata 必须是 dict 类型，当前为: {type(metadata).__name__}") # 过滤敏感数据 safe_metadata = _filter_sensitive_data(metadata or {}) # 创建事件记录 event = { "event_type": event_type, "mode": mode, "timestamp": datetime.now().isoformat(), "metadata": safe_metadata } # 加载现有数据 data = _load_stats_data() data["events"].append(event) data["metadata"]["last_updated"] = datetime.now().isoformat() # 原子写入 _atomic_write(data) return True except Exception as e: print(f"记录事件失败: {e}") return False def get_stats(mode: Optional[str] = None) -> Dict[str, Any]: """ 生成统计报告 Args: mode: 筛选特定模式，None 表示所有模式 Returns: Dict: 统计报告数据 """ data = _load_stats_data() events = data.get("events", []) # 筛选模式 if mode: events = [e for e in events if e.get("mode") == mode] # 统计数据 stats = { "total_events": len(events), "events_by_type": defaultdict(int), "events_by_mode": defaultdict(int), "events_by_date": defaultdict(int), "first_event": None, "last_event": None, "metadata": data.get("metadata", {}) } if events: stats["first_event"] = events[0].get("timestamp") stats["last_event"] = events[-1].get("timestamp") for event in events: event_type = event.get("event_type", "unknown") event_mode = event.get("mode", "unknown") stats["events_by_type"][event_type] += 1 stats["events_by_mode"][event_mode] += 1 # 按日期统计 timestamp = event.get("timestamp", "") date = timestamp[:10] if timestamp else "unknown" # YYYY-MM-DD stats["events_by_date"][date] += 1 # 转换 defaultdict 为普通 dict stats["events_by_type"] = dict(stats["events_by_type"]) stats["events_by_mode"] = dict(stats["events_by_mode"]) stats["events_by_date"] = dict(stats["events_by_date"]) return stats def format_stats_report(stats: Dict[str, Any]) -> str: """ 格式化统计报告为可读文本 Args: stats: 统计数据 Returns: str: 格式化的报告文本 """ lines = [ "=" * 60, "Cross 模式统计报告", "=" * 60, "", f"总事件数: {stats['total_events']}", "" ] if stats["total_events"] == 0: lines.append("暂无统计数据") return "\n".join(lines) # 时间范围 lines.extend([ f"首次事件: {stats['first_event']}", f"最后事件: {stats['last_event']}", "" ]) # 按类型统计 if stats["events_by_type"]: lines.append("按事件类型统计:") for event_type, count in sorted(stats["events_by_type"].items()): lines.append(f" - {event_type}: {count}") lines.append("") # 按模式统计 if stats["events_by_mode"]: lines.append("按模式统计:") for mode, count in sorted(stats["events_by_mode"].items()): lines.append(f" - {mode}: {count}") lines.append("") # 按日期统计（最近7天） if stats["events_by_date"]: lines.append("按日期统计（最近7天）:") sorted_dates = sorted(stats["events_by_date"].items(), reverse=True)[:7] for date, count in sorted_dates: lines.append(f" - {date}: {count}") lines.append("") lines.append("=" * 60) return "\n".join(lines) def clear_stats(confirm: bool = False) -> bool: """ 清空统计数据 Args: confirm: 确认清空 Returns: bool: 是否成功清空 """ if not confirm: return False try: if STATS_FILE.exists(): STATS_FILE.unlink() return True except Exception as e: print(f"清空统计数据失败: {e}") return False if __name__ == "__main__": # 测试代码 print("测试 stats 模块") # 记录一些测试事件 record_event("test_event", mode="cross", metadata={"user": "test", "action": "demo"}) record_event("another_event", mode="normal", metadata={"data": "sample"}) # 生成报告 stats = get_stats() print(format_stats_report(stats))