starmen1265
cee3598d37
- 新增 stats.py 模块：record_event, get_stats, format_stats_report - dev_workflow.py 添加 stats 和 record-event 命令 - 支持事件记录、统计报告、数据清理 - 安全特性：敏感数据过滤、文件锁、权限控制 - Cross-Debate 和 Cross-Review 记录文件
360 lines
10 KiB
Python
360 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
stats.py - Cross 模式统计报告模块
|
|
|
|
用法:
|
|
from stats import record_event, get_stats, format_stats_report
|
|
|
|
# 记录事件
|
|
record_event("debate", "cross", {"agents": 3})
|
|
|
|
# 获取统计
|
|
stats = get_stats()
|
|
print(format_stats_report(stats))
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import fcntl
|
|
import logging
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
# 配置日志
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# 数据存储路径
|
|
DATA_DIR = Path(__file__).parent / "data"
|
|
STATS_FILE = DATA_DIR / "cross_stats.json"
|
|
LOCK_FILE = DATA_DIR / "cross_stats.lock"
|
|
MAX_EVENTS = 10000 # 最大事件数
|
|
|
|
# 白名单
|
|
ALLOWED_EVENT_TYPES = {"debate", "review", "develop", "fix", "other"}
|
|
ALLOWED_MODES = {"cross", "normal"}
|
|
|
|
# 敏感字段
|
|
SENSITIVE_KEYS = {"password", "token", "secret", "key", "credential", "api_key"}
|
|
|
|
|
|
def _ensure_data_dir():
|
|
"""确保数据目录存在"""
|
|
DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
def _filter_sensitive_data(data: dict) -> dict:
|
|
"""过滤敏感数据"""
|
|
if not isinstance(data, dict):
|
|
return {}
|
|
|
|
filtered = {}
|
|
for key, value in data.items():
|
|
key_lower = key.lower()
|
|
# 检查是否是敏感字段
|
|
if any(s in key_lower for s in SENSITIVE_KEYS):
|
|
filtered[key] = "[REDACTED]"
|
|
elif isinstance(value, dict):
|
|
filtered[key] = _filter_sensitive_data(value)
|
|
else:
|
|
filtered[key] = value
|
|
return filtered
|
|
|
|
|
|
def _load_stats_data() -> dict:
|
|
"""加载统计数据（带文件锁）"""
|
|
_ensure_data_dir()
|
|
|
|
if not STATS_FILE.exists():
|
|
return {"events": [], "created_at": datetime.now().isoformat()}
|
|
|
|
try:
|
|
with open(STATS_FILE, "r", encoding="utf-8") as f:
|
|
# 获取共享锁（读锁）
|
|
fcntl.flock(f.fileno(), fcntl.LOCK_SH)
|
|
try:
|
|
data = json.load(f)
|
|
if not isinstance(data, dict) or "events" not in data:
|
|
logger.warning("统计数据格式无效，返回空数据")
|
|
return {"events": [], "created_at": datetime.now().isoformat()}
|
|
return data
|
|
finally:
|
|
fcntl.flock(f.fileno(), fcntl.LOCK_UN)
|
|
except json.JSONDecodeError as e:
|
|
# 备份损坏文件
|
|
backup_file = STATS_FILE.with_suffix(f".json.bak.{datetime.now().strftime('%Y%m%d%H%M%S')}")
|
|
STATS_FILE.rename(backup_file)
|
|
logger.error(f"统计数据损坏，已备份到 {backup_file}，错误: {e}")
|
|
return {"events": [], "created_at": datetime.now().isoformat()}
|
|
except Exception as e:
|
|
logger.error(f"加载统计数据失败: {e}")
|
|
return {"events": [], "created_at": datetime.now().isoformat()}
|
|
|
|
|
|
def _save_stats_data(data: dict):
|
|
"""保存统计数据（带文件锁和原子写入）"""
|
|
_ensure_data_dir()
|
|
|
|
# 原子写入：先写临时文件，再替换
|
|
temp_file = STATS_FILE.with_suffix(".tmp")
|
|
|
|
try:
|
|
with open(temp_file, "w", encoding="utf-8") as f:
|
|
# 获取排他锁（写锁）
|
|
fcntl.flock(f.fileno(), fcntl.LOCK_EX)
|
|
try:
|
|
json.dump(data, f, ensure_ascii=False, indent=2)
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
finally:
|
|
fcntl.flock(f.fileno(), fcntl.LOCK_UN)
|
|
|
|
# 设置文件权限
|
|
os.chmod(temp_file, 0o600)
|
|
|
|
# 原子替换
|
|
temp_file.replace(STATS_FILE)
|
|
os.chmod(STATS_FILE, 0o600)
|
|
|
|
except Exception as e:
|
|
logger.error(f"保存统计数据失败: {e}")
|
|
if temp_file.exists():
|
|
temp_file.unlink()
|
|
raise
|
|
|
|
|
|
def record_event(
|
|
event_type: str,
|
|
mode: str,
|
|
metadata: Optional[dict] = None,
|
|
status: str = "success"
|
|
) -> bool:
|
|
"""
|
|
记录 Cross 模式事件
|
|
|
|
Args:
|
|
event_type: 事件类型 (debate, review, develop, fix, other)
|
|
mode: 模式 (cross, normal)
|
|
metadata: 事件元数据
|
|
status: 状态 (success, failed)
|
|
|
|
Returns:
|
|
是否记录成功
|
|
"""
|
|
# 参数校验
|
|
if event_type not in ALLOWED_EVENT_TYPES:
|
|
logger.warning(f"无效的 event_type: {event_type}，使用 'other'")
|
|
event_type = "other"
|
|
|
|
if mode not in ALLOWED_MODES:
|
|
logger.warning(f"无效的 mode: {mode}，使用 'normal'")
|
|
mode = "normal"
|
|
|
|
# metadata 类型校验
|
|
if metadata is not None and not isinstance(metadata, dict):
|
|
logger.warning(f"metadata 必须是 dict 类型，收到: {type(metadata)}")
|
|
metadata = {}
|
|
|
|
# 过滤敏感数据
|
|
if metadata:
|
|
metadata = _filter_sensitive_data(metadata)
|
|
|
|
# 创建事件
|
|
event = {
|
|
"timestamp": datetime.now().isoformat(),
|
|
"event_type": event_type,
|
|
"mode": mode,
|
|
"status": status,
|
|
"metadata": metadata or {}
|
|
}
|
|
|
|
try:
|
|
# 加载现有数据
|
|
data = _load_stats_data()
|
|
|
|
# 添加事件
|
|
data["events"].append(event)
|
|
data["last_updated"] = datetime.now().isoformat()
|
|
|
|
# 限制事件数量
|
|
if len(data["events"]) > MAX_EVENTS:
|
|
data["events"] = data["events"][-MAX_EVENTS:]
|
|
logger.info(f"事件数量超过 {MAX_EVENTS}，已裁剪")
|
|
|
|
# 保存
|
|
_save_stats_data(data)
|
|
logger.info(f"事件已记录: {event_type}/{mode}/{status}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"记录事件失败: {e}")
|
|
return False
|
|
|
|
|
|
def get_stats() -> dict:
|
|
"""
|
|
获取统计报告
|
|
|
|
Returns:
|
|
统计数据字典
|
|
"""
|
|
data = _load_stats_data()
|
|
events = data.get("events", [])
|
|
|
|
if not events:
|
|
return {
|
|
"total_events": 0,
|
|
"message": "暂无统计数据"
|
|
}
|
|
|
|
# 统计
|
|
stats = {
|
|
"total_events": len(events),
|
|
"first_event": events[0].get("timestamp"),
|
|
"last_event": events[-1].get("timestamp"),
|
|
"by_event_type": {},
|
|
"by_mode": {},
|
|
"by_status": {},
|
|
"by_date": {}
|
|
}
|
|
|
|
for event in events:
|
|
# 按事件类型
|
|
event_type = event.get("event_type", "unknown")
|
|
stats["by_event_type"][event_type] = stats["by_event_type"].get(event_type, 0) + 1
|
|
|
|
# 按模式
|
|
mode = event.get("mode", "unknown")
|
|
stats["by_mode"][mode] = stats["by_mode"].get(mode, 0) + 1
|
|
|
|
# 按状态
|
|
status = event.get("status", "unknown")
|
|
stats["by_status"][status] = stats["by_status"].get(status, 0) + 1
|
|
|
|
# 按日期
|
|
try:
|
|
date = event.get("timestamp", "")[:10]
|
|
if date:
|
|
stats["by_date"][date] = stats["by_date"].get(date, 0) + 1
|
|
except:
|
|
pass
|
|
|
|
# 计算成功率
|
|
total = stats["total_events"]
|
|
success = stats["by_status"].get("success", 0)
|
|
stats["success_rate"] = f"{success / total * 100:.1f}%" if total > 0 else "0%"
|
|
|
|
return stats
|
|
|
|
|
|
def format_stats_report(stats: dict) -> str:
|
|
"""
|
|
格式化统计报告
|
|
|
|
Args:
|
|
stats: 统计数据
|
|
|
|
Returns:
|
|
格式化的报告字符串
|
|
"""
|
|
lines = [
|
|
"=" * 60,
|
|
"Cross 模式统计报告",
|
|
"=" * 60,
|
|
""
|
|
]
|
|
|
|
if stats.get("total_events") == 0:
|
|
lines.append("暂无统计数据")
|
|
lines.append("")
|
|
lines.append("=" * 60)
|
|
return "\n".join(lines)
|
|
|
|
lines.append(f"总事件数: {stats['total_events']}")
|
|
lines.append(f"成功率: {stats.get('success_rate', 'N/A')}")
|
|
lines.append("")
|
|
lines.append(f"首次事件: {stats.get('first_event', 'N/A')}")
|
|
lines.append(f"最后事件: {stats.get('last_event', 'N/A')}")
|
|
lines.append("")
|
|
|
|
# 按事件类型
|
|
if stats.get("by_event_type"):
|
|
lines.append("按事件类型统计:")
|
|
for event_type, count in sorted(stats["by_event_type"].items(), key=lambda x: -x[1]):
|
|
lines.append(f" - {event_type}: {count}")
|
|
lines.append("")
|
|
|
|
# 按模式
|
|
if stats.get("by_mode"):
|
|
lines.append("按模式统计:")
|
|
for mode, count in sorted(stats["by_mode"].items(), key=lambda x: -x[1]):
|
|
lines.append(f" - {mode}: {count}")
|
|
lines.append("")
|
|
|
|
# 按日期（最近7天）
|
|
if stats.get("by_date"):
|
|
lines.append("按日期统计（最近7天）:")
|
|
sorted_dates = sorted(stats["by_date"].items(), key=lambda x: x[0], 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: 必须为 True 才能清空
|
|
|
|
Returns:
|
|
是否清空成功
|
|
"""
|
|
# 身份验证：检查 ADMIN_TOKEN
|
|
admin_token = os.environ.get("ADMIN_TOKEN", "")
|
|
if not admin_token:
|
|
logger.error("清空统计数据需要设置 ADMIN_TOKEN 环境变量")
|
|
return False
|
|
|
|
if not confirm:
|
|
logger.warning("清空统计数据需要 confirm=True")
|
|
return False
|
|
|
|
try:
|
|
_ensure_data_dir()
|
|
|
|
# 备份现有数据
|
|
if STATS_FILE.exists():
|
|
backup_file = STATS_FILE.with_suffix(f".json.cleared.{datetime.now().strftime('%Y%m%d%H%M%S')}")
|
|
STATS_FILE.rename(backup_file)
|
|
logger.info(f"统计数据已备份到 {backup_file}")
|
|
|
|
# 创建空数据
|
|
_save_stats_data({"events": [], "created_at": datetime.now().isoformat()})
|
|
logger.info("统计数据已清空")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"清空统计数据失败: {e}")
|
|
return False
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# 测试
|
|
print("测试 stats 模块...")
|
|
|
|
# 记录几个测试事件
|
|
record_event("debate", "cross", {"agents": 3, "result": "consensus"})
|
|
record_event("review", "cross", {"issues_found": 5})
|
|
record_event("develop", "cross", {"files_changed": 3})
|
|
|
|
# 获取统计
|
|
stats = get_stats()
|
|
print(format_stats_report(stats))
|