aceflow-pateoas-framework-main/.aceflow/scripts/aceflow

#!/usr/bin/env python3
"""
AceFlow v3.0 CLI 工具
主要功能：项目管理、流程控制、AI辅助
"""

import json
import os
import sys
import argparse
import yaml
from datetime import datetime
from pathlib import Path

class AceFlowCLI:
    def __init__(self):
        self.project_root = Path.cwd()
        self.aceflow_dir = self.project_root / ".aceflow"
        self.state_file = self.aceflow_dir / "state" / "project_state.json"
        self.config_file = self.aceflow_dir / "config" / "project.yaml"
    
    def load_state(self):
        """加载项目状态"""
        if self.state_file.exists():
            with open(self.state_file, 'r', encoding='utf-8') as f:
                return json.load(f)
        return {}
    
    def save_state(self, state):
        """保存项目状态"""
        state['last_updated'] = datetime.now().isoformat()
        with open(self.state_file, 'w', encoding='utf-8') as f:
            json.dump(state, f, indent=2, ensure_ascii=False)
    
    def load_config(self):
        """加载项目配置"""
        if self.config_file.exists():
            with open(self.config_file, 'r', encoding='utf-8') as f:
                return yaml.safe_load(f)
        return {}
    
    def init_project(self, mode='smart'):
        """初始化 AceFlow 项目"""
        print(f"🚀 初始化 AceFlow v3.0 项目...")
        
        # 创建目录结构
        dirs = [
            '.aceflow/config',
            '.aceflow/state',
            '.aceflow/scripts',
            '.aceflow/templates',
            '.aceflow/memory',
            'aceflow_result'
        ]
        
        for dir_path in dirs:
            os.makedirs(dir_path, exist_ok=True)
        
        # 初始化状态
        state = {
            "project_id": f"aceflow_{datetime.now().strftime('%Y%m%d_%H%M%S')}",
            "flow_mode": mode,
            "selected_mode": None,
            "current_stage": None,
            "overall_progress": 0,
            "created_at": datetime.now().isoformat(),
            "version": "3.0.0"
        }
        
        self.save_state(state)
        print(f"✅ AceFlow 项目初始化完成！模式: {mode}")
        return state
    
    def status(self, format_type='text', verbose=False):
        """查看项目状态"""
        if not self.aceflow_dir.exists():
            print("❌ 未检测到 AceFlow 项目，请先运行 'aceflow init'")
            return
        
        state = self.load_state()
        config = self.load_config()
        
        if format_type == 'json':
            print(json.dumps(state, indent=2, ensure_ascii=False))
        else:
            self._print_status_text(state, config, verbose)
    
    def _print_status_text(self, state, config, verbose):
        """打印文本格式的状态"""
        print("\n📊 AceFlow 项目状态")
        print("=" * 40)
        print(f"项目ID: {state.get('project_id', 'N/A')}")
        print(f"流程模式: {state.get('flow_mode', 'N/A')}")
        print(f"当前阶段: {state.get('current_stage', '未开始')}")
        print(f"整体进度: {state.get('overall_progress', 0)}%")
        print(f"最后更新: {state.get('last_updated', 'N/A')}")
        
        if verbose:
            print(f"\n📋 详细信息:")
            print(f"版本: {state.get('version', 'N/A')}")
            print(f"创建时间: {state.get('created_at', 'N/A')}")
            print(f"配置文件: {self.config_file}")
            print(f"状态文件: {self.state_file}")
    
    def analyze(self, task_description):
        """AI 任务分析"""
        print(f"🧠 正在分析任务: {task_description}")
        
        # 简单的任务分类逻辑
        keywords = {
            'bug': ['修复', 'fix', 'bug', '问题', '错误'],
            'feature': ['新功能', '开发', '实现', '添加', '功能'],
            'refactor': ['重构', '优化', '改进', '重写'],
            'project': ['项目', '系统', '平台', '架构']
        }
        
        task_type = 'unknown'
        for category, kw_list in keywords.items():
            if any(kw in task_description.lower() for kw in kw_list):
                task_type = category
                break
        
        # 推荐模式
        mode_mapping = {
            'bug': 'minimal',
            'feature': 'standard',
            'refactor': 'standard',
            'project': 'complete',
            'unknown': 'smart'
        }
        
        recommended_mode = mode_mapping.get(task_type, 'smart')
        
        result = {
            'task_description': task_description,
            'task_type': task_type,
            'recommended_mode': recommended_mode,
            'confidence': 0.85,
            'analysis_time': datetime.now().isoformat()
        }
        
        print(f"📊 分析结果:")
        print(f"  任务类型: {task_type}")
        print(f"  推荐模式: {recommended_mode}")
        print(f"  置信度: 85%")
        
        return result
    
    def start(self, description=None, mode=None):
        """开始新的工作流"""
        state = self.load_state()
        
        if not description:
            description = input("请描述您要开始的任务: ")
        
        if not mode:
            # 智能分析推荐模式
            analysis = self.analyze(description)
            mode = analysis['recommended_mode']
        
        # 生成迭代ID
        iteration_id = f"iter_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
        
        # 更新状态
        state.update({
            'selected_mode': mode,
            'current_stage': self._get_first_stage(mode),
            'iteration_id': iteration_id,
            'task_description': description,
            'overall_progress': 0,
            'stage_progress': 0
        })
        
        self.save_state(state)
        
        # 创建结果目录
        result_dir = Path(f"aceflow_result/{iteration_id}")
        result_dir.mkdir(parents=True, exist_ok=True)
        
        print(f"🚀 开始 AceFlow 工作流")
        print(f"  任务描述: {description}")
        print(f"  选择模式: {mode}")
        print(f"  迭代ID: {iteration_id}")
        print(f"  当前阶段: {state['current_stage']}")
        print(f"  结果目录: {result_dir}")
        
        return state
    
    def _get_first_stage(self, mode):
        """获取模式的第一个阶段"""
        stage_mapping = {
            'minimal': 'P',
            'standard': 'P1',
            'complete': 'S1',
            'smart': 'S1'
        }
        return stage_mapping.get(mode, 'S1')
    
    def progress(self, stage, percentage):
        """更新进度"""
        state = self.load_state()
        
        if stage == 'current':
            stage = state.get('current_stage')
        
        if not stage:
            print("❌ 未找到当前阶段")
            return
        
        state['stage_progress'] = percentage
        # 简单的整体进度计算
        state['overall_progress'] = min(percentage, 100)
        
        self.save_state(state)
        
        print(f"📈 进度更新: {stage} -> {percentage}%")
        return state
    
    def complete(self, stage=None):
        """完成阶段"""
        state = self.load_state()
        
        if not stage:
            stage = state.get('current_stage')
        
        if not stage:
            print("❌ 未找到当前阶段")
            return
        
        print(f"✅ 完成阶段: {stage}")
        
        # 更新状态
        state['stage_progress'] = 100
        
        # 移动到下一阶段
        next_stage = self._get_next_stage(stage, state.get('selected_mode'))
        if next_stage:
            state['current_stage'] = next_stage
            state['stage_progress'] = 0
            print(f"➡️  进入下一阶段: {next_stage}")
        else:
            print("🎉 所有阶段完成！")
            state['current_stage'] = None
            state['overall_progress'] = 100
        
        self.save_state(state)
        return state
    
    def _get_next_stage(self, current_stage, mode):
        """获取下一个阶段"""
        stage_flows = {
            'minimal': ['P', 'D', 'R'],
            'standard': ['P1', 'P2', 'D1', 'D2', 'R1'],
            'complete': ['S1', 'S2', 'S3', 'S4', 'S5', 'S6', 'S7', 'S8'],
            'smart': ['S1', 'S2', 'S3', 'S4', 'S5', 'S6', 'S7', 'S8']
        }
        
        flow = stage_flows.get(mode, [])
        if current_stage in flow:
            current_index = flow.index(current_stage)
            if current_index + 1 < len(flow):
                return flow[current_index + 1]
        
        return None

def main():
    parser = argparse.ArgumentParser(description='AceFlow v3.0 CLI 工具')
    subparsers = parser.add_subparsers(dest='command', help='可用命令')
    
    # init 命令
    init_parser = subparsers.add_parser('init', help='初始化项目')
    init_parser.add_argument('--mode', choices=['smart', 'minimal', 'standard', 'complete'], 
                           default='smart', help='流程模式')
    
    # status 命令
    status_parser = subparsers.add_parser('status', help='查看状态')
    status_parser.add_argument('--format', choices=['text', 'json'], default='text', help='输出格式')
    status_parser.add_argument('--verbose', action='store_true', help='详细输出')
    
    # analyze 命令
    analyze_parser = subparsers.add_parser('analyze', help='分析任务')
    analyze_parser.add_argument('task', help='任务描述')
    
    # start 命令
    start_parser = subparsers.add_parser('start', help='开始工作流')
    start_parser.add_argument('--description', help='任务描述')
    start_parser.add_argument('--mode', choices=['smart', 'minimal', 'standard', 'complete'], 
                            help='流程模式')
    
    # progress 命令
    progress_parser = subparsers.add_parser('progress', help='更新进度')
    progress_parser.add_argument('stage', help='阶段名称')
    progress_parser.add_argument('percentage', type=int, help='进度百分比')
    
    # complete 命令
    complete_parser = subparsers.add_parser('complete', help='完成阶段')
    complete_parser.add_argument('stage', nargs='?', default='current', help='阶段名称')
    
    args = parser.parse_args()
    
    cli = AceFlowCLI()
    
    if args.command == 'init':
        cli.init_project(args.mode)
    elif args.command == 'status':
        cli.status(args.format, args.verbose)
    elif args.command == 'analyze':
        cli.analyze(args.task)
    elif args.command == 'start':
        cli.start(args.description, args.mode)
    elif args.command == 'progress':
        cli.progress(args.stage, args.percentage)
    elif args.command == 'complete':
        cli.complete(args.stage)
    else:
        parser.print_help()

if __name__ == '__main__':
    main()