#!/usr/bin/env python3 """ Cross-Debate 统计报告模块 提供事件记录和统计报告功能 """ import json import os import sys import fcntl from datetime import datetime, timezone from pathlib import Path from typing import Dict, List, Any, Optional class StatsManager: """统计管理器""" # 敏感字段列表 SENSITIVE_FIELDS = {'password', 'token', 'secret', 'api_key', 'apikey', 'credential'} def __init__(self, stats_file: str = None): """初始化统计管理器 Args: stats_file: 统计数据文件路径，默认为 ~/.dev-workflow/stats.json """ if stats_file is None: stats_dir = Path.home() / '.dev-workflow' stats_dir.mkdir(parents=True, exist_ok=True) stats_file = stats_dir / 'stats.json' self.stats_file = Path(stats_file) self._ensure_stats_file() def _filter_sensitive_data(self, data: Dict[str, Any]) -> Dict[str, Any]: """过滤敏感数据 Args: data: 原始数据字典 Returns: 过滤后的数据字典 """ filtered = {} for key, value in data.items(): key_lower = key.lower() # 检查是否包含敏感字段 if any(sensitive in key_lower for sensitive in self.SENSITIVE_FIELDS): filtered[key] = '***REDACTED***' elif isinstance(value, dict): filtered[key] = self._filter_sensitive_data(value) elif isinstance(value, list): # 处理列表中的字典元素 filtered[key] = [ self._filter_sensitive_data(item) if isinstance(item, dict) else item for item in value ] else: filtered[key] = value return filtered def _ensure_stats_file(self): """确保统计文件存在并设置正确权限""" if not self.stats_file.exists(): self.stats_file.parent.mkdir(parents=True, exist_ok=True) self._write_data({"events": []}) # 设置文件权限为 600 (仅所有者可读写) self.stats_file.chmod(0o600) def _read_data(self) -> Dict[str, Any]: """读取统计数据（带文件锁） Returns: 统计数据字典 """ if not self.stats_file.exists(): return {"events": []} with open(self.stats_file, 'r') as f: # 获取共享锁（读锁） fcntl.flock(f.fileno(), fcntl.LOCK_SH) try: try: data = json.load(f) except json.JSONDecodeError as e: # 文件损坏时返回空数据结构 print(f"⚠️ 警告: 统计文件损坏，将返回空数据: {e}", file=sys.stderr) return {"events": []} finally: # 释放锁 fcntl.flock(f.fileno(), fcntl.LOCK_UN) return data def _write_data(self, data: Dict[str, Any]): """写入统计数据（带文件锁） Args: data: 要写入的数据字典 """ with open(self.stats_file, 'w') as f: # 获取排他锁（写锁） fcntl.flock(f.fileno(), fcntl.LOCK_EX) try: json.dump(data, f, indent=2, ensure_ascii=False) finally: # 释放锁 fcntl.flock(f.fileno(), fcntl.LOCK_UN) # 确保文件权限为 600 self.stats_file.chmod(0o600) def record_event( self, event_type: str, mode: str, metadata: Optional[Dict[str, Any]] = None, status: str = "success" ) -> bool: """记录事件 Args: event_type: 事件类型 (debate|review|audit) mode: 模式 (cross|normal) metadata: 事件元数据 status: 事件状态 (success|failed) Returns: 是否记录成功 """ # 验证参数 valid_types = {'debate', 'review', 'audit'} valid_modes = {'cross', 'normal'} valid_statuses = {'success', 'failed'} if event_type not in valid_types: raise ValueError(f"Invalid event_type: {event_type}. Must be one of {valid_types}") if mode not in valid_modes: raise ValueError(f"Invalid mode: {mode}. Must be one of {valid_modes}") if status not in valid_statuses: raise ValueError(f"Invalid status: {status}. Must be one of {valid_statuses}") # 过滤敏感数据 if metadata is None: metadata = {} metadata = self._filter_sensitive_data(metadata) # 创建事件记录 event = { "timestamp": datetime.now(timezone.utc).isoformat(), "event_type": event_type, "mode": mode, "status": status, "metadata": metadata } # 读取现有数据 data = self._read_data() # 添加新事件 data["events"].append(event) # 写入数据 self._write_data(data) return True def get_stats(self) -> Dict[str, Any]: """获取统计数据 Returns: 统计数据字典，包含总数、各类型计数等 """ data = self._read_data() events = data.get("events", []) # 基础统计 stats = { "total_events": len(events), "by_type": {}, "by_mode": {}, "by_status": {}, "by_type_and_mode": {}, "recent_events": events[-10:] if events else [] # 最近10条事件 } # 统计各维度 for event in events: event_type = event.get("event_type", "unknown") mode = event.get("mode", "unknown") status = event.get("status", "unknown") # 按类型统计 stats["by_type"][event_type] = stats["by_type"].get(event_type, 0) + 1 # 按模式统计 stats["by_mode"][mode] = stats["by_mode"].get(mode, 0) + 1 # 按状态统计 stats["by_status"][status] = stats["by_status"].get(status, 0) + 1 # 按类型和模式交叉统计 key = f"{event_type}_{mode}" stats["by_type_and_mode"][key] = stats["by_type_and_mode"].get(key, 0) + 1 return stats def format_stats_report(self) -> str: """格式化统计报告 Returns: 格式化的统计报告字符串 """ stats = self.get_stats() report_lines = [ "=" * 50, "Cross-Debate 统计报告", "=" * 50, "", f"📊 总事件数: {stats['total_events']}", "", "📋 按事件类型统计:", ] for event_type, count in sorted(stats["by_type"].items()): report_lines.append(f" - {event_type}: {count}") report_lines.append("") report_lines.append("🔀 按模式统计:") for mode, count in sorted(stats["by_mode"].items()): report_lines.append(f" - {mode}: {count}") report_lines.append("") report_lines.append("✅ 按状态统计:") for status, count in sorted(stats["by_status"].items()): report_lines.append(f" - {status}: {count}") report_lines.append("") report_lines.append("📊 按类型和模式交叉统计:") for key, count in sorted(stats["by_type_and_mode"].items()): report_lines.append(f" - {key}: {count}") report_lines.append("") report_lines.append("=" * 50) return "\n".join(report_lines) # 便捷函数 _manager = None def _get_manager() -> StatsManager: """获取全局统计管理器实例""" global _manager if _manager is None: _manager = StatsManager() return _manager def record_event(event_type: str, mode: str, metadata: Optional[Dict[str, Any]] = None, status: str = "success") -> bool: """记录事件（便捷函数）""" return _get_manager().record_event(event_type, mode, metadata, status) def get_stats() -> Dict[str, Any]: """获取统计数据（便捷函数）""" return _get_manager().get_stats() def format_stats_report() -> str: """格式化统计报告（便捷函数）""" return _get_manager().format_stats_report() if __name__ == "__main__": # 测试代码 print("测试统计模块...") # 记录一些测试事件 record_event("debate", "cross", {"topic": "test1", "result": "win"}) record_event("review", "normal", {"files": 5}) record_event("audit", "cross", {"issues": 3}, status="failed") # 打印报告 print(format_stats_report())