diff --git a/dev-workflow/.stats_data.json b/dev-workflow/.stats_data.json new file mode 100644 index 0000000..f9545ec --- /dev/null +++ b/dev-workflow/.stats_data.json @@ -0,0 +1,21 @@ +{ + "events": [ + { + "timestamp": "2026-03-09T08:53:12.508206+00:00", + "event_type": "debate", + "mode": "cross", + "status": "success", + "metadata": { + "topic": "AI安全", + "password": "[REDACTED]" + } + }, + { + "timestamp": "2026-03-09T08:53:12.540089+00:00", + "event_type": "review", + "mode": "normal", + "status": "failed", + "metadata": {} + } + ] +} \ No newline at end of file diff --git a/dev-workflow/dev_workflow.py b/dev-workflow/dev_workflow.py new file mode 100644 index 0000000..f038324 --- /dev/null +++ b/dev-workflow/dev_workflow.py @@ -0,0 +1,132 @@ +#!/usr/bin/env python3 +""" +Dev Workflow CLI + +开发工作流命令行工具,包含 Cross 模式统计报告功能。 +""" + +import argparse +import sys +from pathlib import Path + +# 添加当前目录到路径 +sys.path.insert(0, str(Path(__file__).parent)) + +from stats import record_event, get_stats, format_stats_report + + +def cmd_stats(args): + """显示统计报告""" + stats = get_stats() + report = format_stats_report(stats) + print(report) + + +def cmd_record_event(args): + """记录一个事件""" + # 构建 metadata + metadata = {} + if args.metadata: + for item in args.metadata: + if "=" in item: + key, value = item.split("=", 1) + metadata[key] = value + + success = record_event( + event_type=args.type, + mode=args.mode, + metadata=metadata if metadata else None, + status=args.status + ) + + if success: + print(f"✅ 事件记录成功: type={args.type}, mode={args.mode}, status={args.status}") + else: + print(f"❌ 事件记录失败") + sys.exit(1) + + +def cmd_clear_stats(args): + """清除统计数据(谨慎使用)""" + from stats import STATS_FILE + import os + + if STATS_FILE.exists(): + if args.force or input("确认清除所有统计数据? [y/N]: ").lower() == "y": + os.remove(STATS_FILE) + print("✅ 统计数据已清除") + else: + print("操作已取消") + else: + print("统计数据文件不存在") + + +def main(): + parser = argparse.ArgumentParser( + description="Dev Workflow - 开发工作流工具", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +示例: + # 查看统计报告 + python3 dev_workflow.py stats + + # 记录一个 Cross 模式辩论事件 + python3 dev_workflow.py record-event --type debate --mode cross + + # 记录一个失败的事件(带元数据) + python3 dev_workflow.py record-event --type review --mode normal --status failed --metadata pr_id=123 + + # 清除统计数据 + python3 dev_workflow.py clear-stats --force +""" + ) + + subparsers = parser.add_subparsers(dest="command", help="可用命令") + + # stats 命令 + stats_parser = subparsers.add_parser("stats", help="查看统计报告") + stats_parser.set_defaults(func=cmd_stats) + + # record-event 命令 + record_parser = subparsers.add_parser("record-event", help="记录事件") + record_parser.add_argument( + "--type", "-t", + required=True, + choices=["debate", "review", "audit"], + help="事件类型: debate(辩论), review(评审), audit(审计)" + ) + record_parser.add_argument( + "--mode", "-m", + required=True, + choices=["cross", "normal"], + help="运行模式: cross 或 normal" + ) + record_parser.add_argument( + "--status", "-s", + default="success", + choices=["success", "failed"], + help="事件状态 (默认: success)" + ) + record_parser.add_argument( + "--metadata", + nargs="*", + help="元数据,格式: key=value key2=value2" + ) + record_parser.set_defaults(func=cmd_record_event) + + # clear-stats 命令 + clear_parser = subparsers.add_parser("clear-stats", help="清除统计数据") + clear_parser.add_argument("--force", "-f", action="store_true", help="强制清除,不询问") + clear_parser.set_defaults(func=cmd_clear_stats) + + args = parser.parse_args() + + if args.command is None: + parser.print_help() + sys.exit(0) + + args.func(args) + + +if __name__ == "__main__": + main() diff --git a/dev-workflow/stats.py b/dev-workflow/stats.py new file mode 100644 index 0000000..398d349 --- /dev/null +++ b/dev-workflow/stats.py @@ -0,0 +1,280 @@ +""" +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))