import ctypes import platform import subprocess import sys import os import threading import time from datetime import datetime from PyQt5.QtCore import Qt, QUrl, QDateTime, QThread, pyqtSignal, pyqtSlot from PyQt5.QtWidgets import QApplication, QMainWindow, QLabel, QFileDialog, QMessageBox, QDateTimeEdit, QPushButton from PyQt5.QtGui import QDesktopServices from app import login_account, run_threading_order, run_daily_job, run_manual_job, cancel_manual_job, stop_order_job, \ write_ticket_info from litter_helper import Ui_menu # 确保这个导入路径是正确的 from tools.utils import save_all_proxy_ip_v3 import winreg from pathlib import Path # 定义 Redis 安装程序路径 REDIS_INSTALLER_PATH = "Redis-x64-5.0.14.1.msi" # 替换为实际的安装程序路径 class ClickableLabel(QLabel): def __init__(self, parent=None, file_path=None): super().__init__(parent) self.file_path = file_path def mousePressEvent(self, event): if event.button() == Qt.LeftButton: self.emit_click_event() def emit_click_event(self): if self.file_path: self.parent().on_label_click(self.file_path) def get_resource_path(relative_path): """获取资源文件的绝对路径""" if getattr(sys, 'frozen', False): # 是否为PyInstaller打包的exe文件 # 返回exe文件所在的绝对路径 base_path = os.path.dirname(sys.executable) else: # 在开发环境下运行 # 返回脚本文件所在的绝对路径 base_path = os.path.dirname(__file__) return os.path.join(base_path, relative_path) def get_dir_path(relative_path): """获取资源文件路径""" base_path = getattr(sys, '_MEIPASS', str(Path(__file__).parent.absolute())) return Path(base_path, relative_path) # 停止定时任务 def on_stop_time_button_click(): print(f"停止定时任务: {time.time()}") cancel_manual_job() # 写入文件 def on_write_file_button_click(): print(f"写入文件：{time.time()}") write_ticket_info("abcd", "1234556") class WorkerThread(QThread): """用于执行下单逻辑的线程""" signal = pyqtSignal(str) # 定义一个信号，用于将信息传递回主线程 def __init__(self, thread_num, accounts_file, net_url): super().__init__() self.thread_num = thread_num self.accounts_file = accounts_file self.net_url = net_url def run(self): try: run_threading_order(self.thread_num, self.accounts_file, self.net_url) self.signal.emit("下单完成") # 发送信号，通知主线程下单完成 except Exception as e: self.signal.emit(f"下单失败: {e}") # 发送信号，通知主线程下单失败 class LoginThread(QThread): """用于执行登录逻辑的线程""" signal = pyqtSignal(str) # 定义一个信号，用于将信息传递回主线程 def __init__(self, thread_num, accounts_file, pocket_url): super().__init__() self.thread_num = thread_num self.accounts_file = accounts_file self.pocket_url = pocket_url def run(self): try: login_account(self.thread_num, self.accounts_file, self.pocket_url) self.signal.emit("登录成功") # 发送信号，通知主线程登录成功 except Exception as e: self.signal.emit(f"登录失败: {e}") # 发送信号，通知主线程登录失败 class MainWindow(QMainWindow, Ui_menu): def __init__(self, param1, param2): super().__init__() self.setupUi(self) # 使用传递的参数 self.param1 = param1 self.param2 = param2 # 设置标签的文本为传递的参数 # self.label.setText(f"参数1: {self.param1}, 参数2: {self.param2}") self.concurrencyLineEdit.setText("1") self.urlLineEdit.setText("https://t.livepocket.jp/e/l3im7") # 登录账号 self.loginPushButton.clicked.connect(self.on_login_button_click) # 开始下单 self.startPushButton.clicked.connect(self.on_start_button_click) # 停止下单 self.stopOrderPushButton.clicked.connect(self.on_stop_button_click) # 定时开始 self.startTimePushButton.clicked.connect(self.on_start_time_button_click) # 停止定时任务 self.stopTimePushButton.clicked.connect(on_stop_time_button_click) # 写入文件 # self.writeFilePushButton.clicked.connect(on_write_file_button_click) # 编辑账号 # 获取 QLabel 控件并替换为 ClickableLabel self.accountsLabel = self.findChild(QLabel, "accountsLabel") if self.accountsLabel: # 替换为 ClickableLabel self.clickableAccountsLabel = ClickableLabel(self, file_path=get_resource_path("accounts.txt")) self.clickableAccountsLabel.setObjectName("accountsLabel") self.clickableAccountsLabel.setText("编辑账号") self.clickableAccountsLabel.setGeometry(self.accountsLabel.geometry()) self.clickableAccountsLabel.show() self.accountsLabel.deleteLater() # 删除原来的 QLabel 控件 # 编辑IP代理池 # 获取 QLabel 控件并替换为 ClickableLabel self.ipPoolLabel = self.findChild(QLabel, "ipPoolLabel") if self.ipPoolLabel: # 替换为 ClickableLabel self.clickableIpPoolLabel = ClickableLabel(self, file_path=get_resource_path("proxy_list.txt")) self.clickableIpPoolLabel.setObjectName("ipPoolLabel") self.clickableIpPoolLabel.setText("编辑代理池") self.clickableIpPoolLabel.setGeometry(self.ipPoolLabel.geometry()) self.clickableIpPoolLabel.show() self.ipPoolLabel.deleteLater() # 删除原来的 QLabel 控件 # 获取 QDateTimeEdit 控件并设置显示格式 self.startDateTimeEdit = self.findChild(QDateTimeEdit, "startDateTimeEdit") if self.startDateTimeEdit: self.startDateTimeEdit.setDisplayFormat("yyyy-MM-dd HH:mm:ss") # 设置当前时间 self.startDateTimeEdit.setDateTime(QDateTime.currentDateTime()) # 获取 QDateTimeEdit 控件并设置显示格式 self.endDateTimeEdit = self.findChild(QDateTimeEdit, "endDateTimeEdit") if self.endDateTimeEdit: self.endDateTimeEdit.setDisplayFormat("yyyy-MM-dd HH:mm:ss") # 设置当前时间 self.endDateTimeEdit.setDateTime(QDateTime.currentDateTime()) # 连接 watchPushButton 点击事件到槽函数 self.watchPushButton = self.findChild(QPushButton, "watchPushButton") if self.watchPushButton: self.watchPushButton.clicked.connect(self.on_watch_button_click) # 检查 Redis 是否安装 self.check_redis_installation() def closeEvent(self, event): """重写 closeEvent 方法，在窗口关闭时终止程序""" reply = QMessageBox.question(self, '退出', "确定要退出吗？", QMessageBox.Yes | QMessageBox.No, QMessageBox.No) if reply == QMessageBox.Yes: event.accept() # 接受关闭事件 sys.exit() # 退出应用程序 else: event.ignore() # 忽略关闭事件 def on_label_click(self, file_path): if os.path.exists(file_path): try: QDesktopServices.openUrl(QUrl.fromLocalFile(file_path)) except Exception as e: QMessageBox.critical(self, "Error", f"无法打开文件: {e}") else: QMessageBox.critical(self, "Error", f"文件不存在: {file_path}") def on_start_button_click(self): print(f"开始下单!") # 在按钮点击时执行的操作 # print(f"按钮被点击了！参数1: {self.param1}, 参数2: {self.param2}") # 获取 QDateTimeEdit 控件的值 if self.startDateTimeEdit: date_time_value = self.startDateTimeEdit.dateTime() # print(f"选中的日期和时间: {date_time_value.toString('yyyy-MM-dd HH:mm:ss')}") # 获取 QDateTimeEdit 控件的值 if self.endDateTimeEdit: date_time_value = self.endDateTimeEdit.dateTime() # print(f"选中的日期和时间: {date_time_value.toString('yyyy-MM-dd HH:mm:ss')}") # 获取并发数 thread_num = 1 if self.concurrencyLineEdit: # print(f"并发数为： {self.concurrencyLineEdit.text()}") thread_num = self.concurrencyLineEdit.text() # url net_url = None if self.urlLineEdit: net_url = self.urlLineEdit.text() # 创建线程并启动 self.worker_thread = WorkerThread(thread_num, get_resource_path("accounts.txt"), net_url) self.worker_thread.signal.connect(self.on_worker_thread_signal) self.worker_thread.start() # 登录账号 def on_login_button_click(self): print(f"登录账号! {self.urlLineEdit.text()}") thread_num = 1 if self.concurrencyLineEdit.text(): thread_num = self.concurrencyLineEdit.text() pocket_url = None if self.urlLineEdit.text(): pocket_url = self.urlLineEdit.text() save_all_proxy_ip_v3(get_resource_path("proxy_list.txt")) # 创建登录线程 self.login_thread = LoginThread(thread_num, get_resource_path("accounts.txt"), pocket_url) self.login_thread.signal.connect(self.on_login_thread_signal) self.login_thread.start() else: QMessageBox.warning(self, '警告', '请输入URL') return # 定时开始 def on_start_time_button_click(self): print(f"定时开始：{time.time()}") net_url = None if self.urlLineEdit: net_url = self.urlLineEdit.text() start_time = self.startDateTimeEdit.text() print(f"start_time:{start_time}") # 使用datetime模块解析时间字符串 dt = datetime.strptime(start_time, '%Y-%m-%d %H:%M:%S') # 提取出时、分、秒 hour = dt.hour minute = dt.minute second = dt.second print(f"Hour: {hour}, Minute: {minute}, Second: {second}") # 获取并发数 thread_num = self.concurrencyLineEdit.text() # 创建一个新的线程来运行 run_manual_job 函数 thread = threading.Thread(target=run_manual_job, args=(hour, minute, thread_num, net_url)) thread.start() else: QMessageBox.warning(self, '警告', '请输入URL') return # 终止下单 def on_stop_button_click(self): print(f"终止下单! 目标网址: {self.urlLineEdit.text()}") stop_order_job() def edit_accounts(self): # 编辑账号 print(f"编辑账号: {self.urlLineEdit.text()}") # 查看结果 def on_watch_button_click(self): # 获取项目当前路径 current_path = sys.path[0] # 拼接文件夹路径 folder_path = get_dir_path("results") # 假设文件夹名为 "results" # 判断文件夹是否存在，不存在则创建 if not folder_path.exists(): try: folder_path.mkdir(parents=True, exist_ok=True) print(f"文件夹 {folder_path} 已创建") except Exception as e: QMessageBox.critical(self, "Error", f"无法创建文件夹: {e}") return # 打开文件夹 folder_path = get_resource_path("results") # 假设文件夹名为 "results" try: # 使用 QDesktopServices.openUrl 打开文件夹 QDesktopServices.openUrl(QUrl.fromLocalFile(folder_path)) except Exception as e: QMessageBox.critical(self, "Error", f"无法打开文件夹: {e}") # 检查 Redis 是否安装 def check_redis_installation(self): # 检查操作系统 system = platform.system() if system == "Windows": # 检查 Redis 服务是否正在运行 try: key = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, r"SYSTEM\CurrentControlSet\Services\Redis") winreg.CloseKey(key) print("Redis 服务已安装并正在运行") return except FileNotFoundError as ex: print(f"Redis 服务未安装或未运行") # 如果 Redis 服务未安装，则启动安装程序 if REDIS_INSTALLER_PATH: try: subprocess.run([REDIS_INSTALLER_PATH], shell=True) print("Redis 安装程序已启动") except Exception as e: QMessageBox.critical(self, "Error", f"无法启动 Redis 安装程序: {e}") else: # 在其他操作系统上，可以使用不同的方法检查 Redis 是否安装 # 例如，可以使用 `which redis-server` 命令 print("当前操作系统不支持自动安装 Redis") # 处理线程信号 @pyqtSlot(str) def on_worker_thread_signal(self, message): QMessageBox.information(self, "下单结果", message) # 处理登录线程信号 @pyqtSlot(str) def on_login_thread_signal(self, message): QMessageBox.information(self, "登录结果", message) if __name__ == "__main__": app = QApplication(sys.argv) # 创建窗口并传递参数 window = MainWindow("Hello", "World") window.show() sys.exit(app.exec_())