#!/usr/bin/env python3 """ Cursor任务执行触发器 这个脚本会读取pending_tasks.json，并以明确格式输出任务信息， 供Cursor识别并自动执行。 支持两种模式： 1. 单次执行：python scripts/trigger_cursor_execution.py --once 2. 定期执行：python scripts/trigger_cursor_execution.py --interval 300 定期执行模式下，脚本会： - 定期检查processing任务 - 自动生成执行指令文件（.cursor/task_execute_instructions.md） - 输出任务信息供Cursor识别 """ import json import sys import time import argparse import logging from pathlib import Path from datetime import datetime # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) logger = logging.getLogger('CursorTaskTrigger') def get_task_file_path(task, workspace): """获取任务文件路径""" import os import re task_file = task.get('task_file') if not task_file and task.get('code_path'): # 从任务名称生成文件名 task_name = task['task_name'] safe_name = re.sub(r'[^\w\u4e00-\u9fff]+', '_', task_name) safe_name = re.sub(r'_+', '_', safe_name).strip('_') code_path = task.get('code_path', 'app/core/data_flow') task_file = os.path.join(workspace, code_path, f"{safe_name}.py") if not os.path.exists(task_file): task_file = None return task_file def create_execute_instructions_file(workspace, processing_tasks): """ 创建任务执行指令文件，供Cursor自动读取 Args: workspace: 工作区路径 processing_tasks: 待处理任务列表 """ instructions_file = workspace / '.cursor' / 'task_execute_instructions.md' instructions_file.parent.mkdir(parents=True, exist_ok=True) with open(instructions_file, 'w', encoding='utf-8') as f: f.write(f"# 🤖 Cursor自动任务执行指令\n\n") f.write(f"**⚠️ 重要：Cursor AI请立即执行此文件中的任务！**\n\n") f.write(f"**生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n") f.write(f"**待执行任务数量**: {len(processing_tasks)}\n\n") f.write(f"**执行状态**: ⏳ 等待执行\n\n") f.write("---\n\n") f.write(f"## 📋 执行指令\n\n") f.write(f"**Cursor AI：请按照以下步骤执行任务：**\n\n") f.write(f"1. 依次处理每个任务\n") f.write(f"2. 打开任务文件并实现功能\n") f.write(f"3. 完成后调用MCP工具更新状态\n\n") f.write("---\n\n") for idx, task in enumerate(processing_tasks, 1): task_file = get_task_file_path(task, workspace) f.write(f"## 🔴 任务 {idx}/{len(processing_tasks)}: [{task['task_id']}] {task['task_name']}\n\n") f.write(f"**执行状态**: ⏳ 待执行\n\n") f.write(f"**任务文件**: `{task_file or '未创建'}`\n\n") f.write(f"### 📝 任务描述\n\n{task['task_description']}\n\n") f.write(f"### ✅ 执行步骤（Cursor AI请按此执行）\n\n") f.write(f"#### 步骤1：打开任务文件\n") f.write(f"- 文件路径: `{task_file or 'N/A'}`\n") f.write(f"- 如果文件不存在，需要先创建\n\n") f.write(f"#### 步骤2：实现功能\n") f.write(f"- 仔细阅读上述任务描述\n") f.write(f"- 根据描述实现完整的功能\n") f.write(f"- 确保代码符合项目规范（PEP 8）\n") f.write(f"- 添加必要的错误处理和日志记录\n\n") f.write(f"#### 步骤3：验证代码\n") f.write(f"- 确保代码可以正常运行\n") f.write(f"- 检查代码逻辑是否正确\n\n") f.write(f"#### 步骤4：更新任务状态\n") f.write(f"- 调用MCP工具 `update_task_status`\n") f.write(f"- 使用以下参数:\n\n") f.write(f"```json\n") f.write(f"{{\n") f.write(f" \"tool\": \"update_task_status\",\n") f.write(f" \"task_id\": {task['task_id']},\n") if task_file: import os file_name = os.path.basename(task_file) file_dir = os.path.dirname(task_file).replace(str(workspace), '').strip('\\').strip('/').replace('\\', '/') f.write(f" \"code_name\": \"{file_name}\",\n") f.write(f" \"code_path\": \"{file_dir}\",\n") f.write(f" \"status\": \"completed\"\n") f.write(f"}}\n") f.write(f"```\n\n") f.write(f"**⚠️ 重要：必须完成上述所有步骤！**\n\n") f.write("---\n\n") logger.info(f"✅ 任务执行指令文件已创建: {instructions_file}") # 创建触发器文件，更新时间戳 trigger_file = workspace / '.cursor' / 'task_trigger.txt' with open(trigger_file, 'w', encoding='utf-8') as f: f.write(f"CURSOR_AUTO_EXECUTE_TASK_TRIGGER\n") f.write(f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") f.write(f"状态: 有 {len(processing_tasks)} 个任务待执行\n\n") f.write(f"此文件用于触发Cursor自动执行任务。\n") f.write(f"当此文件更新时间在5分钟内时，Cursor应该检查 .cursor/task_execute_instructions.md 并执行其中的任务。\n") logger.info(f"✅ 触发器文件已更新: {trigger_file}") def execute_tasks_once(): """ 执行一次任务检查和触发 Returns: 找到的任务数量 """ workspace = Path(__file__).parent.parent pending_tasks_file = workspace / '.cursor' / 'pending_tasks.json' if not pending_tasks_file.exists(): logger.info("⚠️ 没有找到pending任务文件") return 0 try: with open(pending_tasks_file, 'r', encoding='utf-8') as f: pending_tasks = json.load(f) if not pending_tasks: logger.info("✅ 没有pending任务") return 0 # 处理所有任务（如果没有status字段，默认为processing） # 只处理状态为processing或没有status的任务 processing_tasks = [ t for t in pending_tasks if t.get('status', 'processing') == 'processing' or 'status' not in t ] if not processing_tasks: logger.info("✅ 没有processing状态的任务") return 0 logger.info(f"📋 找到 {len(processing_tasks)} 个processing任务，需要执行") # 创建执行指令文件 create_execute_instructions_file(workspace, processing_tasks) # 输出任务信息供Cursor识别 print("\n" + "=" * 80) print(f"🤖 [CURSOR-AUTO-EXECUTE] 发现 {len(processing_tasks)} 个待执行任务") print("=" * 80 + "\n") for task in processing_tasks: task_file = get_task_file_path(task, workspace) print(f"\n🤖 [CURSOR-EXECUTE-TASK]") print(f"任务ID: {task['task_id']}") print(f"任务名称: {task['task_name']}") print(f"任务文件: {task_file or '未创建'}") print(f"\n任务描述:\n{task['task_description']}") print(f"\n执行指令:") print(f"1. 打开任务文件: {task_file or task.get('code_path', 'app/core/data_flow') + '/' + task['task_name'] + '.py'}") print(f"2. 根据任务描述实现功能") print(f"3. 调用MCP工具更新状态:") print(f" 工具: update_task_status") print(f" 参数: {{") print(f" \"task_id\": {task['task_id']},") if task_file: import os file_path = task_file file_name = os.path.basename(file_path) file_dir = os.path.dirname(file_path).replace(str(workspace), '').strip('\\').strip('/').replace('\\', '/') print(f" \"code_name\": \"{file_name}\",") print(f" \"code_path\": \"{file_dir}\",") print(f" \"status\": \"completed\"") print(f" }}") print(f"\n🔚 [END-CURSOR-EXECUTE-TASK]\n") print("=" * 80) print(f"\n💡 提示：任务执行指令已保存到 .cursor/task_execute_instructions.md") print(f"💡 Cursor可以自动读取此文件并执行任务\n") return len(processing_tasks) except Exception as e: logger.error(f"❌ 读取任务文件失败: {e}") return 0 def execute_tasks_loop(interval=300): """ 循环执行任务检查和触发 Args: interval: 检查间隔（秒），默认300秒（5分钟） """ logger.info("=" * 80) logger.info("🚀 Cursor任务执行触发器已启动（定期模式）") logger.info(f"⏰ 检查间隔: {interval}秒 ({interval//60}分钟)") logger.info("按 Ctrl+C 停止服务") logger.info("=" * 80 + "\n") try: while True: try: count = execute_tasks_once() if count > 0: logger.info(f"\n✅ 已触发 {count} 个任务执行") logger.info(f"💡 任务指令已保存到 .cursor/task_execute_instructions.md") logger.info(f"💡 Cursor将自动读取并执行任务\n") logger.info(f"\n⏳ 下次检查时间: {interval}秒后...") time.sleep(interval) except KeyboardInterrupt: raise except Exception as e: logger.error(f"❌ 执行出错: {e}") logger.info(f"⏳ {interval}秒后重试...") time.sleep(interval) except KeyboardInterrupt: logger.info("\n" + "=" * 80) logger.info("⛔ 用户停止了Cursor任务执行触发器") logger.info("=" * 80) def main(): """ 主函数 """ parser = argparse.ArgumentParser( description='Cursor任务执行触发器 - 定期检查并触发任务执行' ) parser.add_argument( '--once', action='store_true', help='只执行一次检查，不持续运行' ) parser.add_argument( '--interval', type=int, default=300, help='检查间隔（秒），默认300秒（5分钟）' ) args = parser.parse_args() if args.once: # 执行一次 count = execute_tasks_once() logger.info(f"\n✅ 完成！找到 {count} 个待执行任务") else: # 循环执行 execute_tasks_loop(interval=args.interval) if __name__ == '__main__': main()