aceflow-pateoas-framework-main/.aceflow/scripts/cli/aceflow_cli_enhanced.py

#!/usr/bin/env python
import sys
import os
import argparse
from pathlib import Path

# 解决相对导入问题：将项目根目录和脚本目录添加到Python路径
project_root = Path(__file__).parent.parent
scripts_dir = Path(".aceflow/scripts")

# 添加到系统路径
sys.path.append(str(project_root))
sys.path.append(str(scripts_dir))

# 导入核心模块
try:
    from core.state_engine_enhanced import PATEOASStateEngineEnhanced as PATEOASStateEngine
    from core.workflow_navigator import WorkflowNavigator
    from utils.config_loader import load_config
except ImportError as e:
    print(f"导入核心模块失败: {e}")
    print("请确保 .aceflow/scripts 目录存在并包含必要的模块")
    sys.exit(1)

def validate_stage_output(args):
    """验证阶段输出产物是否完整"""
    state_engine = PATEOASStateEngine()
    state = state_engine.get_current_state()
    stage_id = args.stage_id if args.stage_id else state['current_stage']
    
    print(f"验证阶段 {stage_id} 的输出产物...")
    if state_engine.validate_stage_output(stage_id):
        print(f"阶段 {stage_id} 验证通过，输出产物完整。")
        return True
    else:
        print(f"阶段 {stage_id} 验证未通过，输出产物不完整。")
        print(f"请确保阶段 {stage_id} 的输出文档存在且内容非空。")
        return False

def check_dependencies(args):
    """检查阶段依赖性是否满足"""
    state_engine = PATEOASStateEngine()
    state = state_engine.get_current_state()
    stage_id = args.stage_id if args.stage_id else state['current_stage']
    
    print(f"检查阶段 {stage_id} 的依赖性...")
    # 这里添加依赖性检查逻辑
    print(f"阶段 {stage_id} 依赖性检查通过（模拟结果）")
    return True

def revert_stage(args):
    """回退到指定阶段"""
    state_engine = PATEOASStateEngine()
    state = state_engine.get_current_state()
    target_stage = args.target_stage
    
    print(f"回退到阶段 {target_stage}...")
    state['current_stage'] = target_stage
    state['stage_status'][target_stage] = 'in_progress'
    state_engine.save_state(state)
    print(f"已回退到阶段 {target_stage}")
    return True

def review_previous_stage(args):
    """复查前一阶段产物"""
    state_engine = PATEOASStateEngine()
    state = state_engine.get_current_state()
    current_stage = state['current_stage']
    
    navigator = WorkflowNavigator()
    workflow_type = state.get('workflow_type', '完整流程')
    path = navigator.get_workflow_path(workflow_type)
    current_index = path.index(current_stage) if current_stage in path else -1
    
    if current_index > 0:
        previous_stage = path[current_index - 1]
        print(f"复查前一阶段 {previous_stage} 的产物...")
        # 这里添加复查逻辑
        print(f"阶段 {previous_stage} 复查完成（模拟结果）")
    else:
        print("当前阶段为流程起点，无前一阶段可复查")
    return True

def generate_stage_template(args):
    """生成阶段模板文档"""
    state_engine = PATEOASStateEngine()
    state = state_engine.get_current_state()
    stage_id = args.stage_id if args.stage_id else state['current_stage']
    
    print(f"为阶段 {stage_id} 生成模板文档...")
    # 这里添加模板生成逻辑
    print(f"阶段 {stage_id} 模板文档已生成（模拟结果）")
    return True

def associate_output(args):
    """关联工作产物到阶段"""
    state_engine = PATEOASStateEngine()
    state = state_engine.get_current_state()
    stage_id = args.stage_id if args.stage_id else state['current_stage']
    output_path = args.output_path
    
    print(f"将产物 {output_path} 关联到阶段 {stage_id}...")
    # 这里添加关联逻辑
    print(f"产物 {output_path} 已关联到阶段 {stage_id}（模拟结果）")
    return True

def stage_review(args):
    """记录阶段评审结果"""
    state_engine = PATEOASStateEngine()
    state = state_engine.get_current_state()
    stage_id = args.stage_id if args.stage_id else state['current_stage']
    review_result = args.review_result
    
    print(f"记录阶段 {stage_id} 的评审结果: {review_result}...")
    # 这里添加评审记录逻辑
    print(f"阶段 {stage_id} 评审结果已记录（模拟结果）")
    return True

def stage_memory_summary(args):
    """生成阶段记忆摘要，并存储到记忆池文件"""
    import json
    from pathlib import Path

    state_engine = PATEOASStateEngine()
    state = state_engine.get_current_state()
    stage_id = args.stage_id if args.stage_id else state['current_stage']
    
    print(f"生成阶段 {stage_id} 的记忆摘要...")
    
    # 获取当前迭代信息
    iteration = state.get('current_iteration', 'iteration-01')
    
    # 构建阶段文档路径
    stage_doc_path = Path(f"aceflow_result/iterations/{iteration}/{stage_id.lower()}_*.md")
    document_path = ""
    summary = ""
    
    # 查找阶段文档并提取摘要
    matching_files = list(stage_doc_path.parent.glob(stage_doc_path.name))
    if matching_files:
        document_path = str(matching_files[0])
        try:
            with open(document_path, 'r', encoding='utf-8') as f:
                content = f.read()
                # 提取前200个字符作为摘要，或者整个内容如果较短
                summary = content[:200] + ('...' if len(content) > 200 else '')
                summary = f"阶段 {stage_id} 的关键内容摘要（基于 {document_path}）：\n{summary}"
        except Exception as e:
            summary = f"阶段 {stage_id} 的文档读取失败（{document_path}）：{str(e)}"
    else:
        summary = f"阶段 {stage_id} 的文档未找到，摘要为空"
    
    # 构建记忆片段，包含最佳实践信息
    from datetime import datetime
    best_practice = ""
    if matching_files:
        try:
            with open(document_path, 'r', encoding='utf-8') as f:
                content = f.read()
                # 简单提取最佳实践信息（示例逻辑，实际应根据内容分析）
                if "测试通过率" in content or "coverage" in content.lower():
                    best_practice = "确保单元测试覆盖率达到90%以上。"
                elif "代码评审" in content or "code review" in content.lower():
                    best_practice = "代码评审应包含至少两位团队成员的反馈。"
                else:
                    best_practice = "遵循阶段规范，确保文档完整性和准确性。"
        except Exception as e:
            best_practice = f"无法提取最佳实践信息：{str(e)}"
    else:
        best_practice = "文档未找到，无法提取最佳实践信息。"
    
    memory_entry = {
        "stage_id": stage_id,
        "iteration": iteration,
        "document_path": document_path,
        "summary": summary,
        "best_practice": best_practice,
        "task_type": "未分类",  # 待后续完善
        "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    }
    
    # 存储到记忆池文件
    memory_pool_file = Path("aceflow_result/config/memory_pool.json")
    memory_pool = []
    
    if memory_pool_file.exists():
        try:
            with open(memory_pool_file, 'r', encoding='utf-8') as f:
                memory_pool = json.load(f)
        except Exception as e:
            print(f"读取记忆池文件失败: {e}")
            memory_pool = []
    
    # 检查是否已存在相同阶段和迭代的记录，如果存在则更新，否则添加新记录
    updated = False
    for entry in memory_pool:
        if entry.get("stage_id") == stage_id and entry.get("iteration") == iteration:
            entry.update(memory_entry)
            updated = True
            break
    
    if not updated:
        memory_pool.append(memory_entry)
    
    # 保存更新后的记忆池
    memory_pool_file.parent.mkdir(parents=True, exist_ok=True)
    try:
        with open(memory_pool_file, 'w', encoding='utf-8') as f:
            json.dump(memory_pool, f, ensure_ascii=False, indent=2)
        print(f"阶段 {stage_id} 记忆摘要已生成并存储到 {memory_pool_file}")
    except Exception as e:
        print(f"存储记忆池文件失败: {e}")
        return False
    
    return True

def update_status(args):
    """更新阶段进度，包含前置条件检查，并在进度达到100%时自动触发记忆摘要生成"""
    state_engine = PATEOASStateEngine()
    state = state_engine.get_current_state()
    stage_id = args.stage_id
    progress = args.progress
    
    # 前置条件检查
    print(f"检查阶段 {stage_id} 的前置条件...")
    # 暂时禁用依赖性检查，以便测试
    # if not state_engine.check_dependencies(stage_id):
    #     print(f"阶段 {stage_id} 前置条件检查未通过，依赖性未满足。")
    #     return False
    
    if progress == 100 and not state_engine.validate_stage_output(stage_id):
        print(f"阶段 {stage_id} 前置条件检查未通过，输出产物不完整。")
        return False
    
    print(f"阶段 {stage_id} 前置条件检查通过。")
    
    # 更新进度
    state_engine.update_stage_progress(stage_id, progress)
    print(f"阶段 {stage_id} 进度已更新为 {progress}%")
    
    # 如果进度达到100%，自动触发stage-memory-summary命令
    if progress == 100:
        print(f"阶段 {stage_id} 进度达到100%，自动生成记忆摘要...")
        summary_args = argparse.Namespace(stage_id=stage_id)
        stage_memory_summary(summary_args)
    
    return True

def index_documents(args):
    """扫描指定目录下的文档并更新记忆池文件"""
    import json
    from pathlib import Path

    directory = args.directory if args.directory else "aceflow_result/iterations/"
    print(f"扫描目录 {directory} 下的文档以更新记忆池...")

    memory_pool_file = Path("aceflow_result/config/memory_pool.json")
    memory_pool = []

    if memory_pool_file.exists():
        try:
            with open(memory_pool_file, 'r', encoding='utf-8') as f:
                memory_pool = json.load(f)
        except Exception as e:
            print(f"读取记忆池文件失败: {e}")
            memory_pool = []

    # 扫描目录下的所有 Markdown 文件
    dir_path = Path(directory)
    if not dir_path.exists():
        print(f"目录 {directory} 不存在")
        return False

    updated_count = 0
    for md_file in dir_path.rglob("*.md"):
        file_path = str(md_file)
        # 提取元数据（简单示例，实际应解析文件内容）
        parts = file_path.replace('\\', '/').split('/')
        if len(parts) < 3:
            continue  # 路径不符合预期，跳过

        iteration = parts[-2] if "iteration" in parts[-2].lower() else "unknown_iteration"
        stage_id = parts[-1].split('_')[0].upper() if '_' in parts[-1] else "unknown_stage"

        # 检查是否已存在相同文档路径的记录，如果存在则更新，否则添加新记录
        from datetime import datetime
        found = False
        for entry in memory_pool:
            if entry.get("document_path") == file_path:
                entry.update({
                    "stage_id": stage_id,
                    "iteration": iteration,
                    "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                })
                updated_count += 1
                found = True
                break

        if not found:
            memory_pool.append({
                "stage_id": stage_id,
                "iteration": iteration,
                "document_path": file_path,
                "summary": f"自动索引的文档摘要（基于 {file_path}）",
                "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            })
            updated_count += 1

    # 保存更新后的记忆池
    memory_pool_file.parent.mkdir(parents=True, exist_ok=True)
    try:
        with open(memory_pool_file, 'w', encoding='utf-8') as f:
            json.dump(memory_pool, f, ensure_ascii=False, indent=2)
        print(f"已更新记忆池文件 {memory_pool_file}，更新或新增 {updated_count} 个文档记录")
    except Exception as e:
        print(f"存储记忆池文件失败: {e}")
        return False

    return True

def request_ai_suggestion(args):
    """引导用户通过 Cline 与 AI 交互获取建议，并检查阶段依赖性"""
    state_engine = PATEOASStateEngine()
    state = state_engine.get_current_state()
    stage_id = args.stage_id if args.stage_id else state['current_stage']
    
    print(f"当前阶段: {stage_id}")
    print(f"检查阶段 {stage_id} 的依赖性...")
    # 调用依赖性检查逻辑
    navigator = WorkflowNavigator()
    workflow_type = state.get('workflow_type', '完整流程')
    path = navigator.get_workflow_path(workflow_type)
    current_index = path.index(stage_id) if stage_id in path else -1
    
    state_engine = PATEOASStateEngine()
    if current_index > 0:
        previous_stage = path[current_index - 1]
        if state_engine.check_dependencies(stage_id):
            print(f"阶段 {stage_id} 依赖性检查通过，前一阶段 {previous_stage} 已完成。")
        else:
            previous_status = state.get('stage_status', {}).get(previous_stage, 'not_started')
            print(f"警告：阶段 {stage_id} 依赖性检查未通过，前一阶段 {previous_stage} 未完成（当前状态：{previous_status}）。")
            print(f"建议先完成前一阶段 {previous_stage} 的工作。")
    else:
        print(f"阶段 {stage_id} 为流程起点，无依赖性问题。")
    
    print("请通过 Cline 界面与 AI 交互，获取关于当前阶段的工具推荐和建议。")
    print("您可以描述您的任务或问题，AI 将基于当前阶段状态和流程规范提供帮助。")
    print("例如，您可以询问：'如何完成当前阶段的工作？' 或 '推荐适合当前阶段的工具命令。'")
    return True

def auto_progress(args):
    """自动评估阶段进度并提供更新建议"""
    import json
    from pathlib import Path
    
    state_engine = PATEOASStateEngine()
    state = state_engine.get_current_state()
    stage_id = args.stage_id if args.stage_id else state['current_stage']
    
    print(f"自动评估阶段 {stage_id} 的进度...")
    
    # 检查记忆池文件以获取历史数据
    memory_pool_file = Path("aceflow_result/config/memory_pool.json")
    memory_pool = []
    
    if memory_pool_file.exists():
        try:
            with open(memory_pool_file, 'r', encoding='utf-8') as f:
                memory_pool = json.load(f)
        except Exception as e:
            print(f"读取记忆池文件失败: {e}")
            memory_pool = []
    
    # 查找当前阶段的记忆记录
    current_iteration = state.get('current_iteration', 'iteration-01')
    stage_memory = None
    for entry in memory_pool:
        if entry.get("stage_id") == stage_id and entry.get("iteration") == current_iteration:
            stage_memory = entry
            break
    
    # 检查阶段文档是否存在
    stage_doc_path = Path(f"aceflow_result/iterations/{current_iteration}/{stage_id.lower()}_*.md")
    matching_files = list(stage_doc_path.parent.glob(stage_doc_path.name))
    
    if matching_files and stage_memory:
        print(f"阶段 {stage_id} 的文档存在，评估进度...")
        # 简单评估逻辑：如果文档存在且有最佳实践信息，建议进度为100%
        if stage_memory.get("best_practice") and stage_memory["best_practice"] != "文档未找到，无法提取最佳实践信息。":
            suggested_progress = 100
            reason = "文档存在且包含最佳实践信息，表明阶段工作已完成。"
        else:
            suggested_progress = 50
            reason = "文档存在但可能不完整，建议进一步完善。"
    else:
        suggested_progress = 0
        reason = "阶段文档未找到或记忆记录不存在，表明阶段工作尚未开始或未完成。"
    
    print(f"阶段 {stage_id} 的建议进度：{suggested_progress}%")
    print(f"原因：{reason}")
    print("请确认是否更新进度，或通过 Cline 界面与 AI 交互获取进一步建议。")
    return True

def main():
    parser = argparse.ArgumentParser(description="AceFlow-PATEOAS CLI 工具")
    subparsers = parser.add_subparsers(dest="command")
    
    # 验证阶段输出命令
    parser_validate = subparsers.add_parser("validate-stage-output", help="验证阶段输出产物是否完整")
    parser_validate.add_argument("--stage-id", help="指定阶段ID")
    
    # 检查依赖性命令
    parser_check = subparsers.add_parser("check-dependencies", help="检查阶段依赖性是否满足")
    parser_check.add_argument("--stage-id", help="指定阶段ID")
    
    # 回退阶段命令
    parser_revert = subparsers.add_parser("revert-stage", help="回退到指定阶段")
    parser_revert.add_argument("target_stage", help="目标阶段ID")
    
    # 复查前一阶段命令
    parser_review = subparsers.add_parser("review-previous-stage", help="复查前一阶段产物")
    
    # 生成阶段模板命令
    parser_generate = subparsers.add_parser("generate-stage-template", help="生成阶段模板文档")
    parser_generate.add_argument("--stage-id", help="指定阶段ID")
    
    # 关联输出产物命令
    parser_associate = subparsers.add_parser("associate-output", help="关联工作产物到阶段")
    parser_associate.add_argument("--stage-id", help="指定阶段ID")
    parser_associate.add_argument("output_path", help="输出产物路径")
    
    # 阶段评审命令
    parser_stage_review = subparsers.add_parser("stage-review", help="记录阶段评审结果")
    parser_stage_review.add_argument("--stage-id", help="指定阶段ID")
    parser_stage_review.add_argument("review_result", help="评审结果")
    
    # 阶段记忆摘要命令
    parser_memory_summary = subparsers.add_parser("stage-memory-summary", help="生成阶段记忆摘要")
    parser_memory_summary.add_argument("--stage-id", help="指定阶段ID")
    
    # 索引文档命令
    parser_index = subparsers.add_parser("index-documents", help="扫描指定目录下的文档并更新记忆池")
    parser_index.add_argument("--directory", help="指定扫描目录，默认为 aceflow_result/iterations/")
    
    # 请求 AI 建议命令
    parser_suggestion = subparsers.add_parser("request-ai-suggestion", help="引导用户通过 Cline 与 AI 交互获取建议")
    parser_suggestion.add_argument("--stage-id", help="指定阶段ID")
    
    # 自动进度评估命令
    parser_auto_progress = subparsers.add_parser("auto-progress", help="自动评估阶段进度并提供更新建议")
    parser_auto_progress.add_argument("--stage-id", help="指定阶段ID")
    
    # 更新状态命令（增强版）
    parser_update = subparsers.add_parser("update-status", help="更新阶段进度")
    parser_update.add_argument("stage_id", help="阶段ID")
    parser_update.add_argument("progress", type=int, help="进度百分比")
    
    args = parser.parse_args()
    
    if args.command == "validate-stage-output":
        validate_stage_output(args)
    elif args.command == "check-dependencies":
        check_dependencies(args)
    elif args.command == "revert-stage":
        revert_stage(args)
    elif args.command == "review-previous-stage":
        review_previous_stage(args)
    elif args.command == "generate-stage-template":
        generate_stage_template(args)
    elif args.command == "associate-output":
        associate_output(args)
    elif args.command == "stage-review":
        stage_review(args)
    elif args.command == "stage-memory-summary":
        stage_memory_summary(args)
    elif args.command == "index-documents":
        index_documents(args)
    elif args.command == "request-ai-suggestion":
        request_ai_suggestion(args)
    elif args.command == "auto-progress":
        auto_progress(args)
    elif args.command == "update-status":
        update_status(args)
    else:
        parser.print_help()

if __name__ == "__main__":
    main()