import ctypes import time import os import sys import platform import win32gui import win32con import win32api import win32clipboard as clipboard from PIL import ImageGrab import requests import io import tkinter as tk from tkinter import ttk, scrolledtext, messagebox import threading import json lastKeyWord = "" # ========== 获取程序目录 ========== def get_app_dir(): if getattr(sys, 'frozen', False): return os.path.dirname(sys.executable) else: return os.path.dirname(os.path.abspath(__file__)) APP_DIR = get_app_dir() # ========== 全局变量 ========== is_running = False stop_flag = False config_file = os.path.join(APP_DIR, "config.json") # ========== 幽灵键鼠加载 ========== def load_ghost_key_mouse(): if platform.architecture()[0] == "64bit": dll_path = os.path.join(APP_DIR, "gbild64.dll") else: dll_path = os.path.join(APP_DIR, "gbild32.dll") if not os.path.exists(dll_path): return None try: dll = ctypes.windll.LoadLibrary(dll_path) return dll except Exception as e: return None 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 pressandreleasemousebutton(mbtn): return ghost_dll.pressandreleasemousebutton(mbtn) def combinationkey(key_sequence): return ghost_dll.combinationkey(key_sequence) def pressandreleasekeybyname(key_name): return ghost_dll.pressandreleasekeybyname(key_name) device_id = opendevice(0) if device_id == 0: ghost_available = False else: ghost_available = True else: ghost_available = False # ========== 核心功能类 ========== class AuctionBot: def __init__(self, log_callback): self.log_callback = log_callback self.target_hwnd = None self.left = 0 self.top = 0 self.is_running = False def log(self, message): if self.log_callback: self.log_callback(message) def get_hwnds_by_title_contains(self, title_contains): hwnds = [] def enum_callback(hwnd, _): if win32gui.IsWindowVisible(hwnd): window_title = win32gui.GetWindowText(hwnd) if title_contains in window_title: hwnds.append(hwnd) return True win32gui.EnumWindows(enum_callback, None) return hwnds def get_client_rect(self, hwnd): rect = win32gui.GetClientRect(hwnd) point = win32gui.ClientToScreen(hwnd, (rect[0], rect[1])) return (point[0], point[1], point[0] + rect[2], point[1] + rect[3]) def activate_window(self, hwnd): try: if win32gui.IsIconic(hwnd): win32gui.ShowWindow(hwnd, win32con.SW_RESTORE) win32gui.SetForegroundWindow(hwnd) win32gui.BringWindowToTop(hwnd) time.sleep(0.2) return True except Exception as e: self.log(f"激活窗口失败: {e}") return False def capture_window_area_to_bytes(self, hwnd, x1, y1, x2, y2): try: client_rect = self.get_client_rect(hwnd) left, top, right, bottom = client_rect screen_x1 = left + x1 screen_y1 = top + y1 screen_x2 = left + x2 screen_y2 = top + y2 screenshot = ImageGrab.grab(bbox=(screen_x1, screen_y1, screen_x2, screen_y2)) img_bytes = io.BytesIO() screenshot.save(img_bytes, format='PNG') img_bytes.seek(0) return img_bytes.getvalue() except Exception as e: self.log(f"截图失败: {e}") return None def ocr_price(self, image_bytes): try: # url = "http://111.229.45.38:19100/ocr" url = "http://192.168.1.101:10001/ocr" files = {'image': ('screenshot.png', image_bytes, 'image/png')} self.log("正在识别价格...") response = requests.post(url, files=files, timeout=30) if response.status_code != 200: self.log(f"OCR请求失败: HTTP {response.status_code}") return None result = response.json() if not result.get('success', False): self.log(f"OCR识别失败: {result}") return None texts = result.get('texts', []) if not texts: self.log("未识别到任何文字") return None all_texts = [] for item in texts: rec_texts = item.get('rec_texts', []) all_texts.extend(rec_texts) for text in all_texts: try: price = float(text) self.log(f"识别到价格: {price}") return price except ValueError: continue self.log(f"未识别到价格，识别到的文字: {all_texts}") return None except Exception as e: self.log(f"OCR识别异常: {e}") return None def mouse_click_at(self, x, y, button=0): try: # 1. 移动物理光标（部分游戏会校验系统光标位置） win32api.SetCursorPos((x, y)) time.sleep(0.1) # 2. 准备 SendInput 结构体（64/32位兼容） ULONG_PTR = ctypes.c_uint64 if ctypes.sizeof(ctypes.c_void_p) == 8 else ctypes.c_uint32 class MOUSEINPUT(ctypes.Structure): _fields_ = [ ("dx", ctypes.c_long), ("dy", ctypes.c_long), ("mouseData", ctypes.c_ulong), ("dwFlags", ctypes.c_ulong), ("time", ctypes.c_ulong), ("dwExtraInfo", ULONG_PTR), ] class KEYBDINPUT(ctypes.Structure): _fields_ = [ ("wVk", ctypes.c_ushort), ("wScan", ctypes.c_ushort), ("dwFlags", ctypes.c_ulong), ("time", ctypes.c_ulong), ("dwExtraInfo", ULONG_PTR), ] class HARDWAREINPUT(ctypes.Structure): _fields_ = [ ("uMsg", ctypes.c_ulong), ("wParamL", ctypes.c_ushort), ("wParamH", ctypes.c_ushort), ] class DUMMYUNIONNAME(ctypes.Union): _fields_ = [("mi", MOUSEINPUT), ("ki", KEYBDINPUT), ("hi", HARDWAREINPUT)] class INPUT(ctypes.Structure): _fields_ = [("type", ctypes.c_ulong), ("DUMMYUNIONNAME", DUMMYUNIONNAME)] INPUT_MOUSE = 0 screen_w = win32api.GetSystemMetrics(win32con.SM_CXSCREEN) screen_h = win32api.GetSystemMetrics(win32con.SM_CYSCREEN) # 3. 转换为绝对坐标（0-65535） abs_x = int(x * 65535 / (screen_w - 1)) if screen_w > 1 else 0 abs_y = int(y * 65535 / (screen_h - 1)) if screen_h > 1 else 0 # 4. 选择按键 if button == 0: down_flag = win32con.MOUSEEVENTF_LEFTDOWN up_flag = win32con.MOUSEEVENTF_LEFTUP else: down_flag = win32con.MOUSEEVENTF_RIGHTDOWN up_flag = win32con.MOUSEEVENTF_RIGHTUP move_abs = win32con.MOUSEEVENTF_ABSOLUTE | win32con.MOUSEEVENTF_MOVE # 5. 按下（带绝对坐标移动） inp_down = INPUT() inp_down.type = INPUT_MOUSE inp_down.DUMMYUNIONNAME.mi.dx = abs_x inp_down.DUMMYUNIONNAME.mi.dy = abs_y inp_down.DUMMYUNIONNAME.mi.mouseData = 0 inp_down.DUMMYUNIONNAME.mi.dwFlags = move_abs | down_flag inp_down.DUMMYUNIONNAME.mi.time = 0 inp_down.DUMMYUNIONNAME.mi.dwExtraInfo = 0 ctypes.windll.user32.SendInput(1, ctypes.byref(inp_down), ctypes.sizeof(INPUT)) # 关键：保持按下状态至少100ms，确保游戏能轮询到 time.sleep(0.1) # 6. 释放 inp_up = INPUT() inp_up.type = INPUT_MOUSE inp_up.DUMMYUNIONNAME.mi.dx = abs_x inp_up.DUMMYUNIONNAME.mi.dy = abs_y inp_up.DUMMYUNIONNAME.mi.mouseData = 0 inp_up.DUMMYUNIONNAME.mi.dwFlags = move_abs | up_flag inp_up.DUMMYUNIONNAME.mi.time = 0 inp_up.DUMMYUNIONNAME.mi.dwExtraInfo = 0 ctypes.windll.user32.SendInput(1, ctypes.byref(inp_up), ctypes.sizeof(INPUT)) time.sleep(0.1) return True except Exception as e: self.log(f"鼠标点击失败: {e}") return False def keyboard_ctrl_v(self): try: if ghost_available: return combinationkey(b"ctrl+v") == 1 else: win32api.keybd_event(win32con.VK_CONTROL, 0, 0, 0) time.sleep(0.05) win32api.keybd_event(ord('V'), 0, 0, 0) time.sleep(0.05) win32api.keybd_event(ord('V'), 0, win32con.KEYEVENTF_KEYUP, 0) time.sleep(0.05) win32api.keybd_event(win32con.VK_CONTROL, 0, win32con.KEYEVENTF_KEYUP, 0) return True except Exception as e: self.log(f"Ctrl+V失败: {e}") return False def clipboard_set_text(self, text): try: clipboard.OpenClipboard() clipboard.EmptyClipboard() clipboard.SetClipboardText(text, clipboard.CF_TEXT) clipboard.CloseClipboard() return True except Exception as e: self.log(f"设置剪贴板失败: {e}") return False def process_item(self, keyword, target_price): global lastKeyWord """处理单个关键词""" if stop_flag: return False self.log(f"========== 处理: {keyword} (目标价格: {target_price}) ==========") if (lastKeyWord != keyword): # 点击重置 click_x = self.left + 728 click_y = self.top + 152 self.log(f"点击重置: ({click_x}, {click_y})") self.mouse_click_at(click_x, click_y, 0) time.sleep(1) # 设置剪贴板 self.log(f"设置剪贴板: '{keyword}'") self.clipboard_set_text(keyword) # 点击输入框粘贴 click_x = self.left + 568 click_y = self.top + 152 self.log(f"点击输入框: ({click_x}, {click_y})") self.mouse_click_at(click_x, click_y, 0) time.sleep(1) # Ctrl+V self.log("执行 Ctrl+V") self.keyboard_ctrl_v() time.sleep(1) else: self.log(f"搜索关键词一样，自动忽略重置") lastKeyWord = keyword # 点击搜索按钮 click_x = self.left + 664 click_y = self.top + 152 self.log(f"点击搜索按钮: ({click_x}, {click_y})") self.mouse_click_at(click_x, click_y, 0) time.sleep(2) # 截图识别价格 self.log("截图识别价格...") img_bytes = self.capture_window_area_to_bytes(self.target_hwnd, 580, 190 - 30, 634, 211 - 30) if not img_bytes: self.log("截图失败") return False price = self.ocr_price(img_bytes) if price is None: self.log("价格识别失败") return False self.log(f"识别到价格: {price}, 目标价格: {target_price}") if price <= target_price and price > 1: self.log(f"价格 {price} <= {target_price}，执行购买!") # 选择购买项 click_x = self.left + 506 click_y = self.top + 202 self.log(f"选择购买项: ({click_x}, {click_y})") self.mouse_click_at(click_x, click_y, 0) time.sleep(0.5) # 购买 click_x = self.left + 729 click_y = self.top + 545 self.log(f"点击购买: ({click_x}, {click_y})") self.mouse_click_at(click_x, click_y, 0) time.sleep(1) # 确认购买 click_x = self.left + 516 click_y = self.top + 286 self.log(f"确认购买: ({click_x}, {click_y})") self.mouse_click_at(click_x, click_y, 0) self.log(f"✅ {keyword} 购买成功! 价格: {price}") return True else: self.log(f"价格 {price} > {target_price}，不购买") return False def run(self, keywords_prices, interval): """主循环""" global stop_flag stop_flag = False self.log("="*60) self.log("开始自动拍卖操作") self.log("="*60) # 查找窗口 self.log("查找新天龙八部窗口...") window_list = self.get_hwnds_by_title_contains("新天龙八部") if not window_list: self.log("未找到新天龙八部窗口！") return self.target_hwnd = window_list[0] window_title = win32gui.GetWindowText(self.target_hwnd) self.log(f"找到窗口: {window_title}") # 获取窗口位置 client_rect = self.get_client_rect(self.target_hwnd) self.left, self.top, right, bottom = client_rect self.top = self.top - 30 self.log(f"窗口位置: left={self.left}, top={self.top}") # 激活窗口 self.log("激活窗口...") self.activate_window(self.target_hwnd) time.sleep(1) round_count = 0 while not stop_flag: round_count += 1 self.log(f"\n========== 第 {round_count} 轮 ==========") for keyword, target_price in keywords_prices: if stop_flag: break try: self.process_item(keyword, target_price) time.sleep(1) except Exception as e: self.log(f"处理 {keyword} 时出错: {e}") if not stop_flag: self.log(f"等待 {interval} 秒后继续...") # 分段等待，便于检测停止信号 for _ in range(interval): if stop_flag: break time.sleep(1) self.log("程序已停止") # ========== GUI程序 ========== class AuctionGUI: def __init__(self, root): self.root = root self.root.title("天龙八部自动拍卖机器人") self.root.geometry("800x700") self.root.resizable(True, True) self.bot = AuctionBot(self.log_message) self.keywords_list = [] self.running = False self.setup_ui() self.load_config() def setup_ui(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)) # 配置区域 config_frame = ttk.LabelFrame(main_frame, text="配置", padding="10") config_frame.grid(row=0, column=0, sticky=(tk.W, tk.E), pady=(0, 10)) # 关键词列表 ttk.Label(config_frame, text="关键词和价格:").grid(row=0, column=0, sticky=tk.W) # 关键词表格 columns = ('关键词', '目标价格') self.tree = ttk.Treeview(config_frame, columns=columns, show='headings', height=6) self.tree.heading('关键词', text='关键词') self.tree.heading('目标价格', text='目标价格') self.tree.column('关键词', width=150) self.tree.column('目标价格', width=100) self.tree.grid(row=1, column=0, columnspan=4, sticky=(tk.W, tk.E), pady=(5, 5)) # 滚动条 scrollbar = ttk.Scrollbar(config_frame, orient=tk.VERTICAL, command=self.tree.yview) scrollbar.grid(row=1, column=4, sticky=(tk.N, tk.S)) self.tree.configure(yscrollcommand=scrollbar.set) # 添加关键词 ttk.Label(config_frame, text="关键词:").grid(row=2, column=0, sticky=tk.W, pady=(5, 0)) self.keyword_entry = ttk.Entry(config_frame, width=15) self.keyword_entry.grid(row=3, column=0, sticky=tk.W, pady=(0, 5)) ttk.Label(config_frame, text="目标价格:").grid(row=2, column=1, sticky=tk.W, pady=(5, 0)) self.price_entry = ttk.Entry(config_frame, width=10) self.price_entry.grid(row=3, column=1, sticky=tk.W, pady=(0, 5)) ttk.Button(config_frame, text="添加", command=self.add_keyword).grid(row=3, column=2, padx=(5, 0)) ttk.Button(config_frame, text="删除选中", command=self.delete_keyword).grid(row=3, column=3, padx=(5, 0)) # 循环间隔 ttk.Label(config_frame, text="循环间隔(秒):").grid(row=4, column=0, sticky=tk.W, pady=(5, 0)) self.interval_var = tk.StringVar(value="10") self.interval_entry = ttk.Entry(config_frame, textvariable=self.interval_var, width=10) self.interval_entry.grid(row=5, column=0, sticky=tk.W, pady=(0, 5)) # 控制按钮 control_frame = ttk.Frame(config_frame) control_frame.grid(row=5, column=1, columnspan=3, sticky=tk.E, pady=(0, 5)) self.start_btn = ttk.Button(control_frame, text="开始", command=self.start_bot) self.start_btn.grid(row=0, column=0, padx=(0, 5)) self.stop_btn = ttk.Button(control_frame, text="停止", command=self.stop_bot, state=tk.DISABLED) self.stop_btn.grid(row=0, column=1, padx=(0, 5)) ttk.Button(control_frame, text="保存配置", command=self.save_config).grid(row=0, column=2, padx=(0, 5)) ttk.Button(control_frame, text="加载配置", command=self.load_config).grid(row=0, column=3) # 日志区域 log_frame = ttk.LabelFrame(main_frame, text="日志", padding="10") log_frame.grid(row=1, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) self.log_text = scrolledtext.ScrolledText(log_frame, height=20, width=80) self.log_text.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S)) # 配置网格权重 self.root.columnconfigure(0, weight=1) self.root.rowconfigure(0, weight=1) main_frame.columnconfigure(0, weight=1) main_frame.rowconfigure(1, weight=1) log_frame.columnconfigure(0, weight=1) log_frame.rowconfigure(0, weight=1) def add_keyword(self): keyword = self.keyword_entry.get().strip() price_str = self.price_entry.get().strip() if not keyword: messagebox.showwarning("警告", "请输入关键词") return try: price = float(price_str) except ValueError: messagebox.showwarning("警告", "请输入有效的价格") return self.tree.insert('', 'end', values=(keyword, price)) self.keyword_entry.delete(0, tk.END) self.price_entry.delete(0, tk.END) def delete_keyword(self): selected = self.tree.selection() if not selected: messagebox.showwarning("警告", "请先选中要删除的项目") return for item in selected: self.tree.delete(item) def get_keywords_list(self): items = self.tree.get_children() result = [] for item in items: values = self.tree.item(item)['values'] if values: result.append((values[0], float(values[1]))) return result def log_message(self, message): self.log_text.insert(tk.END, f"{time.strftime('%H:%M:%S')} - {message}\n") self.log_text.see(tk.END) self.root.update_idletasks() def start_bot(self): keywords = self.get_keywords_list() if not keywords: messagebox.showwarning("警告", "请至少添加一个关键词") return try: interval = int(self.interval_var.get()) if interval < 1: raise ValueError except ValueError: messagebox.showwarning("警告", "请输入有效的间隔秒数（大于0）") return if not ghost_available: result = messagebox.askyesno("警告", "幽灵键鼠未连接，点击将使用系统API，是否继续？") if not result: return self.running = True self.start_btn.config(state=tk.DISABLED) self.stop_btn.config(state=tk.NORMAL) # 在新线程中运行 self.bot_thread = threading.Thread( target=self.bot.run, args=(keywords, interval) ) self.bot_thread.daemon = True self.bot_thread.start() def stop_bot(self): global stop_flag stop_flag = True self.log_message("正在停止...") self.start_btn.config(state=tk.NORMAL) self.stop_btn.config(state=tk.DISABLED) self.running = False def save_config(self): keywords = self.get_keywords_list() config = { 'keywords': keywords, 'interval': self.interval_var.get() } try: with open(config_file, 'w', encoding='utf-8') as f: json.dump(config, f, ensure_ascii=False, indent=2) self.log_message(f"配置已保存到: {config_file}") messagebox.showinfo("成功", "配置已保存") except Exception as e: messagebox.showerror("错误", f"保存配置失败: {e}") def load_config(self): try: if not os.path.exists(config_file): return with open(config_file, 'r', encoding='utf-8') as f: config = json.load(f) # 清空当前列表 for item in self.tree.get_children(): self.tree.delete(item) # 加载关键词 for keyword, price in config.get('keywords', []): self.tree.insert('', 'end', values=(keyword, price)) # 加载间隔 if 'interval' in config: self.interval_var.set(config['interval']) self.log_message("配置已加载") except Exception as e: messagebox.showerror("错误", f"加载配置失败: {e}") def main(): root = tk.Tk() app = AuctionGUI(root) root.mainloop() if __name__ == "__main__": try: main() except Exception as e: print(f"程序出错: {e}") import traceback traceback.print_exc() finally: if ghost_available: try: closedevice() except: pass