import array import json import os import socket import struct import time import win32con import win32gui import constant from utils import ths_ocr_util, invalid_hwnds_manager import setting from utils import socket_util from utils import ths_util import win32_util # freeze_support() import logging import multiprocessing import threading import wx import wx.html2 as webview import concurrent.futures APP_TITLE = "卖票看板" APP_ICON = "" SOCKET_PORT = 11008 HTTP_PORT = 9004 def show_warning(content, click): toastone = wx.MessageDialog(None, content, "提示", wx.YES_DEFAULT | wx.ICON_WARNING) if toastone.ShowModal() == wx.ID_YES: # 如果点击了提示框的确定按钮 if click is not None: click() toastone.Destroy() def show_info(content, click): toastone = wx.MessageDialog(None, content, "提示", wx.YES_DEFAULT | wx.ICON_INFORMATION) if toastone.ShowModal() == wx.ID_YES: # 如果点击了提示框的确定按钮 click() toastone.Destroy() def show_sure(content, callback): toastone = wx.MessageDialog(None, content, "提示", wx.YES_NO | wx.ICON_INFORMATION) result = toastone.ShowModal() if result == wx.ID_YES: # 如果点击了提示框的确定按钮 callback(True) toastone.Destroy() elif result == wx.ID_NO: callback(False) toastone.Destroy() class JueJinSettingFrame(wx.Frame): def __init__(self, position): wx.Frame.__init__(self, None, -1, "掘金参数设置", style=wx.SYSTEM_MENU ^ wx.CLOSE_BOX ^ wx.CAPTION ^ wx.STAY_ON_TOP, size=(450, 200)) self.SetBackgroundColour(wx.Colour(224, 224, 224)) self.SetPosition(wx.Point(position[0] - self.GetSize()[0] / 2, position[1] - self.GetSize()[1] / 2)) boxsier = wx.BoxSizer() flex = wx.FlexGridSizer(rows=3, cols=2, vgap=10, hgap=10) # 策略 label = wx.StaticText(self, label="策略ID：") flex.Add(label, flag=wx.ALIGN_RIGHT) self.edit_celue = wx.TextCtrl(self, size=(300, -1)) flex.Add(self.edit_celue, flag=wx.EXPAND) # token label = wx.StaticText(self, label="Token：") flex.Add(label) self.edit_token = wx.TextCtrl(self, size=(300, -1), style=wx.TE_MULTILINE) flex.Add(self.edit_token) # 占位 flex.Add(wx.StaticText(self, label="")) # 确定按钮 ID_SURE = wx.NewId() self.btn_sure = wx.Button(self, label='确定', id=ID_SURE) self.btn_sure.Bind(wx.EVT_BUTTON, self.__sure_btn) flex.Add(self.btn_sure, 1, wx.TOP | wx.LEFT, 20) boxsier.Add(flex, 1, wx.TOP | wx.LEFT, 20) self.SetSizer(boxsier) # 初始化数据 self.__init_data() def __init_data(self): strategy_id, token = setting.get_juejin_params() self.edit_celue.SetLabelText(strategy_id) self.edit_token.SetLabelText(token) pass def __sure_btn(self, event): strategy_id = self.edit_celue.GetValue() token = self.edit_token.GetValue() setting.set_juejin_params(strategy_id, token) toastone = wx.MessageDialog(None, "更改成功", "提示", wx.YES_DEFAULT | wx.ICON_INFORMATION) if toastone.ShowModal() == wx.ID_YES: # 如果点击了提示框的确定按钮 self.Close() toastone.Destroy() class FobiddenCodesFrame(wx.Frame): def __init__(self, position): wx.Frame.__init__(self, None, -1, "添加禁止交易代码", style=wx.SYSTEM_MENU ^ wx.CLOSE_BOX ^ wx.CAPTION ^ wx.STAY_ON_TOP, size=(170, 200)) self.SetBackgroundColour(wx.Colour(224, 224, 224)) self.SetPosition(wx.Point(position[0] - self.GetSize()[0] / 2, position[1] - self.GetSize()[1] / 2)) boxsier = wx.BoxSizer(wx.VERTICAL) # 代码 label = wx.StaticText(self, label="代码：") boxsier.Add(label, flag=wx.ALIGN_LEFT, border=10) self.edit_codes = wx.TextCtrl(self, size=(150, 100), style=wx.TE_MULTILINE) boxsier.Add(self.edit_codes) # 确定按钮 ID_SURE = wx.NewId() self.btn_sure = wx.Button(self, label='确定', id=ID_SURE) self.btn_sure.Bind(wx.EVT_BUTTON, self.__sure_btn) boxsier.Add(self.btn_sure) root_boxsier = wx.BoxSizer(wx.HORIZONTAL) root_boxsier.Add(boxsier, 1, wx.TOP | wx.TOP, 10) self.SetSizer(root_boxsier) def __request(self, codes): client = socket.socket() # 生成socket，连接server ip_port = (constant.SERVER_HOST, SOCKET_PORT) # server地址和端口号（最好是10000以后） client.connect(ip_port) data = {"type": 201, "data": {"codes": codes}} client.send(json.dumps(data).encode("utf-8")) client.close() def __sure_btn(self, event): codes_str = self.edit_codes.GetValue() if codes_str: codes_str = codes_str.strip() codes = codes_str.split("\n") codes_result = [] for code in codes: if code.strip() and len(code.strip()) == 6: codes_result.append(code.strip()) if len(codes_result) == 0: show_warning("请填写正确的代码！", self.Close) return try: self.__request(codes_result) except Exception as e: show_warning("添加出错：" + str(e), None) return toastone = wx.MessageDialog(None, "添加成功", "提示", wx.YES_DEFAULT | wx.ICON_INFORMATION) if toastone.ShowModal() == wx.ID_YES: # 如果点击了提示框的确定按钮 self.Close() toastone.Destroy() class THSPositionSettingFrame(wx.Frame): def __init__(self, position): wx.Frame.__init__(self, None, -1, "同花顺坐标设置", style=wx.SYSTEM_MENU ^ wx.CLOSE_BOX ^ wx.CAPTION ^ wx.STAY_ON_TOP, size=(170, 400)) self.SetBackgroundColour(wx.Colour(224, 224, 224)) self.SetPosition(wx.Point(position[0] - self.GetSize()[0] / 2, position[1] - self.GetSize()[1] / 2)) boxsier = wx.BoxSizer(wx.VERTICAL) label = wx.StaticText(self, label="交易刷新事件间隔(ms)：") boxsier.Add(label) self.edit_refresh_time_space = wx.TextCtrl(self, size=(150, -1)) boxsier.Add(self.edit_refresh_time_space) label = wx.StaticText(self, label="点击时间间隔(ms)：") boxsier.Add(label) self.edit_time_space = wx.TextCtrl(self, size=(150, -1)) boxsier.Add(self.edit_time_space) # 代码 label = wx.StaticText(self, label="坐标：") boxsier.Add(label) self.edit_codes = wx.TextCtrl(self, size=(150, 200), style=wx.TE_MULTILINE) boxsier.Add(self.edit_codes) # 确定按钮 ID_SURE = wx.NewId() self.btn_sure = wx.Button(self, label='确定', id=ID_SURE) self.btn_sure.Bind(wx.EVT_BUTTON, self.__sure_btn) boxsier.Add(self.btn_sure) root_boxsier = wx.BoxSizer(wx.HORIZONTAL) root_boxsier.Add(boxsier, 1, wx.TOP | wx.TOP, 10) self.SetSizer(root_boxsier) # 初始化数据 self.__init_data() def __init_data(self): pos = setting.get_ths_auto_click_positions() self.edit_codes.SetValue("\n".join(pos)) space = setting.get_ths_auto_click_time_space() if space is None: space = 500 self.edit_time_space.SetValue(f"{space}") space = setting.get_ths_auto_refresh_time_space() if space is None: space = 500 self.edit_refresh_time_space.SetValue(f"{space}") def __sure_btn(self, event): codes_str = self.edit_codes.GetValue() ps = codes_str.split("\n") result = [] for p in ps: if p.strip(): result.append(p.strip()) setting.set_ths_auto_click_positions(result) time_space = self.edit_time_space.GetValue() if len(time_space) == 0: show_warning("点击时间间隔不正确", None) return refresh_time_space = self.edit_refresh_time_space.GetValue() if len(refresh_time_space) == 0: show_warning("刷新时间间隔不正确", None) return setting.set_ths_auto_click_time_space(time_space) setting.set_ths_auto_refresh_time_space(refresh_time_space) toastone = wx.MessageDialog(None, "更改成功", "提示", wx.YES_DEFAULT | wx.ICON_INFORMATION) if toastone.ShowModal() == wx.ID_YES: # 如果点击了提示框的确定按钮 self.Close() toastone.Destroy() class SocketApiUtil: @classmethod def __request(cls, data_json): socket_util.encryp_client_params_sign(data_json) client = socket.socket() # 生成socket，连接server ip_port = (constant.SERVER_HOST, 11009) # server地址和端口号（最好是10000以后） client.connect(ip_port) client.send(socket_util.load_header(json.dumps(data_json).encode("utf-8"))) result_str, header = socket_util.recv_data(client) return result_str @classmethod def get_cost_price(cls, code): result_str = cls.__request({"type": "get_cost_price", "data": {"code": code}}) print(result_str) result = json.loads(result_str) if result['code'] == 0: return result['data']['price'] raise Exception(result['msg']) class mainApp(wx.App): def __init_data(self): pass def OnInit(self): self.SetAppName(APP_TITLE) self.__init_data() # self.__show_main_frame() return True def ths_auto_click(): hwnd = ths_util.get_ths_second_screen_menu_hwnd() while True: try: if hwnd is None or not win32gui.IsWindowVisible(hwnd): # print("未找到同花顺副屏句柄") hwnd = ths_util.get_ths_second_screen_menu_hwnd() if hwnd is None: continue if not setting.is_ths_auto_click(): continue ps = setting.get_ths_auto_click_positions() if not ps: continue ps_new = [] for p in ps: p = eval(p) ps_new.append(p) space = setting.get_ths_auto_click_time_space() if space is None: space = 500 ths_util.betch_click(hwnd, ps_new, round(space / 1000, 4)) except Exception as e: pass finally: time.sleep(0.02) def ths_auto_refresh(): hwnd = ths_util.get_trade_refesh_hwnd() count = 0 while True: count += 1 if count > 10: count = 0 hwnd = ths_util.get_trade_refesh_hwnd() try: if hwnd is None: continue # 测试 if not setting.is_ths_trade_auto_refresh(): continue rect = win32gui.GetWindowRect(hwnd) win32_util.visual_click(hwnd, (160, (rect[3] - rect[1]) // 2)) time_space = setting.get_ths_auto_refresh_time_space() if time_space is None: time_space = 500 time.sleep(round(time_space / 1000, 4)) except Exception as e: pass finally: time.sleep(0.02) if __name__ == "__main__1": print(webview.WebView.IsBackendAvailable(webview.WebViewBackendEdge)) set_codes_thread_pool = concurrent.futures.ThreadPoolExecutor(max_workers=10) # 接受来自其他进程的数据 def recieve_data(pipe): while True: try: value = pipe.recv() data = json.loads(value) print("接受到数据：", data) type_ = data["type"] if type_ == "juejin_setting": wx.CallAfter(lambda: JueJinSettingFrame(data["pos"]).Show()) elif type_ == "manage_ths_pos": wx.CallAfter(lambda: THSPositionSettingFrame(data["pos"]).Show()) elif type_ == "set_code": code = data["code"] set_codes_thread_pool.submit(set_xfdp_codes, code) elif type_ == "show_dead_hwnds": hwnds = win32_util.text_hwnds_doing print("死亡窗口", hwnds) if hwnds: for hwnd in hwnds: invalid_hwnds_manager.add_hwnd(hwnd) elif type_ == "exit": os._exit(0) except Exception as e: logging.exception(e) from pynput.mouse import Listener ocr_settings = {} def set_xfdp_codes(code): """ 设置悬浮盯盘中的代码 :param code: :return: """ def focus(hwnd, code_): try: int_buffer = array.array("L", [0]) char_buffer = array.array('b', json.dumps({"type": "set_code", "data": {"code": code_, "time": int(time.time() * 1000)}}).encode()) int_buffer_address = int_buffer.buffer_info()[0] char_buffer_address, char_buffer_size = char_buffer.buffer_info() copy_struct = struct.pack("PLP", int_buffer_address, char_buffer_size, char_buffer_address) win32gui.SendMessage(hwnd, win32con.WM_COPYDATA, None, copy_struct) # pythoncom.CoInitialize() # shell = client.Dispatch("WScript.Shell") # shell.SendKeys('%') # # win32gui.ShowWindow(hwnd, win32con.SW_RESTORE) # win32gui.SetForegroundWindow(hwnd) # win32gui.PumpMessages() # win32gui.SetFocus(hwnd) # lib = CDLL("D:/workspace/GP/dll/Dll/x64/Debug/Dll.dll") # win32gui.SendMessage(hwnd, win32con.WM_SETFOCUS, 0, 0) # print("执行结果", lib.focus(hwnd)) # remote_thread = win32process.GetWindowThreadProcessId(hwnd)[0] # tid = win32api.GetCurrentThreadId() # win32process.AttachThreadInput(tid, remote_thread, True) # prev_handle = win32gui.SetFocus(hwnd) # print("之前的焦点句柄", prev_handle,hwnd) except Exception as e: logging.exception(e) pass # 获取代码属性描述 # auto_focus = setting.get_float_frame_auto_focus() if True: # 窗口显示在最前面 hwnds = win32_util.search_window("悬浮盯盘") if hwnds: print("句柄", hwnds[0]) time.sleep(0.1) focus(hwnds[0], code) def on_mouse_click(x, y, button, pressed): """ 鼠标点击事件 :param x: :param y: :param button: :param pressed: :return: """ if pressed: # 点击事件过后1s内可截图识别代码 ocr_settings["expire_time"] = time.time() + 1 # print(f"鼠标在 ({x}, {y}) 位置按下，按钮: {button}") else: # print(f"鼠标在 ({x}, {y}) 位置释放，按钮: {button}") pass def __rec_code(pipe): time.sleep(5) while True: if "expire_time" in ocr_settings and ocr_settings["expire_time"] > time.time(): try: code = ths_ocr_util.ocr_ths_code() if not code: time.sleep(0.1) continue pipe.send(json.dumps({"code": code, "type": "code"})) except Exception as e: # logging.exception(e) # print(str(e)) pass time.sleep(0.005) def run(pipe): global p2 p1, p2 = multiprocessing.Pipe() threading.Thread(target=lambda: __rec_code(pipe), daemon=True).start() threading.Thread(target=lambda: recieve_data(pipe), daemon=True).start() threading.Thread(target=lambda: ths_auto_click(), daemon=True).start() threading.Thread(target=lambda: ths_auto_refresh(), daemon=True).start() app = mainApp() app.MainLoop() with Listener(on_click=on_mouse_click) as listener: listener.join() while True: time.sleep(2) if __name__ == "__main__": app = mainApp() app.MainLoop()