diff --git a/dev-workflow/.gitignore b/dev-workflow/.gitignore new file mode 100644 index 0000000..d55038a --- /dev/null +++ b/dev-workflow/.gitignore @@ -0,0 +1,41 @@ +# 统计数据文件 +stats.json +*.json.bak + +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg + +# 虚拟环境 +venv/ +ENV/ +env/ + +# IDE +.vscode/ +.idea/ +*.swp +*.swo +*~ + +# 操作系统 +.DS_Store +Thumbs.db diff --git a/dev-workflow/dev_workflow.py b/dev-workflow/dev_workflow.py new file mode 100644 index 0000000..16ff155 --- /dev/null +++ b/dev-workflow/dev_workflow.py @@ -0,0 +1,121 @@ +#!/usr/bin/env python3 +""" +Dev Workflow - 开发工作流工具 +提供 Cross-Debate 统计报告等功能 +""" + +import sys +import argparse +from pathlib import Path + +# 添加当前目录到 Python 路径 +sys.path.insert(0, str(Path(__file__).parent)) + +from stats import record_event, get_stats, format_stats_report + + +def cmd_stats(args): + """查看统计报告""" + print(format_stats_report()) + return 0 + + +def cmd_record_event(args): + """记录事件""" + try: + # 构建 metadata + metadata = {} + if args.metadata: + for item in args.metadata: + if '=' in item: + key, value = item.split('=', 1) + metadata[key] = value + + # 记录事件 + record_event( + event_type=args.type, + mode=args.mode, + metadata=metadata, + status=args.status + ) + + print(f"✅ 事件已记录: type={args.type}, mode={args.mode}, status={args.status}") + return 0 + + except ValueError as e: + print(f"❌ 错误: {e}", file=sys.stderr) + return 1 + except Exception as e: + print(f"❌ 记录事件失败: {e}", file=sys.stderr) + return 1 + + +def main(): + """主函数""" + parser = argparse.ArgumentParser( + description='Dev Workflow - 开发工作流工具', + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +示例: + # 查看统计报告 + python3 dev_workflow.py stats + + # 记录一个 cross 模式的 debate 事件 + python3 dev_workflow.py record-event --type debate --mode cross --status success + + # 记录带元数据的事件 + python3 dev_workflow.py record-event --type review --mode normal --metadata files=5 issues=2 + """ + ) + + 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', + required=True, + choices=['debate', 'review', 'audit'], + help='事件类型 (debate|review|audit)' + ) + record_parser.add_argument( + '--mode', + required=True, + choices=['cross', 'normal'], + help='模式 (cross|normal)' + ) + record_parser.add_argument( + '--status', + default='success', + choices=['success', 'failed'], + help='事件状态 (success|failed),默认为 success' + ) + record_parser.add_argument( + '--metadata', + nargs='*', + help='元数据,格式为 key=value(可多个)' + ) + record_parser.set_defaults(func=cmd_record_event) + + # 解析参数 + args = parser.parse_args() + + # 如果没有指定命令,显示帮助 + if not args.command: + parser.print_help() + return 0 + + # 执行命令 + if hasattr(args, 'func'): + return args.func(args) + else: + parser.print_help() + return 0 + + +if __name__ == '__main__': + sys.exit(main()) diff --git a/dev-workflow/stats.py b/dev-workflow/stats.py new file mode 100644 index 0000000..3b5d1e8 --- /dev/null +++ b/dev-workflow/stats.py @@ -0,0 +1,279 @@ +#!/usr/bin/env python3 +""" +Cross-Debate 统计报告模块 +提供事件记录和统计报告功能 +""" + +import json +import os +import fcntl +from datetime import datetime, timezone +from pathlib import Path +from typing import Dict, List, Any, Optional + + +class StatsManager: + """统计管理器""" + + # 敏感字段列表 + SENSITIVE_FIELDS = {'password', 'token', 'secret', 'api_key', 'apikey', 'credential'} + + def __init__(self, stats_file: str = None): + """初始化统计管理器 + + Args: + stats_file: 统计数据文件路径,默认为 ~/.dev-workflow/stats.json + """ + if stats_file is None: + stats_dir = Path.home() / '.dev-workflow' + stats_dir.mkdir(parents=True, exist_ok=True) + stats_file = stats_dir / 'stats.json' + + self.stats_file = Path(stats_file) + self._ensure_stats_file() + + def _filter_sensitive_data(self, data: Dict[str, Any]) -> Dict[str, Any]: + """过滤敏感数据 + + Args: + data: 原始数据字典 + + Returns: + 过滤后的数据字典 + """ + filtered = {} + for key, value in data.items(): + key_lower = key.lower() + # 检查是否包含敏感字段 + if any(sensitive in key_lower for sensitive in self.SENSITIVE_FIELDS): + filtered[key] = '***REDACTED***' + elif isinstance(value, dict): + filtered[key] = self._filter_sensitive_data(value) + else: + filtered[key] = value + return filtered + + def _ensure_stats_file(self): + """确保统计文件存在并设置正确权限""" + if not self.stats_file.exists(): + self.stats_file.parent.mkdir(parents=True, exist_ok=True) + self._write_data({"events": []}) + + # 设置文件权限为 600 (仅所有者可读写) + self.stats_file.chmod(0o600) + + def _read_data(self) -> Dict[str, Any]: + """读取统计数据(带文件锁) + + Returns: + 统计数据字典 + """ + if not self.stats_file.exists(): + return {"events": []} + + with open(self.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 _write_data(self, data: Dict[str, Any]): + """写入统计数据(带文件锁) + + Args: + data: 要写入的数据字典 + """ + with open(self.stats_file, 'w') as f: + # 获取排他锁(写锁) + fcntl.flock(f.fileno(), fcntl.LOCK_EX) + try: + json.dump(data, f, indent=2, ensure_ascii=False) + finally: + # 释放锁 + fcntl.flock(f.fileno(), fcntl.LOCK_UN) + + # 确保文件权限为 600 + self.stats_file.chmod(0o600) + + def record_event( + self, + event_type: str, + mode: str, + metadata: Optional[Dict[str, Any]] = None, + status: str = "success" + ) -> bool: + """记录事件 + + Args: + event_type: 事件类型 (debate|review|audit) + mode: 模式 (cross|normal) + metadata: 事件元数据 + status: 事件状态 (success|failed) + + Returns: + 是否记录成功 + """ + # 验证参数 + valid_types = {'debate', 'review', 'audit'} + valid_modes = {'cross', 'normal'} + valid_statuses = {'success', 'failed'} + + if event_type not in valid_types: + raise ValueError(f"Invalid event_type: {event_type}. Must be one of {valid_types}") + if mode not in valid_modes: + raise ValueError(f"Invalid mode: {mode}. Must be one of {valid_modes}") + if status not in valid_statuses: + raise ValueError(f"Invalid status: {status}. Must be one of {valid_statuses}") + + # 过滤敏感数据 + if metadata is None: + metadata = {} + metadata = self._filter_sensitive_data(metadata) + + # 创建事件记录 + event = { + "timestamp": datetime.now(timezone.utc).isoformat(), + "event_type": event_type, + "mode": mode, + "status": status, + "metadata": metadata + } + + # 读取现有数据 + data = self._read_data() + + # 添加新事件 + data["events"].append(event) + + # 写入数据 + self._write_data(data) + + return True + + def get_stats(self) -> Dict[str, Any]: + """获取统计数据 + + Returns: + 统计数据字典,包含总数、各类型计数等 + """ + data = self._read_data() + events = data.get("events", []) + + # 基础统计 + stats = { + "total_events": len(events), + "by_type": {}, + "by_mode": {}, + "by_status": {}, + "by_type_and_mode": {}, + "recent_events": events[-10:] if events else [] # 最近10条事件 + } + + # 统计各维度 + for event in events: + event_type = event.get("event_type", "unknown") + mode = event.get("mode", "unknown") + status = event.get("status", "unknown") + + # 按类型统计 + stats["by_type"][event_type] = stats["by_type"].get(event_type, 0) + 1 + + # 按模式统计 + stats["by_mode"][mode] = stats["by_mode"].get(mode, 0) + 1 + + # 按状态统计 + stats["by_status"][status] = stats["by_status"].get(status, 0) + 1 + + # 按类型和模式交叉统计 + key = f"{event_type}_{mode}" + stats["by_type_and_mode"][key] = stats["by_type_and_mode"].get(key, 0) + 1 + + return stats + + def format_stats_report(self) -> str: + """格式化统计报告 + + Returns: + 格式化的统计报告字符串 + """ + stats = self.get_stats() + + report_lines = [ + "=" * 50, + "Cross-Debate 统计报告", + "=" * 50, + "", + f"📊 总事件数: {stats['total_events']}", + "", + "📋 按事件类型统计:", + ] + + for event_type, count in sorted(stats["by_type"].items()): + report_lines.append(f" - {event_type}: {count}") + + report_lines.append("") + report_lines.append("🔀 按模式统计:") + + for mode, count in sorted(stats["by_mode"].items()): + report_lines.append(f" - {mode}: {count}") + + report_lines.append("") + report_lines.append("✅ 按状态统计:") + + for status, count in sorted(stats["by_status"].items()): + report_lines.append(f" - {status}: {count}") + + report_lines.append("") + report_lines.append("📊 按类型和模式交叉统计:") + + for key, count in sorted(stats["by_type_and_mode"].items()): + report_lines.append(f" - {key}: {count}") + + report_lines.append("") + report_lines.append("=" * 50) + + return "\n".join(report_lines) + + +# 便捷函数 +_manager = None + +def _get_manager() -> StatsManager: + """获取全局统计管理器实例""" + global _manager + if _manager is None: + _manager = StatsManager() + return _manager + + +def record_event(event_type: str, mode: str, metadata: Optional[Dict[str, Any]] = None, status: str = "success") -> bool: + """记录事件(便捷函数)""" + return _get_manager().record_event(event_type, mode, metadata, status) + + +def get_stats() -> Dict[str, Any]: + """获取统计数据(便捷函数)""" + return _get_manager().get_stats() + + +def format_stats_report() -> str: + """格式化统计报告(便捷函数)""" + return _get_manager().format_stats_report() + + +if __name__ == "__main__": + # 测试代码 + print("测试统计模块...") + + # 记录一些测试事件 + record_event("debate", "cross", {"topic": "test1", "result": "win"}) + record_event("review", "normal", {"files": 5}) + record_event("audit", "cross", {"issues": 3}, status="failed") + + # 打印报告 + print(format_stats_report())