326 lines
11 KiB
Python
326 lines
11 KiB
Python
#!/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
|
||
如果提供相对路径，将解析到 ~/.dev-workflow/ 目录内
|
||
如果提供绝对路径，必须在 ~/.dev-workflow/ 目录内
|
||
"""
|
||
# 定义安全目录
|
||
safe_dir = Path.home() / '.dev-workflow'
|
||
safe_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
if stats_file is None:
|
||
self.stats_file = safe_dir / 'stats.json'
|
||
else:
|
||
stats_path = Path(stats_file)
|
||
|
||
# 安全校验：防止路径遍历攻击
|
||
if stats_path.is_absolute():
|
||
# 绝对路径：必须解析到安全目录内
|
||
try:
|
||
# 解析并验证路径在安全目录内
|
||
resolved_path = stats_path.resolve()
|
||
safe_dir_resolved = safe_dir.resolve()
|
||
|
||
# 检查路径是否以安全目录开头
|
||
if not str(resolved_path).startswith(str(safe_dir_resolved)):
|
||
raise ValueError(
|
||
f"安全错误：stats_file 路径必须在 ~/.dev-workflow/ 目录内，"
|
||
f"提供的路径：{stats_file}"
|
||
)
|
||
|
||
self.stats_file = resolved_path
|
||
except (OSError, RuntimeError) as e:
|
||
raise ValueError(f"无效的 stats_file 路径：{stats_file}") from e
|
||
else:
|
||
# 相对路径：检查是否包含路径遍历
|
||
if '..' in stats_path.parts:
|
||
raise ValueError(
|
||
f"安全错误：stats_file 路径不能包含 '..'，"
|
||
f"提供的路径：{stats_file}"
|
||
)
|
||
|
||
# 将相对路径解析到安全目录内
|
||
self.stats_file = safe_dir / stats_path
|
||
|
||
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())
|