import tkinter as tk from tkinter import ttk, messagebox import time import threading import random import win32gui import win32con import win32api import ctypes from ctypes import wintypes import platform import os import sys import json import websocket # ========== 幽灵键鼠加载 ========== def load_ghost_key_mouse(): """加载幽灵键鼠DLL""" script_dir = os.path.dirname(os.path.abspath(__file__)) # 根据系统架构选择DLL if platform.architecture()[0] == "64bit": dll_path = os.path.join(script_dir, "gbild64.dll") else: dll_path = os.path.join(script_dir, "gbild32.dll") if not os.path.exists(dll_path): # 如果当前目录没有，尝试从exe所在目录加载 if getattr(sys, 'frozen', False): script_dir = os.path.dirname(sys.executable) if platform.architecture()[0] == "64bit": dll_path = os.path.join(script_dir, "gbild64.dll") else: dll_path = os.path.join(script_dir, "gbild32.dll") try: dll = ctypes.windll.LoadLibrary(dll_path) print(f"幽灵键鼠DLL加载成功: {dll_path}") return dll except Exception as e: print(f"幽灵键鼠DLL加载失败: {e}") return None # 加载DLL ghost_dll = load_ghost_key_mouse() if ghost_dll: # 设置接口返回值类型 ghost_dll.getmodel.restype = ctypes.c_char_p ghost_dll.getserialnumber.restype = ctypes.c_char_p ghost_dll.getproductiondate.restype = ctypes.c_char_p ghost_dll.getfirmwareversion.restype = ctypes.c_char_p ghost_dll.getclientscreenresolution.restype = ctypes.c_char_p ghost_dll.readstring.restype = ctypes.c_char_p ghost_dll.encryptstring.restype = ctypes.c_char_p ghost_dll.decryptstring.restype = ctypes.c_char_p ghost_dll.getproductname.restype = ctypes.c_char_p ghost_dll.sdktype.restype = ctypes.c_char_p ghost_dll.sdkversion.restype = ctypes.c_char_p # ================ 设备操作 ================ def opendevice(index=0): """打开设备（根据设备序号）""" return ghost_dll.opendevice(index) def closedevice(): """关闭设备""" return ghost_dll.closedevice() def isconnected(): """检查设备是否连接""" return ghost_dll.isconnected() # ================ 鼠标操作 ================ def movemouseto(x, y): """移动鼠标到指定坐标""" return ghost_dll.movemouseto(x, y) def pressandreleasemousebutton(mbtn): """按下并释放鼠标键 (1:左键, 2:右键, 3:中键)""" return ghost_dll.pressandreleasemousebutton(mbtn) def pressmousebutton(mbtn): """按下鼠标键""" return ghost_dll.pressmousebutton(mbtn) def releasemousebutton(mbtn): """释放鼠标键""" return ghost_dll.releasemousebutton(mbtn) def getmousex(): """获取鼠标当前X坐标""" return ghost_dll.getmousex() def getmousey(): """获取鼠标当前Y坐标""" return ghost_dll.getmousey() def setmousemovementdelay(maxd, mind): """设置鼠标移动延时""" return ghost_dll.setmousemovementdelay(maxd, mind) def setmousemovementspeed(speedvalue): """设置鼠标移动速度""" return ghost_dll.setmousemovementspeed(speedvalue) # ================ 键盘操作 ================ def presskeybyname(key_name): """按下键（通过键名）""" return ghost_dll.presskeybyname(key_name) def releasekeybyname(key_name): """释放键（通过键名）""" return ghost_dll.releasekeybyname(key_name) def combinationkey(key_sequence): """组合键（如 b"ctrl+c"）""" return ghost_dll.combinationkey(key_sequence) def pressandreleasekeybyname(key_name): """按下并释放键""" return ghost_dll.pressandreleasekeybyname(key_name) def presskeybycode(key_code): """按下键（通过键码）""" return ghost_dll.presskeybycode(key_code) def releasekeybycode(key_code): """释放键（通过键码）""" return ghost_dll.releasekeybycode(key_code) def pressandreleasekeybycode(key_code): """按下并释放键（通过键码）""" return ghost_dll.pressandreleasekeybycode(key_code) def clearkeys(): """清除所有按下的键""" return ghost_dll.clearkeys() # 尝试打开设备 device_id = opendevice(0) if device_id == 0: print("幽灵键鼠设备连接失败!") ghost_available = False else: print("幽灵键鼠设备连接成功") ghost_available = True else: ghost_available = False print("幽灵键鼠不可用，将使用系统API") class WebSocketClient: """WebSocket客户端""" def __init__(self, channel_type, message_callback, log_callback): self.channel_type = channel_type self.message_callback = message_callback self.log_callback = log_callback self.ws = None self.is_connected = False self.stop_flag = False self.thread = None self.user_id = None def connect(self): """连接WebSocket服务器""" def on_message(ws, message): try: data = json.loads(message) msg_type = data.get("type") # 处理用户信息（登录成功） if msg_type == "userInfo": self.is_connected = True self.user_id = data.get("id") if self.log_callback: self.log_callback(f"WS登录成功，用户ID: {self.user_id}") return # 处理其他消息 if self.message_callback: self.message_callback(data) except Exception as e: if self.log_callback: self.log_callback(f"WebSocket消息解析失败: {e}") def on_error(ws, error): if self.log_callback: self.log_callback(f"WebSocket错误: {error}") def on_close(ws, close_status_code, close_msg): self.is_connected = False if self.log_callback: self.log_callback("WebSocket连接已关闭") # 自动重连 if not self.stop_flag: if self.log_callback: self.log_callback("3秒后尝试重新连接...") time.sleep(3) self.connect() def on_open(ws): if self.log_callback: self.log_callback(f"WebSocket连接成功，信道: {self.channel_type}") # 发送登录信息 login_msg = { "route": "login", "type": self.channel_type, "admin": True } ws.send(json.dumps(login_msg)) try: # 连接到本地服务器 self.ws = websocket.WebSocketApp( 'wss://ws.lamp.run', # 你的服务器地址 on_open=on_open, on_message=on_message, on_error=on_error, on_close=on_close ) # 在新线程中运行 self.thread = threading.Thread(target=self.ws.run_forever, daemon=True) self.thread.start() except Exception as e: if self.log_callback: self.log_callback(f"WebSocket连接失败: {e}") def disconnect(self): """断开WebSocket连接""" self.stop_flag = True if self.ws: self.ws.close() self.is_connected = False def send_message(self, route, value): """发送消息 - 群发模式（不指定id）""" if not self.ws or not self.is_connected: if self.log_callback: self.log_callback(f"⚠️ WS未连接，无法发送消息") return False if not self.user_id: if self.log_callback: self.log_callback(f"⚠️ 未获取到用户ID，无法发送消息") return False try: # 构建消息格式 - 不包含id字段，实现群发 msg = { "route": route, "type": self.channel_type, "userID": self.user_id, "value": value } # ===== 打印原始消息 ===== msg_json = json.dumps(msg) if self.log_callback: self.log_callback(f"📤 发送原始消息: {msg_json}") self.ws.send(msg_json) if self.log_callback: self.log_callback(f"📤 群发WS消息: route={route}, value={value}") return True except Exception as e: if self.log_callback: self.log_callback(f"发送消息失败: {e}") return False class VmControlGUI: def __init__(self, root): self.root = root self.root.title("虚拟机按键循环工具 - 双模式") self.root.geometry("850x750") self.root.resizable(True, True) # ===== 脚本选择 ===== self.script_mode = tk.StringVar(value="A") # ===== 脚本A变量 ===== self.vm_window_title = tk.StringVar() self.vm_windows = [] self.vm_hwnd = None self.window_rect = None # ===== 脚本A运行状态 ===== self.is_running = False self.stop_flag = False self.thread = None # ===== 脚本A参数 ===== self.key_interval_min = tk.StringVar(value="300") self.key_interval_max = tk.StringVar(value="800") self.round_interval_min = tk.StringVar(value="3") self.round_interval_max = tk.StringVar(value="5") self.pause_interval_min = tk.StringVar(value="30") self.pause_interval_max = tk.StringVar(value="50") self.pause_duration = tk.StringVar(value="60") # ===== 脚本B变量 ===== self.channel_type = tk.StringVar(value="控制虚拟机_1") self.ws_client = None self.is_b_running = False self.b_stop_flag = False self.b_thread = None # 脚本B参数 self.b_key_interval_min = tk.StringVar(value="30") self.b_key_interval_max = tk.StringVar(value="50") self.b_stop_delay_min = tk.StringVar(value="10") self.b_stop_delay_max = tk.StringVar(value="30") self.b_wait_after_stop_min = tk.StringVar(value="200") self.b_wait_after_stop_max = tk.StringVar(value="500") self.b_pause_duration = tk.StringVar(value="20") # ===== 激活坐标 ===== self.ACTIVATE_X = 300 self.ACTIVATE_Y = 300 # ===== 鼠标移动延迟 ===== self.MOUSE_MOVE_DELAY = 0.1 # ===== 幽灵键鼠状态 ===== self.ghost_available = ghost_available if self.ghost_available: setmousemovementspeed(5) setmousemovementdelay(10, 5) self.create_widgets() self.scan_vm_windows() # 显示幽灵键鼠状态 if self.ghost_available: self.log("✅ 幽灵键鼠已连接") else: self.log("⚠️ 幽灵键鼠未连接，使用系统API") def create_widgets(self): main_frame = ttk.Frame(self.root, padding="10") main_frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) # ===== 脚本模式选择 ===== mode_frame = ttk.LabelFrame(main_frame, text="脚本模式", padding="10") mode_frame.grid(row=0, column=0, columnspan=4, sticky=(tk.W, tk.E), pady=5) ttk.Radiobutton(mode_frame, text="脚本A - 主控 (按键循环)", variable=self.script_mode, value="A", command=self.on_mode_change).pack(side=tk.LEFT, padx=10) ttk.Radiobutton(mode_frame, text="脚本B - 分机 (WS监听)", variable=self.script_mode, value="B", command=self.on_mode_change).pack(side=tk.LEFT, padx=10) # 信道输入框 ttk.Label(mode_frame, text="信道:").pack(side=tk.LEFT, padx=(20, 5)) ttk.Entry(mode_frame, textvariable=self.channel_type, width=15).pack(side=tk.LEFT, padx=5) ttk.Button(mode_frame, text="连接WS", command=self.connect_websocket, width=8).pack(side=tk.LEFT, padx=5) ttk.Button(mode_frame, text="断开WS", command=self.disconnect_websocket, width=8).pack(side=tk.LEFT, padx=5) # WS状态 self.ws_status_label = ttk.Label(mode_frame, text="WS: 未连接", foreground="gray") self.ws_status_label.pack(side=tk.LEFT, padx=10) # 发送指令按钮（群发） ttk.Button(mode_frame, text="📤 群发开始", command=self.send_start_command, width=10).pack(side=tk.LEFT, padx=5) ttk.Button(mode_frame, text="📤 群发停止", command=self.send_stop_command, width=10).pack(side=tk.LEFT, padx=5) # ===== 脚本A界面 ===== self.frame_a = ttk.Frame(main_frame) self.frame_a.grid(row=1, column=0, columnspan=4, sticky=(tk.W, tk.E, tk.N, tk.S)) # 窗口搜索 ttk.Button(self.frame_a, text="🔍 搜索 VMware 窗口", command=self.scan_vm_windows, width=20).grid( row=0, column=0, sticky=tk.W, pady=5 ) ttk.Label(self.frame_a, text="选择虚拟机窗口:").grid(row=0, column=1, sticky=tk.W, pady=5, padx=10) self.vm_combo = ttk.Combobox(self.frame_a, textvariable=self.vm_window_title, width=45) self.vm_combo.grid(row=0, column=2, sticky=(tk.W, tk.E), pady=5, padx=5) self.window_info_label = ttk.Label(self.frame_a, text="窗口状态: 未选择", foreground="gray") self.window_info_label.grid(row=1, column=0, columnspan=4, sticky=tk.W, pady=5) # 脚本A参数 param_frame_a = ttk.LabelFrame(self.frame_a, text="脚本A参数设置", padding="10") param_frame_a.grid(row=2, column=0, columnspan=4, sticky=(tk.W, tk.E), pady=10) ttk.Label(param_frame_a, text="按键间隔 (毫秒):").grid(row=0, column=0, sticky=tk.W, pady=3) ttk.Label(param_frame_a, text="最小:").grid(row=0, column=1, sticky=tk.W, pady=3, padx=(10, 0)) ttk.Entry(param_frame_a, textvariable=self.key_interval_min, width=8).grid(row=0, column=1, sticky=tk.W, pady=3, padx=(30, 0)) ttk.Label(param_frame_a, text="最大:").grid(row=0, column=2, sticky=tk.W, pady=3, padx=(10, 0)) ttk.Entry(param_frame_a, textvariable=self.key_interval_max, width=8).grid(row=0, column=2, sticky=tk.W, pady=3, padx=(30, 0)) ttk.Label(param_frame_a, text="轮次间隔 (秒):").grid(row=1, column=0, sticky=tk.W, pady=3) ttk.Label(param_frame_a, text="最小:").grid(row=1, column=1, sticky=tk.W, pady=3, padx=(10, 0)) ttk.Entry(param_frame_a, textvariable=self.round_interval_min, width=8).grid(row=1, column=1, sticky=tk.W, pady=3, padx=(30, 0)) ttk.Label(param_frame_a, text="最大:").grid(row=1, column=2, sticky=tk.W, pady=3, padx=(10, 0)) ttk.Entry(param_frame_a, textvariable=self.round_interval_max, width=8).grid(row=1, column=2, sticky=tk.W, pady=3, padx=(30, 0)) ttk.Label(param_frame_a, text="暂停间隔 (分钟):").grid(row=2, column=0, sticky=tk.W, pady=3) ttk.Label(param_frame_a, text="最小:").grid(row=2, column=1, sticky=tk.W, pady=3, padx=(10, 0)) ttk.Entry(param_frame_a, textvariable=self.pause_interval_min, width=8).grid(row=2, column=1, sticky=tk.W, pady=3, padx=(30, 0)) ttk.Label(param_frame_a, text="最大:").grid(row=2, column=2, sticky=tk.W, pady=3, padx=(10, 0)) ttk.Entry(param_frame_a, textvariable=self.pause_interval_max, width=8).grid(row=2, column=2, sticky=tk.W, pady=3, padx=(30, 0)) ttk.Label(param_frame_a, text="暂停时长(秒):").grid(row=2, column=3, sticky=tk.W, pady=3, padx=(10, 0)) ttk.Entry(param_frame_a, textvariable=self.pause_duration, width=8).grid(row=2, column=3, sticky=tk.W, pady=3, padx=(5, 0)) # 脚本A按钮 btn_frame_a = ttk.Frame(self.frame_a) btn_frame_a.grid(row=3, column=0, columnspan=4, pady=10) self.start_btn = ttk.Button(btn_frame_a, text="▶ 开始循环", command=self.start_loop, width=15) self.start_btn.pack(side=tk.LEFT, padx=5) self.stop_btn = ttk.Button(btn_frame_a, text="⏹ 停止循环", command=self.stop_loop, width=15, state=tk.DISABLED) self.stop_btn.pack(side=tk.LEFT, padx=5) ttk.Button(btn_frame_a, text="测试点击", command=self.test_click, width=15).pack(side=tk.LEFT, padx=5) ttk.Button(btn_frame_a, text="测试按键", command=self.test_keys, width=15).pack(side=tk.LEFT, padx=5) # ===== 脚本B界面 ===== self.frame_b = ttk.Frame(main_frame) self.frame_b.grid(row=1, column=0, columnspan=4, sticky=(tk.W, tk.E, tk.N, tk.S)) self.frame_b.grid_remove() # 脚本B参数 param_frame_b = ttk.LabelFrame(self.frame_b, text="脚本B参数设置", padding="10") param_frame_b.grid(row=0, column=0, columnspan=4, sticky=(tk.W, tk.E), pady=10) ttk.Label(param_frame_b, text="按键间隔 (秒):").grid(row=0, column=0, sticky=tk.W, pady=3) ttk.Label(param_frame_b, text="最小:").grid(row=0, column=1, sticky=tk.W, pady=3, padx=(10, 0)) ttk.Entry(param_frame_b, textvariable=self.b_key_interval_min, width=8).grid(row=0, column=1, sticky=tk.W, pady=3, padx=(30, 0)) ttk.Label(param_frame_b, text="最大:").grid(row=0, column=2, sticky=tk.W, pady=3, padx=(10, 0)) ttk.Entry(param_frame_b, textvariable=self.b_key_interval_max, width=8).grid(row=0, column=2, sticky=tk.W, pady=3, padx=(30, 0)) ttk.Label(param_frame_b, text="停止延迟 (秒):").grid(row=1, column=0, sticky=tk.W, pady=3) ttk.Label(param_frame_b, text="最小:").grid(row=1, column=1, sticky=tk.W, pady=3, padx=(10, 0)) ttk.Entry(param_frame_b, textvariable=self.b_stop_delay_min, width=8).grid(row=1, column=1, sticky=tk.W, pady=3, padx=(30, 0)) ttk.Label(param_frame_b, text="最大:").grid(row=1, column=2, sticky=tk.W, pady=3, padx=(10, 0)) ttk.Entry(param_frame_b, textvariable=self.b_stop_delay_max, width=8).grid(row=1, column=2, sticky=tk.W, pady=3, padx=(30, 0)) ttk.Label(param_frame_b, text="停止后延迟 (毫秒):").grid(row=2, column=0, sticky=tk.W, pady=3) ttk.Label(param_frame_b, text="最小:").grid(row=2, column=1, sticky=tk.W, pady=3, padx=(10, 0)) ttk.Entry(param_frame_b, textvariable=self.b_wait_after_stop_min, width=8).grid(row=2, column=1, sticky=tk.W, pady=3, padx=(30, 0)) ttk.Label(param_frame_b, text="最大:").grid(row=2, column=2, sticky=tk.W, pady=3, padx=(10, 0)) ttk.Entry(param_frame_b, textvariable=self.b_wait_after_stop_max, width=8).grid(row=2, column=2, sticky=tk.W, pady=3, padx=(30, 0)) ttk.Label(param_frame_b, text="暂停延续(秒):").grid(row=3, column=0, sticky=tk.W, pady=3) ttk.Label(param_frame_b, text="最小:").grid(row=3, column=1, sticky=tk.W, pady=3, padx=(10, 0)) ttk.Entry(param_frame_b, textvariable=self.b_pause_duration, width=8).grid(row=3, column=1, sticky=tk.W, pady=3, padx=(30, 0)) # 脚本B状态 self.b_status_label = ttk.Label(self.frame_b, text="脚本B状态: 等待启动信号", foreground="gray") self.b_status_label.grid(row=1, column=0, columnspan=4, sticky=tk.W, pady=5) # ===== 公共区域 ===== self.status_label = ttk.Label(main_frame, text="状态: 停止", foreground="gray") self.status_label.grid(row=2, column=0, columnspan=4, sticky=tk.W, pady=5) ttk.Label(main_frame, text="执行日志:").grid(row=3, column=0, sticky=tk.W, pady=5) self.log_text = tk.Text(main_frame, height=14, width=100, font=("Consolas", 9)) self.log_text.grid(row=4, column=0, columnspan=4, pady=5, sticky=(tk.W, tk.E, tk.N, tk.S)) scrollbar = ttk.Scrollbar(main_frame, orient=tk.VERTICAL, command=self.log_text.yview) scrollbar.grid(row=4, column=4, sticky=(tk.N, tk.S)) self.log_text.config(yscrollcommand=scrollbar.set) ttk.Button(main_frame, text="清空日志", command=self.clear_log, width=15).grid(row=5, column=0, columnspan=4, pady=5) main_frame.columnconfigure(2, weight=1) main_frame.rowconfigure(4, weight=1) self.root.columnconfigure(0, weight=1) self.root.rowconfigure(0, weight=1) def on_mode_change(self): """切换脚本模式""" mode = self.script_mode.get() if mode == "A": self.frame_a.grid() self.frame_b.grid_remove() else: self.frame_a.grid_remove() self.frame_b.grid() def log(self, message): import datetime timestamp = datetime.datetime.now().strftime("%H:%M:%S") self.log_text.insert(tk.END, f"[{timestamp}] {message}\n") self.log_text.see(tk.END) self.root.update() def clear_log(self): self.log_text.delete(1.0, tk.END) # ========== WebSocket相关 ========== def connect_websocket(self): """连接WebSocket""" channel = self.channel_type.get().strip() if not channel: messagebox.showerror("错误", "请输入信道名称！") return if self.ws_client and self.ws_client.is_connected: self.log("WebSocket已连接") return self.ws_client = WebSocketClient( channel_type=channel, message_callback=self.on_ws_message, log_callback=self.log ) self.ws_client.connect() self.ws_status_label.config(text="WS: 连接中...", foreground="orange") def disconnect_websocket(self): """断开WebSocket""" if self.ws_client: self.ws_client.disconnect() self.ws_client = None self.ws_status_label.config(text="WS: 已断开", foreground="gray") self.log("WebSocket已断开") def send_start_command(self): """群发开始指令""" if not self.ws_client or not self.ws_client.is_connected: self.log("⚠️ WS未连接，无法发送指令") messagebox.showerror("错误", "请先连接WebSocket！") return # 群发 - 不指定id self.ws_client.send_message("start", "开始执行") self.log("📤 已群发开始指令") def send_stop_command(self): """群发停止指令""" if not self.ws_client or not self.ws_client.is_connected: self.log("⚠️ WS未连接，无法发送指令") messagebox.showerror("错误", "请先连接WebSocket！") return # 群发 - 不指定id self.ws_client.send_message("stop", "停止执行") self.log("📤 已群发停止指令") def send_pause_command(self): """群发暂停指令（让B执行停止序列）""" if not self.ws_client or not self.ws_client.is_connected: self.log("⚠️ WS未连接，无法发送指令") return self.ws_client.send_message("pause", "暂停执行") self.log("📤 已群发暂停指令") def on_ws_message(self, data): """处理WebSocket消息""" try: self.log(f"📨 收到原始消息: {json.dumps(data)}") msg_type = data.get("type") value = data.get("value") user_id = data.get("userID") self.log(f"📨 解析结果: type={msg_type}, value={value}, from={user_id}") if msg_type == "start": self.log(f"📨 收到开始指令") self.start_script_b() elif msg_type == "stop": self.log(f"📨 收到停止指令") self.stop_script_b() elif msg_type == "pause": self.log(f"📨 收到暂停指令") # 执行停止序列（2次），但不结束脚本B self.execute_stop_sequence() elif msg_type == "close": self.log(f"📨 收到关闭指令，可能是ID冲突") else: self.log(f"收到其他消息: {msg_type}") except Exception as e: self.log(f"处理WS消息失败: {e}") def execute_stop_sequence(self): """执行停止序列（2次），不结束脚本B""" try: # 暂停时继续执行的秒数（默认20秒） pause_duration = int(self.b_pause_duration.get()) # ===== 1. 继续执行20秒（按- 和 2） ===== self.log(f"⏳ 暂停指令，继续执行 {pause_duration} 秒（按- 和 2）") # 按键间隔（30-50秒） interval_min = int(self.b_key_interval_min.get()) interval_max = int(self.b_key_interval_max.get()) # 记录上次按2的时间 last_key2_time = time.time() pause_start_time = time.time() while time.time() - pause_start_time < pause_duration: # 按 - (减号键) - 每300-800ms一次 self.log("发送: - (减号) [暂停延续中]") self.press_key(0xBD, False) # 减号间隔 300-800ms key_delay = random.uniform(0.3, 0.8) time.sleep(key_delay) # 检查是否该按 2（30-50秒一次） current_time = time.time() if current_time - last_key2_time >= random.uniform(interval_min, interval_max): self.log("发送: 2 (数字2) [暂停延续中]") self.press_key(0x32, False) last_key2_time = current_time self.log(f"⏳ 延续结束，执行停止序列（2次）") # ===== 2. 执行停止序列（2次） ===== for i in range(2): self.log(f"停止序列 {i+1}/2") # 按 = 等待2-2.5秒 wait_time = random.uniform(2.0, 2.5) self.log(f" 按 = (延迟 {wait_time:.1f}秒)") self.press_key(0xBB, False) time.sleep(wait_time) # 按 小键盘2 等待2-2.7秒 wait_time = random.uniform(2.0, 2.7) self.log(f" 按 小键盘2 (延迟 {wait_time:.1f}秒)") self.press_key(0x62, True) time.sleep(wait_time) # 按 ESC self.log(f" 按 ESC") self.press_key(0x1B, False) # 每次之间等待200-800ms（最后一次不等待） if i < 1: between_delay = random.uniform(0.2, 0.8) self.log(f" 等待 {between_delay*1000:.0f}ms 后执行下一次") time.sleep(between_delay) self.log("✅ 暂停停止序列执行完成") except Exception as e: self.log(f"执行暂停停止序列失败: {e}") # ========== 脚本B相关 ========== def start_script_b(self): """启动脚本B""" if self.is_b_running: self.log("脚本B已在运行中") return self.b_stop_flag = False self.is_b_running = True self.b_status_label.config(text="脚本B状态: 运行中", foreground="green") self.b_thread = threading.Thread(target=self.script_b_worker, daemon=True) self.b_thread.start() self.log("🚀 脚本B已启动") def stop_script_b(self): """停止脚本B""" if not self.is_b_running: self.log("脚本B未运行") return self.b_stop_flag = True self.b_status_label.config(text="脚本B状态: 正在停止...", foreground="orange") self.log("⏹ 脚本B正在停止...") def script_b_worker(self): """脚本B工作线程 - 大键盘单键循环""" try: # 按键间隔（30-50秒） interval_min = int(self.b_key_interval_min.get()) interval_max = int(self.b_key_interval_max.get()) # 停止延迟（10-30秒）- 现在表示继续执行的时间 stop_delay_min = int(self.b_stop_delay_min.get()) stop_delay_max = int(self.b_stop_delay_max.get()) # 停止序列中的延迟（200-500毫秒） wait_min = int(self.b_wait_after_stop_min.get()) / 1000.0 wait_max = int(self.b_wait_after_stop_max.get()) / 1000.0 self.log("=" * 50) self.log("🚀 脚本B - 大键盘单键循环开始") self.log(f"按键1: - (减号) 每300-800ms一次") self.log(f"按键2: 2 (数字2) 每{interval_min}-{interval_max}秒一次") self.log("=" * 50) # 记录上次按2的时间 last_key2_time = time.time() # ===== 正常运行 ===== while not self.b_stop_flag: # 按 - (减号键) - 每300-800ms一次 self.log("发送: - (减号)") self.press_key(0xBD, False) # 减号间隔 300-800ms key_delay = random.uniform(0.3, 0.8) time.sleep(key_delay) # 检查是否该按 2（30-50秒一次） current_time = time.time() if current_time - last_key2_time >= random.uniform(interval_min, interval_max): self.log("发送: 2 (数字2)") self.press_key(0x62, False) last_key2_time = current_time # ===== 收到停止指令后，继续执行10-30秒 ===== stop_duration = random.uniform(stop_delay_min, stop_delay_max) self.log(f"⏳ 收到停止指令，继续执行 {stop_duration:.1f} 秒后停止") stop_start_time = time.time() while time.time() - stop_start_time < stop_duration: # 继续按 - (减号键) self.log("发送: - (减号) [停止延迟中]") self.press_key(0xBD, False) # 减号间隔 300-800ms key_delay = random.uniform(0.3, 0.8) time.sleep(key_delay) # 检查是否该按 2（30-50秒一次） current_time = time.time() if current_time - last_key2_time >= random.uniform(interval_min, interval_max): self.log("发送: 2 (数字2) [停止延迟中]") self.press_key(0x62, False) last_key2_time = current_time self.log(f"⏳ 延迟结束，开始执行停止序列") # ===== 执行停止序列（共3次） ===== self.log("脚本B停止序列开始") # 执行3次停止序列 for i in range(3): self.log(f"停止序列 {i+1}/3") # 按 = 等待2-2.5秒 wait_time = random.uniform(2.0, 2.5) self.log(f" 按 = (延迟 {wait_time:.1f}秒)") self.press_key(0xBB, False) time.sleep(wait_time) # 按 小键盘2 等待2-2.7秒 wait_time = random.uniform(2.0, 2.7) self.log(f" 按 小键盘2 (延迟 {wait_time:.1f}秒)") self.press_key(0x62, True) time.sleep(wait_time) # 按 ESC self.log(f" 按 ESC") self.press_key(0x1B, False) # 每次之间等待200-800ms（最后一次不等待） if i < 2: between_delay = random.uniform(0.2, 0.8) self.log(f" 等待 {between_delay*1000:.0f}ms 后执行下一次") time.sleep(between_delay) self.log("✅ 脚本B停止序列完成") self.b_status_label.config(text="脚本B状态: 已停止", foreground="gray") except Exception as e: self.log(f"脚本B出错: {e}") finally: self.is_b_running = False self.b_status_label.config(text="脚本B状态: 已停止", foreground="gray") self.log("脚本B已结束") # ========== 窗口扫描 ========== def scan_vm_windows(self): self.log("正在扫描 VMware 窗口...") self.vm_windows = [] def enum_callback(hwnd, _): if win32gui.IsWindowVisible(hwnd): title = win32gui.GetWindowText(hwnd) if "VMware Workstation" in title and title.strip(): self.vm_windows.append((title, hwnd)) return True win32gui.EnumWindows(enum_callback, None) if self.vm_windows: titles = [f"{i+1}. {t[0]}" for i, t in enumerate(self.vm_windows)] self.vm_combo["values"] = titles self.vm_combo.current(0) self.vm_window_title.set(titles[0]) self.log(f"找到 {len(self.vm_windows)} 个 VMware 窗口") self.window_info_label.config(text=f"找到 {len(self.vm_windows)} 个窗口", foreground="green") self.select_window(0) else: self.log("未找到 VMware 窗口") self.window_info_label.config(text="未找到 VMware 窗口", foreground="red") self.vm_combo["values"] = [] self.vm_combo.bind("<>", self.on_window_selected) def on_window_selected(self, event=None): selection = self.vm_combo.current() if selection >= 0: self.select_window(selection) def select_window(self, index): if index < len(self.vm_windows): title, hwnd = self.vm_windows[index] self.vm_hwnd = hwnd self.vm_window_title.set(f"{index+1}. {title}") try: rect = win32gui.GetWindowRect(hwnd) self.window_rect = rect self.window_info_label.config( text=f"已选择: {title} (大小: {rect[2]-rect[0]}x{rect[3]-rect[1]})", foreground="blue" ) self.log(f"已选择窗口: {title}") except Exception as e: self.log(f"获取窗口信息失败: {e}") # ========== 坐标转换 ========== def get_absolute_coords(self, x, y): if self.window_rect is None: return None, None left, top, right, bottom = self.window_rect offset_x = 8 offset_y = 30 return left + offset_x + x, top + offset_y + y def activate_window(self): if self.vm_hwnd is None: return False try: if win32gui.IsIconic(self.vm_hwnd): win32gui.ShowWindow(self.vm_hwnd, win32con.SW_RESTORE) win32gui.SetForegroundWindow(self.vm_hwnd) win32gui.BringWindowToTop(self.vm_hwnd) for _ in range(5): if win32gui.GetForegroundWindow() == self.vm_hwnd: break time.sleep(0.1) win32gui.SetForegroundWindow(self.vm_hwnd) rect = win32gui.GetWindowRect(self.vm_hwnd) self.window_rect = rect return True except Exception as e: self.log(f"激活窗口失败: {e}") return False def click_at(self, x, y, button="left"): """使用幽灵键鼠在指定屏幕坐标点击""" try: if self.ghost_available: ret = movemouseto(x, y) if ret != 1: self.log(f"幽灵键鼠移动失败: {ret}") return False time.sleep(0.05) if button == "left": ret = pressandreleasemousebutton(1) else: ret = pressandreleasemousebutton(2) if ret != 1: self.log(f"幽灵键鼠点击失败: {ret}") return False return True else: win32api.SetCursorPos((x, y)) time.sleep(0.05) if button == "left": win32api.mouse_event(win32con.MOUSEEVENTF_LEFTDOWN, x, y, 0, 0) time.sleep(0.05) win32api.mouse_event(win32con.MOUSEEVENTF_LEFTUP, x, y, 0, 0) else: win32api.mouse_event(win32con.MOUSEEVENTF_RIGHTDOWN, x, y, 0, 0) time.sleep(0.05) win32api.mouse_event(win32con.MOUSEEVENTF_RIGHTUP, x, y, 0, 0) return True except Exception as e: self.log(f"点击失败: {e}") return False def press_key(self, key_code, extended=False): """使用幽灵键鼠模拟按键""" try: # 按键名称映射 key_name_map = { 0x68: b"num8", # 小键盘8 0x62: b"num2", # 小键盘2 0x25: b"left", # 左箭头 0x20: b"space", # 空格 0x26: b"up", # 上箭头 0x28: b"down", # 下箭头 0x27: b"right", # 右箭头 0x1B: b"esc", # ESC 0xBB: b"=", # = 号 0xBD: b"-", # - 号 0x32: b"2", # 主键盘2（如果确实需要主键盘的） } if self.ghost_available: if key_code in key_name_map: pressandreleasekeybyname(key_name_map[key_code]) else: if extended: win32api.keybd_event(key_code, 0, win32con.KEYEVENTF_EXTENDEDKEY, 0) time.sleep(0.03) win32api.keybd_event(key_code, 0, win32con.KEYEVENTF_EXTENDEDKEY | win32con.KEYEVENTF_KEYUP, 0) else: win32api.keybd_event(key_code, 0, 0, 0) time.sleep(0.03) win32api.keybd_event(key_code, 0, win32con.KEYEVENTF_KEYUP, 0) return True except Exception as e: self.log(f"按键失败: {e}") return False # ========== 脚本A函数 ========== def test_click(self): self.log("=" * 50) self.log("测试点击 (300, 300)...") if self.vm_hwnd is None: self.log("错误: 未选择窗口") return if not self.activate_window(): self.log("激活窗口失败") return abs_x, abs_y = self.get_absolute_coords(self.ACTIVATE_X, self.ACTIVATE_Y) if abs_x is None: return self.log(f"屏幕坐标: ({abs_x}, {abs_y})") self.log(f"使用: {'幽灵键鼠' if self.ghost_available else '系统API'}") if self.click_at(abs_x, abs_y): self.log("点击成功！") else: self.log("点击失败") self.log("=" * 50) def test_keys(self): self.log("=" * 50) self.log("测试按键序列") if self.vm_hwnd is None: self.log("错误: 未选择窗口") return if not self.activate_window(): self.log("激活窗口失败") return abs_x, abs_y = self.get_absolute_coords(self.ACTIVATE_X, self.ACTIVATE_Y) if abs_x is None: return self.log("点击激活虚拟机...") self.click_at(abs_x, abs_y) time.sleep(0.5) keys = [ (0x68, True, "小键盘8"), (0x62, True, "小键盘2"), (0x25, False, "左键"), (0x20, False, "空格"), ] self.log(f"使用: {'幽灵键鼠' if self.ghost_available else '系统API'}") for key_code, extended, name in keys: self.log(f"发送: {name}") self.press_key(key_code, extended) time.sleep(0.3) self.log("测试完成") self.log("=" * 50) def click_at_current_position(self): """在当前鼠标位置点击左键""" try: if self.ghost_available: # 幽灵键鼠：获取当前坐标并点击 x = getmousex() y = getmousey() ret = movemouseto(x, y) if ret == 1: ret = pressandreleasemousebutton(1) # 左键 return ret == 1 return False else: # 系统API：获取当前位置并点击 x, y = win32api.GetCursorPos() win32api.mouse_event(win32con.MOUSEEVENTF_LEFTDOWN, x, y, 0, 0) time.sleep(0.05) win32api.mouse_event(win32con.MOUSEEVENTF_LEFTUP, x, y, 0, 0) return True except Exception as e: self.log(f"鼠标左键点击失败: {e}") return False def loop_worker(self): # 按键定义：(类型, 键码/标识, 扩展标志, 名称) keys = [ ("keyboard", 0x68, True, "小键盘8"), # 小键盘8 ("keyboard", 0x62, True, "小键盘2"), # 小键盘2 ("mouse", None, None, "鼠标左键"), # 鼠标左键（原来的0x25左箭头改成鼠标左键） ("keyboard", 0x20, False, "空格"), # 空格 ] key_min = int(self.key_interval_min.get()) / 1000.0 key_max = int(self.key_interval_max.get()) / 1000.0 round_min = int(self.round_interval_min.get()) round_max = int(self.round_interval_max.get()) pause_min = int(self.pause_interval_min.get()) * 60 pause_max = int(self.pause_interval_max.get()) * 60 pause_dur = int(self.pause_duration.get()) round_count = 0 total_keys = 0 last_pause_time = time.time() self.log("=" * 50) self.log("🚀 脚本A循环开始！") self.log(f"使用: {'幽灵键鼠' if self.ghost_available else '系统API'}") self.log(f"按键: 小键盘8 -> 小键盘2 -> 鼠标左键 -> 空格") self.log("=" * 50) # 只在开始前执行一次鼠标点击激活窗口 if not self.stop_flag: try: if win32gui.IsWindow(self.vm_hwnd): self.activate_window() abs_x, abs_y = self.get_absolute_coords(self.ACTIVATE_X, self.ACTIVATE_Y) if abs_x is not None: self.log(f"初始点击激活虚拟机 ({abs_x}, {abs_y})") self.click_at(abs_x, abs_y) time.sleep(0.3) except Exception as e: self.log(f"初始点击失败: {e}") while not self.stop_flag: try: if not win32gui.IsWindow(self.vm_hwnd): self.log("窗口已关闭，停止循环") break # 执行一轮按键 for key_type, key_code, extended, key_name in keys: if self.stop_flag: break self.log(f"发送: {key_name}") if key_type == "mouse": self.click_at_current_position() else: self.press_key(key_code, extended) total_keys += 1 interval = random.uniform(key_min, key_max) time.sleep(interval) round_count += 1 if round_count % 10 == 0: self.log(f"📊 已执行 {round_count} 轮") # ===== 暂停检查 ===== current_time = time.time() if current_time - last_pause_time >= random.uniform(pause_min, pause_max): self.log(f"⏸ 到达暂停时间，通知B执行停止序列") # 通知B执行停止序列（2次） if self.ws_client and self.ws_client.is_connected: self.ws_client.send_message("pause", "暂停执行，执行停止序列") self.log("📤 已群发暂停指令到同信道脚本B") else: self.log("⚠️ WebSocket未连接，无法通知脚本B") # A自己暂停 self.log(f"⏸ A暂停 {pause_dur} 秒") self.status_label.config(text=f"状态: 暂停中 ({pause_dur}s)", foreground="orange") for _ in range(int(pause_dur / 0.5)): if self.stop_flag: break time.sleep(0.5) self.log("▶ A暂停结束，继续循环") self.status_label.config(text="状态: 运行中", foreground="green") # 通知B继续（发送start） if self.ws_client and self.ws_client.is_connected: self.ws_client.send_message("start", "继续执行") self.log("📤 已群发继续指令到同信道脚本B") else: self.log("⚠️ WebSocket未连接，无法通知脚本B") last_pause_time = time.time() continue # 轮次间隔 round_interval = random.uniform(round_min, round_max) sleep_chunks = max(1, int(round_interval / 0.5)) for _ in range(sleep_chunks): if self.stop_flag: break time.sleep(0.5) except Exception as e: self.log(f"循环出错: {e}") time.sleep(2) self.log(f"🏁 脚本A结束！共执行 {round_count} 轮") self.status_label.config(text="状态: 已停止", foreground="gray") self.root.after(0, self.on_loop_stopped) def start_loop(self): if self.vm_hwnd is None: messagebox.showerror("错误", "请先选择虚拟机窗口！") return if self.is_running: return if not self.ghost_available: if not messagebox.askyesno("提示", "幽灵键鼠未连接，将使用系统API模拟按键。\n继续吗？"): return # ===== 新增：通过WebSocket通知同信道的脚本B ===== if self.ws_client and self.ws_client.is_connected: self.ws_client.send_message("start", "开始执行") self.log("📤 已群发开始指令到同信道脚本B") else: self.log("⚠️ WebSocket未连接，无法通知脚本B") self.is_running = True self.stop_flag = False self.status_label.config(text="状态: 运行中", foreground="green") self.start_btn.config(state=tk.DISABLED) self.stop_btn.config(state=tk.NORMAL) self.thread = threading.Thread(target=self.loop_worker, daemon=True) self.thread.start() def stop_loop(self): self.log("⏹ 正在停止脚本A...") # ===== 新增：通过WebSocket通知同信道的脚本B停止 ===== if self.ws_client and self.ws_client.is_connected: self.ws_client.send_message("stop", "停止执行") self.log("📤 已群发停止指令到同信道脚本B") else: self.log("⚠️ WebSocket未连接，无法通知脚本B") self.stop_flag = True self.status_label.config(text="状态: 正在停止...", foreground="orange") self.start_btn.config(state=tk.NORMAL) self.stop_btn.config(state=tk.DISABLED) def on_loop_stopped(self): self.is_running = False self.start_btn.config(state=tk.NORMAL) self.stop_btn.config(state=tk.DISABLED) self.status_label.config(text="状态: 已停止", foreground="gray") def __del__(self): if self.ws_client: self.ws_client.disconnect() if self.ghost_available: try: closedevice() except: pass if __name__ == "__main__": # 检查依赖 if not ghost_available: try: import win32api except ImportError: print("需要安装 pywin32: pip install pywin32") sys.exit(1) # 检查websocket-client try: import websocket except ImportError: print("需要安装 websocket-client: pip install websocket-client") sys.exit(1) root = tk.Tk() app = VmControlGUI(root) root.mainloop()