DataOps-platform/scripts/auto_execute_tasks.py

#!/usr/bin/env python3
"""
自动任务执行核心调度脚本

工作流程：
1. 从 PostgreSQL 数据库 task_list 表中读取 pending 任务
2. 生成 tasks/task_execute_instructions.md 执行指令文件
3. 更新任务状态为 processing，并维护 tasks/pending_tasks.json
4. 更新 tasks/task_trigger.txt 触发器文件
5. 启动新的 Cursor Agent 并发送执行指令
6. Cursor Agent 完成任务后，更新 pending_tasks.json 状态为 completed
7. 调度脚本检测到任务完成后，同步数据库并关闭 Agent

使用方式：
  # 执行一次任务检查
  python scripts/auto_execute_tasks.py --once

  # 循环模式（每 5 分钟检查）
  python scripts/auto_execute_tasks.py --interval 300

  # 刷新触发器文件（用于现有 processing 任务）
  python scripts/auto_execute_tasks.py --refresh-trigger

  # 【推荐】使用 Agent 模式执行任务（自动打开/关闭 Agent）
  python scripts/auto_execute_tasks.py --agent-run

  # 【推荐】启动 Agent 循环模式（有任务时自动启动 Agent，完成后自动关闭）
  python scripts/auto_execute_tasks.py --chat-loop --use-agent

  # 启动传统 Chat 模式（不使用 Agent）
  python scripts/auto_execute_tasks.py --chat-loop --no-agent

  # 立即发送 Chat 消息触发 Cursor 执行
  python scripts/auto_execute_tasks.py --send-chat-now

  # 设置 Agent 超时时间（默认 3600 秒）
  python scripts/auto_execute_tasks.py --agent-run --agent-timeout 7200

  # 任务完成后不自动关闭 Agent
  python scripts/auto_execute_tasks.py --agent-run --no-auto-close

  # 禁用自动部署功能
  python scripts/auto_execute_tasks.py --chat-loop --use-agent --no-deploy

  # 立即部署指定任务ID的脚本到生产服务器
  python scripts/auto_execute_tasks.py --deploy-now 123

  # 测试到生产服务器的 SSH 连接
  python scripts/auto_execute_tasks.py --test-connection
"""

from __future__ import annotations

import argparse
import json
import logging
import time
from datetime import datetime
from pathlib import Path
from typing import Any

# ============================================================================
# 日志配置
# ============================================================================
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger("AutoExecuteTasks")

# ============================================================================
# Windows GUI 自动化依赖（可选）
# ============================================================================
HAS_CURSOR_GUI = False
HAS_PYPERCLIP = False

try:
    import pyautogui
    import win32con
    import win32gui

    pyautogui.FAILSAFE = True
    pyautogui.PAUSE = 0.5
    HAS_CURSOR_GUI = True

    try:
        import pyperclip

        HAS_PYPERCLIP = True
    except ImportError:
        pass
except ImportError:
    logger.info(
        "未安装 Windows GUI 自动化依赖（pywin32/pyautogui），"
        "将禁用自动 Cursor Chat 功能。"
    )

# ============================================================================
# 全局配置
# ============================================================================
WORKSPACE_ROOT = Path(__file__).parent.parent
TASKS_DIR = WORKSPACE_ROOT / "tasks"
PENDING_TASKS_FILE = TASKS_DIR / "pending_tasks.json"
INSTRUCTIONS_FILE = TASKS_DIR / "task_execute_instructions.md"
TRIGGER_FILE = TASKS_DIR / "task_trigger.txt"

# 生产服务器配置
PRODUCTION_SERVER = {
    "host": "192.168.3.143",
    "port": 22,
    "username": "ubuntu",
    "password": "citumxl2357",
    "script_path": "/opt/dataops-platform/datafactory/scripts",
    "workflow_path": "/opt/dataops-platform/n8n/workflows",
}

# 命令行参数控制的全局变量
ENABLE_CHAT: bool = False
CHAT_MESSAGE: str = "请阅读 tasks/task_execute_instructions.md 并执行任务。"
CHAT_INPUT_POS: tuple[int, int] | None = None
ENABLE_AUTO_DEPLOY: bool = True  # 默认启用自动部署


# ============================================================================
# 数据库操作
# ============================================================================
def get_db_connection():
    """获取数据库连接（使用 production 环境配置）"""
    try:
        import sys

        import psycopg2

        sys.path.insert(0, str(WORKSPACE_ROOT))
        from app.config.config import config

        # 强制使用 production 环境的数据库配置
        app_config = config["production"]
        db_uri = app_config.SQLALCHEMY_DATABASE_URI
        return psycopg2.connect(db_uri)

    except ImportError as e:
        logger.error(f"导入依赖失败: {e}")
        return None
    except Exception as e:
        logger.error(f"连接数据库失败: {e}")
        return None


def get_pending_tasks() -> list[dict[str, Any]]:
    """从数据库获取所有 pending 任务"""
    try:
        from psycopg2.extras import RealDictCursor

        conn = get_db_connection()
        if not conn:
            return []

        cursor = conn.cursor(cursor_factory=RealDictCursor)
        cursor.execute("""
            SELECT task_id, task_name, task_description, status,
                   code_name, code_path, create_time, create_by
            FROM task_list
            WHERE status = 'pending'
            ORDER BY create_time ASC
        """)

        tasks = cursor.fetchall()
        cursor.close()
        conn.close()

        return [dict(task) for task in tasks]

    except Exception as e:
        logger.error(f"获取 pending 任务失败: {e}")
        return []


def update_task_status(
    task_id: int,
    status: str,
    code_name: str | None = None,
    code_path: str | None = None,
) -> bool:
    """更新任务状态"""
    try:
        conn = get_db_connection()
        if not conn:
            return False

        cursor = conn.cursor()

        if code_name and code_path:
            cursor.execute(
                """
                UPDATE task_list
                SET status = %s, code_name = %s, code_path = %s,
                    update_time = CURRENT_TIMESTAMP
                WHERE task_id = %s
            """,
                (status, code_name, code_path, task_id),
            )
        else:
            cursor.execute(
                """
                UPDATE task_list
                SET status = %s, update_time = CURRENT_TIMESTAMP
                WHERE task_id = %s
            """,
                (status, task_id),
            )

        conn.commit()
        updated = cursor.rowcount > 0
        cursor.close()
        conn.close()

        if updated:
            logger.info(f"✅ 任务 {task_id} 状态已更新为: {status}")
        return updated

    except Exception as e:
        logger.error(f"更新任务状态失败: {e}")
        return False


# ============================================================================
# 触发器文件管理
# ============================================================================
def update_trigger_file(
    task_count: int = 0,
    status: str = "有待执行任务",
    task_ids: list[int] | None = None,
) -> None:
    """
    更新触发器文件 tasks/task_trigger.txt

    Cursor Rules 会检测此文件的更新时间，如果在 5 分钟内更新，
    则触发自动任务执行。
    """
    TASKS_DIR.mkdir(parents=True, exist_ok=True)

    gen_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    task_ids_str = ", ".join(str(tid) for tid in (task_ids or []))

    content = f"""CURSOR_AUTO_EXECUTE_TASK_TRIGGER
生成时间: {gen_time}
状态: {status}
待处理任务数: {task_count}
任务ID列表: [{task_ids_str}]

此文件用于触发Cursor自动执行任务。
当此文件更新时间在5分钟内时，Cursor应该检查 tasks/task_execute_instructions.md 并执行其中的任务。

【重要】Cursor AI 请立即执行以下操作：
1. 读取 tasks/task_execute_instructions.md 文件
2. 按照文件中的任务描述，依次完成每个任务
3. 完成后更新 tasks/pending_tasks.json 中对应任务的状态为 completed

"""

    with TRIGGER_FILE.open("w", encoding="utf-8") as f:
        f.write(content)

    logger.info(f"✅ 触发器文件已更新: {TRIGGER_FILE}")


def get_processing_task_ids() -> list[int]:
    """从 pending_tasks.json 获取所有 processing 状态的任务 ID"""
    if not PENDING_TASKS_FILE.exists():
        return []

    try:
        with PENDING_TASKS_FILE.open("r", encoding="utf-8") as f:
            tasks = json.load(f)
        return [
            t.get("task_id")
            for t in tasks
            if t.get("status") == "processing" and t.get("task_id")
        ]
    except Exception:
        return []


# ============================================================================
# 任务文件生成
# ============================================================================
def write_pending_tasks_json(tasks: list[dict[str, Any]]) -> None:
    """将任务列表写入 tasks/pending_tasks.json"""
    TASKS_DIR.mkdir(parents=True, exist_ok=True)

    # 读取现有任务
    existing_tasks = []
    if PENDING_TASKS_FILE.exists():
        try:
            with PENDING_TASKS_FILE.open("r", encoding="utf-8") as f:
                existing_tasks = json.load(f)
        except Exception:
            existing_tasks = []

    existing_ids = {t["task_id"] for t in existing_tasks if "task_id" in t}

    # 添加新任务
    for task in tasks:
        if task["task_id"] not in existing_ids:
            task_info = {
                "task_id": task["task_id"],
                "task_name": task["task_name"],
                "code_path": task.get("code_path", ""),
                "code_name": task.get("code_name", ""),
                "status": "processing",
                "notified_at": datetime.now().isoformat(),
                "code_file": task.get("code_file", ""),
            }
            existing_tasks.append(task_info)

    with PENDING_TASKS_FILE.open("w", encoding="utf-8") as f:
        json.dump(existing_tasks, f, indent=2, ensure_ascii=False)

    logger.info(f"✅ pending_tasks.json 已更新，任务数: {len(existing_tasks)}")


def create_execute_instructions(tasks: list[dict[str, Any]]) -> None:
    """生成任务执行指令文件 tasks/task_execute_instructions.md"""
    TASKS_DIR.mkdir(parents=True, exist_ok=True)

    with INSTRUCTIONS_FILE.open("w", encoding="utf-8") as f:
        f.write("# 🤖 Cursor 自动任务执行指令\n\n")
        f.write("**⚠️ 重要：请立即执行以下任务！**\n\n")
        gen_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        f.write(f"**生成时间**: {gen_time}\n\n")
        f.write(f"**待执行任务数量**: {len(tasks)}\n\n")

        f.write("## 📋 任务完成后的操作\n\n")
        f.write("完成每个任务后，请更新 `tasks/pending_tasks.json` 中")
        f.write("对应任务的 `status` 为 `completed`，\n")
        f.write("并填写 `code_name`（代码文件名）和 `code_path`（代码路径）。\n\n")
        f.write("调度脚本会自动将完成的任务同步到数据库。\n\n")

        f.write("## ⚠️ 任务约束要求\n\n")
        f.write("**重要约束**：完成脚本创建后，**不需要生成任务总结文件**。\n\n")
        f.write("- ❌ 不要创建任何 summary、report、总结类的文档文件\n")
        f.write("- ❌ 不要生成 task_summary.md、execution_report.md 等总结文件\n")
        f.write("- ✅ 只需创建任务要求的功能脚本文件\n")
        f.write("- ✅ 只需更新 `tasks/pending_tasks.json` 中的任务状态\n\n")

        f.write("---\n\n")

        for idx, task in enumerate(tasks, 1):
            task_id = task["task_id"]
            task_name = task["task_name"]
            task_desc = task["task_description"]

            create_time = task.get("create_time", "")
            if hasattr(create_time, "strftime"):
                create_time = create_time.strftime("%Y-%m-%d %H:%M:%S")

            f.write(f"## 🔴 任务 {idx}: {task_name}\n\n")
            f.write(f"- **任务ID**: `{task_id}`\n")
            f.write(f"- **创建时间**: {create_time}\n")
            f.write(f"- **创建者**: {task.get('create_by', 'unknown')}\n\n")
            f.write(f"### 📝 任务描述\n\n{task_desc}\n\n")
            f.write("---\n\n")

    logger.info(f"✅ 执行指令文件已创建: {INSTRUCTIONS_FILE}")


# ============================================================================
# Neo4j 独立连接（不依赖 Flask 应用上下文）
# ============================================================================
def get_neo4j_driver():
    """获取 Neo4j 驱动（独立于 Flask 应用上下文）"""
    try:
        import sys

        from neo4j import GraphDatabase

        sys.path.insert(0, str(WORKSPACE_ROOT))
        from app.config.config import config

        # 强制使用 production 环境的配置
        app_config = config["production"]
        uri = app_config.NEO4J_URI
        user = app_config.NEO4J_USER
        password = app_config.NEO4J_PASSWORD

        driver = GraphDatabase.driver(uri, auth=(user, password))
        return driver

    except ImportError as e:
        logger.error(f"导入 Neo4j 驱动失败: {e}")
        return None
    except Exception as e:
        logger.error(f"连接 Neo4j 失败: {e}")
        return None


# ============================================================================
# 状态同步
# ============================================================================
def update_dataflow_script_path(task_name: str, script_path: str) -> bool:
    """
    更新 DataFlow 节点的 script_path 字段（独立于 Flask 应用上下文）

    Args:
        task_name: 任务名称（对应 DataFlow 的 name_zh）
        script_path: Python 脚本的完整路径

    Returns:
        是否更新成功
    """
    try:
        from datetime import datetime

        driver = get_neo4j_driver()
        if not driver:
            logger.error("无法获取 Neo4j 驱动")
            return False

        query = """
        MATCH (n:DataFlow {name_zh: $name_zh})
        SET n.script_path = $script_path, n.updated_at = $updated_at
        RETURN n
        """

        updated_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

        with driver.session() as session:
            result = session.run(
                query,
                name_zh=task_name,
                script_path=script_path,
                updated_at=updated_at,
            ).single()

            driver.close()

            if result:
                logger.info(f"成功更新 DataFlow 脚本路径: {task_name} -> {script_path}")
                return True
            else:
                logger.warning(f"未找到 DataFlow 节点: {task_name}")
                return False

    except Exception as e:
        logger.error(f"更新 DataFlow script_path 失败: {e}")
        return False


def sync_completed_tasks_to_db() -> int:
    """将 pending_tasks.json 中 completed 的任务同步到数据库"""
    if not PENDING_TASKS_FILE.exists():
        return 0

    try:
        with PENDING_TASKS_FILE.open("r", encoding="utf-8") as f:
            tasks = json.load(f)
    except Exception as e:
        logger.error(f"读取 pending_tasks.json 失败: {e}")
        return 0

    if not isinstance(tasks, list):
        return 0

    updated = 0
    remaining_tasks = []

    for t in tasks:
        if t.get("status") == "completed":
            task_id = t.get("task_id")
            if not task_id:
                continue

            task_name = t.get("task_name")
            code_name = t.get("code_name")
            code_path = t.get("code_path")

            # 统一处理：code_path 始终为 "datafactory/scripts"
            # code_name 保持原样（可能是新建的脚本名或 import_resource_data.py）
            code_path = "datafactory/scripts"

            # 只处理 Python 脚本文件，跳过 JSON 等其他文件
            is_python_script = code_name and code_name.endswith(".py")

            if is_python_script:
                logger.info(f"任务 {task_id} 使用 Python 脚本: {code_path}/{code_name}")
            else:
                logger.info(
                    f"任务 {task_id} 的 code_name ({code_name}) 不是 Python 脚本，跳过 DataFlow 更新"
                )

            if update_task_status(task_id, "completed", code_name, code_path):
                updated += 1
                logger.info(f"已同步任务 {task_id} 为 completed")

                # 只有 Python 脚本才更新 DataFlow 节点的 script_path
                if task_name and is_python_script:
                    full_script_path = f"{code_path}/{code_name}"
                    if update_dataflow_script_path(task_name, full_script_path):
                        logger.info(
                            f"已更新 DataFlow 脚本路径: {task_name} -> {full_script_path}"
                        )
                    else:
                        logger.warning(f"更新 DataFlow 脚本路径失败: {task_name}")

                    # 自动部署到生产服务器（如果启用）
                    if ENABLE_AUTO_DEPLOY:
                        logger.info(f"🚀 开始自动部署任务 {task_id} 到生产服务器...")
                        if auto_deploy_completed_task(t):
                            logger.info(f"✅ 任务 {task_id} 已成功部署到生产服务器")
                        else:
                            logger.warning(f"⚠️ 任务 {task_id} 部署到生产服务器失败")
                    else:
                        logger.info(f"ℹ️ 自动部署已禁用，跳过任务 {task_id} 的部署")
            else:
                remaining_tasks.append(t)
        else:
            remaining_tasks.append(t)

    if updated > 0:
        with PENDING_TASKS_FILE.open("w", encoding="utf-8") as f:
            json.dump(remaining_tasks, f, indent=2, ensure_ascii=False)
        logger.info(f"本次共同步 {updated} 个 completed 任务到数据库")

    return updated


# ============================================================================
# 生产服务器部署功能
# ============================================================================
def get_ssh_connection():
    """获取 SSH 连接到生产服务器"""
    try:
        import paramiko  # type: ignore

        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

        logger.info(
            f"正在连接生产服务器 {PRODUCTION_SERVER['username']}@"
            f"{PRODUCTION_SERVER['host']}:{PRODUCTION_SERVER['port']}..."
        )

        ssh.connect(
            hostname=PRODUCTION_SERVER["host"],
            port=PRODUCTION_SERVER["port"],
            username=PRODUCTION_SERVER["username"],
            password=PRODUCTION_SERVER["password"],
            timeout=10,
        )

        logger.info("✅ SSH 连接成功")
        return ssh

    except ImportError:
        logger.error("未安装 paramiko 库，请运行: pip install paramiko")
        return None
    except Exception as e:
        logger.error(f"SSH 连接失败: {e}")
        return None


def test_ssh_connection() -> bool:
    """
    测试 SSH 连接到生产服务器

    Returns:
        连接是否成功
    """
    logger.info("=" * 60)
    logger.info("🔍 测试生产服务器连接")
    logger.info("=" * 60)

    ssh = get_ssh_connection()
    if not ssh:
        logger.error("❌ SSH 连接测试失败")
        return False

    try:
        # 测试执行命令
        _, stdout, _ = ssh.exec_command("echo 'Connection test successful'")
        output = stdout.read().decode().strip()
        logger.info(f"✅ 命令执行成功: {output}")

        # 检查目标目录是否存在
        _, stdout, _ = ssh.exec_command(
            f"test -d {PRODUCTION_SERVER['script_path']} && echo 'exists' || echo 'not exists'"
        )
        result = stdout.read().decode().strip()

        if result == "exists":
            logger.info(f"✅ 脚本目录存在: {PRODUCTION_SERVER['script_path']}")
        else:
            logger.warning(f"⚠️ 脚本目录不存在: {PRODUCTION_SERVER['script_path']}")
            logger.info("将在首次部署时自动创建")

        ssh.close()
        logger.info("=" * 60)
        logger.info("✅ 连接测试完成")
        logger.info("=" * 60)
        return True

    except Exception as e:
        logger.error(f"❌ 测试执行命令失败: {e}")
        ssh.close()
        return False


def deploy_script_to_production(
    local_script_path: str, remote_filename: str | None = None
) -> bool:
    """
    部署脚本文件到生产服务器

    Args:
        local_script_path: 本地脚本文件路径（相对或绝对）
        remote_filename: 远程文件名（可选，默认使用本地文件名）

    Returns:
        是否部署成功
    """
    try:
        import importlib.util

        if importlib.util.find_spec("paramiko") is None:
            logger.error("未安装 paramiko 库,请运行: pip install paramiko")
            return False

        # 转换为绝对路径
        local_path = Path(local_script_path)
        if not local_path.is_absolute():
            local_path = WORKSPACE_ROOT / local_path

        if not local_path.exists():
            logger.error(f"本地文件不存在: {local_path}")
            return False

        # 确定远程文件名
        if not remote_filename:
            remote_filename = local_path.name

        remote_path = f"{PRODUCTION_SERVER['script_path']}/{remote_filename}"

        # 建立 SSH 连接
        ssh = get_ssh_connection()
        if not ssh:
            return False

        try:
            # 创建 SFTP 客户端
            sftp = ssh.open_sftp()

            # 确保远程目录存在
            try:
                sftp.stat(PRODUCTION_SERVER["script_path"])
            except FileNotFoundError:
                logger.info(f"创建远程目录: {PRODUCTION_SERVER['script_path']}")
                _, stdout, _ = ssh.exec_command(
                    f"mkdir -p {PRODUCTION_SERVER['script_path']}"
                )
                stdout.channel.recv_exit_status()

            # 上传文件
            logger.info(f"正在上传: {local_path} -> {remote_path}")
            sftp.put(str(local_path), remote_path)

            # 设置文件权限为可执行
            sftp.chmod(remote_path, 0o755)

            logger.info(f"✅ 脚本部署成功: {remote_path}")

            sftp.close()
            ssh.close()
            return True

        except Exception as e:
            logger.error(f"文件传输失败: {e}")
            ssh.close()
            return False

    except ImportError:
        logger.error("未安装 paramiko 库，请运行: pip install paramiko")
        return False
    except Exception as e:
        logger.error(f"部署脚本失败: {e}")
        return False


def deploy_n8n_workflow_to_production(workflow_file: str) -> bool:
    """
    部署 n8n 工作流文件到生产服务器

    Args:
        workflow_file: 本地工作流 JSON 文件路径

    Returns:
        是否部署成功
    """
    try:
        import importlib.util

        if importlib.util.find_spec("paramiko") is None:
            logger.error("未安装 paramiko 库,请运行: pip install paramiko")
            return False

        # 转换为绝对路径
        local_path = Path(workflow_file)
        if not local_path.is_absolute():
            local_path = WORKSPACE_ROOT / local_path

        if not local_path.exists():
            logger.error(f"工作流文件不存在: {local_path}")
            return False

        remote_path = f"{PRODUCTION_SERVER['workflow_path']}/{local_path.name}"

        # 建立 SSH 连接
        ssh = get_ssh_connection()
        if not ssh:
            return False

        try:
            # 创建 SFTP 客户端
            sftp = ssh.open_sftp()

            # 确保远程目录存在
            try:
                sftp.stat(PRODUCTION_SERVER["workflow_path"])
            except FileNotFoundError:
                logger.info(f"创建远程目录: {PRODUCTION_SERVER['workflow_path']}")
                _, stdout, _ = ssh.exec_command(
                    f"mkdir -p {PRODUCTION_SERVER['workflow_path']}"
                )
                stdout.channel.recv_exit_status()

            # 上传工作流文件
            logger.info(f"正在上传工作流: {local_path} -> {remote_path}")
            sftp.put(str(local_path), remote_path)

            logger.info(f"✅ 工作流部署成功: {remote_path}")

            sftp.close()
            ssh.close()
            return True

        except Exception as e:
            logger.error(f"工作流传输失败: {e}")
            ssh.close()
            return False

    except ImportError:
        logger.error("未安装 paramiko 库，请运行: pip install paramiko")
        return False
    except Exception as e:
        logger.error(f"部署工作流失败: {e}")
        return False


def auto_deploy_completed_task(task_info: dict[str, Any]) -> bool:
    """
    自动部署已完成任务的脚本和工作流到生产服务器

    Args:
        task_info: 任务信息字典，包含 code_name, code_path 等

    Returns:
        是否部署成功
    """
    code_name = task_info.get("code_name")
    code_path = task_info.get("code_path")
    task_name = task_info.get("task_name", "未知任务")

    if not code_name or not code_path:
        logger.warning(f"任务 {task_name} 缺少代码文件信息，跳过部署")
        return False

    logger.info("=" * 60)
    logger.info(f"🚀 开始自动部署任务: {task_name}")
    logger.info("=" * 60)

    deploy_success = True

    # 1. 部署 Python 脚本
    if code_name.endswith(".py"):
        script_path = f"{code_path}/{code_name}"
        logger.info(f"📦 部署 Python 脚本: {script_path}")

        if deploy_script_to_production(script_path):
            logger.info(f"✅ 脚本 {code_name} 部署成功")
        else:
            logger.error(f"❌ 脚本 {code_name} 部署失败")
            deploy_success = False

    # 2. 查找并部署相关的 n8n 工作流文件
    # 工作流文件通常与脚本在同一目录或 n8n_workflows 目录
    workflow_files = []

    # 查找模式1: 与脚本同目录的工作流文件
    script_dir = WORKSPACE_ROOT / code_path
    if script_dir.exists() and script_dir.is_dir():
        for wf_file in script_dir.glob("n8n_workflow_*.json"):
            if wf_file.is_file():
                workflow_files.append(wf_file)

    # 查找模式2: datafactory/n8n_workflows 目录
    n8n_workflows_dir = WORKSPACE_ROOT / "datafactory" / "n8n_workflows"
    if n8n_workflows_dir.exists():
        for wf_file in n8n_workflows_dir.glob("*.json"):
            if wf_file.is_file() and wf_file not in workflow_files:
                workflow_files.append(wf_file)

    # 查找模式3: 根据任务名称匹配工作流文件
    if task_name and task_name != "未知任务":
        # 将任务名称转换为可能的文件名模式
        task_name_pattern = task_name.replace(" ", "_").lower()
        for wf_file in (WORKSPACE_ROOT / "datafactory").rglob(
            f"*{task_name_pattern}*.json"
        ):
            if (
                wf_file.is_file()
                and "n8n" in wf_file.name.lower()
                and wf_file not in workflow_files
            ):
                workflow_files.append(wf_file)

    if workflow_files:
        logger.info(f"📦 发现 {len(workflow_files)} 个工作流文件")
        for wf_file in workflow_files:
            logger.info(f"📦 部署工作流: {wf_file.name}")
            if deploy_n8n_workflow_to_production(str(wf_file)):
                logger.info(f"✅ 工作流 {wf_file.name} 部署成功")
            else:
                logger.error(f"❌ 工作流 {wf_file.name} 部署失败")
                deploy_success = False
    else:
        logger.info("ℹ️ 未发现相关工作流文件")

    logger.info("=" * 60)
    if deploy_success:
        logger.info(f"✅ 任务 {task_name} 部署完成")
    else:
        logger.warning(f"⚠️ 任务 {task_name} 部署过程中出现错误")
    logger.info("=" * 60)

    return deploy_success


# ============================================================================
# Cursor Chat 自动化（Agent 模式）
# ============================================================================

# Agent 会话状态
AGENT_SESSION_ACTIVE: bool = False
AGENT_START_TIME: float = 0


def get_all_cursor_windows() -> list[dict[str, Any]]:
    """获取所有 Cursor 窗口信息"""
    if not HAS_CURSOR_GUI:
        return []

    cursor_windows: list[dict[str, Any]] = []

    def enum_windows_callback(hwnd, _extra):
        if win32gui.IsWindowVisible(hwnd):
            title = win32gui.GetWindowText(hwnd) or ""
            class_name = win32gui.GetClassName(hwnd) or ""

            is_cursor = "cursor" in title.lower()
            if class_name and "chrome_widgetwin" in class_name.lower():
                is_cursor = True

            if is_cursor:
                left, top, right, bottom = win32gui.GetWindowRect(hwnd)
                area = (right - left) * (bottom - top)
                cursor_windows.append(
                    {
                        "hwnd": hwnd,
                        "title": title,
                        "class_name": class_name,
                        "area": area,
                    }
                )
        return True

    win32gui.EnumWindows(enum_windows_callback, None)
    return cursor_windows


def find_cursor_window() -> int | None:
    """查找 Cursor 主窗口句柄"""
    if not HAS_CURSOR_GUI:
        return None

    cursor_windows = get_all_cursor_windows()

    if not cursor_windows:
        logger.warning("未找到 Cursor 窗口")
        return None

    # 按面积排序，返回最大的窗口（主窗口）
    cursor_windows.sort(key=lambda x: x["area"], reverse=True)
    return cursor_windows[0]["hwnd"]


def activate_window(hwnd: int) -> bool:
    """激活指定窗口"""
    if not HAS_CURSOR_GUI:
        return False

    try:
        win32gui.ShowWindow(hwnd, win32con.SW_RESTORE)
        time.sleep(0.3)
        win32gui.SetForegroundWindow(hwnd)
        time.sleep(0.5)
        return True
    except Exception as e:
        logger.error(f"激活窗口失败: {e}")
        return False


def open_new_agent() -> bool:
    """
    在 Cursor 中打开新的 Agent 窗口

    使用快捷键 Ctrl+Shift+I 打开新的 Composer/Agent
    """
    global AGENT_SESSION_ACTIVE, AGENT_START_TIME

    if not HAS_CURSOR_GUI:
        logger.warning("当前环境不支持 Cursor GUI 自动化")
        return False

    hwnd = find_cursor_window()
    if not hwnd:
        return False

    if not activate_window(hwnd):
        return False

    try:
        # 使用 Ctrl+Shift+I 打开新的 Agent/Composer
        # 这是 Cursor 默认的快捷键
        logger.info("🚀 正在打开新的 Agent...")
        pyautogui.hotkey("ctrl", "shift", "i")
        time.sleep(2.0)  # 等待 Agent 窗口打开

        AGENT_SESSION_ACTIVE = True
        AGENT_START_TIME = time.time()
        logger.info("✅ 新的 Agent 已打开")
        return True

    except Exception as e:
        logger.error(f"打开 Agent 失败: {e}")
        return False


def close_current_agent() -> bool:
    """
    关闭当前的 Agent 会话

    使用 Escape 键关闭 Agent 或使用快捷键
    """
    global AGENT_SESSION_ACTIVE

    if not HAS_CURSOR_GUI:
        return False

    if not AGENT_SESSION_ACTIVE:
        logger.info("没有活动的 Agent 会话")
        return True

    hwnd = find_cursor_window()
    if not hwnd:
        return False

    if not activate_window(hwnd):
        return False

    try:
        logger.info("🔄 正在关闭 Agent...")

        # 方法1: 按 Escape 键关闭 Agent
        pyautogui.press("escape")
        time.sleep(0.5)

        # 再按一次确保关闭
        pyautogui.press("escape")
        time.sleep(0.3)

        AGENT_SESSION_ACTIVE = False
        logger.info("✅ Agent 已关闭")
        return True

    except Exception as e:
        logger.error(f"关闭 Agent 失败: {e}")
        return False


def type_message_to_agent(message: str) -> bool:
    """
    向 Agent 输入消息

    Args:
        message: 要发送的消息内容

    Returns:
        是否成功发送
    """
    if not HAS_CURSOR_GUI:
        return False

    try:
        # 等待 Agent 输入框获得焦点
        time.sleep(0.5)

        # 使用剪贴板粘贴（更可靠地处理中文和特殊字符）
        if HAS_PYPERCLIP:
            try:
                pyperclip.copy(message)
                pyautogui.hotkey("ctrl", "v")
                time.sleep(0.5)
            except Exception:
                # 回退到逐字符输入
                pyautogui.write(message, interval=0.03)
        else:
            pyautogui.write(message, interval=0.03)

        time.sleep(0.3)

        # 按 Enter 发送消息
        pyautogui.press("enter")
        logger.info("✅ 消息已发送到 Agent")
        return True

    except Exception as e:
        logger.error(f"发送消息到 Agent 失败: {e}")
        return False


def wait_for_agent_completion(
    timeout: int = 3600,
    check_interval: int = 30,
) -> bool:
    """
    等待 Agent 完成任务

    通过检查 pending_tasks.json 中的任务状态来判断是否完成

    Args:
        timeout: 超时时间（秒），默认 1 小时
        check_interval: 检查间隔（秒），默认 30 秒

    Returns:
        是否所有任务都已完成
    """
    start_time = time.time()
    logger.info(f"⏳ 等待 Agent 完成任务（超时: {timeout}s）...")

    while time.time() - start_time < timeout:
        processing_ids = get_processing_task_ids()

        if not processing_ids:
            elapsed = int(time.time() - start_time)
            logger.info(f"✅ 所有任务已完成！耗时: {elapsed}s")
            return True

        remaining = len(processing_ids)
        elapsed = int(time.time() - start_time)
        logger.info(
            f"⏳ 仍有 {remaining} 个任务进行中... (已等待 {elapsed}s / {timeout}s)"
        )

        time.sleep(check_interval)

    logger.warning("⚠️ 等待超时，仍有未完成的任务")
    return False


def run_agent_session(
    message: str,
    wait_completion: bool = True,
    timeout: int = 3600,
    auto_close: bool = True,
) -> bool:
    """
    运行完整的 Agent 会话

    流程：
    1. 打开新的 Agent
    2. 发送任务消息
    3. （可选）等待任务完成
    4. （可选）关闭 Agent

    Args:
        message: 要发送给 Agent 的消息
        wait_completion: 是否等待任务完成
        timeout: 等待超时时间（秒）
        auto_close: 任务完成后是否自动关闭 Agent

    Returns:
        是否成功完成会话
    """
    logger.info("=" * 50)
    logger.info("🤖 开始 Agent 会话")
    logger.info("=" * 50)

    # 1. 打开新的 Agent
    if not open_new_agent():
        logger.error("❌ 无法打开 Agent")
        return False

    # 2. 发送消息
    if not type_message_to_agent(message):
        logger.error("❌ 无法发送消息到 Agent")
        close_current_agent()
        return False

    logger.info(f"📤 已发送消息: {message[:50]}...")

    # 3. 等待任务完成
    if wait_completion:
        completed = wait_for_agent_completion(timeout=timeout)

        # 同步已完成的任务到数据库
        sync_completed_tasks_to_db()

        if completed:
            logger.info("✅ Agent 已完成所有任务")
        else:
            logger.warning("⚠️ Agent 未能在超时时间内完成所有任务")

    # 4. 关闭 Agent
    if auto_close:
        close_current_agent()

    logger.info("=" * 50)
    logger.info("🏁 Agent 会话结束")
    logger.info("=" * 50)

    return True


def send_chat_message(message: str, input_pos: tuple[int, int] | None) -> bool:
    """
    在 Cursor Chat 中发送消息（传统方式，保留向后兼容）

    推荐使用 run_agent_session() 替代此函数
    """
    if not HAS_CURSOR_GUI:
        logger.warning("当前环境不支持 Cursor GUI 自动化")
        return False

    hwnd = find_cursor_window()
    if not hwnd:
        return False

    if not activate_window(hwnd):
        return False

    # 点击输入框或使用快捷键
    if input_pos:
        x, y = input_pos
        pyautogui.click(x, y)
        time.sleep(0.4)
    else:
        pyautogui.hotkey("ctrl", "l")
        time.sleep(1.0)

    pyautogui.hotkey("ctrl", "a")
    time.sleep(0.2)

    # 输入消息
    if HAS_PYPERCLIP:
        try:
            pyperclip.copy(message)
            pyautogui.hotkey("ctrl", "v")
            time.sleep(0.5)
        except Exception:
            pyautogui.write(message, interval=0.03)
    else:
        pyautogui.write(message, interval=0.03)

    time.sleep(0.3)
    pyautogui.press("enter")
    logger.info("✅ 消息已发送到 Cursor Chat")
    return True


def send_chat_for_tasks(force: bool = False, use_agent: bool = True) -> bool:
    """
    向 Cursor Chat 发送任务提醒

    Args:
        force: 强制发送，忽略 ENABLE_CHAT 设置
        use_agent: 是否使用新 Agent 模式（推荐）

    Returns:
        是否成功发送
    """
    if not force and not ENABLE_CHAT:
        return False

    if not HAS_CURSOR_GUI:
        logger.warning("未安装 GUI 自动化依赖，无法发送 Chat 消息")
        return False

    processing_ids = get_processing_task_ids()
    if not processing_ids:
        logger.info("没有 processing 任务，跳过发送")
        return False

    logger.info(f"📤 发送任务提醒（{len(processing_ids)} 个任务）...")

    if use_agent:
        # 使用新的 Agent 模式
        return run_agent_session(
            message=CHAT_MESSAGE,
            wait_completion=False,  # 在循环模式中不阻塞等待
            auto_close=False,  # 保持 Agent 打开，让其完成任务
        )
    else:
        # 传统模式
        return send_chat_message(CHAT_MESSAGE, CHAT_INPUT_POS)


def chat_trigger_loop(
    interval: int = 60,
    use_agent: bool = True,
    auto_close_on_complete: bool = True,
) -> None:
    """
    循环模式：定期检查 processing 任务并发送 Chat 消息

    Args:
        interval: 检查间隔（秒），默认 60 秒
        use_agent: 是否使用新 Agent 模式
        auto_close_on_complete: 任务完成后是否自动关闭 Agent
    """
    global AGENT_SESSION_ACTIVE

    logger.info("=" * 60)
    logger.info("🚀 Cursor Chat 自动触发服务已启动")
    logger.info(f"⏰ 检查间隔: {interval} 秒")
    logger.info(f"🤖 Agent 模式: {'已启用' if use_agent else '未启用'}")
    logger.info(f"📝 发送消息: {CHAT_MESSAGE}")
    logger.info("按 Ctrl+C 停止服务")
    logger.info("=" * 60)

    last_sent_time = 0
    min_send_interval = 120  # 最小发送间隔（秒），避免频繁打扰

    try:
        while True:
            try:
                # 1. 先同步已完成任务
                sync_completed_tasks_to_db()

                # 2. 从数据库拉取 pending 任务并转为 processing
                pending_tasks = get_pending_tasks()
                if pending_tasks:
                    logger.info(f"📥 从数据库发现 {len(pending_tasks)} 个 pending 任务")
                    # 更新任务状态为 processing
                    for task in pending_tasks:
                        update_task_status(task["task_id"], "processing")
                    # 写入 pending_tasks.json
                    write_pending_tasks_json(pending_tasks)
                    # 生成执行指令文件
                    create_execute_instructions(pending_tasks)
                    logger.info("✅ 已将 pending 任务转为 processing")

                # 3. 检查是否有 processing 任务
                processing_ids = get_processing_task_ids()

                if processing_ids:
                    current_time = time.time()
                    time_since_last = current_time - last_sent_time

                    # 如果有活动的 Agent 会话，不需要重新发送
                    if AGENT_SESSION_ACTIVE:
                        logger.info(
                            f"🤖 Agent 正在执行中，剩余 {len(processing_ids)} 个任务"
                        )
                    elif time_since_last >= min_send_interval:
                        logger.info(f"📋 发现 {len(processing_ids)} 个待处理任务")

                        # 更新触发器文件
                        update_trigger_file(
                            task_count=len(processing_ids),
                            status="有待执行任务",
                            task_ids=processing_ids,
                        )

                        # 发送 Chat 消息（启动 Agent）
                        if use_agent:
                            if open_new_agent():
                                if type_message_to_agent(CHAT_MESSAGE):
                                    last_sent_time = current_time
                                    logger.info("✅ 已启动 Agent 并发送执行提醒")
                                else:
                                    logger.warning("⚠️ 发送消息失败")
                                    close_current_agent()
                            else:
                                logger.warning("⚠️ 启动 Agent 失败")
                        else:
                            if send_chat_message(CHAT_MESSAGE, CHAT_INPUT_POS):
                                last_sent_time = current_time
                                logger.info("✅ 已发送执行提醒")
                            else:
                                logger.warning("⚠️ 发送失败，将在下次重试")
                    else:
                        remaining = int(min_send_interval - time_since_last)
                        logger.info(
                            f"⏳ 距离上次发送不足 {min_send_interval}s，"
                            f"还需等待 {remaining}s"
                        )
                else:
                    # 没有待处理任务
                    if AGENT_SESSION_ACTIVE and auto_close_on_complete:
                        logger.info("✅ 所有任务已完成，正在关闭 Agent...")
                        close_current_agent()
                        logger.info("✅ Agent 已关闭")
                    else:
                        logger.info("✅ 没有待处理任务")

                logger.info(f"⏳ {interval} 秒后再次检查...")
                time.sleep(interval)

            except KeyboardInterrupt:
                raise
            except Exception as e:
                logger.error(f"❌ 执行出错: {e}")
                time.sleep(interval)

    except KeyboardInterrupt:
        # 退出时关闭 Agent
        if AGENT_SESSION_ACTIVE:
            logger.info("正在关闭 Agent...")
            close_current_agent()
        logger.info("\n⛔ 服务已停止")


# ============================================================================
# 主执行流程
# ============================================================================
def auto_execute_tasks_once() -> int:
    """执行一次任务检查和处理"""
    # 1. 先同步已完成任务到数据库
    sync_completed_tasks_to_db()

    # 2. 获取 pending 任务
    logger.info("🔍 检查 pending 任务...")
    tasks = get_pending_tasks()

    # 3. 检查是否有现有的 processing 任务
    existing_processing_ids = get_processing_task_ids()

    if not tasks and not existing_processing_ids:
        logger.info("✅ 没有 pending 或 processing 任务")
        # 更新触发器文件为"已完成"状态
        update_trigger_file(0, "所有任务已完成", [])
        return 0

    if tasks:
        logger.info(f"📋 找到 {len(tasks)} 个 pending 任务")

        # 4. 更新任务状态为 processing
        for task in tasks:
            update_task_status(task["task_id"], "processing")

        # 5. 写入 pending_tasks.json
        write_pending_tasks_json(tasks)

        # 6. 生成执行指令文件
        create_execute_instructions(tasks)

    # 7. 更新触发器文件（关键：触发 Cursor 自动执行）
    all_processing_ids = get_processing_task_ids()
    if all_processing_ids:
        update_trigger_file(
            task_count=len(all_processing_ids),
            status="有待执行任务",
            task_ids=all_processing_ids,
        )
        logger.info(
            f"🔔 已更新触发器，等待 Cursor 执行 {len(all_processing_ids)} 个任务"
        )

    return len(tasks)


def auto_execute_tasks_loop(interval: int = 300) -> None:
    """循环执行任务检查"""
    logger.info("=" * 60)
    logger.info("🚀 自动任务执行服务已启动")
    logger.info(f"⏰ 检查间隔: {interval} 秒")
    logger.info(f"💬 自动 Chat: {'已启用' if ENABLE_CHAT else '未启用'}")
    logger.info("按 Ctrl+C 停止服务")
    logger.info("=" * 60)

    try:
        while True:
            try:
                count = auto_execute_tasks_once()
                if count > 0:
                    send_chat_for_tasks()
                    logger.info(f"✅ 已处理 {count} 个任务")

                logger.info(f"⏳ {interval} 秒后再次检查...")
                time.sleep(interval)

            except KeyboardInterrupt:
                raise
            except Exception as e:
                logger.error(f"❌ 执行出错: {e}")
                time.sleep(interval)

    except KeyboardInterrupt:
        logger.info("\n⛔ 服务已停止")


def refresh_trigger_only() -> None:
    """仅刷新触发器文件（用于现有 processing 任务）"""
    processing_ids = get_processing_task_ids()

    if not processing_ids:
        logger.info("没有 processing 状态的任务")
        update_trigger_file(0, "所有任务已完成", [])
        return

    logger.info(f"📋 发现 {len(processing_ids)} 个 processing 任务")

    # 重新生成指令文件（从 pending_tasks.json 读取）
    if PENDING_TASKS_FILE.exists():
        try:
            with PENDING_TASKS_FILE.open("r", encoding="utf-8") as f:
                all_tasks = json.load(f)

            processing_tasks = [t for t in all_tasks if t.get("status") == "processing"]

            if processing_tasks:
                # 重新生成执行指令文件
                create_execute_instructions_from_json(processing_tasks)
        except Exception as e:
            logger.error(f"读取任务文件失败: {e}")

    # 更新触发器文件
    update_trigger_file(
        task_count=len(processing_ids),
        status="有待执行任务",
        task_ids=processing_ids,
    )

    logger.info("✅ 触发器已刷新，等待 Cursor 执行任务")


def create_execute_instructions_from_json(tasks: list[dict[str, Any]]) -> None:
    """从 pending_tasks.json 格式的任务列表生成执行指令文件"""
    TASKS_DIR.mkdir(parents=True, exist_ok=True)

    with INSTRUCTIONS_FILE.open("w", encoding="utf-8") as f:
        f.write("# 🤖 Cursor 自动任务执行指令\n\n")
        f.write("**⚠️ 重要：请立即执行以下任务！**\n\n")
        gen_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        f.write(f"**生成时间**: {gen_time}\n\n")
        f.write(f"**待执行任务数量**: {len(tasks)}\n\n")

        f.write("## 📋 任务完成后的操作\n\n")
        f.write("完成每个任务后，请更新 `tasks/pending_tasks.json` 中")
        f.write("对应任务的 `status` 为 `completed`，\n")
        f.write("并填写 `code_name`（代码文件名）和 `code_path`（代码路径）。\n\n")
        f.write("调度脚本会自动将完成的任务同步到数据库。\n\n")

        f.write("## ⚠️ 任务约束要求\n\n")
        f.write("**重要约束**：完成脚本创建后，**不需要生成任务总结文件**。\n\n")
        f.write("- ❌ 不要创建任何 summary、report、总结类的文档文件\n")
        f.write("- ❌ 不要生成 task_summary.md、execution_report.md 等总结文件\n")
        f.write("- ✅ 只需创建任务要求的功能脚本文件\n")
        f.write("- ✅ 只需更新 `tasks/pending_tasks.json` 中的任务状态\n\n")

        f.write("---\n\n")

        for idx, task in enumerate(tasks, 1):
            task_id = task.get("task_id", "N/A")
            task_name = task.get("task_name", "未命名任务")
            task_desc = task.get("task_description", "无描述")
            notified_at = task.get("notified_at", "")

            f.write(f"## 🔴 任务 {idx}: {task_name}\n\n")
            f.write(f"- **任务ID**: `{task_id}`\n")
            f.write(f"- **通知时间**: {notified_at}\n")
            f.write(f"- **代码路径**: {task.get('code_path', 'N/A')}\n\n")
            f.write(f"### 📝 任务描述\n\n{task_desc}\n\n")
            f.write("---\n\n")

    logger.info(f"✅ 执行指令文件已更新: {INSTRUCTIONS_FILE}")


def main() -> None:
    """主函数"""
    parser = argparse.ArgumentParser(
        description="自动任务执行调度脚本",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例:
  # 执行一次任务检查
  python scripts/auto_execute_tasks.py --once

  # 使用 Agent 模式执行任务（自动部署）
  python scripts/auto_execute_tasks.py --agent-run

  # 启动 Agent 循环模式（推荐，自动部署）
  python scripts/auto_execute_tasks.py --chat-loop --use-agent

  # 禁用自动部署功能
  python scripts/auto_execute_tasks.py --chat-loop --use-agent --no-deploy

  # 立即部署指定任务到生产服务器
  python scripts/auto_execute_tasks.py --deploy-now 123

  # 测试生产服务器连接
  python scripts/auto_execute_tasks.py --test-connection
        """,
    )
    parser.add_argument("--once", action="store_true", help="只执行一次")
    parser.add_argument("--interval", type=int, default=300, help="检查间隔（秒）")
    parser.add_argument(
        "--enable-chat", action="store_true", help="启用自动 Cursor Chat"
    )
    parser.add_argument("--chat-input-pos", type=str, help='Chat 输入框位置 "x,y"')
    parser.add_argument(
        "--chat-message",
        type=str,
        default="请阅读 tasks/task_execute_instructions.md 并执行任务。",
        help="发送到 Chat 的消息",
    )
    parser.add_argument(
        "--refresh-trigger",
        action="store_true",
        help="仅刷新触发器文件（用于现有 processing 任务）",
    )
    parser.add_argument(
        "--chat-loop",
        action="store_true",
        help="启动 Chat 自动触发循环（定期发送执行消息）",
    )
    parser.add_argument(
        "--chat-interval",
        type=int,
        default=60,
        help="Chat 触发循环的检查间隔（秒），默认 60",
    )
    parser.add_argument(
        "--send-chat-now",
        action="store_true",
        help="立即发送一次 Chat 消息（用于手动触发）",
    )
    # Agent 模式相关参数
    parser.add_argument(
        "--use-agent",
        action="store_true",
        default=True,
        help="使用新 Agent 模式（默认启用）",
    )
    parser.add_argument(
        "--no-agent",
        action="store_true",
        help="禁用 Agent 模式，使用传统 Chat 方式",
    )
    parser.add_argument(
        "--agent-run",
        action="store_true",
        help="立即启动 Agent 会话执行任务",
    )
    parser.add_argument(
        "--agent-timeout",
        type=int,
        default=3600,
        help="Agent 等待任务完成的超时时间（秒），默认 3600",
    )
    parser.add_argument(
        "--no-auto-close",
        action="store_true",
        help="任务完成后不自动关闭 Agent",
    )
    # 自动部署相关参数
    parser.add_argument(
        "--enable-deploy",
        action="store_true",
        default=True,
        help="启用自动部署到生产服务器（默认启用）",
    )
    parser.add_argument(
        "--no-deploy",
        action="store_true",
        help="禁用自动部署功能",
    )
    parser.add_argument(
        "--deploy-now",
        type=str,
        metavar="TASK_ID",
        help="立即部署指定任务ID的脚本到生产服务器",
    )
    parser.add_argument(
        "--test-connection",
        action="store_true",
        help="测试到生产服务器的 SSH 连接",
    )

    args = parser.parse_args()

    global ENABLE_CHAT, CHAT_INPUT_POS, CHAT_MESSAGE, ENABLE_AUTO_DEPLOY
    ENABLE_CHAT = bool(args.enable_chat)
    CHAT_MESSAGE = args.chat_message
    ENABLE_AUTO_DEPLOY = args.enable_deploy and not args.no_deploy

    # 确定是否使用 Agent 模式
    use_agent = args.use_agent and not args.no_agent
    auto_close = not args.no_auto_close

    if args.chat_input_pos:
        try:
            x, y = args.chat_input_pos.split(",")
            CHAT_INPUT_POS = (int(x.strip()), int(y.strip()))
        except Exception:
            pass

    # 测试 SSH 连接
    if args.test_connection:
        if test_ssh_connection():
            logger.info("✅ 连接测试成功")
        else:
            logger.error("❌ 连接测试失败")
        return

    # 立即部署指定任务
    if args.deploy_now:
        try:
            task_id = int(args.deploy_now)
            logger.info(f"🚀 开始部署任务 {task_id}...")

            # 从 pending_tasks.json 查找任务信息
            if PENDING_TASKS_FILE.exists():
                with PENDING_TASKS_FILE.open("r", encoding="utf-8") as f:
                    tasks = json.load(f)

                task_found = None
                for t in tasks:
                    if t.get("task_id") == task_id:
                        task_found = t
                        break

                if task_found:
                    if auto_deploy_completed_task(task_found):
                        logger.info(f"✅ 任务 {task_id} 部署成功")
                    else:
                        logger.error(f"❌ 任务 {task_id} 部署失败")
                else:
                    logger.error(f"未找到任务 {task_id}")
            else:
                logger.error("pending_tasks.json 文件不存在")

        except ValueError:
            logger.error(f"无效的任务ID: {args.deploy_now}")
        return

    # 仅刷新触发器
    if args.refresh_trigger:
        refresh_trigger_only()
        return

    # 立即启动 Agent 会话
    if args.agent_run:
        processing_ids = get_processing_task_ids()
        if not processing_ids:
            # 先检查是否有 pending 任务
            auto_execute_tasks_once()
            processing_ids = get_processing_task_ids()

        if not processing_ids:
            logger.info("没有待执行的任务")
            return

        logger.info(f"🚀 启动 Agent 执行 {len(processing_ids)} 个任务...")
        success = run_agent_session(
            message=CHAT_MESSAGE,
            wait_completion=True,
            timeout=args.agent_timeout,
            auto_close=auto_close,
        )

        if success:
            logger.info("✅ Agent 会话完成")
        else:
            logger.error("❌ Agent 会话失败")
        return

    # 立即发送一次 Chat 消息
    if args.send_chat_now:
        if use_agent:
            # 使用 Agent 模式
            processing_ids = get_processing_task_ids()
            if not processing_ids:
                auto_execute_tasks_once()
                processing_ids = get_processing_task_ids()

            if processing_ids:
                if open_new_agent():
                    if type_message_to_agent(CHAT_MESSAGE):
                        logger.info("✅ Agent 已启动并发送消息")
                    else:
                        logger.error("❌ 发送消息失败")
                else:
                    logger.error("❌ 启动 Agent 失败")
            else:
                logger.info("没有待执行的任务")
        else:
            if send_chat_for_tasks(force=True, use_agent=False):
                logger.info("✅ 消息已发送")
            else:
                logger.error("❌ 发送失败")
        return

    # Chat 自动触发循环模式
    if args.chat_loop:
        chat_trigger_loop(
            interval=args.chat_interval,
            use_agent=use_agent,
            auto_close_on_complete=auto_close,
        )
        return

    if args.once:
        count = auto_execute_tasks_once()
        if count > 0:
            send_chat_for_tasks(use_agent=use_agent)
        logger.info(f"✅ 完成！处理了 {count} 个任务")
    else:
        auto_execute_tasks_loop(interval=args.interval)


if __name__ == "__main__":
    main()