starmen1265
f86bb7043c
- stats.py: record_event, get_stats, format_stats_report - dev_workflow.py: stats 和 record-event CLI 命令 - 安全特性: 文件权限 600, 文件锁, 敏感数据过滤 - .gitignore: 排除数据文件
280 lines
8.5 KiB
Python
280 lines
8.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Cross-Debate 统计报告模块
|
|
提供事件记录和统计报告功能
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
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)
|
|
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:
|
|
data = json.load(f)
|
|
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())
|