aceflow-pateoas-framework-main/.aceflow/ai/cli/agent_cli.py

#!/usr/bin/env python3
"""
AceFlow v2.0 CLI工具 - Agent集成版本
为Cursor、Cline、Copilot等AI工具提供智能工作流建议
"""

import sys
import json
import yaml
import argparse
from pathlib import Path
from typing import Dict, Any, List, Optional
from datetime import datetime

# 导入决策引擎
sys.path.append(str(Path(__file__).parent.parent))
from engines.rule_based_engine import get_decision_engine, DecisionResult

class AceFlowCLI:
    """AceFlow CLI工具"""
    
    def __init__(self):
        self.project_root = Path.cwd()
        self.aceflow_dir = self.project_root / ".aceflow"
        self.version = "2.0.0"
        self.engine = get_decision_engine(self.project_root)
    
    def describe(self, output_format: str = "json") -> str:
        """描述工具能力，供Agent发现和理解"""
        description = {
            "name": "AceFlow",
            "version": self.version,
            "description": "AI驱动的软件开发工作流管理工具",
            "capabilities": [
                "智能工作流程推荐",
                "任务分类和优先级建议",
                "开发阶段管理",
                "进度跟踪和预测",
                "项目复杂度分析",
                "团队协作优化"
            ],
            "use_cases": [
                "项目规划和流程选择",
                "任务管理和跟踪",
                "开发效率优化",
                "质量保证流程",
                "敏捷开发支持"
            ],
            "trigger_patterns": [
                "工作流",
                "流程",
                "项目规划",
                "任务管理",
                "开发流程",
                "敏捷开发",
                "项目管理"
            ],
            "integration": {
                "cli_commands": {
                    "suggest": {
                        "description": "智能工作流推荐",
                        "usage": "aceflow suggest --task '任务描述' [选项]",
                        "example": "aceflow suggest --task '修复登录bug' --format json"
                    },
                    "plan": {
                        "description": "项目规划建议",
                        "usage": "aceflow plan --project-type web --team-size 5 [选项]",
                        "example": "aceflow plan --project-type web --team-size 5 --format json"
                    },
                    "track": {
                        "description": "进度跟踪",
                        "usage": "aceflow track --stage current [选项]",
                        "example": "aceflow track --stage current --format json"
                    },
                    "status": {
                        "description": "项目状态查询",
                        "usage": "aceflow status [选项]",
                        "example": "aceflow status --format json"
                    }
                },
                "output_formats": ["json", "yaml", "text"],
                "response_schema": {
                    "type": "object",
                    "properties": {
                        "recommended_flow": {"type": "string"},
                        "confidence": {"type": "number"},
                        "reasoning": {"type": "string"},
                        "steps": {"type": "array"},
                        "estimated_hours": {"type": "integer"},
                        "alternatives": {"type": "array"}
                    }
                }
            },
            "when_to_use": {
                "scenarios": [
                    "用户询问如何组织开发流程",
                    "需要制定项目计划",
                    "想要标准化开发流程",
                    "需要跟踪项目进度",
                    "选择合适的开发模式"
                ],
                "keywords": [
                    "workflow", "process", "planning", "management",
                    "工作流", "流程", "规划", "管理"
                ]
            }
        }
        
        if output_format == "yaml":
            return yaml.dump(description, default_flow_style=False, allow_unicode=True)
        elif output_format == "text":
            return self._format_description_text(description)
        else:
            return json.dumps(description, indent=2, ensure_ascii=False)
    
    def suggest(self, task: str, **kwargs) -> Dict[str, Any]:
        """智能工作流推荐"""
        if not task:
            raise ValueError("任务描述不能为空")
        
        # 构建上下文
        context = {}
        if kwargs.get("team_size"):
            context["team_size"] = int(kwargs["team_size"])
        if kwargs.get("project_type"):
            context["project_type"] = kwargs["project_type"]
        if kwargs.get("complexity"):
            context["complexity"] = kwargs["complexity"]
        if kwargs.get("urgency"):
            context["urgency"] = kwargs["urgency"]
        
        # 获取决策结果
        result = self.engine.make_decision(task, context)
        
        # 格式化输出
        return self._format_decision_result(result)
    
    def plan(self, **kwargs) -> Dict[str, Any]:
        """项目规划建议"""
        # 构建虚拟任务描述
        project_type = kwargs.get("project_type", "web")
        team_size = kwargs.get("team_size", 5)
        
        task_description = f"为{project_type}项目制定开发计划"
        
        context = {
            "project_type": project_type,
            "team_size": int(team_size),
            "urgency": kwargs.get("urgency", "medium")
        }
        
        if kwargs.get("complexity"):
            context["complexity"] = kwargs["complexity"]
        
        # 获取决策结果
        result = self.engine.make_decision(task_description, context)
        
        # 增加项目规划特定信息
        planning_result = self._format_decision_result(result)
        planning_result["project_recommendations"] = self._generate_project_recommendations(context)
        
        return planning_result
    
    def track(self, **kwargs) -> Dict[str, Any]:
        """进度跟踪"""
        stage = kwargs.get("stage", "current")
        
        # 读取项目状态
        project_state = self._load_project_state()
        
        if not project_state:
            return {
                "error": "未找到项目状态，请先初始化项目",
                "suggestion": "运行 'aceflow init' 初始化项目"
            }
        
        # 构建跟踪结果
        tracking_result = {
            "project_id": project_state.get("project_id", "unknown"),
            "flow_mode": project_state.get("flow_mode", "unknown"),
            "current_stage": project_state.get("current_stage", "unknown"),
            "overall_progress": project_state.get("progress", {}).get("overall", 0),
            "stage_states": project_state.get("stage_states", {}),
            "last_updated": project_state.get("last_updated", "unknown"),
            "recommendations": []
        }
        
        # 添加进度建议
        if tracking_result["overall_progress"] < 30:
            tracking_result["recommendations"].append("项目处于早期阶段，建议专注于需求分析和设计")
        elif tracking_result["overall_progress"] < 70:
            tracking_result["recommendations"].append("项目进展良好，建议保持当前开发节奏")
        else:
            tracking_result["recommendations"].append("项目接近完成，建议重点关注测试和部署准备")
        
        return tracking_result
    
    def status(self, **kwargs) -> Dict[str, Any]:
        """项目状态查询"""
        # 读取项目状态和配置
        project_state = self._load_project_state()
        project_config = self._load_project_config()
        
        # 分析项目特征
        project_profile = self.engine.project_analyzer.analyze_project()
        
        status_result = {
            "project_info": {
                "name": project_config.get("project", {}).get("name", "unknown"),
                "type": project_profile.project_type,
                "team_size": project_profile.team_size,
                "complexity": project_profile.complexity.value,
                "tech_stack": project_profile.tech_stack
            },
            "current_state": project_state or {},
            "project_health": {
                "has_tests": project_profile.has_tests,
                "has_ci_cd": project_profile.has_ci_cd,
                "has_documentation": project_profile.has_documentation,
                "file_count": project_profile.file_count,
                "git_activity": project_profile.git_activity
            },
            "recommendations": []
        }
        
        # 生成健康建议
        if not project_profile.has_tests:
            status_result["recommendations"].append("建议添加测试用例以提高代码质量")
        
        if not project_profile.has_ci_cd:
            status_result["recommendations"].append("建议配置CI/CD流程以自动化部署")
        
        if not project_profile.has_documentation:
            status_result["recommendations"].append("建议完善项目文档以提高可维护性")
        
        return status_result
    
    def memory(self, action: str, **kwargs) -> Dict[str, Any]:
        """记忆管理"""
        memory_dir = self.aceflow_dir / "memory"
        
        if action == "list":
            return self._list_memories(memory_dir)
        elif action == "search":
            query = kwargs.get("query", "")
            return self._search_memories(memory_dir, query)
        elif action == "clean":
            return self._clean_memories(memory_dir)
        else:
            return {"error": f"不支持的记忆操作: {action}"}
    
    def _format_decision_result(self, result: DecisionResult) -> Dict[str, Any]:
        """格式化决策结果"""
        return {
            "recommended_flow": result.recommended_flow,
            "confidence": result.confidence,
            "reasoning": result.reasoning,
            "steps": result.steps,
            "estimated_hours": result.estimated_hours,
            "alternatives": result.alternatives,
            "timestamp": datetime.now().isoformat(),
            "task_type": result.metadata.get("task_type", "unknown")
        }
    
    def _generate_project_recommendations(self, context: Dict[str, Any]) -> List[str]:
        """生成项目规划建议"""
        recommendations = []
        
        team_size = context.get("team_size", 1)
        project_type = context.get("project_type", "web")
        
        # 团队规模建议
        if team_size <= 3:
            recommendations.append("小团队建议使用敏捷开发模式，快速迭代")
        elif team_size > 8:
            recommendations.append("大团队建议制定详细的协作流程和代码规范")
        
        # 项目类型建议
        if project_type == "web":
            recommendations.append("Web项目建议重点关注性能优化和用户体验")
        elif project_type == "mobile":
            recommendations.append("移动应用建议关注多平台适配和离线功能")
        elif project_type == "api":
            recommendations.append("API项目建议重点关注接口设计和文档完善")
        
        return recommendations
    
    def _load_project_state(self) -> Optional[Dict[str, Any]]:
        """加载项目状态"""
        state_file = self.aceflow_dir / "state" / "project_state.json"
        
        if state_file.exists():
            try:
                with open(state_file, 'r', encoding='utf-8') as f:
                    return json.load(f)
            except Exception:
                return None
        
        return None
    
    def _load_project_config(self) -> Optional[Dict[str, Any]]:
        """加载项目配置"""
        config_file = self.aceflow_dir / "config.yaml"
        
        if config_file.exists():
            try:
                with open(config_file, 'r', encoding='utf-8') as f:
                    return yaml.safe_load(f)
            except Exception:
                return None
        
        return None
    
    def _list_memories(self, memory_dir: Path) -> Dict[str, Any]:
        """列出记忆"""
        if not memory_dir.exists():
            return {"memories": [], "count": 0}
        
        memories = []
        for memory_file in memory_dir.glob("*.json"):
            try:
                with open(memory_file, 'r', encoding='utf-8') as f:
                    memory_data = json.load(f)
                    memories.append({
                        "id": memory_file.stem,
                        "timestamp": memory_data.get("timestamp", "unknown"),
                        "content_preview": memory_data.get("content", "")[:100] + "..." if len(memory_data.get("content", "")) > 100 else memory_data.get("content", ""),
                        "keywords": memory_data.get("keywords", [])
                    })
            except Exception:
                continue
        
        return {"memories": memories, "count": len(memories)}
    
    def _search_memories(self, memory_dir: Path, query: str) -> Dict[str, Any]:
        """搜索记忆"""
        if not memory_dir.exists():
            return {"results": [], "count": 0}
        
        results = []
        query_lower = query.lower()
        
        for memory_file in memory_dir.glob("*.json"):
            try:
                with open(memory_file, 'r', encoding='utf-8') as f:
                    memory_data = json.load(f)
                    
                    # 简单的关键词匹配
                    content = memory_data.get("content", "").lower()
                    keywords = [k.lower() for k in memory_data.get("keywords", [])]
                    
                    if query_lower in content or any(query_lower in keyword for keyword in keywords):
                        results.append({
                            "id": memory_file.stem,
                            "timestamp": memory_data.get("timestamp", "unknown"),
                            "content": memory_data.get("content", ""),
                            "relevance": self._calculate_relevance(query_lower, content, keywords)
                        })
            except Exception:
                continue
        
        # 按相关性排序
        results.sort(key=lambda x: x["relevance"], reverse=True)
        
        return {"results": results, "count": len(results)}
    
    def _calculate_relevance(self, query: str, content: str, keywords: List[str]) -> float:
        """计算相关性分数"""
        score = 0.0
        
        # 内容匹配
        if query in content:
            score += 1.0
        
        # 关键词匹配
        for keyword in keywords:
            if query in keyword:
                score += 0.5
        
        return score
    
    def _clean_memories(self, memory_dir: Path) -> Dict[str, Any]:
        """清理记忆"""
        if not memory_dir.exists():
            return {"message": "记忆目录不存在", "cleaned": 0}
        
        cleaned_count = 0
        
        # 清理空文件或损坏文件
        for memory_file in memory_dir.glob("*.json"):
            try:
                with open(memory_file, 'r', encoding='utf-8') as f:
                    memory_data = json.load(f)
                    
                    # 检查是否为空或无效记忆
                    if not memory_data.get("content") or len(memory_data.get("content", "")) < 10:
                        memory_file.unlink()
                        cleaned_count += 1
            except Exception:
                # 删除损坏的文件
                memory_file.unlink()
                cleaned_count += 1
        
        return {"message": f"清理完成，删除了{cleaned_count}个无效记忆", "cleaned": cleaned_count}
    
    def _format_description_text(self, description: Dict[str, Any]) -> str:
        """格式化文本描述"""
        text = f"{description['name']} v{description['version']}\n"
        text += f"{description['description']}\n\n"
        
        text += "核心能力:\n"
        for capability in description['capabilities']:
            text += f"  - {capability}\n"
        
        text += "\n使用场景:\n"
        for use_case in description['use_cases']:
            text += f"  - {use_case}\n"
        
        text += "\n触发关键词:\n"
        text += f"  {', '.join(description['trigger_patterns'])}\n"
        
        return text

def main():
    """主函数"""
    parser = argparse.ArgumentParser(
        description="AceFlow v2.0 - AI驱动的软件开发工作流管理工具",
        formatter_class=argparse.RawDescriptionHelpFormatter
    )
    
    parser.add_argument("--version", action="version", version="AceFlow v2.0.0")
    parser.add_argument("--format", choices=["json", "yaml", "text"], default="json",
                       help="输出格式")
    parser.add_argument("--verbose", action="store_true", help="详细输出")
    parser.add_argument("--quiet", action="store_true", help="静默模式")
    
    subparsers = parser.add_subparsers(dest="command", help="可用命令")
    
    # describe命令
    describe_parser = subparsers.add_parser("describe", help="描述工具能力")
    describe_parser.add_argument("--format", choices=["json", "yaml", "text"], default="json")
    
    # suggest命令
    suggest_parser = subparsers.add_parser("suggest", help="智能工作流推荐")
    suggest_parser.add_argument("--task", required=True, help="任务描述")
    suggest_parser.add_argument("--team-size", type=int, help="团队规模")
    suggest_parser.add_argument("--project-type", help="项目类型")
    suggest_parser.add_argument("--complexity", choices=["simple", "moderate", "complex", "enterprise"], help="项目复杂度")
    suggest_parser.add_argument("--urgency", choices=["low", "medium", "high"], default="medium", help="紧急程度")
    
    # plan命令
    plan_parser = subparsers.add_parser("plan", help="项目规划建议")
    plan_parser.add_argument("--project-type", default="web", help="项目类型")
    plan_parser.add_argument("--team-size", type=int, default=5, help="团队规模")
    plan_parser.add_argument("--complexity", choices=["simple", "moderate", "complex", "enterprise"], help="项目复杂度")
    plan_parser.add_argument("--urgency", choices=["low", "medium", "high"], default="medium", help="紧急程度")
    
    # track命令
    track_parser = subparsers.add_parser("track", help="进度跟踪")
    track_parser.add_argument("--stage", default="current", help="阶段")
    
    # status命令
    status_parser = subparsers.add_parser("status", help="项目状态查询")
    
    # memory命令
    memory_parser = subparsers.add_parser("memory", help="记忆管理")
    memory_parser.add_argument("action", choices=["list", "search", "clean"], help="操作类型")
    memory_parser.add_argument("--query", help="搜索查询")
    
    args = parser.parse_args()
    
    if not args.command:
        parser.print_help()
        return
    
    try:
        cli = AceFlowCLI()
        
        if args.command == "describe":
            result = cli.describe(args.format)
            print(result)
        
        elif args.command == "suggest":
            result = cli.suggest(
                task=args.task,
                team_size=args.team_size,
                project_type=args.project_type,
                complexity=args.complexity,
                urgency=args.urgency
            )
            print(json.dumps(result, indent=2, ensure_ascii=False))
        
        elif args.command == "plan":
            result = cli.plan(
                project_type=args.project_type,
                team_size=args.team_size,
                complexity=args.complexity,
                urgency=args.urgency
            )
            print(json.dumps(result, indent=2, ensure_ascii=False))
        
        elif args.command == "track":
            result = cli.track(stage=args.stage)
            print(json.dumps(result, indent=2, ensure_ascii=False))
        
        elif args.command == "status":
            result = cli.status()
            print(json.dumps(result, indent=2, ensure_ascii=False))
        
        elif args.command == "memory":
            result = cli.memory(args.action, query=args.query)
            print(json.dumps(result, indent=2, ensure_ascii=False))
    
    except Exception as e:
        if args.verbose:
            import traceback
            traceback.print_exc()
        else:
            print(json.dumps({"error": str(e)}, indent=2, ensure_ascii=False))
        sys.exit(1)

if __name__ == "__main__":
    main()