371 lines
11 KiB
Python
371 lines
11 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Cross 模式统计报告模块
|
||
|
||
提供事件记录和统计报告生成功能
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import tempfile
|
||
import fcntl
|
||
import logging
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Dict, List, Any, Optional
|
||
from collections import defaultdict
|
||
|
||
# 配置日志
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# 统计数据存储路径
|
||
STATS_DIR = Path(__file__).parent / "data"
|
||
STATS_FILE = STATS_DIR / "stats_data.json"
|
||
|
||
# 锁文件路径
|
||
LOCK_FILE = STATS_DIR / "stats_data.lock"
|
||
|
||
# 最大事件数量限制
|
||
MAX_EVENTS = 10000
|
||
|
||
# 敏感字段列表（不记录这些字段）
|
||
SENSITIVE_FIELDS = {"password", "token", "secret", "key", "credential", "auth"}
|
||
|
||
|
||
def _ensure_stats_dir():
|
||
"""确保统计目录存在"""
|
||
STATS_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
|
||
def _acquire_lock(lock_file_path: Path):
|
||
"""获取文件锁
|
||
|
||
Raises:
|
||
OSError: 无法创建或锁定文件时
|
||
"""
|
||
try:
|
||
lock_file_path.parent.mkdir(parents=True, exist_ok=True)
|
||
# 使用 os.open 创建文件并设置权限
|
||
fd = os.open(lock_file_path, os.O_WRONLY | os.O_CREAT, 0o600)
|
||
lock_fd = os.fdopen(fd, 'w')
|
||
fcntl.flock(lock_fd.fileno(), fcntl.LOCK_EX)
|
||
return lock_fd
|
||
except (OSError, IOError) as e:
|
||
logger.error(f"获取文件锁失败: {e}")
|
||
raise
|
||
|
||
|
||
def _release_lock(lock_fd):
|
||
"""释放文件锁"""
|
||
if lock_fd is None:
|
||
return
|
||
try:
|
||
fcntl.flock(lock_fd.fileno(), fcntl.LOCK_UN)
|
||
lock_fd.close()
|
||
except (OSError, IOError) as e:
|
||
logger.warning(f"释放文件锁时发生错误: {e}")
|
||
try:
|
||
lock_fd.close()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _load_stats_data() -> Dict[str, Any]:
|
||
"""加载统计数据（带锁保护）
|
||
|
||
Returns:
|
||
Dict: 统计数据，加载失败时返回空结构
|
||
"""
|
||
if not STATS_FILE.exists():
|
||
return {"events": [], "metadata": {"created_at": datetime.now().isoformat()}}
|
||
|
||
lock_fd = None
|
||
try:
|
||
lock_fd = _acquire_lock(LOCK_FILE)
|
||
with open(STATS_FILE, 'r', encoding='utf-8') as f:
|
||
return json.load(f)
|
||
except json.JSONDecodeError as e:
|
||
logger.warning(f"JSON 解析失败，将备份损坏文件: {e}")
|
||
# 备份损坏的文件
|
||
backup_path = STATS_FILE.with_suffix(f'.corrupted.{datetime.now().strftime("%Y%m%d_%H%M%S")}.bak')
|
||
try:
|
||
STATS_FILE.rename(backup_path)
|
||
logger.info(f"已备份损坏文件到: {backup_path}")
|
||
except OSError as backup_error:
|
||
logger.error(f"备份损坏文件失败: {backup_error}")
|
||
return {"events": [], "metadata": {"created_at": datetime.now().isoformat()}}
|
||
except (OSError, IOError) as e:
|
||
logger.error(f"读取统计文件失败: {e}")
|
||
return {"events": [], "metadata": {"created_at": datetime.now().isoformat()}}
|
||
except Exception as e:
|
||
logger.error(f"加载统计数据时发生未知错误: {e}")
|
||
return {"events": [], "metadata": {"created_at": datetime.now().isoformat()}}
|
||
finally:
|
||
_release_lock(lock_fd)
|
||
|
||
|
||
def _atomic_write(data: Dict[str, Any]):
|
||
"""原子写入统计数据（带锁保护）
|
||
|
||
Raises:
|
||
OSError: 文件操作失败时
|
||
TypeError: 数据无法序列化为 JSON 时
|
||
"""
|
||
_ensure_stats_dir()
|
||
|
||
lock_fd = None
|
||
temp_path = None
|
||
try:
|
||
lock_fd = _acquire_lock(LOCK_FILE)
|
||
|
||
# 写入临时文件（使用 os.open 设置权限）
|
||
fd = os.open(
|
||
tempfile.mktemp(suffix='.tmp', dir=STATS_DIR),
|
||
os.O_WRONLY | os.O_CREAT | os.O_EXCL,
|
||
0o600
|
||
)
|
||
temp_path = f"/proc/self/fd/{fd}"
|
||
|
||
with os.fdopen(fd, 'w', encoding='utf-8') as temp_file:
|
||
json.dump(data, temp_file, ensure_ascii=False, indent=2)
|
||
temp_path = temp_file.name
|
||
|
||
# 原子重命名
|
||
os.replace(temp_path, STATS_FILE)
|
||
# 确保最终文件权限正确
|
||
os.chmod(STATS_FILE, 0o600)
|
||
temp_path = None # 防止在 finally 中再次删除
|
||
except (OSError, IOError) as e:
|
||
logger.error(f"写入统计文件失败: {e}")
|
||
raise
|
||
except (TypeError, ValueError) as e:
|
||
logger.error(f"JSON 序列化失败: {e}")
|
||
raise
|
||
finally:
|
||
_release_lock(lock_fd)
|
||
# 清理临时文件
|
||
if temp_path and os.path.exists(temp_path):
|
||
try:
|
||
os.unlink(temp_path)
|
||
except OSError as e:
|
||
logger.warning(f"清理临时文件失败: {e}")
|
||
|
||
|
||
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)
|
||
|
||
# 事件数量限制：超过 MAX_EVENTS 时移除最旧的事件
|
||
if len(data["events"]) > MAX_EVENTS:
|
||
data["events"] = data["events"][-MAX_EVENTS:]
|
||
|
||
data["metadata"]["last_updated"] = datetime.now().isoformat()
|
||
|
||
# 原子写入
|
||
_atomic_write(data)
|
||
|
||
return True
|
||
except TypeError:
|
||
raise # 重新抛出类型错误
|
||
except Exception as e:
|
||
logger.error(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].get("timestamp")
|
||
stats["last_event"] = events[-1].get("timestamp")
|
||
|
||
for event in events:
|
||
event_type = event.get("event_type", "unknown")
|
||
event_mode = event.get("mode", "unknown")
|
||
stats["events_by_type"][event_type] += 1
|
||
stats["events_by_mode"][event_mode] += 1
|
||
|
||
# 按日期统计
|
||
timestamp = event.get("timestamp", "")
|
||
date = timestamp[:10] if timestamp else "unknown" # 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
|
||
|
||
# 安全检查：验证管理员令牌
|
||
admin_token = os.getenv("ADMIN_TOKEN")
|
||
if not admin_token:
|
||
logger.error("清空统计数据需要设置 ADMIN_TOKEN 环境变量")
|
||
return False
|
||
|
||
try:
|
||
if STATS_FILE.exists():
|
||
STATS_FILE.unlink()
|
||
return True
|
||
except Exception as e:
|
||
logger.error(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))
|