DataOps-platform/scripts/trigger_cursor_execution.py

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