261 lines
7.0 KiB
Python
261 lines
7.0 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Cross 模式统计报告模块
|
||
|
||
提供事件记录和统计报告生成功能
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import tempfile
|
||
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)
|
||
|
||
|
||
def _load_stats_data() -> Dict[str, Any]:
|
||
"""加载统计数据"""
|
||
if not STATS_FILE.exists():
|
||
return {"events": [], "metadata": {"created_at": datetime.now().isoformat()}}
|
||
|
||
try:
|
||
with open(STATS_FILE, 'r', encoding='utf-8') as f:
|
||
return json.load(f)
|
||
except (json.JSONDecodeError, IOError):
|
||
return {"events": [], "metadata": {"created_at": datetime.now().isoformat()}}
|
||
|
||
|
||
def _atomic_write(data: Dict[str, Any]):
|
||
"""原子写入统计数据"""
|
||
_ensure_stats_dir()
|
||
|
||
# 写入临时文件
|
||
fd, temp_path = tempfile.mkstemp(dir=STATS_DIR, suffix='.tmp')
|
||
try:
|
||
with os.fdopen(fd, 'w', encoding='utf-8') as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
|
||
# 原子重命名
|
||
os.replace(temp_path, STATS_FILE)
|
||
except Exception:
|
||
# 清理临时文件
|
||
if os.path.exists(temp_path):
|
||
os.unlink(temp_path)
|
||
raise
|
||
|
||
|
||
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]["timestamp"]
|
||
stats["last_event"] = events[-1]["timestamp"]
|
||
|
||
for event in events:
|
||
stats["events_by_type"][event["event_type"]] += 1
|
||
stats["events_by_mode"][event["mode"]] += 1
|
||
|
||
# 按日期统计
|
||
date = event["timestamp"][:10] # 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))
|