diff --git a/.planning/debate.md b/.planning/debate.md new file mode 100644 index 0000000..b39fbe1 --- /dev/null +++ b/.planning/debate.md @@ -0,0 +1,55 @@ +# Cross-Debate 结果 + +**创建时间:** 2026-03-09 14:33 +**需求:** + +--- + +## Agent 方案 + +### Claude Code（保守派） + +**方案:** + + +**理由:** + + +--- + +### Codex（全面派） + +**方案:** + + +**理由:** + + +--- + +### Gemini（快速派） + +**方案:** + + +**理由:** + + +--- + +## 共识点 + +- ✅ +- ✅ + +## 分歧点 + +| Agent | 方案 | 理由 | +|-------|------|------| +| | | | + +## 用户决策 + +**选择方案:** + +--- diff --git a/.planning/spec.md b/.planning/spec.md new file mode 100644 index 0000000..f349fd0 --- /dev/null +++ b/.planning/spec.md @@ -0,0 +1,155 @@ +# Spec: Cross 模式统计报告 + +**创建时间:** 2026-03-09 14:38 +**状态:** approved +**分支:** feature/cross-stats +**来源:** Cross-Debate + +--- + +## 概述 + +给 dev-workflow 添加统计报告功能，显示 Cross 模式（Cross-Debate + Cross-Review）的使用情况。 + +--- + +## 背景 + +Cross 模式是 dev-workflow 的核心功能，需要了解其使用情况： +- 调用频率 +- 成功率 +- 常见场景 +- 问题分布 + +--- + +## Cross-Debate 统一方案 + +### 共识 + +三个 Agent（保守派、全面派、快速派）经过交叉审查后达成共识： + +> **"快速不等于简陋，结构化 + 缓存 + 权限是底线"** + +### 渐进式架构 + +#### 短期（MVP） +``` +日志埋点 → 结构化 JSON 存储 → Redis 缓存（TTL 5分钟） +``` + +#### 中期（加固） +``` +Grafana 可视化 + 权限隔离（API 层最小校验） +``` + +#### 长期（完善） +``` +数据脱敏 + 告警通知 + 定期归档 +``` + +--- + +## 需求 + +### 功能需求 + +- [ ] **REQ-01**: 记录 Cross-Debate 调用事件（时间、场景、Agent 参与情况） +- [ ] **REQ-02**: 记录 Cross-Review 调用事件（时间、轮次、问题数量） +- [ ] **REQ-03**: 统计报告命令：`python3 dev_workflow.py stats` +- [ ] **REQ-04**: 输出格式：JSON + Markdown +- [ ] **REQ-05**: 统计维度：日/周/月、按场景、按成功率 + +### 非功能需求 + +- 性能：查询响应 < 100ms（通过 Redis 缓存） +- 安全：API 层最小权限校验 +- 兼容：不破坏现有 dev_workflow.py 接口 + +--- + +## 技术设计 + +### 数据结构 + +```json +{ + "cross_events": [ + { + "timestamp": "2026-03-09T14:38:00Z", + "type": "debate" | "review", + "scenario": "功能设计" | "Bug修复" | "重构", + "agents": ["claude-code", "codex", "gemini"], + "duration_ms": 45000, + "status": "success" | "failed", + "issues_found": 3, + "rounds": 2 + } + ] +} +``` + +### 文件结构 + +``` +dev-workflow/ +├── dev_workflow.py # 添加 stats 命令 +├── stats.py # 新增：统计逻辑 +└── .data/ + └── cross_events.json # 事件存储 +``` + +### 实现步骤 + +1. 添加 `record_event()` 函数记录事件 +2. 在 Cross-Debate 和 Cross-Review 流程中埋点 +3. 实现 `stats` 命令输出报告 +4. 添加 Redis 缓存（可选） + +### 依赖 + +- 无新增外部依赖 +- 可选：Redis（用于缓存） + +--- + +## 测试计划 + +### 单元测试 + +- [ ] `record_event()` 正常记录 +- [ ] `record_event()` 异常处理 +- [ ] `get_stats()` 空数据返回 +- [ ] `get_stats()` 正常统计 + +### 集成测试 + +- [ ] 完整 Cross-Debate 流程后查看统计 +- [ ] 完整 Cross-Review 流程后查看统计 + +--- + +## 验收标准 + +1. 执行 `python3 dev_workflow.py stats` 输出统计报告 +2. 报告包含：总调用次数、成功率、Top 5 场景 +3. 数据持久化到 JSON 文件 +4. 不影响现有功能 + +--- + +## 风险与缓解 + +| 风险 | 缓解措施 | +|------|----------| +| 并发写入冲突 | 使用文件锁或队列 | +| 存储增长 | 定期归档旧数据 | +| 敏感信息泄露 | 不记录用户输入内容 | + +--- + +## 变更日志 + +| 日期 | 变更 | +|------|------| +| 2026-03-09 14:38 | 初始创建（来自 Cross-Debate 共识） | diff --git a/dev-workflow/.gitignore b/dev-workflow/.gitignore new file mode 100644 index 0000000..d9af100 --- /dev/null +++ b/dev-workflow/.gitignore @@ -0,0 +1,14 @@ +# 数据文件 +data/ +*.json + +# Python +__pycache__/ +*.pyc +*.pyo +*.egg-info/ + +# 临时文件 +*.tmp +*.bak +*.lock diff --git a/dev-workflow/dev_workflow.py b/dev-workflow/dev_workflow.py index 703956c..14b94fa 100644 --- a/dev-workflow/dev_workflow.py +++ b/dev-workflow/dev_workflow.py @@ -428,6 +428,16 @@ def main(): # status subparsers.add_parser('status', help='查看状态') + # stats + subparsers.add_parser('stats', help='查看 Cross 模式统计报告') + + # record-event + event_parser = subparsers.add_parser('record-event', help='记录 Cross 模式事件') + event_parser.add_argument('--type', required=True, choices=['debate', 'review', 'develop', 'fix', 'other'], help='事件类型') + event_parser.add_argument('--mode', default='cross', choices=['cross', 'normal'], help='模式') + event_parser.add_argument('--metadata', default='{}', help='元数据（JSON）') + event_parser.add_argument('--status', default='success', help='状态') + args = parser.parse_args() if args.command == 'create-spec': @@ -442,6 +452,22 @@ def main(): result = record_review(args.round, args.issues) elif args.command == 'status': result = get_status() + elif args.command == 'stats': + # 导入 stats 模块 + sys.path.insert(0, str(Path(__file__).parent)) + from stats import get_stats, format_stats_report + stats = get_stats() + print(format_stats_report(stats)) + result = stats + elif args.command == 'record-event': + # 导入 stats 模块 + sys.path.insert(0, str(Path(__file__).parent)) + from stats import record_event + try: + metadata = json.loads(args.metadata) + except: + metadata = {} + result = {"success": record_event(args.type, args.mode, metadata, args.status)} else: parser.print_help() return diff --git a/dev-workflow/stats.py b/dev-workflow/stats.py new file mode 100644 index 0000000..78de33c --- /dev/null +++ b/dev-workflow/stats.py @@ -0,0 +1,359 @@ +#!/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))