feat: 添加 Cross 模式统计报告功能

- 添加 stats.py 统计模块 - 添加 dev_workflow.py CLI 工具 - 实现 record_event() 记录事件 - 实现 get_stats() 生成报告 - 支持敏感数据自动过滤 - 使用原子写入保证数据安全 - 添加 README 文档
2026-03-09 14:44:04 +08:00
parent 512f7fdf41
commit 3efe98e17d
4 changed files with 562 additions and 0 deletions
@@ -0,0 +1,27 @@
+# 数据文件
+data/
+
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+.Python
+*.egg-info/
+dist/
+build/
+
+# 测试
+.pytest_cache/
+.coverage
+htmlcov/
+
+# IDE
+.vscode/
+.idea/
+*.swp
+*.swo
+
+# 临时文件
+*.tmp
+*.bak
@@ -0,0 +1,146 @@
+# Dev Workflow - 开发工作流工具
+
+Cross 模式统计报告功能
+
+## 功能特性
+
+- ✅ 事件记录和统计
+- ✅ 多模式支持（cross, normal 等）
+- ✅ 敏感数据自动过滤
+- ✅ 原子写入保证数据安全
+- ✅ 格式化报告输出
+
+## 使用方法
+
+### 查看统计报告
+
+```bash
+python3 dev-workflow/dev_workflow.py stats
+```
+
+### 筛选特定模式
+
+```bash
+python3 dev-workflow/dev_workflow.py stats --mode cross
+```
+
+### 记录事件（用于集成）
+
+```bash
+python3 dev-workflow/dev_workflow.py record "event_type" --mode cross --description "事件描述"
+```
+
+### 清空统计数据
+
+```bash
+python3 dev-workflow/dev_workflow.py clear --confirm
+```
+
+## API 使用
+
+### Python 代码中记录事件
+
+```python
+from stats import record_event, get_stats, format_stats_report
+
+# 记录事件
+record_event(
+    "task_completed",
+    mode="cross",
+    metadata={
+        "task_name": "开发功能",
+        "duration": 120
+    }
+)
+
+# 获取统计数据
+stats = get_stats(mode="cross")
+print(format_stats_report(stats))
+```
+
+## 安全特性
+
+### 敏感数据过滤
+
+以下字段会被自动过滤（替换为 `[REDACTED]`）：
+- password
+- token
+- secret
+- key
+- credential
+- auth
+
+示例：
+```python
+record_event("login", metadata={
+    "username": "user123",      # 正常记录
+    "password": "secret123"     # 记录为 [REDACTED]
+})
+```
+
+### 原子写入
+
+所有数据写入都使用原子操作（临时文件 + 重命名），确保：
+- 并发写入不会损坏数据
+- 系统崩溃不会丢失数据
+- 临时文件自动清理
+
+## 数据存储
+
+- 位置：`dev-workflow/data/stats_data.json`
+- 格式：JSON
+- 结构：
+  ```json
+  {
+    "events": [
+      {
+        "event_type": "string",
+        "mode": "string",
+        "timestamp": "ISO8601",
+        "metadata": {}
+      }
+    ],
+    "metadata": {
+      "created_at": "ISO8601",
+      "last_updated": "ISO8601"
+    }
+  }
+  ```
+
+## 统计报告内容
+
+报告包含以下信息：
+- 总事件数
+- 时间范围（首次和最后事件）
+- 按事件类型统计
+- 按模式统计
+- 按日期统计（最近7天）
+
+## 开发说明
+
+### 文件结构
+
+```
+dev-workflow/
+├── dev_workflow.py    # CLI 工具
+├── stats.py           # 统计模块
+├── README.md          # 本文档
+└── data/
+    └── stats_data.json  # 数据存储
+```
+
+### 扩展事件类型
+
+可以在代码中定义新的事件类型，例如：
+- `task_started` - 任务开始
+- `task_completed` - 任务完成
+- `code_committed` - 代码提交
+- `error_occurred` - 错误发生
+- `stats_viewed` - 查看统计
+
+## 测试
+
+运行测试：
+```bash
+python3 dev-workflow/stats.py
+```
@@ -0,0 +1,136 @@
+#!/usr/bin/env python3
+"""
+Dev Workflow - 开发工作流工具
+
+提供多种工作模式和统计功能
+"""
+
+import sys
+import argparse
+from pathlib import Path
+
+# 添加当前目录到路径
+sys.path.insert(0, str(Path(__file__).parent))
+
+from stats import record_event, get_stats, format_stats_report, clear_stats
+
+
+def cmd_stats(args):
+    """stats 命令处理函数"""
+    # 记录查看统计的事件
+    record_event(
+        "stats_viewed",
+        mode="cross",
+        metadata={"filter_mode": args.mode}
+    )
+    
+    # 获取统计数据
+    stats = get_stats(mode=args.mode)
+    
+    # 输出报告
+    report = format_stats_report(stats)
+    print(report)
+    
+    return 0
+
+
+def cmd_record(args):
+    """record 命令处理函数 - 用于测试"""
+    success = record_event(
+        args.event_type,
+        mode=args.mode,
+        metadata={"description": args.description} if args.description else None
+    )
+    
+    if success:
+        print(f"✓ 事件已记录: {args.event_type}")
+        return 0
+    else:
+        print("✗ 记录事件失败")
+        return 1
+
+
+def cmd_clear(args):
+    """clear 命令处理函数"""
+    if not args.confirm:
+        print("警告: 此操作将清空所有统计数据!")
+        print("使用 --confirm 参数确认清空")
+        return 1
+    
+    success = clear_stats(confirm=True)
+    if success:
+        print("✓ 统计数据已清空")
+        return 0
+    else:
+        print("✗ 清空失败")
+        return 1
+
+
+def main():
+    """主函数"""
+    parser = argparse.ArgumentParser(
+        description="Dev Workflow - 开发工作流工具",
+        formatter_class=argparse.RawDescriptionHelpFormatter
+    )
+    
+    subparsers = parser.add_subparsers(dest="command", help="可用命令")
+    
+    # stats 命令
+    stats_parser = subparsers.add_parser("stats", help="显示统计报告")
+    stats_parser.add_argument(
+        "--mode", "-m",
+        help="筛选特定模式（如 cross, normal）"
+    )
+    stats_parser.set_defaults(func=cmd_stats)
+    
+    # record 命令（用于测试）
+    record_parser = subparsers.add_parser("record", help="记录事件（测试用）")
+    record_parser.add_argument(
+        "event_type",
+        help="事件类型"
+    )
+    record_parser.add_argument(
+        "--mode", "-m",
+        default="cross",
+        help="工作模式（默认: cross）"
+    )
+    record_parser.add_argument(
+        "--description", "-d",
+        help="事件描述"
+    )
+    record_parser.set_defaults(func=cmd_record)
+    
+    # clear 命令
+    clear_parser = subparsers.add_parser("clear", help="清空统计数据")
+    clear_parser.add_argument(
+        "--confirm",
+        action="store_true",
+        help="确认清空"
+    )
+    clear_parser.set_defaults(func=cmd_clear)
+    
+    # 解析参数
+    args = parser.parse_args()
+    
+    # 如果没有指定命令，默认显示 stats
+    if not args.command:
+        args.command = "stats"
+        args.func = cmd_stats
+        args.mode = None
+    
+    # 执行命令
+    try:
+        return args.func(args)
+    except AttributeError:
+        parser.print_help()
+        return 1
+    except KeyboardInterrupt:
+        print("\n操作已取消")
+        return 130
+    except Exception as e:
+        print(f"错误: {e}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())
@@ -0,0 +1,253 @@
+#!/usr/bin/env python3
+"""
+Cross 模式统计报告模块
+
+提供事件记录和统计报告生成功能
+"""
+
+import json
+import os
+import tempfile
+from datetime import datetime
+from pathlib import Path
+from typing import Dict, List, Any, Optional
+from collections import defaultdict
+
+# 统计数据存储路径
+STATS_DIR = Path(__file__).parent / "data"
+STATS_FILE = STATS_DIR / "stats_data.json"
+
+# 敏感字段列表（不记录这些字段）
+SENSITIVE_FIELDS = {"password", "token", "secret", "key", "credential", "auth"}
+
+
+def _ensure_stats_dir():
+    """确保统计目录存在"""
+    STATS_DIR.mkdir(parents=True, exist_ok=True)
+
+
+def _load_stats_data() -> Dict[str, Any]:
+    """加载统计数据"""
+    if not STATS_FILE.exists():
+        return {"events": [], "metadata": {"created_at": datetime.now().isoformat()}}
+    
+    try:
+        with open(STATS_FILE, 'r', encoding='utf-8') as f:
+            return json.load(f)
+    except (json.JSONDecodeError, IOError):
+        return {"events": [], "metadata": {"created_at": datetime.now().isoformat()}}
+
+
+def _atomic_write(data: Dict[str, Any]):
+    """原子写入统计数据"""
+    _ensure_stats_dir()
+    
+    # 写入临时文件
+    fd, temp_path = tempfile.mkstemp(dir=STATS_DIR, suffix='.tmp')
+    try:
+        with os.fdopen(fd, 'w', encoding='utf-8') as f:
+            json.dump(data, f, ensure_ascii=False, indent=2)
+        
+        # 原子重命名
+        os.replace(temp_path, STATS_FILE)
+    except Exception:
+        # 清理临时文件
+        if os.path.exists(temp_path):
+            os.unlink(temp_path)
+        raise
+
+
+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: 记录是否成功
+    """
+    try:
+        # 过滤敏感数据
+        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)
+        data["metadata"]["last_updated"] = datetime.now().isoformat()
+        
+        # 原子写入
+        _atomic_write(data)
+        
+        return True
+    except Exception as e:
+        print(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]["timestamp"]
+        stats["last_event"] = events[-1]["timestamp"]
+        
+        for event in events:
+            stats["events_by_type"][event["event_type"]] += 1
+            stats["events_by_mode"][event["mode"]] += 1
+            
+            # 按日期统计
+            date = event["timestamp"][:10]  # 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
+    
+    try:
+        if STATS_FILE.exists():
+            STATS_FILE.unlink()
+        return True
+    except Exception as e:
+        print(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))