281 lines
8.9 KiB
Python
281 lines
8.9 KiB
Python
"""
|
|
Cross 模式统计报告功能
|
|
|
|
提供事件记录和统计报告生成功能，支持 Cross 和 Normal 两种模式。
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import fcntl
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
# 统计数据文件路径
|
|
STATS_FILE = Path(__file__).parent / ".stats_data.json"
|
|
|
|
# 敏感字段列表（用于过滤）
|
|
SENSITIVE_FIELDS = {"password", "token", "secret", "api_key", "apikey", "credential"}
|
|
|
|
|
|
def _filter_sensitive_data(data: dict) -> dict:
|
|
"""过滤敏感数据
|
|
|
|
递归过滤字典中的敏感字段，将值替换为 [REDACTED]
|
|
"""
|
|
if not isinstance(data, dict):
|
|
return data
|
|
|
|
filtered = {}
|
|
for key, value in data.items():
|
|
key_lower = key.lower().replace("-", "_").replace(" ", "_")
|
|
|
|
# 检查是否为敏感字段
|
|
is_sensitive = any(s in key_lower for s in SENSITIVE_FIELDS)
|
|
|
|
if is_sensitive:
|
|
filtered[key] = "[REDACTED]"
|
|
elif isinstance(value, dict):
|
|
filtered[key] = _filter_sensitive_data(value)
|
|
else:
|
|
filtered[key] = value
|
|
|
|
return filtered
|
|
|
|
|
|
def _ensure_stats_file():
|
|
"""确保统计数据文件存在，并设置正确权限"""
|
|
if not STATS_FILE.exists():
|
|
STATS_FILE.write_text(json.dumps({"events": []}, ensure_ascii=False, indent=2))
|
|
os.chmod(STATS_FILE, 0o600)
|
|
|
|
|
|
def _load_stats() -> dict:
|
|
"""加载统计数据（带文件锁）"""
|
|
_ensure_stats_file()
|
|
|
|
with open(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 _save_stats(data: dict):
|
|
"""保存统计数据（带文件锁）"""
|
|
_ensure_stats_file()
|
|
|
|
with open(STATS_FILE, "w") as f:
|
|
fcntl.flock(f.fileno(), fcntl.LOCK_EX) # 排他锁
|
|
try:
|
|
json.dump(data, f, ensure_ascii=False, indent=2)
|
|
finally:
|
|
fcntl.flock(f.fileno(), fcntl.LOCK_UN)
|
|
|
|
# 确保文件权限为 600
|
|
os.chmod(STATS_FILE, 0o600)
|
|
|
|
|
|
def record_event(
|
|
event_type: str,
|
|
mode: str,
|
|
metadata: Optional[dict] = None,
|
|
status: str = "success"
|
|
) -> bool:
|
|
"""记录一个事件
|
|
|
|
Args:
|
|
event_type: 事件类型 (debate, review, audit)
|
|
mode: 运行模式 (cross, normal)
|
|
metadata: 事件元数据（可选）
|
|
status: 事件状态 (success, failed)
|
|
|
|
Returns:
|
|
bool: 记录是否成功
|
|
"""
|
|
# 验证参数
|
|
valid_types = {"debate", "review", "audit"}
|
|
valid_modes = {"cross", "normal"}
|
|
valid_status = {"success", "failed"}
|
|
|
|
if event_type not in valid_types:
|
|
print(f"错误: 无效的事件类型 '{event_type}'，有效值: {valid_types}")
|
|
return False
|
|
|
|
if mode not in valid_modes:
|
|
print(f"错误: 无效的模式 '{mode}'，有效值: {valid_modes}")
|
|
return False
|
|
|
|
if status not in valid_status:
|
|
print(f"错误: 无效的状态 '{status}'，有效值: {valid_status}")
|
|
return False
|
|
|
|
# 过滤敏感数据
|
|
if metadata:
|
|
metadata = _filter_sensitive_data(metadata)
|
|
else:
|
|
metadata = {}
|
|
|
|
# 创建事件记录
|
|
event = {
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
"event_type": event_type,
|
|
"mode": mode,
|
|
"status": status,
|
|
"metadata": metadata
|
|
}
|
|
|
|
try:
|
|
data = _load_stats()
|
|
data["events"].append(event)
|
|
_save_stats(data)
|
|
return True
|
|
except Exception as e:
|
|
print(f"记录事件失败: {e}")
|
|
return False
|
|
|
|
|
|
def get_stats() -> dict:
|
|
"""获取统计汇总数据
|
|
|
|
Returns:
|
|
dict: 统计汇总数据
|
|
"""
|
|
data = _load_stats()
|
|
events = data.get("events", [])
|
|
|
|
# 初始化统计结构
|
|
stats = {
|
|
"total_events": len(events),
|
|
"by_type": {
|
|
"debate": {"total": 0, "success": 0, "failed": 0},
|
|
"review": {"total": 0, "success": 0, "failed": 0},
|
|
"audit": {"total": 0, "success": 0, "failed": 0}
|
|
},
|
|
"by_mode": {
|
|
"cross": {"total": 0, "success": 0, "failed": 0},
|
|
"normal": {"total": 0, "success": 0, "failed": 0}
|
|
},
|
|
"cross_mode_breakdown": {
|
|
"debate": {"total": 0, "success": 0, "failed": 0},
|
|
"review": {"total": 0, "success": 0, "failed": 0},
|
|
"audit": {"total": 0, "success": 0, "failed": 0}
|
|
},
|
|
"normal_mode_breakdown": {
|
|
"debate": {"total": 0, "success": 0, "failed": 0},
|
|
"review": {"total": 0, "success": 0, "failed": 0},
|
|
"audit": {"total": 0, "success": 0, "failed": 0}
|
|
},
|
|
"success_rate": 0.0,
|
|
"first_event": None,
|
|
"last_event": None
|
|
}
|
|
|
|
if not events:
|
|
return stats
|
|
|
|
# 统计各类数据
|
|
for event in events:
|
|
event_type = event["event_type"]
|
|
mode = event["mode"]
|
|
status = event["status"]
|
|
|
|
# 按类型统计
|
|
stats["by_type"][event_type]["total"] += 1
|
|
stats["by_type"][event_type][status] += 1
|
|
|
|
# 按模式统计
|
|
stats["by_mode"][mode]["total"] += 1
|
|
stats["by_mode"][mode][status] += 1
|
|
|
|
# 按模式+类型交叉统计
|
|
breakdown_key = f"{mode}_mode_breakdown"
|
|
stats[breakdown_key][event_type]["total"] += 1
|
|
stats[breakdown_key][event_type][status] += 1
|
|
|
|
# 计算成功率
|
|
total = stats["total_events"]
|
|
success_count = sum(t["success"] for t in stats["by_type"].values())
|
|
stats["success_rate"] = round(success_count / total * 100, 2) if total > 0 else 0.0
|
|
|
|
# 时间范围
|
|
stats["first_event"] = events[0]["timestamp"]
|
|
stats["last_event"] = events[-1]["timestamp"]
|
|
|
|
return stats
|
|
|
|
|
|
def format_stats_report(stats: dict) -> str:
|
|
"""格式化统计报告为可读字符串
|
|
|
|
Args:
|
|
stats: get_stats() 返回的统计数据
|
|
|
|
Returns:
|
|
str: 格式化的统计报告
|
|
"""
|
|
lines = []
|
|
lines.append("=" * 50)
|
|
lines.append(" Cross 模式统计报告")
|
|
lines.append("=" * 50)
|
|
|
|
# 总览
|
|
lines.append(f"\n📊 总览")
|
|
lines.append(f" 总事件数: {stats['total_events']}")
|
|
lines.append(f" 成功率: {stats['success_rate']}%")
|
|
if stats['first_event']:
|
|
lines.append(f" 时间范围: {stats['first_event']} ~ {stats['last_event']}")
|
|
|
|
# 按模式统计
|
|
lines.append(f"\n📈 按模式统计")
|
|
for mode, data in stats["by_mode"].items():
|
|
mode_name = "Cross 模式" if mode == "cross" else "Normal 模式"
|
|
rate = round(data["success"] / data["total"] * 100, 2) if data["total"] > 0 else 0
|
|
lines.append(f" {mode_name}:")
|
|
lines.append(f" 总数: {data['total']} | 成功: {data['success']} | 失败: {data['failed']} | 成功率: {rate}%")
|
|
|
|
# 按事件类型统计
|
|
lines.append(f"\n📋 按事件类型统计")
|
|
for event_type, data in stats["by_type"].items():
|
|
type_name = {"debate": "辩论", "review": "评审", "audit": "审计"}.get(event_type, event_type)
|
|
rate = round(data["success"] / data["total"] * 100, 2) if data["total"] > 0 else 0
|
|
lines.append(f" {type_name}:")
|
|
lines.append(f" 总数: {data['total']} | 成功: {data['success']} | 失败: {data['failed']} | 成功率: {rate}%")
|
|
|
|
# Cross 模式详细分解
|
|
lines.append(f"\n🔍 Cross 模式详细分解")
|
|
for event_type, data in stats["cross_mode_breakdown"].items():
|
|
type_name = {"debate": "辩论", "review": "评审", "audit": "审计"}.get(event_type, event_type)
|
|
rate = round(data["success"] / data["total"] * 100, 2) if data["total"] > 0 else 0
|
|
lines.append(f" {type_name}:")
|
|
lines.append(f" 总数: {data['total']} | 成功: {data['success']} | 失败: {data['failed']} | 成功率: {rate}%")
|
|
|
|
# Normal 模式详细分解
|
|
lines.append(f"\n🔍 Normal 模式详细分解")
|
|
for event_type, data in stats["normal_mode_breakdown"].items():
|
|
type_name = {"debate": "辩论", "review": "评审", "audit": "审计"}.get(event_type, event_type)
|
|
rate = round(data["success"] / data["total"] * 100, 2) if data["total"] > 0 else 0
|
|
lines.append(f" {type_name}:")
|
|
lines.append(f" 总数: {data['total']} | 成功: {data['success']} | 失败: {data['failed']} | 成功率: {rate}%")
|
|
|
|
lines.append("\n" + "=" * 50)
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# 测试代码
|
|
print("测试 stats.py 模块...")
|
|
|
|
# 记录一些测试事件
|
|
record_event("debate", "cross", {"topic": "AI 安全"}, "success")
|
|
record_event("review", "normal", {"pr_id": 123}, "success")
|
|
record_event("audit", "cross", {"password": "secret123"}, "failed") # 敏感数据会被过滤
|
|
|
|
# 获取并打印统计
|
|
stats = get_stats()
|
|
print(format_stats_report(stats))
|