#!/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()