#!/usr/bin/env python3 """ Cross 模式统计报告模块 提供事件记录和统计报告生成功能 """ import json import os import tempfile import fcntl import logging from datetime import datetime from pathlib import Path from typing import Dict, List, Any, Optional from collections import defaultdict # 配置日志 logger = logging.getLogger(__name__) # 统计数据存储路径 STATS_DIR = Path(__file__).parent / "data" STATS_FILE = STATS_DIR / "stats_data.json" # 锁文件路径 LOCK_FILE = STATS_DIR / "stats_data.lock" # 最大事件数量限制 MAX_EVENTS = 10000 # 敏感字段列表（不记录这些字段） SENSITIVE_FIELDS = {"password", "token", "secret", "key", "credential", "auth"} def _ensure_stats_dir(): """确保统计目录存在""" STATS_DIR.mkdir(parents=True, exist_ok=True) def _acquire_lock(lock_file_path: Path): """获取文件锁 Raises: OSError: 无法创建或锁定文件时 """ try: 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 except (OSError, IOError) as e: logger.error(f"获取文件锁失败: {e}") raise def _release_lock(lock_fd): """释放文件锁""" if lock_fd is None: return try: fcntl.flock(lock_fd.fileno(), fcntl.LOCK_UN) lock_fd.close() except (OSError, IOError) as e: logger.warning(f"释放文件锁时发生错误: {e}") try: lock_fd.close() except Exception: pass def _load_stats_data() -> Dict[str, Any]: """加载统计数据（带锁保护） Returns: Dict: 统计数据，加载失败时返回空结构 """ 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: logger.warning(f"JSON 解析失败，将备份损坏文件: {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) logger.info(f"已备份损坏文件到: {backup_path}") except OSError as backup_error: logger.error(f"备份损坏文件失败: {backup_error}") return {"events": [], "metadata": {"created_at": datetime.now().isoformat()}} except (OSError, IOError) as e: logger.error(f"读取统计文件失败: {e}") return {"events": [], "metadata": {"created_at": datetime.now().isoformat()}} except Exception as e: logger.error(f"加载统计数据时发生未知错误: {e}") return {"events": [], "metadata": {"created_at": datetime.now().isoformat()}} finally: _release_lock(lock_fd) def _atomic_write(data: Dict[str, Any]): """原子写入统计数据（带锁保护） Raises: OSError: 文件操作失败时 TypeError: 数据无法序列化为 JSON 时 """ _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 中再次删除 except (OSError, IOError) as e: logger.error(f"写入统计文件失败: {e}") raise except (TypeError, ValueError) as e: logger.error(f"JSON 序列化失败: {e}") raise finally: _release_lock(lock_fd) # 清理临时文件 if temp_path and os.path.exists(temp_path): try: os.unlink(temp_path) except OSError as e: logger.warning(f"清理临时文件失败: {e}") 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) # 事件数量限制：超过 MAX_EVENTS 时移除最旧的事件 if len(data["events"]) > MAX_EVENTS: data["events"] = data["events"][-MAX_EVENTS:] data["metadata"]["last_updated"] = datetime.now().isoformat() # 原子写入 _atomic_write(data) return True except TypeError: raise # 重新抛出类型错误 except Exception as e: logger.error(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 # 安全检查：验证管理员令牌 admin_token = os.getenv("ADMIN_TOKEN") if not admin_token: logger.error("清空统计数据需要设置 ADMIN_TOKEN 环境变量") return False try: if STATS_FILE.exists(): STATS_FILE.unlink() return True except Exception as e: logger.error(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))