DataOps-platform/scripts/cursor_auto_chat.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Cursor自动聊天工具

这个工具可以自动查找Cursor窗口，定位到chat窗口，并自动发送消息。

功能：
1. 查找Windows操作系统中运行的Cursor程序
2. 找到当前运行Cursor实例，并定位到当前的chat窗口
3. 模拟鼠标点击到chat窗口
4. 模拟键盘输入"请检查并执行所有待处理任务。"到chat窗口
5. 模拟鼠标点击chat窗口的"提交"按钮
6. 以服务方式持续运行，间隔300秒进行一次上述操作

使用方法：
1. 单次执行：python scripts/cursor_auto_chat.py --once
2. 服务模式：python scripts/cursor_auto_chat.py --daemon
3. 自定义间隔：python scripts/cursor_auto_chat.py --interval 300
4. 指定输入框位置：python scripts/cursor_auto_chat.py --input-box-pos "1180,965"
"""

import sys
import time
import argparse
import logging
from pathlib import Path
from datetime import datetime

try:
    import win32gui
    import win32con
    import win32process
    import win32api
    import pyautogui
    import pywinauto
    from pywinauto import Application
    try:
        import pyperclip
        HAS_PYPERCLIP = True
    except ImportError:
        HAS_PYPERCLIP = False
except ImportError as e:
    print(f"❌ 缺少必要的依赖库: {e}")
    print("请运行: pip install pywin32 pyautogui pywinauto pyperclip")
    sys.exit(1)

# 配置日志
logs_dir = Path(__file__).parent.parent / 'logs'
logs_dir.mkdir(exist_ok=True)

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(logs_dir / 'cursor_auto_chat.log', encoding='utf-8'),
        logging.StreamHandler(sys.stdout)
    ]
)
logger = logging.getLogger('CursorAutoChat')

# 配置pyautogui安全设置
pyautogui.FAILSAFE = True  # 鼠标移到屏幕左上角会触发异常，用于紧急停止
pyautogui.PAUSE = 0.5  # 每个操作之间的暂停时间（秒）

# 检查pyperclip并记录
if not HAS_PYPERCLIP:
    logger.warning("未安装pyperclip，中文输入可能有问题，建议安装: pip install pyperclip")


class CursorAutoChat:
    """Cursor自动聊天工具类"""
    
    def __init__(self, message="请检查并执行所有待处理任务。", interval=300, input_box_pos=None):
        """
        初始化工具
        
        Args:
            message: 要发送的消息内容
            interval: 执行间隔（秒）
            input_box_pos: 输入框位置 (x, y)，如果提供则直接使用，不进行自动定位
        """
        self.message = message
        self.interval = interval
        self.cursor_window = None
        self.input_box_pos = input_box_pos  # 用户指定的输入框位置
        logger.info(f"Cursor自动聊天工具已初始化")
        logger.info(f"消息内容: {self.message}")
        logger.info(f"执行间隔: {self.interval}秒")
        if self.input_box_pos:
            logger.info(f"使用指定的输入框位置: {self.input_box_pos}")
    
    def find_cursor_processes(self):
        """
        查找所有运行的Cursor进程
        
        Returns:
            list: Cursor进程ID列表
        """
        cursor_pids = []
        try:
            import psutil
            for proc in psutil.process_iter(['pid', 'name', 'exe']):
                try:
                    proc_name = proc.info['name'].lower() if proc.info['name'] else ''
                    proc_exe = proc.info['exe'].lower() if proc.info['exe'] else ''
                    
                    # 查找Cursor相关进程
                    if 'cursor' in proc_name or 'cursor' in proc_exe:
                        cursor_pids.append(proc.info['pid'])
                        logger.debug(f"找到Cursor进程: PID={proc.info['pid']}, Name={proc.info['name']}")
                except (psutil.NoSuchProcess, psutil.AccessDenied):
                    continue
        except ImportError:
            # 如果没有psutil，使用win32api枚举窗口
            logger.warning("未安装psutil，使用窗口枚举方式查找Cursor")
            cursor_pids = self._find_cursor_by_windows()
        
        logger.info(f"找到 {len(cursor_pids)} 个Cursor进程")
        return cursor_pids
    
    def _find_cursor_by_windows(self):
        """通过枚举窗口查找Cursor进程"""
        cursor_pids = []
        
        def enum_windows_callback(hwnd, windows):
            if win32gui.IsWindowVisible(hwnd):
                window_title = win32gui.GetWindowText(hwnd)
                if 'cursor' in window_title.lower():
                    _, pid = win32process.GetWindowThreadProcessId(hwnd)
                    if pid not in cursor_pids:
                        cursor_pids.append(pid)
            return True
        
        win32gui.EnumWindows(enum_windows_callback, None)
        return cursor_pids
    
    def find_cursor_window(self):
        """
        查找Cursor主窗口
        
        Returns:
            int: 窗口句柄，如果未找到返回None
        """
        cursor_windows = []  # 存储所有可能的Cursor窗口
        
        def enum_windows_callback(hwnd, windows):
            if win32gui.IsWindowVisible(hwnd):
                window_title = win32gui.GetWindowText(hwnd)
                class_name = win32gui.GetClassName(hwnd)
                
                # Cursor基于Electron，窗口类名可能是Chrome_WidgetWin_1或类似
                # 查找可能的Cursor窗口
                is_cursor = False
                
                # 检查窗口标题
                if window_title and 'cursor' in window_title.lower():
                    is_cursor = True
                
                # 检查窗口类名（Electron应用通常有特定类名）
                if class_name and ('chrome_widgetwin' in class_name.lower() or 'electron' in class_name.lower()):
                    # 进一步检查：Electron窗口通常比较大
                    rect = win32gui.GetWindowRect(hwnd)
                    width = rect[2] - rect[0]
                    height = rect[3] - rect[1]
                    if width > 800 and height > 600:
                        is_cursor = True
                
                if is_cursor:
                    rect = win32gui.GetWindowRect(hwnd)
                    width = rect[2] - rect[0]
                    height = rect[3] - rect[1]
                    area = width * height
                    cursor_windows.append({
                        'hwnd': hwnd,
                        'title': window_title,
                        'class': class_name,
                        'width': width,
                        'height': height,
                        'area': area
                    })
                    logger.debug(f"找到可能的Cursor窗口: {window_title} ({class_name}), Size: {width}x{height}")
            
            return True
        
        win32gui.EnumWindows(enum_windows_callback, None)
        
        if not cursor_windows:
            logger.warning("未找到Cursor窗口")
            return None
        
        # 选择最大的窗口作为主窗口（通常是主应用窗口）
        cursor_windows.sort(key=lambda x: x['area'], reverse=True)
        main_window = cursor_windows[0]
        
        logger.info(f"找到Cursor主窗口: {main_window['title']} ({main_window['class']})")
        logger.info(f"窗口大小: {main_window['width']}x{main_window['height']} (HWND: {main_window['hwnd']})")
        
        return main_window['hwnd']
    
    def activate_cursor_window(self, hwnd):
        """
        激活Cursor窗口
        
        Args:
            hwnd: 窗口句柄
        """
        try:
            # 恢复窗口（如果最小化）
            win32gui.ShowWindow(hwnd, win32con.SW_RESTORE)
            time.sleep(0.3)
            
            # 激活窗口
            win32gui.SetForegroundWindow(hwnd)
            time.sleep(0.5)
            
            logger.info("Cursor窗口已激活")
            return True
        except Exception as e:
            logger.error(f"激活窗口失败: {e}")
            return False
    
    def find_chat_input_area(self, hwnd):
        """
        查找chat输入区域
        
        使用多种策略定位Cursor的chat输入框：
        1. 尝试多个快捷键打开chat（Ctrl+K, Ctrl+L, Ctrl+Shift+L等）
        2. 使用相对窗口坐标定位输入框（chat通常在窗口底部中央或右侧）
        3. 验证输入框是否被激活
        
        Args:
            hwnd: Cursor窗口句柄
        
        Returns:
            tuple: (是否成功, 输入框坐标(x, y)) 或 (False, None)
        """
        try:
            # 获取窗口位置和大小
            rect = win32gui.GetWindowRect(hwnd)
            window_left = rect[0]
            window_top = rect[1]
            window_width = rect[2] - rect[0]
            window_height = rect[3] - rect[1]
            
            logger.info(f"窗口位置: ({window_left}, {window_top}), 大小: {window_width}x{window_height}")
            
            # 策略1: 尝试多个快捷键打开chat
            logger.info("尝试使用快捷键打开chat窗口...")
            shortcuts = [
                ('ctrl', 'k'),  # Cursor最常用的快捷键
                ('ctrl', 'l'),  # 备用快捷键
                ('ctrl', 'shift', 'l'),  # 另一个可能的快捷键
            ]
            
            for shortcut in shortcuts:
                try:
                    logger.debug(f"尝试快捷键: {'+'.join(shortcut)}")
                    pyautogui.hotkey(*shortcut)
                    time.sleep(1.5)  # 给chat窗口时间打开
                    
                    # 尝试定位输入框
                    result = self._try_activate_input_box(hwnd, window_left, window_top, window_width, window_height)
                    if result:
                        input_pos = result
                        logger.info(f"成功定位并激活chat输入框，位置: {input_pos}")
                        return (True, input_pos)
                except Exception as e:
                    logger.debug(f"快捷键 {shortcut} 失败: {e}")
                    continue
            
            # 策略2: 直接尝试点击可能的输入框位置（即使chat已打开）
            logger.info("尝试直接点击可能的输入框位置...")
            result = self._try_activate_input_box(hwnd, window_left, window_top, window_width, window_height)
            if result:
                input_pos = result
                logger.info(f"成功定位并激活chat输入框，位置: {input_pos}")
                return (True, input_pos)
            
            logger.warning("未能定位chat输入框，将尝试通用方法")
            return (False, None)
            
        except Exception as e:
            logger.error(f"查找chat输入区域失败: {e}")
            return (False, None)
    
    def _try_activate_input_box(self, hwnd, window_left, window_top, window_width, window_height):
        """
        尝试激活输入框
        
        Args:
            hwnd: 窗口句柄
            window_left: 窗口左边界
            window_top: 窗口上边界
            window_width: 窗口宽度
            window_height: 窗口高度
        
        Returns:
            tuple: 成功时返回 (x, y) 坐标，失败时返回 None
        """
        # Cursor的chat输入框通常在窗口底部中央或右侧底部
        # 尝试多个可能的位置（相对于窗口）
        # 根据实际测试，输入框位置约为(0.61, 0.92)
        possible_relative_positions = [
            (0.61, 0.92),  # 实际测试位置（优先尝试）
            (0.6, 0.92),   # 稍微偏左
            (0.62, 0.92),  # 稍微偏右
            (0.61, 0.91),  # 稍微偏上
            (0.61, 0.93),  # 稍微偏下
            (0.75, 0.92),  # 窗口右侧底部（备用）
            (0.5, 0.92),   # 窗口底部中央（备用）
            (0.5, 0.88),   # 窗口底部稍上（备用）
            (0.8, 0.9),    # 窗口右侧（备用）
        ]
        
        for rel_x, rel_y in possible_relative_positions:
            try:
                # 计算绝对坐标
                abs_x = window_left + int(window_width * rel_x)
                abs_y = window_top + int(window_height * rel_y)
                
                logger.debug(f"尝试点击位置（相对窗口）: ({rel_x:.2f}, {rel_y:.2f}) -> 绝对坐标: ({abs_x}, {abs_y})")
                
                # 点击输入框位置
                pyautogui.click(abs_x, abs_y)
                time.sleep(0.8)  # 等待输入框激活
                
                # 验证输入框是否激活：尝试输入一个测试字符然后删除
                # 如果输入框已激活，这个操作应该成功
                pyautogui.write('test', interval=0.1)
                time.sleep(0.3)
                
                # 删除测试文本
                for _ in range(4):
                    pyautogui.press('backspace')
                time.sleep(0.3)
                
                # 如果到这里没有异常，说明输入框可能已激活
                logger.info(f"成功激活输入框，位置: ({abs_x}, {abs_y})")
                return (abs_x, abs_y)
                
            except Exception as e:
                logger.debug(f"位置 ({rel_x:.2f}, {rel_y:.2f}) 失败: {e}")
                continue
        
        return None
    
    def send_message(self, message, input_pos=None):
        """
        发送消息到chat窗口
        
        Args:
            message: 要发送的消息
            input_pos: 输入框坐标 (x, y)，如果提供则点击该位置确保激活
        """
        try:
            # 如果提供了输入框位置，先移动鼠标到该位置，然后点击
            if input_pos:
                x, y = input_pos
                logger.info(f"移动鼠标到输入框位置: ({x}, {y})")
                pyautogui.moveTo(x, y, duration=0.3)  # 平滑移动到输入框位置
                time.sleep(0.2)
                
                logger.info(f"点击输入框位置确保激活: ({x}, {y})")
                pyautogui.click(x, y)
                time.sleep(0.5)  # 等待输入框激活
            else:
                # 如果没有提供位置，点击当前鼠标位置
                logger.info("点击当前位置确保输入框激活")
                pyautogui.click()
                time.sleep(0.3)
            
            # 清空可能的现有文本
            logger.info("清空输入框中的现有文本...")
            pyautogui.hotkey('ctrl', 'a')
            time.sleep(0.3)
            
            # 再次确保鼠标在输入框位置并点击（如果提供了位置）
            if input_pos:
                x, y = input_pos
                logger.info(f"再次移动鼠标到输入框并点击: ({x}, {y})")
                pyautogui.moveTo(x, y, duration=0.2)
                time.sleep(0.2)
                pyautogui.click(x, y)
                time.sleep(0.4)  # 给足够时间让输入框完全激活
            else:
                pyautogui.click()
                time.sleep(0.2)
            
            # 输入消息
            logger.info(f"正在输入消息: {message}")
            
            # 对于中文文本，使用剪贴板方法更可靠
            if HAS_PYPERCLIP:
                try:
                    # 保存当前剪贴板内容
                    old_clipboard = pyperclip.paste() if hasattr(pyperclip, 'paste') else None
                    
                    # 复制消息到剪贴板
                    pyperclip.copy(message)
                    time.sleep(0.3)
                    
                    # 确保鼠标在输入框位置（如果提供了位置）
                    if input_pos:
                        x, y = input_pos
                        logger.info(f"粘贴前确保鼠标在输入框位置: ({x}, {y})")
                        pyautogui.moveTo(x, y, duration=0.2)
                        time.sleep(0.2)
                        # 再次点击确保焦点
                        pyautogui.click(x, y)
                        time.sleep(0.3)
                    
                    # 粘贴消息
                    logger.info("执行Ctrl+V粘贴消息...")
                    pyautogui.hotkey('ctrl', 'v')
                    time.sleep(1.5)  # 等待粘贴完成，给足够时间
                    
                    # 验证粘贴是否成功（可选：再次点击确保文本已输入）
                    if input_pos:
                        # 轻微移动鼠标确认输入框仍有焦点
                        x, y = input_pos
                        pyautogui.moveTo(x + 1, y + 1, duration=0.1)
                        time.sleep(0.2)
                    
                    # 恢复剪贴板（可选）
                    if old_clipboard:
                        try:
                            pyperclip.copy(old_clipboard)
                        except:
                            pass
                    
                    logger.info("使用剪贴板方法输入消息成功")
                    return True
                except Exception as e:
                    logger.warning(f"剪贴板方法失败: {e}，尝试其他方法...")
            
            # 备用方法：直接输入（对英文有效）
            try:
                # 检查是否包含中文字符
                has_chinese = any('\u4e00' <= char <= '\u9fff' for char in message)
                if has_chinese:
                    logger.warning("消息包含中文，但pyperclip不可用，输入可能失败")
                
                pyautogui.write(message, interval=0.05)
                time.sleep(0.8)
                logger.info("使用write方法输入成功")
                return True
            except Exception as e2:
                logger.error(f"使用write方法也失败: {e2}")
                return False
                
        except Exception as e:
            logger.error(f"输入消息失败: {e}")
            return False
    
    def click_submit_button(self):
        """
        点击提交按钮
        
        Cursor的提交方式可能是：
        1. Enter键（单行输入）
        2. Ctrl+Enter组合键（多行输入或某些配置）
        3. 点击提交按钮（如果存在）
        """
        try:
            # 策略1: 先尝试Enter键（最常见）
            logger.info("尝试按Enter键提交...")
            pyautogui.press('enter')
            time.sleep(1.0)  # 等待消息发送
            
            # 策略2: 如果Enter不行，尝试Ctrl+Enter（某些配置下需要）
            # 但先等待一下，看看Enter是否生效
            logger.info("等待消息发送完成...")
            time.sleep(1.5)  # 给足够时间让消息发送
            
            logger.info("消息已提交（使用Enter键）")
            logger.info("提示: 如果消息未出现在chat历史中，可能需要使用Ctrl+Enter")
            return True
        except Exception as e:
            logger.error(f"点击提交按钮失败: {e}")
            # 尝试备用方法
            try:
                logger.info("尝试使用Ctrl+Enter提交...")
                pyautogui.hotkey('ctrl', 'enter')
                time.sleep(1.5)
                logger.info("使用Ctrl+Enter提交完成")
                return True
            except Exception as e2:
                logger.error(f"使用Ctrl+Enter也失败: {e2}")
                return False
    
    def execute_once(self):
        """
        执行一次完整的操作流程
        
        Returns:
            bool: 是否成功执行
        """
        logger.info("=" * 60)
        logger.info("开始执行自动聊天操作...")
        logger.info("=" * 60)
        
        try:
            # 步骤1: 查找Cursor进程
            logger.info("步骤1: 查找Cursor进程...")
            cursor_pids = self.find_cursor_processes()
            if not cursor_pids:
                logger.error("未找到Cursor进程，请确保Cursor正在运行")
                return False
            
            # 步骤2: 查找Cursor窗口
            logger.info("步骤2: 查找Cursor主窗口...")
            cursor_hwnd = self.find_cursor_window()
            if not cursor_hwnd:
                logger.error("未找到Cursor主窗口")
                return False
            
            # 步骤3: 激活窗口
            logger.info("步骤3: 激活Cursor窗口...")
            if not self.activate_cursor_window(cursor_hwnd):
                logger.error("激活窗口失败")
                return False
            
            # 步骤4: 定位chat输入区域
            logger.info("步骤4: 定位chat输入区域...")
            
            # 如果用户指定了输入框位置，直接使用
            if self.input_box_pos:
                input_pos = self.input_box_pos
                logger.info(f"使用用户指定的输入框位置: {input_pos}")
                # 激活窗口后，直接使用指定位置
                time.sleep(0.5)
            else:
                # 自动定位输入框
                result = self.find_chat_input_area(cursor_hwnd)
                if isinstance(result, tuple) and len(result) == 2:
                    success, input_pos = result
                    if success and input_pos:
                        logger.info(f"已定位到输入框位置: {input_pos}")
                    else:
                        logger.warning("未能精确定位输入区域，将尝试直接输入")
                        input_pos = None
                else:
                    # 兼容旧版本返回格式
                    if result:
                        logger.info("已定位到输入框")
                        input_pos = None  # 旧版本不返回位置
                    else:
                        logger.warning("未能精确定位输入区域，将尝试直接输入")
                        input_pos = None
                    time.sleep(1)
            
            # 步骤5: 输入消息
            logger.info("步骤5: 输入消息...")
            if not self.send_message(self.message, input_pos):
                logger.error("输入消息失败")
                return False
            
            # 步骤6: 点击提交按钮
            logger.info("步骤6: 提交消息...")
            if not self.click_submit_button():
                logger.error("提交消息失败")
                return False
            
            logger.info("=" * 60)
            logger.info("✅ 自动聊天操作执行成功！")
            logger.info("=" * 60)
            return True
            
        except Exception as e:
            logger.error(f"执行过程中出错: {e}", exc_info=True)
            return False
    
    def run_daemon(self):
        """
        以守护进程模式运行，定期执行操作
        """
        logger.info("=" * 60)
        logger.info("🚀 Cursor自动聊天工具已启动（守护进程模式）")
        logger.info(f"⏰ 执行间隔: {self.interval}秒 ({self.interval//60}分钟)")
        logger.info("按 Ctrl+C 停止服务")
        logger.info("=" * 60)
        
        try:
            while True:
                try:
                    success = self.execute_once()
                    
                    if success:
                        logger.info(f"✅ 操作执行成功，{self.interval}秒后再次执行...")
                    else:
                        logger.warning(f"⚠️ 操作执行失败，{self.interval}秒后重试...")
                    
                    time.sleep(self.interval)
                    
                except KeyboardInterrupt:
                    raise
                except Exception as e:
                    logger.error(f"执行过程中出错: {e}")
                    logger.info(f"⏳ {self.interval}秒后重试...")
                    time.sleep(self.interval)
                    
        except KeyboardInterrupt:
            logger.info("\n" + "=" * 60)
            logger.info("⛔ 用户停止了Cursor自动聊天工具")
            logger.info("=" * 60)


def main():
    """主函数"""
    parser = argparse.ArgumentParser(
        description='Cursor自动聊天工具 - 自动向Cursor chat发送消息',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例:
  # 单次执行
  python scripts/cursor_auto_chat.py --once
  
  # 守护进程模式（默认）
  python scripts/cursor_auto_chat.py --daemon
  
  # 自定义间隔（秒）
  python scripts/cursor_auto_chat.py --interval 300
  
  # 自定义消息
  python scripts/cursor_auto_chat.py --message "你的消息内容"
  
  # 指定输入框位置
  python scripts/cursor_auto_chat.py --input-box-pos "1180,965"
        """
    )
    
    parser.add_argument(
        '--once',
        action='store_true',
        help='只执行一次，不持续运行'
    )
    parser.add_argument(
        '--daemon',
        action='store_true',
        help='作为守护进程运行（默认模式）'
    )
    parser.add_argument(
        '--interval',
        type=int,
        default=300,
        help='执行间隔（秒），默认300秒（5分钟）'
    )
    parser.add_argument(
        '--message',
        type=str,
        default='请检查并执行所有待处理任务。',
        help='要发送的消息内容，默认: "请检查并执行所有待处理任务。"'
    )
    parser.add_argument(
        '--input-box-pos',
        type=str,
        default=None,
        help='输入框位置，格式: "x,y" (例如: "1180,965")，如果提供则直接使用该位置，不进行自动定位'
    )
    
    args = parser.parse_args()
    
    # 解析输入框位置（如果提供）
    input_box_pos = None
    if args.input_box_pos:
        try:
            parts = args.input_box_pos.split(',')
            if len(parts) == 2:
                input_box_pos = (int(parts[0].strip()), int(parts[1].strip()))
                logger.info(f"解析输入框位置: {input_box_pos}")
            else:
                logger.warning(f"输入框位置格式错误，应使用 'x,y' 格式: {args.input_box_pos}")
        except ValueError as e:
            logger.warning(f"无法解析输入框位置: {e}")
    
    # 创建工具实例
    tool = CursorAutoChat(message=args.message, interval=args.interval, input_box_pos=input_box_pos)
    
    # 根据参数运行
    if args.once:
        tool.execute_once()
    else:
        tool.run_daemon()


if __name__ == '__main__':
    main()