Python一键完成奥智利AZL-212C系列分机刷机示例
# -*- coding: utf-8 -*- ################################################################################ ## Form generated from reading UI file 'app.ui' ## ## Created by: Qt User Interface Compiler version 6.6.2 ## ## WARNING! All changes made in this file will be lost when recompiling UI file! ################################################################################ from PySide6.QtCore import (QCoreApplication, QDate, QDateTime, QLocale, QMetaObject, QObject, QPoint, QRect, QSize, QTime, QUrl, Qt) from PySide6.QtGui import (QBrush, QColor, QConicalGradient, QCursor, QFont, QFontDatabase, QGradient, QIcon, QImage, QKeySequence, QLinearGradient, QPainter, QPalette, QPixmap, QRadialGradient, QTransform) from PySide6.QtWidgets import (QApplication, QComboBox, QLabel, QMainWindow, QPushButton, QSizePolicy, QStatusBar, QTextEdit, QWidget) class Ui_MainWindow(object): def setupUi(self, MainWindow): if not MainWindow.objectName(): MainWindow.setObjectName(u"MainWindow") MainWindow.resize(611, 630) MainWindow.setStyleSheet(u"font: 9pt \"Microsoft YaHei UI\";") self.centralwidget = QWidget(MainWindow) self.centralwidget.setObjectName(u"centralwidget") self.comboBox_networkcard = QComboBox(self.centralwidget) self.comboBox_networkcard.setObjectName(u"comboBox_networkcard") self.comboBox_networkcard.setGeometry(QRect(90, 10, 260, 22)) self.label_networkcard = QLabel(self.centralwidget) self.label_networkcard.setObjectName(u"label_networkcard") self.label_networkcard.setGeometry(QRect(20, 10, 53, 20)) self.label_class = QLabel(self.centralwidget) self.label_class.setObjectName(u"label_class") self.label_class.setGeometry(QRect(20, 50, 53, 20)) self.comboBox_class = QComboBox(self.centralwidget) self.comboBox_class.setObjectName(u"comboBox_class") self.comboBox_class.setGeometry(QRect(90, 50, 261, 22)) self.textEdit_networkadapter = QTextEdit(self.centralwidget) self.textEdit_networkadapter.setObjectName(u"textEdit_networkadapter") self.textEdit_networkadapter.setGeometry(QRect(20, 130, 161, 180)) self.label_networkadapter = QLabel(self.centralwidget) self.label_networkadapter.setObjectName(u"label_networkadapter") self.label_networkadapter.setGeometry(QRect(20, 100, 131, 20)) self.label_device = QLabel(self.centralwidget) self.label_device.setObjectName(u"label_device") self.label_device.setGeometry(QRect(190, 100, 121, 20)) self.textEdit_device = QTextEdit(self.centralwidget) self.textEdit_device.setObjectName(u"textEdit_device") self.textEdit_device.setGeometry(QRect(190, 130, 161, 180)) self.textEdit_log = QTextEdit(self.centralwidget) self.textEdit_log.setObjectName(u"textEdit_log") self.textEdit_log.setGeometry(QRect(20, 330, 571, 241)) self.textEdit_configlabel = QTextEdit(self.centralwidget) self.textEdit_configlabel.setObjectName(u"textEdit_configlabel") self.textEdit_configlabel.setGeometry(QRect(360, 10, 231, 301)) self.label_log = QLabel(self.centralwidget) self.label_log.setObjectName(u"label_log") self.label_log.setGeometry(QRect(20, 310, 121, 20)) self.pushButton_update = QPushButton(self.centralwidget) self.pushButton_update.setObjectName(u"pushButton_update") self.pushButton_update.setGeometry(QRect(20, 580, 571, 31)) MainWindow.setCentralWidget(self.centralwidget) self.statusbar = QStatusBar(MainWindow) self.statusbar.setObjectName(u"statusbar") MainWindow.setStatusBar(self.statusbar) self.retranslateUi(MainWindow) QMetaObject.connectSlotsByName(MainWindow) # setupUi def retranslateUi(self, MainWindow): MainWindow.setWindowTitle(QCoreApplication.translate("MainWindow", u"\u9ad8\u6e05\u53ef\u89c6\u5206\u673a(C\u673a)\u81ea\u52a8\u5237\u673a\u5de5\u5177", None)) self.label_networkcard.setText(QCoreApplication.translate("MainWindow", u"\u9009\u62e9\u7f51\u5361", None)) self.label_class.setText(QCoreApplication.translate("MainWindow", u"\u9009\u62e9\u90e8\u95e8", None)) self.label_networkadapter.setText(QCoreApplication.translate("MainWindow", u"\u672c\u673a\u7f51\u7edc\u9002\u914d\u5668\u914d\u7f6e\u4fe1\u606f", None)) self.label_device.setText(QCoreApplication.translate("MainWindow", u"\u540c\u6bb5\u53ef\u89c6\u5bf9\u8bb2\u8bbe\u5907\u4fe1\u606f", None)) self.label_log.setText(QCoreApplication.translate("MainWindow", u"\u8fd0\u884c\u65e5\u5fd7", None)) self.pushButton_update.setText(QCoreApplication.translate("MainWindow", u"\u4e00\u952e\u5237\u673a", None)) # retranslateUiimport binascii import os import socket import subprocess import ipaddress import sys import time from pathlib import Path from lxml import etree import pandas as pd import psutil from PySide6.QtCore import QThread, Signal from PySide6.QtWidgets import QApplication, QMainWindow from app import Ui_MainWindow # 初始化全局变量 config_table_list = [] network_card_list = [] device_config_list = [] device_information_list = [] def Parse_Device_Xml(data: str) -> list: root = etree.fromstring(data.encode('utf-8')) code = root.findtext('CODE') devname = root.findtext('DEVNAME') ip = root.findtext('IP') mask = root.findtext('MASK') gateway = root.findtext('GATEWAY') mac = root.findtext('MAC') version = root.findtext('VERSION') device_information_list.clear() device_information_list.extend([code, devname, ip, mask, gateway, mac, version]) return device_information_list def Get_Networkcard_Ip_And_Netmask(interface_name: str) -> list: """ 获取指定网卡的IP地址和子网掩码 :param interface_name: str,连接名称 :return: list,ip和子网掩码 """ interface_info = psutil.net_if_addrs()[interface_name] # 查找IPv4地址，子网掩码 for snic in interface_info: if snic.family == socket.AddressFamily.AF_INET: ip_address = snic.address netmask = snic.netmask return [ip_address, netmask] def Check_If_Key_Files_Exist() -> bool: """ 检测软件运行目录是否存在xml文件和xls文件 :return: str,检测结果 """ xls_path = [str(p) for p in Path(os.getcwd()).rglob("*.xls")] xml_path = [str(p) for p in Path(os.getcwd()).rglob("*.xml")] if len(xls_path) > 0 and len(xml_path) > 0: return True else: return False def Extension_Number_Packaging(number: str) -> str: """ 分机号数据包组包 :param number: str,分机号 :return: str,十六进制格式数据包 """ data = ''.join([h.replace('0x', '00') for h in [hex(ord(c)) for c in number]]) package = "BBBB666601100A00000000000000000020000000A4F619" + data + "0000002D02C8F6190040884300" return str(package) def Disable_Network_Adapter(adapter_name: str) -> bool: """ 禁用网络适配器 :param adapter_name: str,连接名称 :return: 如果命令执行成功，则为True，否则为False """ try: # 使用 subprocess 调用命令 result = os.system(f'wmic path win32_networkadapter where NetConnectionID="{adapter_name}" call disable') return result == 0 # 0 表示命令执行成功 except Exception as e: print(f"禁用网络适配器时出错: {e}") return False def Ping(ip: str) -> bool: """ 测试与指定IP能否正常通信 :param ip: str,ip :return: 如果命令执行成功，则为True，否则为False """ try: # 使用ping命令，发送1个包 response = os.system("ping -c 1 " + ip) # 检查响应值，如果为0，表示可以ping通 return response == 0 # 0 表示命令执行成功 except Exception as e: print(f"执行Ping命令时出错: {e}") return False def Calculate_Gateway_Address(ip: str, netmask: str) -> str: """ 计算网关地址 :param ip: ip :param netmask: 子网掩码 :return: """ # 给定的IP地址和子网掩码 _ip = ipaddress.IPv4Address(ip) _netmask = ipaddress.IPv4Address(netmask) # 计算网络地址 network = ipaddress.IPv4Network(_ip.exploded + '/' + _netmask.exploded, strict=False) # 打印子网的第二个IP地址 return str(network.network_address + 1) def Read_Config_Xls_To_List() -> list: """ 读取xls配置表到列表 :return: list,分机号开头为1的C机配置表 """ xls_path = [str(p) for p in Path(os.getcwd()).rglob("*.xls")] df = pd.read_excel(xls_path[0]) for i in range(df.shape[0]): data = str(df.iloc[i, 0]) if data[0] != "1": continue else: key = str(df.iloc[i, 1]) value = str(df.iloc[i, 0]) + "|" + str(df.iloc[i, 2]) + "|" + str(df.iloc[i, 3]) + "|" + str(df.iloc[i, 4]) config_table_list.append({key: value}) return config_table_list def Read_Config_Xml(): """ 读取xml配置表 :return: hex,配置表数据字节集 """ xml_path = [str(p) for p in Path(os.getcwd()).rglob("*.xml")] with open(xml_path[0], 'rb') as f: data = f.read() hex_data = binascii.hexlify(data).decode().upper() return hex_data def Get_Pc_Network_Card(): """ 获取本计算机所有网卡连接名称 :return: list,本机网卡连接名称列表 """ interfaces = psutil.net_if_addrs() for interface_name in interfaces: # 检查每个网络接口是否有IPv4地址 for snic in interfaces[interface_name]: if snic.family == socket.AF_INET: network_card_list.append(interface_name) return network_card_list def Send_Data_To_Device(ip: str, packets: list) -> bool: """ 发送数据到可视对讲分机 :param ip: 设备IP :param packets: list,数据包(可多个，使用列表格式传入) :return: bool,成功返回True，失败返回False """ s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: connection_status = Ping(ip) while True: if connection_status: # 如果可以ping通分机则建立TCP连接 s.connect((ip, 34918)) for packet in packets: # 发送数据包 data = binascii.unhexlify(packet) sent = s.send(data) while True: if sent == len(data): # 校验数据包是否发送成功，失败则重发 break else: # 重新发送数据包 sent = s.send(data) return True else: # 如果无法ping通分机，则等1s后再次尝试 time.sleep(1) connection_status = Ping(ip) except socket.error as e: print("将数据包传输到分机设备失败，错误信息:", e) return False finally: # 关闭套接字 s.close() def Search_Device_Info(): """ 向局域网发送搜索设备组播消息 :return: """ # 创建一个udp套接字 udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # 设置socket属性，允许发送广播数据 udp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) # 准备接收方的地址 dest_addr = ('255.255.255.255', 3001) # 注意这是一个元组 # 自定义广播数据 send_data = ('<?xml version="1.0" encoding="GB2312" ' '?><XML_MSG_BODY><XML_MSG_TYPE>SEARCH_DEV_INFO</XML_MSG_TYPE><XML_LOCAL_IP>45.91.3.146' '</XML_LOCAL_IP></XML_MSG_BODY>') try: # 可以使用套接字收发数据 udp_socket.sendto(send_data.encode('GB2312'), dest_addr) except socket.timeout: print("发送搜索分机广播消息超时，正在重试...") time.sleep(1) except OSError as e: if e.winerror == 10065: # [WinError 10065] No route to host print("发送搜索分机广播消息时没有到主机的路由，正在重试...") time.sleep(1) else: print(f"发送搜索分机广播消息时出现意外错误: {e}") finally: # 关闭套接字 udp_socket.close() def Listen_Device_Info() -> str: """ 监听局域网中分机返回的消息 :return: str,分机返回的消息数据 """ # 1.创建一个udp套接字 udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # 2.绑定本地的相关信息，如果一个网络程序不绑定，则系统会随机分配 local_addr = ("", 3002) udp_socket.bind(local_addr) udp_socket.settimeout(1) # 设置套接字超时时间为1秒 try: while True: try: # 3. 等待接收对方发送的数据 recv_data = udp_socket.recvfrom(1024) # 1024表示本次接收的最大字节数 # 接收到的数据recv_data是一个元组,第1个元素是对方发送的数据,第2个元素是对方的ip和端口 print(f"接收到分机消息，{recv_data}") return str(recv_data[0].decode('GB2312')) except socket.timeout: print("未接收到分机消息，正在重试。。。") Search_Device_Info() finally: # 3.关闭套接字 udp_socket.close() def Set_Computer_Network_Card_Dhcp(name: str) -> bool: """ 设置本计算机指定网卡为DHCP模式 :param name: 连接名称 :return: 如果命令都成功执行，则为True，否则为False """ print(name) try: # 设置IP地址为DHCP ip_result = subprocess.run(f'netsh interface ip set address name="{name}" source=dhcp', shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) # 设置DNS为DHCP dns_result = subprocess.run(f'netsh interface ip set dns name="{name}" source=dhcp', shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) print(f"设置指定网卡DHCP返回值: {ip_result.returncode},{dns_result.returncode}") return True except Exception as e: print(f"Error setting network adapter to DHCP: {e}") return False def Set_Computer_Network_Card_Ip(name: str, addr: str, mask: str, gateway: str) -> bool: """ 设置本计算机指定网卡的ip地址，子网掩码，网关地址 :param name: str,连接名称 :param addr: str,ip :param mask: str,子网掩码 :param gateway: str,网关地址 :return: 如果命令都成功执行，则为True，否则为False """ try: # 设置IP地址为静态 ip_result = subprocess.run( f'netsh interface ip set address name="{name}" source=static addr={addr} mask={mask} gateway={gateway} gwmetric=1', shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) print(f"设置指定网卡IP返回值: {ip_result.returncode}") return True except Exception as e: print(f"Error setting network adapter IP: {e}") return False class Worker(QThread): """ 自定义多线程类 """ finished_int = Signal(int) # 定义一个带有一个int参数的信号 finished_str = Signal(str) # 定义一个带有一个str参数的信号 finished_list = Signal(list) # 定义一个带有一个list参数的信号 finished_bool = Signal(bool) # 定义一个带有一个bool参数的信号 def __init__(self, func, return_type, args=()): super(Worker, self).__init__() self.func = func # 保存业务函数的引用 self.args = args # 保存业务函数的参数 self.return_type = return_type # 保存返回类型的引用 def run(self): # 调用业务函数 result = self.func(*self.args) # 根据返回类型发出相应的信号 if self.return_type == int: self.finished_int.emit(result) elif self.return_type == str: self.finished_str.emit(result) elif self.return_type == list: self.finished_list.emit(result) elif self.return_type == bool: self.finished_bool.emit(result) class MyMainWindow(QMainWindow, Ui_MainWindow): def __init__(self, parent=None): super(MyMainWindow, self).__init__(parent) self.worker_set_all_networkcard_dhcp = None self.worker_update_device_config_table = None self.worker_set_networkcard_config = None self.worker_parsing_device_message = None self.worker_waiting_device_connection = None self.worker_pushButton_update_clicked = None self.worker_comboBox_class_currentIndex_Changed = None self.worker_comboBox_networkcard_currentIndex_Changed = None self.worker_update_comboBox_networkcard = None self.worker_update_comboBox_class = None self.worker_initialization = None self.setupUi(self) # 初始化 self.initialization() # self.set_all_networkcard_dhcp() # 组合框网卡信息列表项被选择 self.comboBox_networkcard.currentIndexChanged.connect(self.comboBox_networkcard_currentIndex_Changed) # 组合框部门信息列表项被选择 self.comboBox_class.currentIndexChanged.connect(self.comboBox_class_currentIndex_Changed) # 按钮一键刷机被点击 self.pushButton_update.clicked.connect(self.pushButton_update_clicked) # 禁用编辑框用户输入,仅用于显示信息。 self.textEdit_device.setEnabled(False) self.textEdit_configlabel.setEnabled(False) self.textEdit_networkadapter.setEnabled(False) self.textEdit_log.setEnabled(False) def initialization(self): """ 子线程:初始化。UI界面创建完成后,检测关键文件是否缺失. :return: 无返回 """ self.worker_initialization = Worker(Check_If_Key_Files_Exist, bool) self.worker_initialization.finished_bool.connect(self.initialization_callback) self.worker_initialization.start() def set_all_networkcard_dhcp(self): """ 子线程:设置计算机所有可用网卡为DHCP模式 :return: 无返回 """ self.textEdit_log.append( f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}->为避免您的网络配置' f'影响分机连接,已经自动将已启用的网卡设置为DHCP模式。') for i in range(self.comboBox_class.count()): networkcardname = self.comboBox_class.itemText(i) self.worker_set_all_networkcard_dhcp = Worker(Set_Computer_Network_Card_Dhcp, bool, (networkcardname,)) self.worker_set_all_networkcard_dhcp.finished_bool.connect(self.set_all_networkcard_dhcp_callback) self.worker_set_all_networkcard_dhcp.start() self.textEdit_log.append( f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}->设置网卡{networkcardname}' f'为DHCP模式。🆗') def update_device_config_table(self, ip: str, data: list, mode: int): """ 子线程:向可视对讲分机下发配置表,在set_networkcard_config_callback_1函数执行后调用 :return: 无返回 """ if mode == 1: self.worker_update_device_config_table = Worker(Send_Data_To_Device, bool, (ip, data)) self.worker_update_device_config_table.finished_bool.connect(self.update_device_config_table_callback_1) self.worker_update_device_config_table.start() elif mode == 2: self.worker_update_device_config_table = Worker(Send_Data_To_Device, bool, (ip, data)) self.worker_update_device_config_table.finished_bool.connect(self.update_device_config_table_callback_2) self.worker_update_device_config_table.start() def set_networkcard_ip(self, networkcard_name: str, ip: str, mask: str, gateway: str, mode: int): """ 子线程:设置网卡配置,当mode为1时在parsing_device_message_callback函数执行后调用,当mode为2时 :return: 无返回 """ if mode == 1: self.worker_set_networkcard_config = Worker(Set_Computer_Network_Card_Ip, bool, (networkcard_name, ip, mask, gateway)) self.worker_set_networkcard_config.finished_bool.connect(self.set_networkcard_config_callback_1) self.worker_set_networkcard_config.start() elif mode == 2: self.worker_set_networkcard_config = Worker(Set_Computer_Network_Card_Ip, bool, (networkcard_name, ip, mask, gateway)) self.worker_set_networkcard_config.finished_bool.connect(self.set_networkcard_config_callback_2) self.worker_set_networkcard_config.start() def parsing_device_message(self, data, mode: int): """ 子线程:解析分机返回消息，在waiting_device_connection_callback函数执行后调用。 :param mode: :param data: str,分机返回的数据包 :return: 无返回 """ if mode == 1: self.worker_parsing_device_message = Worker(Parse_Device_Xml, list, (data,)) self.worker_parsing_device_message.finished_list.connect(self.parsing_device_message_callback_1) self.worker_parsing_device_message.start() elif mode == 2: self.worker_parsing_device_message = Worker(Parse_Device_Xml, list, (data,)) self.worker_parsing_device_message.finished_list.connect(self.parsing_device_message_callback_2) self.worker_parsing_device_message.start() elif mode == 3: self.worker_parsing_device_message = Worker(Parse_Device_Xml, list, (data,)) self.worker_parsing_device_message.finished_list.connect(self.parsing_device_message_callback_3) self.worker_parsing_device_message.start() def waiting_device_connection(self, mode: int): """ 子线程:监听可视对讲分机回执消息,等待连接,在pushButton_update_clicked_callback函数执行后调用。 :return: """ if mode == 1: self.worker_waiting_device_connection = Worker(Listen_Device_Info, str) self.worker_waiting_device_connection.finished_str.connect(self.waiting_device_connection_callback_1) self.worker_waiting_device_connection.start() elif mode == 2: self.worker_waiting_device_connection = Worker(Listen_Device_Info, str) self.worker_waiting_device_connection.finished_str.connect(self.waiting_device_connection_callback_2) self.worker_waiting_device_connection.start() elif mode == 3: self.worker_waiting_device_connection = Worker(Listen_Device_Info, str) self.worker_waiting_device_connection.finished_str.connect(self.waiting_device_connection_callback_3) self.worker_waiting_device_connection.start() def update_comboBox_class(self): """ 子线程:更新部门组合框列表项,在initialization_callback函数执行后调用。 :return: 无返回 """ self.worker_update_comboBox_class = Worker(Read_Config_Xls_To_List, list) self.worker_update_comboBox_class.finished_list.connect(self.update_comboBox_class_callback) self.worker_update_comboBox_class.start() def update_comboBox_networkcard(self): """ 子线程:更新网卡组合框列表项,在initialization_callback函数执行后调用。 :return: 无返回 """ self.worker_update_comboBox_networkcard = Worker(Get_Pc_Network_Card, list) self.worker_update_comboBox_networkcard.finished_list.connect(self.update_comboBox_networkcard_callback) self.worker_update_comboBox_networkcard.start() def comboBox_networkcard_currentIndex_Changed(self): """ 子线程:组合框网卡信息列表项被选择,在MyMainWindow中调用 :return: 无返回 """ self.worker_comboBox_networkcard_currentIndex_Changed = Worker(Get_Networkcard_Ip_And_Netmask, list, (self.comboBox_networkcard.currentText(),)) self.worker_comboBox_networkcard_currentIndex_Changed.finished_list.connect( self.comboBox_networkcard_currentIndex_Changed_callback) self.worker_comboBox_networkcard_currentIndex_Changed.start() def set_networkcard_dhcp(self, mode: int): """ 子线程:设置指定连接名称的网卡为DHCP模式,当mode为1时在pushButton_update_clicked函数执行后调用,当mode为2时 :param mode: :return: """ if mode == 1: self.worker_pushButton_update_clicked = Worker(Set_Computer_Network_Card_Dhcp, bool, (self.comboBox_networkcard.currentText(),)) self.worker_pushButton_update_clicked.finished_bool.connect(self.set_networkcard_dhcp_callback_1) self.worker_pushButton_update_clicked.start() elif mode == 2: self.worker_pushButton_update_clicked = Worker(Set_Computer_Network_Card_Dhcp, bool, (self.comboBox_networkcard.currentText(),)) self.worker_pushButton_update_clicked.finished_bool.connect(self.set_networkcard_dhcp_callback_2) self.worker_pushButton_update_clicked.start() elif mode == 3: self.worker_pushButton_update_clicked = Worker(Set_Computer_Network_Card_Dhcp, bool, (self.comboBox_networkcard.currentText(),)) self.worker_pushButton_update_clicked.finished_bool.connect(self.set_networkcard_dhcp_callback_3) self.worker_pushButton_update_clicked.start() def pushButton_update_clicked(self): """ 子线程:按钮一键刷机被点击 :return: 无返回 """ # 禁用网卡和部门组合框选项 self.comboBox_class.setEnabled(False) self.comboBox_networkcard.setEnabled(False) # 禁用一键刷机按钮点击 self.pushButton_update.setEnabled(False) # 情况编辑框内容 self.textEdit_log.clear() self.textEdit_device.clear() # 启动线程 self.set_networkcard_dhcp(1) def set_all_networkcard_dhcp_callback(self): pass def comboBox_class_currentIndex_Changed(self): """ 组合框部门列表项被选择,在MyMainWindow中调用 :return: 无返回 """ self.comboBox_class_currentIndex_Changed_callback() def initialization_callback(self, result): """ 执行initialization函数回调 :param result: 接收initialization程返回值 :return: 无返回值 """ if result: self.update_comboBox_class() self.update_comboBox_networkcard() self.textEdit_log.append( f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}->环境初始化完成,' f'请选择网卡和部门后执行刷机工作。🆗') else: # 如果配置表缺失，则禁止点击相关组件 self.pushButton_update.setEnabled(False) self.comboBox_class.setEnabled(False) self.textEdit_log.append( f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}->未找到配置表,' f'请将本软件复制到配置表同级目录后重新运行。🤡') def update_comboBox_class_callback(self, result): """ 执行update_comboBox_class函数回调 :param result: 接收update_comboBox_class返回值 :return: 无返回值 """ for key in result: self.comboBox_class.addItems(key.keys()) # 在槽函数中向comboBox中添加项 self.textEdit_log.append( f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}->请选择需要刷入的部门名称,' f'该选择与写入分机号关联。😊') def update_comboBox_networkcard_callback(self, result): """ 执行update_comboBox_networkcard函数回调 :param result: 接收update_comboBox_networkcard返回值 :return: 无返回 """ self.comboBox_networkcard.addItems(result) self.textEdit_log.append( f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}->请选择与可视对讲分机连接的网络适配器,' f'错选可能搜索不到分机！⚠') def comboBox_networkcard_currentIndex_Changed_callback(self, result): """ 执行comboBox_networkcard_currentIndex_Changed函数回调 :param result: 接收comboBox_networkcard_currentIndex_Changed返回值 :return: 无返回 """ # 先清空内容,在写入数据。 self.textEdit_networkadapter.clear() if len(result) > 0: self.textEdit_networkadapter.append(f'IP:{result[0]} 子网掩码:{result[1]}') def comboBox_class_currentIndex_Changed_callback(self): """ 执行comboBox_class_currentIndex_Changed函数回调 :return: 无返回 """ # 先清空内容,在写入数据。 self.textEdit_configlabel.clear() class_name = self.comboBox_class.currentText() for con_dict in config_table_list: if class_name in con_dict: value = con_dict[class_name] split_text = value.split('|') self.textEdit_configlabel.append( f'分机号:{split_text[0]} IP地址:{split_text[1]} 子网掩码:{split_text[2]}<br' f'/>网关地址:{split_text[3]}') device_config_list.clear() device_config_list.extend([split_text[0], split_text[1], split_text[2], split_text[3]]) def set_networkcard_dhcp_callback_1(self, result): if result: self.textEdit_log.append( f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}->设置网卡:{self.comboBox_networkcard.currentText()}DHCP模式成功。') self.textEdit_log.append( f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}->等待可视对讲分机连接中。') self.waiting_device_connection(1) else: self.textEdit_log.append( f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}->设置网卡:{self.comboBox_networkcard.currentText()}DHCP模式失败,请检查该网卡运行状态！') def set_networkcard_dhcp_callback_2(self, result): if result: self.textEdit_log.append( f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}->设置网卡:{self.comboBox_networkcard.currentText()}DHCP模式成功。') self.textEdit_log.append( f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}->等待可视对讲分机连接中。') self.waiting_device_connection(2) else: self.textEdit_log.append( f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}->设置网卡:{self.comboBox_networkcard.currentText()}DHCP模式失败,请检查该网卡运行状态！') def set_networkcard_dhcp_callback_3(self, result): if result: self.textEdit_log.append( f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}->设置网卡:{self.comboBox_networkcard.currentText()}DHCP模式成功。') self.textEdit_log.append( f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}->等待可视对讲分机连接中。') self.waiting_device_connection(3) else: self.textEdit_log.append( f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}->设置网卡:{self.comboBox_networkcard.currentText()}DHCP模式失败,请检查该网卡运行状态！') def waiting_device_connection_callback_1(self, result): self.parsing_device_message(result, 1) def waiting_device_connection_callback_2(self, result): self.parsing_device_message(result, 2) def waiting_device_connection_callback_3(self, result): self.parsing_device_message(result, 3) def parsing_device_message_callback_1(self, result): self.textEdit_log.append( f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}->来自{result[2]}的分机:{result[1]}-已连接。\n') self.textEdit_device.append( f'设备ID:{result[0]} 设备名称:{result[1]} IP地址:{result[2]} 子网掩码:{result[3]} 网关地址:{result[4]}') self.set_networkcard_ip(self.comboBox_networkcard.currentText(), str(ipaddress.ip_address(result[2]) + 1), result[3], result[4], 1) def parsing_device_message_callback_2(self, result): self.textEdit_log.append( f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}->来自{result[2]}的分机:{result[1]}-已连接。\n') self.textEdit_device.append( f'设备ID:{result[0]} 设备名称:{result[1]} IP地址:{result[2]} 子网掩码:{result[3]} 网关地址:{result[4]}') self.set_networkcard_ip(self.comboBox_networkcard.currentText(), str(ipaddress.ip_address(result[2]) + 1), result[3], result[4], 2) def parsing_device_message_callback_3(self, result): self.textEdit_log.append( f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}->刷机完成,新设备分区号:{result[0]}' f'-设备名称:{result[1]}-IP地址:{result[2]}-子网掩码:{result[3]}\n') self.comboBox_class.setEnabled(True) self.comboBox_networkcard.setEnabled(True) self.pushButton_update.setEnabled(True) def set_networkcard_config_callback_1(self, result): self.textEdit_log.append( f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}->配置网卡:{self.comboBox_networkcard.currentText()}为静态IP地址。\n') if result: self.update_device_config_table(device_information_list[2], ['BBBB666604100A0000000000000000000000000000000000', '367A0100', Read_Config_Xml()], 1) def set_networkcard_config_callback_2(self, result): self.textEdit_log.append( f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}->配置网卡:{self.comboBox_networkcard.currentText()}为静态IP地址。\n') if result: self.update_device_config_table(device_information_list[2], [Extension_Number_Packaging(device_config_list[0])], 2) def update_device_config_table_callback_1(self, result): if result: self.textEdit_log.append( f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}->向可视对讲分机下发配置表成功,' f'等待设备重启中。\n') self.set_networkcard_dhcp(2) else: self.textEdit_log.append( f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}->向可视对讲分机下发配置表失败,' f'要不试试看官方的刷机工具叭😂。\n') self.comboBox_class.setEnabled(True) self.comboBox_networkcard.setEnabled(True) self.pushButton_update.setEnabled(True) def update_device_config_table_callback_2(self, result): if result: self.textEdit_log.append( f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}->写入可视对讲分机号成功,' f'等待设备重启中。\n') self.set_networkcard_dhcp(3) else: self.textEdit_log.append( f'{time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())}->写入可视对讲分机号失败,' f'要不试试看官方的刷机工具叭😂。\n') if __name__ == "__main__": app = QApplication(sys.argv) window = MyMainWindow() window.show() sys.exit(app.exec())
""" import asyncio async def func(): print("喵") if __name__ == '__main__': # # 协程对象要想运行，必须借助于event_loop # # 拿到协程对象 f = func() # # 拿到事件循环 # event_loop = asyncio.get_event_loop() # # event_loop执行协程对象，直到对象内的内容执行完毕为止. # event_loop.run_until_complete(f) asyncio.run(f) """ ''' import asyncio import time async def func1(): print("我是func1开始") await asyncio.sleep(1) print("我是func1结束") async def func2(): print("我是func2开始") await asyncio.sleep(2) print("我是func2结束") async def func3(): print("我是func3开始") await asyncio.sleep(3) print("我是func3结束") if __name__ == '__main__': start = time.time() f1 = func1() f2 = func2() f3 = func3() # 将三个任务放在一起，组合成任务队列 tasks = [f1, f2, f3] asyncio.run(asyncio.wait(tasks)) print(time.time()-start) ''' ''' import asyncio import time async def download(url, t): print("开始下载了") await asyncio.sleep(t) print("下载完了") async def main(): urls = [ 'http://www.baidu.com', 'http://www.abab.com', 'http://www.lll.com', ] tasks = [] for url in urls: task = asyncio.create_task(download(url, 3)) tasks.append(task) await asyncio.wait(tasks) if __name__ == '__main__': asyncio.run(main()) ''' ''' import asyncio async def func1(): print("我是func1开始") await asyncio.sleep(1) print("我是func1结束") return 'func1的返回值' async def func2(): print("我是func2开始") await asyncio.sleep(2) print("我是func2结束") return 'func2的返回值' async def func3(): print("我是func3开始") await asyncio.sleep(3) print("我是func3结束") return 'func3的返回值' async def main(): f1 = func1() f2 = func2() f3 = func3() tasks = [asyncio.create_task(f1), asyncio.create_task(f3), asyncio.create_task(f2)] done, pending = await asyncio.wait(tasks) for res in done: print(res.result()) res = await asyncio.gather(*tasks) print(res) if __name__ == '__main__': asyncio.run(main()) ''' import asyncio import aiohttp import aiofiles async def download(url): print(f'开始下载{url}') file_name = url.split('/')[-1] async with aiohttp.ClientSession() as session: async with session.get(url) as res: content = await res.content.read() async with aiofiles.open(file_name, mode='wb') as f: await f.write(content) print(f'下载完成{url}') async def main(): url_list = [ 'https://oss-img.mengguzhiai.com/tutuji/20231018/jkyqd3icg24.jpg', 'https://oss-img.mengguzhiai.com/tutuji/20230626/qviw1aaxajt.jpg', 'https://oss-img.mengguzhiai.com/tutuji/20230407/ldsf23o44jg.jpg', ] tasks = [] for url in url_list: t = asyncio.create_task(download(url)) tasks.append(t) await asyncio.wait(tasks) if __name__ == '__main__': asyncio.run(main())
import time import fastdeploy as fd import cv2 class YoloV8Prediction: def __init__(self, ModelPath: str, PreheatingImagesPath: str): option = fd.RuntimeOption() # 切换使用CPU/GPU # option.use_cpu() option.use_gpu() # 切换不同后端 # option.use_paddle_infer_backend() # option.use_paddle_lite_backend() # option.use_paddle_backend() option.use_openvino_backend() # option.use_ort_backend() option.use_trt_backend() # option.use_poros_backend() self.yolov8 = fd.vision.detection.YOLOv8(ModelPath, runtime_option=option) self.yolov8.preprocessor.size = [640, 640] img = cv2.imread(PreheatingImagesPath) s = time.time() for i in range(100): self.yolov8.predict(img) print(f'模型预热成功，预测100张图片耗时{time.time() - s}秒') def Start_Prediction(self, Images: str): return self.yolov8.predict(Images)from ultralytics import YOLO import torch import glob import os import xml.etree.ElementTree as ET import random import shutil import time from window.windowclass_en import WindowManager import cv2 import Draw.D3Gui from yolov8.YoloV8_Prediction import YoloV8Prediction class YoloV8: def __init__(self, ModelPath: str, ParentWindowHandle, WindowClassName: str, WindowTitle: str, PreheatingImagesPath: str, Mode: int = 1): """ 使用YOLOV8时常用的函数，mode为0时为训练模型，为1时为预测模式（训练和预测的环境不一样，训练支支持64位python。训练是cuda+torch+安装ultralytics，预测是TensorRT+fastdeploy） :param ModelPath: str,模型路径，训练模式为pt模型，预测模式为onnx模型 :param ParentWindowHandle: int，父窗口句柄 :param WindowClassName: str，窗口类名 :param WindowTitle: str，窗口标题 :param PreheatingImagesPath: str，用于预热预测模型的图片路径 :param Mode: int，运行模式（训练/预测） """ self.mode = Mode self.window = WindowManager() self.handle = self.window.Find_Windows(ParentWindowHandle, WindowClassName, WindowTitle) if Mode == 0: self.yolo = YOLO(model=ModelPath) elif Mode == 1: self.yolo = YoloV8Prediction(ModelPath, PreheatingImagesPath) pass @staticmethod # 转换一个xml文件为txt def single_xml_to_txt(xml_file, class_names): tree = ET.parse(xml_file) root = tree.getroot() # 保存txt文件路径 text_dir, xml_name = os.path.split(xml_file) txt_name = f'{xml_name[:-4]}.txt' txt_file = os.path.join(text_dir, txt_name) with open(txt_file, 'w') as txt_file: for member in root.findall('object'): # 从xml获取图像的宽和高 picture_width = int(root.find('size')[0].text) picture_height = int(root.find('size')[1].text) class_name = member[0].text # 类名对应的index class_num = class_names.index(class_name) box_x_min = int(member[4][0].text) # 左上角横坐标 box_y_min = int(member[4][1].text) # 左上角纵坐标 box_x_max = int(member[4][2].text) # 右下角横坐标 box_y_max = int(member[4][3].text) # 右下角纵坐标 # 转成相对位置和宽高（所有值处于0~1之间） x_center = (box_x_min + box_x_max) / (2 * picture_width) y_center = (box_y_min + box_y_max) / (2 * picture_height) width = (box_x_max - box_x_min) / picture_width height = (box_y_max - box_y_min) / picture_height print(class_num, x_center, y_center, width, height) txt_file.write( str(class_num) + ' ' + str(x_center) + ' ' + str(y_center) + ' ' + str(width) + ' ' + str( height) + '\n') @staticmethod def Batch_Rename(dir: str, ExtensionName: str): """ 批量重命名 :param dir: str,文件目录 :param ExtensionName: str,文件拓展名 :return: """ files = os.listdir(dir) i = 1 for file in files: if ExtensionName in file: new_name = f"{i}.{ExtensionName}" i += 1 old_path = os.path.join(dir, file) new_path = os.path.join(dir, new_name) os.rename(old_path, new_path) @staticmethod def Delete_obsolete_images(image_folder: str, label_folder: str): """ 匹配txt和jpg的数量,删除废图 :param image_folder: str，图集路径 :param label_folder: str,文本集路径 :return: """ # 获取label文件夹中的所有.txt文件 label_files = os.listdir(label_folder) # 构建一个集合来存储存在于label中的文件名（不包括扩展名） label_file_names = {os.path.splitext(file)[0] for file in label_files} # 获取image文件夹中的所有.jpg文件 image_files = os.listdir(image_folder) # 遍历每个.jpg文件 for image_file in image_files: # 提取文件名（不包括扩展名） image_file_name = os.path.splitext(image_file)[0] # 检查文件名是否存在于label中 if image_file_name not in label_file_names: # 如果文件名不在label中，输出提示信息并删除该.jpg文件 print(f"删除多余的图像文件: {os.path.join(image_folder, image_file)}") os.remove(os.path.join(image_folder, image_file)) @staticmethod def Data_Partitioning(image_folder, label_folder): """ 按照7：2：1划分train：val：test，不会删除源文件 :param image_folder: str,图集路径 :param label_folder: str,标注集路径 :return: """ # 创建train、val和test子文件夹 os.makedirs(os.path.join(image_folder, 'train'), exist_ok=True) os.makedirs(os.path.join(image_folder, 'val'), exist_ok=True) os.makedirs(os.path.join(image_folder, 'test'), exist_ok=True) os.makedirs(os.path.join(label_folder, 'train'), exist_ok=True) os.makedirs(os.path.join(label_folder, 'val'), exist_ok=True) os.makedirs(os.path.join(label_folder, 'test'), exist_ok=True) # 获取image文件夹中的所有.jpg文件和label文件夹中的所有.txt文件 image_files = [file for file in os.listdir(image_folder) if file.endswith('.jpg')] label_files = [file for file in os.listdir(label_folder) if file.endswith('.txt')] # 随机打乱文件列表 random.shuffle(image_files) random.shuffle(label_files) # 计算划分的数量 total_files = len(image_files) train_ratio = 0.7 val_ratio = 0.2 test_ratio = 0.1 train_count = int(train_ratio * total_files) val_count = int(val_ratio * total_files) # 将文件拷贝到对应的子文件夹中 for i, image_file in enumerate(image_files): if i < train_count: dest_folder = 'train' elif i < train_count + val_count: dest_folder = 'val' else: dest_folder = 'test' # 拷贝图像文件 shutil.copy(os.path.join(image_folder, image_file), os.path.join(image_folder, dest_folder, image_file)) # 构建对应的label文件名 label_file = image_file.replace('.jpg', '.txt') # 拷贝label文件 shutil.copy(os.path.join(label_folder, label_file), os.path.join(label_folder, dest_folder, label_file)) @staticmethod def Training_Model(ModelPath: str, DataPath: str, Epochs: int, Imgsz: int, Batch: int): """ 加载预先训练的模型用于训练，支持断点训练 :param ModelPath: str,模型路径 :param DataPath: str,数据集路径，eg:xxx/slide.yaml :param Epochs: int,训练轮数 :param Imgsz: int,图像大小，eg：320，640 :param Batch: int,显存大小，eg:2,4,6,8 :return: """ model = YOLO(ModelPath) # 根据YAML构建并传递权重 model.train(data=DataPath, epochs=Epochs, imgsz=Imgsz, batch=Batch, cache=True) @staticmethod def Build_Weights_And_Train(WeightFilePath: str, ModelPath: str, DataPath: str, Epochs: int, Imgsz: int, Batch: int): """ 根据YAML构建并传递权重，然后开始训练 :param WeightFilePath: str,权重路径 :param ModelPath: str,模型路径 :param DataPath: str,数据集路径，eg:xxx/slide.yaml :param Epochs: int,训练轮数 :param Imgsz: int,图像大小，eg：320，640 :param Batch: int,显存大小，eg:2,4,6,8 :return: """ model = YOLO(WeightFilePath).load(ModelPath) # 根据YAML构建并传递权重 model.train(data=DataPath, epochs=Epochs, imgsz=Imgsz, batch=Batch, cache=True) def Dir_Xml_To_Txt(self, path: str, class_names): """ 将xml标注转换为yolo标注 :param path: str,标注集路径, 后面的斜杠要带 :param class_names: 类名，eg：class_names = ['fei', 'jing'] :return: """ # 转换文件夹下的所有xml文件为txt for xml_file in glob.glob(path + '*.xml'): # 查找符合规定的文件路径名 print(xml_file) self.single_xml_to_txt(xml_file, class_names) def Using_CPU_Prediction(self, ImagePath: str): """ 使用CPU预测(仅用于训练时的预测测试) :param ImagePath: str,要预测的图片路径 :return: """ if self.mode == 0: self.yolo.predict(ImagePath, save=True) else: print("为保证性能预测模式不支持调用该函数") def Using_GPU_Prediction(self, ImagePath: str): """ 使用GPU预测(仅用于训练时的预测测试) :param ImagePath: str,要预测的图片路径 :return: """ if self.mode == 0: torch.cuda.is_available() self.yolo.predict(ImagePath, device="0") else: print("为保证性能预测模式不支持调用该函数") def Screenshot_Always(self, Path: str): """ 窗口截图，一直截（如果黑屏请更换模式Capture_Win32_Gpu（handle,mode）） :param Path: str，图片保存路径，最后不要带/ :return: """ if not self.handle: print("获取窗口句柄失败") exit() i = 0 try: while True: # 无限循环 image = self.window.Capture_Win32_Gpu(self.handle[0], 3) cv2.imwrite(f"{Path}/{i}.jpg", image) i += 1 time.sleep(1) except KeyboardInterrupt: print("用户手动停止截图") def Screenshot_Breakpoint(self, start: int, end: int, Path: str): """ 窗口截图，指定数量（如果黑屏请更换模式Capture_Win32_Gpu（handle,mode）） :param start: int，从几开始 :param end: int，到几结束 :param Path: str，图片保存路径，最后不要带/ :return: """ if not self.handle: print("获取窗口句柄失败") exit() else: for i in range(start, end + 1): image = self.window.Capture_Win32_Gpu(self.handle[0], 3) cv2.imwrite(f"{Path}/{i}.jpg", image) time.sleep(1) def Draw_Rectangle(self): """ 矩形绘制 :return: """ if self.mode == 0: if not self.handle: print("获取窗口句柄失败") exit() draw = Draw.D3Gui.ExecDraw(self.handle[0]) # 初始化模块 while True: image = self.window.Capture_Win32_Gpu(self.handle[0], 3) result = self.yolo.Start_Prediction(image) if result: for box in result.boxes: # 获取四个顶点坐标 left = int(box[0]) top = int(box[1]) right = int(box[2]) bottom = int(box[3]) draw.startLoop() # 开始绘制段 draw.drawRect(left, top, right, bottom, 5, (255, 254, 0)) # 绘制矩形框 draw.endLoop() # 结束绘制段 else: print("为保证性能预测模式不支持调用该函数") def Automatic_Dimension(self, ImagesPath: str, LabelsPath: str, ImageSuffix: str): """ 自动标注 :return: """ if self.mode == 1: image_files = [file for file in os.listdir(ImagesPath) if file.endswith(ImageSuffix)] for file_name in image_files: # 分割文件名和后缀 parts = file_name.split('.') # 检查是否有两个部分，并且后缀是指定后缀 if len(parts) == 2 and parts[1] == ImageSuffix.replace(".", ""): imagepath = f'{ImagesPath}/{file_name}' img = cv2.imread(imagepath) img_height, img_width, _ = img.shape result = self.yolo.Start_Prediction(img) if result: with open(f'{LabelsPath}/{parts[0]}.txt', "a") as file: for a in range(len(result.label_ids)): c = result.label_ids[a] x_min = result.boxes[a][0] y_min = result.boxes[a][1] x_max = result.boxes[a][2] y_max = result.boxes[a][3] # 计算归一化坐标 x_center = (x_min + x_max) / (2 * img_width) y_center = (y_min + y_max) / (2 * img_height) width = (x_max - x_min) / img_width height = (y_max - y_min) / img_height # 将归一化坐标写入文件 normalized_box_str = f'{c} {x_center} {y_center} {width} {height}\n' file.write(normalized_box_str) else: print("为保证性能预测模式不支持调用该函数") if __name__ == '__main__': yolo = YoloV8('models/csgo_0.onnx', None, 'Valve001', 'Counter-Strike', 'CSGO.jpg', 1) yolo.Automatic_Dimension(r'C:\Users\devcat\Desktop\CSGO\images\train', r'C:\Users\devcat\Desktop\瓦\datasets' r'\slide\YOLO\labels', '.jpg')
# -*- coding: utf-8 -*- import ctypes import os import time from ctypes import wintypes code_dict = { "1": 49, "2": 50, "3": 51, "4": 52, "5": 53, "6": 54, "7": 55, "8": 56, "9": 57, "0": 48, "-": 189, "=": 187, "back": 8, "a": 65, "b": 66, "c": 67, "d": 68, "e": 69, "f": 70, "g": 71, "h": 72, "i": 73, "j": 74, "k": 75, "l": 76, "m": 77, "n": 78, "o": 79, "p": 80, "q": 81, "r": 82, "s": 83, "t": 84, "u": 85, "v": 86, "w": 87, "x": 88, "y": 89, "z": 90, "ctrl": 17, "alt": 18, "shift": 16, "win": 91, "space": 32, "cap": 20, "tab": 9, "~": 192, "esc": 27, "enter": 13, "up": 38, "down": 40, "left": 37, "right": 39, "option": 93, "print": 44, "delete": 46, "home": 36, "end": 35, "pgup": 33 , "pgdn": 34, "f1": 112, "f2": 113, "f3": 114, "f4": 115, "f5": 116, "f6": 117, "f7": 118, "f8": 119, "f9": 120, "f10": 121, "f11": 122, "f12": 123, "[": 219, "]": 221, "\\": 220, ";": 186, "'": 222, ",": 188, ".": 190, "/": 191, } class YJS(): def __init__(self, w, h, move_flag=0, KeyDelay=None): # 初始化参数 self.w, self.h = w, h self.move_flag = move_flag self.KeyDelay = KeyDelay # 初始化易建鼠dll VID, PID = 0xC216, 0x0301 __path = os.getcwd() + os.path.sep + "msdk.dll" self.objdll = ctypes.windll.LoadLibrary(__path) # 注册dll self.objdll.M_Open_VidPid.restype = wintypes.LPHANDLE self.hdl = self.objdll.M_Open_VidPid(VID, PID) # 获取usb键鼠 self.__ResolutionUsed() if self.hdl == -1: raise "打开失败" # 设置鼠标轨迹 self.__EnableRealMouse(self.move_flag) # 设置键盘点击延迟随机 self.__EnableRealKeypad(KeyDelay) # 设置随机延迟 # self.__ResolutionUsed() def __del__(self): self.objdll.M_ReleaseAllKey(self.hdl) def __ResolutionUsed(self): # 如果使用绝对移动,则需要初始化分辨率 if self.move_flag == 1: # hdl = ctypes.c_int(self.hdl) self.objdll.M_ResolutionUsed(self.hdl, self.w, self.h) def GetSn(self): len_ = wintypes.DWORD() buf = wintypes.CHAR() self.objdll.M_GetDevSn(self.hdl,ctypes.byref(len_),ctypes.byref(buf)) print(len_.value) def __EnableRealMouse(self, move_flag: int): # 默认每次移动100-127个像素,时间间隔10-20ms self.move_flag = move_flag def __EnableRealKeypad(self, KeyDelay): # 默认50-80ms if KeyDelay is None: return # 上下延迟50% self.__SetKeypadDelay(0, KeyDelay * 0.5, KeyDelay * 1.5) def __SetKeypadDelay(self, __type, mix_delay, max_delay): self.objdll.M_SetParam(self.hdl, __type, mix_delay, max_delay) def KeyPress(self, code): self.objdll.M_KeyPress2(self.hdl, code, 1) def KeyPressChar(self, __str): self.KeyPress(code_dict[__str]) def KeyDown(self, code): self.objdll.M_KeyDown2(self.hdl, code, 1) def KeyUp(self, code): self.objdll.M_KeyUp2(self.hdl, code, 1) def KeyDownChar(self, __str): self.KeyDown(code_dict[__str]) def KeyUpChar(self, __str): self.KeyUp(code_dict[__str]) def MoveTo(self, x: int, y: int): # 瞬间移动 if self.move_flag == 0: # hdl = ctypes.c_char_p(self.hdl) self.objdll.M_MoveTo3_D(self.hdl, x, y) # 模拟移动 elif self.move_flag == 1: self.objdll.M_MoveTo3(self.hdl, x, y) def LeftClick(self): self.objdll.M_LeftClick(self.hdl, 1) def RightClick(self): self.objdll.M_RightClick(self.hdl, 1) def LeftDown(self): self.objdll.M_LeftDown(self.hdl) def LeftUp(self): self.objdll.M_LeftUp(self.hdl) def RightDown(self): self.objdll.M_RightDown(self.hdl) def RightUp(self): self.objdll.M_RightUp(self.hdl) def LeftDoubleClick(self): self.objdll.M_LeftDoubleClick(self.hdl, 1) def KeyPressStr(self,str_:str): bt_str = str_.encode(encoding="gbk") len_ = len(bt_str) p_str = ctypes.c_char_p(bt_str) self.objdll.M_KeyInputStringGBK(self.hdl,p_str,len_) if __name__ == '__main__': time.sleep(1) yjs = YJS(1920, 1080, 1) yjs.MoveTo(500, 500) yjs.KeyPressStr("sfdasd")
import time from ctypes import wintypes import ctypes import win32gui import win32con import win32ui import cv2 import numpy as np class WindowManager: @staticmethod def Capture_Win32_Gpu(handle: int, model: int = 3): """ 通过win32方式截图，并且无视硬件加速 https://stackoverflow.com/questions/19695214/screenshot-of-inactive-window-printwindow-win32gui """ ctypes.windll.user32.SetProcessDPIAware() # 抑制缩放 rect = win32gui.GetWindowRect(handle) width, height = rect[2] - rect[0], rect[3] - rect[1] hwnd_dc = win32gui.GetWindowDC(handle) mfc_dc = win32ui.CreateDCFromHandle(hwnd_dc) save_dc = mfc_dc.CreateCompatibleDC() save_bit_map = win32ui.CreateBitmap() save_bit_map.CreateCompatibleBitmap(mfc_dc, width, height) save_dc.SelectObject(save_bit_map) ctypes.windll.user32.PrintWindow(handle, save_dc.GetSafeHdc(), model) # 如果仍然是黑屏，尝试修改数字 ‘3’ (从0开始...) bmpinfo = save_bit_map.GetInfo() bmpstr = save_bit_map.GetBitmapBits(True) capture = np.frombuffer(bmpstr, dtype=np.uint8).reshape((bmpinfo["bmHeight"], bmpinfo["bmWidth"], 4)) capture = np.ascontiguousarray(capture)[..., :-1] win32gui.DeleteObject(save_bit_map.GetHandle()) save_dc.DeleteDC() mfc_dc.DeleteDC() win32gui.ReleaseDC(handle, hwnd_dc) capture = cv2.cvtColor(capture, cv2.COLOR_RGBA2RGB) return capture @staticmethod def Get_Handle(): """ 取窗口句柄 :return: int,窗口句柄 """ return win32gui.GetForegroundWindow() @staticmethod def Get_Window_Position_And_Size(handle): """ 窗口取位置和大小 :param handle: int,窗口句柄 :return: """ return win32gui.GetWindowRect(handle) class WindowRect(ctypes.Structure): _fields_ = [ ("left", ctypes.c_long), ("top", ctypes.c_long), ("right", ctypes.c_long), ("bottom", ctypes.c_long) ] def __init__(self): """ 调用Windows官方API实现对window的操作 """ self.user32 = ctypes.windll.user32 self.kernel32 = ctypes.windll.kernel32 # self.GetLastError = ctypes.windll.kernel32.GetLastError() self.gdi32 = ctypes.WinDLL("Gdi32.dll") def Find_Windows(self, parent_handle: int = None, class_name: str = None, window_title: str = None): """ 模糊遍历窗口 :param parent_handle: str,父窗口句柄 :param class_name: str,欲寻找窗口类名 :param window_title: str,欲寻找窗口标题 :return: list,窗口句柄列表 """ handle_list = [] if parent_handle is None: parent_handle = self.user32.GetDesktopWindow() handle = self.user32.GetWindow(parent_handle, 5) while handle != 0: valid = True if class_name is not None: actual_class_name = str(self.Get_Class_Name(handle)).lower() if actual_class_name.find(class_name.lower()) == -1: valid = False time.sleep(0) if window_title is not None: actual_window_title = str(self.Get_Window_Title(handle)).lower() if actual_window_title.find(window_title.lower()) == -1: valid = False time.sleep(0) if valid: handle_list.append(handle) handle = self.user32.GetWindow(handle, 2) return handle_list def Get_Class_Name(self, handle: int): """ 取窗口类名 :param handle: int,窗口句柄 :return: str,窗口类名 """ GetClassNameA = self.user32.GetClassNameA GetClassNameA.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int] GetClassNameA.restype = ctypes.c_int buffer_size = 256 name_buffer = ctypes.create_string_buffer(buffer_size) result = GetClassNameA(handle, name_buffer, buffer_size) if result != 0: class_name = name_buffer.value.decode('utf-8') return class_name else: return "" def Get_Window_Title(self, handle: int): """ 取窗口标题 :param handle: int,窗口句柄 :return: str,窗口标题 """ GetWindowTextLengthW = self.user32.GetWindowTextLengthW GetWindowTextLengthW.argtypes = [ctypes.c_int] GetWindowTextLengthW.restype = ctypes.c_int GetWindowTextW = self.user32.GetWindowTextW GetWindowTextW.argtypes = [ctypes.c_int, ctypes.c_wchar_p, ctypes.c_int] GetWindowTextW.restype = ctypes.c_int length = GetWindowTextLengthW(handle) if length > 0: buffer_size = length + 1 title_buffer = ctypes.create_unicode_buffer(buffer_size) result = GetWindowTextW(handle, title_buffer, buffer_size) if result != 0: title = title_buffer.value return title else: return "" else: return "" def Is_Window_Visible(self, handle: int): """ 窗口是否可见 :param handle: int,窗口句柄 :return: bool,如果该函数成功，则返回值为非零值。如果函数失败，则返回值为零。 """ return self.user32.IsWindowVisible(handle) def Set_Window_Position_And_Size(self, handle, left=None, top=None, width=None, height=None): """ 设置窗口位置和大小 :param handle: int,窗口句柄 :param left: int,窗口左侧的新位置 :param top: int,窗口顶部的新位置 :param width: int,窗口的新宽度 :param height: int,窗口的新高度 :return: bool,如果该函数成功，则返回值为非零值。如果函数失败，则返回值为零 """ rect = self.WindowRect() self.user32.GetWindowRect(handle, ctypes.byref(rect)) if left is None: left = rect.left if top is None: top = rect.top if width is None: width = rect.right - rect.left if height is None: height = rect.bottom - rect.top return self.user32.MoveWindow(handle, left, top, width, height, True) def Set_Window_Topmost(self, handle, activate_window=True): """ 窗口置顶 :param handle: int,窗口句柄 :param activate_window: bool,是否激活窗口 :return: bool,如果该函数成功，则返回值为非零值。如果函数失败，则返回值为零。 """ HWND_TOPMOST = -1 SWP_SHOWWINDOW = 0x40 SWP_NOSIZE = 0x1 SWP_NOMOVE = 0x2 SWP_NOACTIVATE = 0x10 flags = SWP_SHOWWINDOW | SWP_NOSIZE | SWP_NOMOVE if not activate_window: flags |= SWP_NOACTIVATE result = self.user32.SetWindowPos(handle, HWND_TOPMOST, 0, 0, 0, 0, flags) return result != 0 def Set_Window_Display_Affinity(self, handle, affinity): """ 窗口反截图，仅支持自身调用 :param handle: int,顶级窗口的句柄。 窗口必须属于当前进程。 :param affinity: int，0:取消设置;1:黑框;2:透明 :return: bool,如果该函数成功，则返回值为非零值。如果函数失败，则返回值为零。 """ HWND = ctypes.c_void_p DWORD = ctypes.c_ulong BOOL = ctypes.c_int self.user32.SetWindowDisplayAffinity.argtypes = (HWND, DWORD) self.user32.SetWindowDisplayAffinity.restype = BOOL AFFINITY_CONSTANTS = { 0: 0x00000000, # WDA_NONE 1: 0x00000001, # WDA_MONITOR 2: 0x00000011 # WDA_EXCLUDEFROMCAPTURE } affinity_value = AFFINITY_CONSTANTS.get(affinity, 0) result = self.user32.SetWindowDisplayAffinity(handle, affinity_value) print(self.kernel32.GetLastError()) return result def Enum_Child_Window_Handles(self, hwnd): """ 枚举给定主窗口句柄的子窗口句柄，并返回一个包含子窗口句柄的列表。 :param hwnd: int，主窗口句柄 :return: list，子窗口句柄列表 """ # 定义Windows API函数的参数类型 EnumChildProc = ctypes.WINFUNCTYPE(wintypes.BOOL, wintypes.HWND, wintypes.LPARAM) EnumChildWindows = self.user32.EnumChildWindows # 定义回调函数，用于处理枚举到的子窗口 def enum_child_proc(hwnd, lparam): # 将子窗口句柄添加到列表中 ctypes.cast(lparam, ctypes.POINTER(ctypes.py_object)).contents.value.append(hwnd) return True # 创建一个空的子窗口句柄列表 child_handles = [] # 枚举子窗口句柄 EnumChildWindows(hwnd, EnumChildProc(enum_child_proc), ctypes.byref(ctypes.py_object(child_handles))) # 返回子窗口句柄列表 return child_handles def Window_Draw_Border_Rectangle(self, handle: int, BorderThickness: int = 2, BorderColor: int = 0xFF0000, rect_left: int = 0, rect_top: int = 0, rect_right: int = 100, rect_bottom: int = 100, isTransparent: bool = False): """ 窗口绘制矩形 :param handle: int,窗口句柄 :param BorderThickness: int,矩形边框粗细 :param BorderColor: int,矩形边框颜色 :param rect_left: int,左边 :param rect_top: int,顶边 :param rect_right: int,右边 :param rect_bottom: int,底边 :param isTransparent: bool,是否透明 :return: """ # 获取窗口DC hdc = win32gui.GetDC(handle) # 创建画笔 pen = win32gui.CreatePen(win32con.PS_SOLID, BorderThickness, BorderColor) # 创建画刷 if isTransparent: brush = win32gui.GetStockObject(win32con.HOLLOW_BRUSH) # 空画刷，不填充 # 将背景模式设置为透明 win32gui.SetBkMode(hdc, win32con.TRANSPARENT) else: brush = win32gui.CreateSolidBrush(0x00000000) # 0x00000000 表示透明 try: # 使用上下文管理器管理画笔和画刷 with self._SelectObjects(hdc, pen, brush): # 绘制矩形 result = self.gdi32.Rectangle(hdc, rect_left, rect_top, rect_right, rect_bottom) finally: # 释放窗口DC win32gui.ReleaseDC(handle, hdc) return result # 自定义上下文管理器来管理GDI对象 class _SelectObjects: def __init__(self, hdc, *objects): self.hdc = hdc self.objects = objects def __enter__(self): for obj in self.objects: win32gui.SelectObject(self.hdc, obj) return self def __exit__(self, exc_type, exc_value, traceback): for obj in self.objects: win32gui.DeleteObject(obj) if __name__ == "__main__": # 示例用法： window_manager = WindowManager() # 查找所有可见窗口 # visible_windows = window_manager.Find_Windows() # for handle in visible_windows: # class_name = window_manager.Get_Class_Name(handle) # window_title = window_manager.Get_Window_Title(handle) # print(f"Handle: {handle}, Class Name: {class_name}, Window Title: {window_title}") # 设置窗口置顶 # window_handle = 6556218 # 替换为实际窗口句柄 # window_manager.Set_Window_Topmost(window_handle) # 设置窗口位置和大小 # window_manager.Set_Window_Position_And_Size(window_handle, left=100, top=100, width=800, height=600) # 示例用法 # handle = 1509402 # 获取当前窗口句柄 # RectangleThickness = 2 # 矩形边框线宽 # RectangleColor = 0xFF0000 # 矩形颜色 (红色) # 指定矩形的坐标和大小 # rect_left = 100 # rect_top = 100 # rect_right = 200 # rect_bottom = 200 # 绘制矩形 # result = window_manager.Window_Draw_Border_Rectangle(handle, RectangleThickness, RectangleColor, rect_left, # rect_top, # rect_right, # rect_bottom, True)
import os import subprocess class VM: def __init__(self, path): os.putenv("path", path) def run_command(self, command): # 取DOS执行结果 """运行cmd命令""" result = subprocess.run(command, capture_output=True, shell=True, encoding='gbk') return result.stdout def StartVirtualMachine(self, VirtualMachinePath: str, gui: bool): """启动指定虚拟机""" if gui: command = f'vmrun.exe -T ws start "{VirtualMachinePath}" gui' else: command = f'vmrun.exe -T ws start "{VirtualMachinePath}" nogui' return self.run_command(command) if __name__ == '__main__': vm = VM(r'E:\VMwareWorkstation') vm.StartVirtualMachine(r"D:\WinXp_52Pojie_2.0\WinXp_52Pojie_2.0.vmx", True)
import base64 import io import subprocess import psutil from datetime import datetime import os from pathlib import Path import win32com.client import win32process import cv2 import numpy as np import win32ui import ctypes import win32gui import win32con import win32api import time from PIL import Image # 将Base64数据解码为图片 def base64_to_image(base64_data): """base64_png""" image_data = base64.b64decode(base64_data) image = Image.open(io.BytesIO(image_data)) return image def find_template(screenshot, template_path, threshold=0.8): """在截图中查找模板图片，返回匹配位置""" # 读取模板图片 template = cv2.imread(template_path, 0) if template is None: print(f"无法读取模板图片: {template_path}") return None # 转换截图为灰度图 screenshot_gray = cv2.cvtColor(screenshot, cv2.COLOR_BGR2GRAY) # 模板匹配 result = cv2.matchTemplate(screenshot_gray, template, cv2.TM_CCOEFF_NORMED) min_val, max_val, min_loc, max_loc = cv2.minMaxLoc(result) # 如果匹配度超过阈值，返回中心点坐标 if max_val >= threshold: h, w = template.shape center_x = max_loc[0] + w // 2 center_y = max_loc[1] + h // 2 return int(center_x), int(center_y), max_val else: print(f"未找到匹配项，最高匹配度: {max_val:.2f}") return None def click_in_window(hwnd, x, y): """在指定窗口的相对坐标处执行鼠标单击""" if not hwnd: return False # 获取窗口位置和大小 left, top, right, bottom = win32gui.GetClientRect(hwnd) window_width = right - left window_height = bottom - top # 检查坐标是否在窗口内 if x < 0 or x >= window_width or y < 0 or y >= window_height: print(f"坐标 ({x}, {y}) 超出窗口范围 ({window_width}, {window_height})") return False # 将客户端坐标转换为屏幕坐标 screen_x, screen_y = win32gui.ClientToScreen(hwnd, (x, y)) # 激活窗口 win32gui.SetForegroundWindow(hwnd) time.sleep(0.1) # 等待窗口激活 # 执行鼠标单击 win32api.SetCursorPos((screen_x, screen_y)) win32api.mouse_event(win32con.MOUSEEVENTF_LEFTDOWN, screen_x, screen_y, 0, 0) time.sleep(0.05) # 模拟真实点击的按下时间 win32api.mouse_event(win32con.MOUSEEVENTF_LEFTUP, screen_x, screen_y, 0, 0) return True def Capture_Win32_Gpu(handle: int, model: int = 3): """ 通过win32方式截图，并且无视硬件加速 https://stackoverflow.com/questions/19695214/screenshot-of-inactive-window-printwindow-win32gui """ ctypes.windll.user32.SetProcessDPIAware() # 抑制缩放 rect = win32gui.GetWindowRect(handle) width, height = rect[2] - rect[0], rect[3] - rect[1] hwnd_dc = win32gui.GetWindowDC(handle) mfc_dc = win32ui.CreateDCFromHandle(hwnd_dc) save_dc = mfc_dc.CreateCompatibleDC() save_bit_map = win32ui.CreateBitmap() save_bit_map.CreateCompatibleBitmap(mfc_dc, width, height) save_dc.SelectObject(save_bit_map) ctypes.windll.user32.PrintWindow(handle, save_dc.GetSafeHdc(), model) # 如果仍然是黑屏，尝试修改数字 ‘3’ (从0开始...) bmpinfo = save_bit_map.GetInfo() bmpstr = save_bit_map.GetBitmapBits(True) capture = np.frombuffer(bmpstr, dtype=np.uint8).reshape((bmpinfo["bmHeight"], bmpinfo["bmWidth"], 4)) capture = np.ascontiguousarray(capture)[..., :-1] win32gui.DeleteObject(save_bit_map.GetHandle()) save_dc.DeleteDC() mfc_dc.DeleteDC() win32gui.ReleaseDC(handle, hwnd_dc) capture = cv2.cvtColor(capture, cv2.COLOR_RGBA2RGB) return capture def get_desktop_path(): """获取当前用户的桌面路径""" return os.path.join(os.path.expanduser("~"), "Desktop") def get_shortcut_target(lnk_path): """获取快捷方式指向的目标路径""" try: shell = win32com.client.Dispatch("WScript.Shell") shortcut = shell.CreateShortCut(lnk_path) return shortcut.TargetPath except Exception as e: print(f"Error reading shortcut {lnk_path}: {e}") return None def find_shortcuts_by_name(query, directory=None): """ 根据名称搜索桌面快捷方式并返回目标路径 参数: query (str): 要搜索的快捷方式名称(不包含.lnk扩展名) directory (str, optional): 要搜索的目录路径，默认为桌面 返回: list: 包含匹配的快捷方式及其目标路径的列表 """ if directory is None: directory = get_desktop_path() results = [] try: # 遍历目录下所有.lnk文件 for lnk_file in Path(directory).rglob("*.lnk"): # 获取不包含扩展名的文件名 file_name = lnk_file.stem # 检查文件名是否包含查询字符串(不区分大小写) if query.lower() in file_name.lower(): target = get_shortcut_target(str(lnk_file)) if target: results.append({ 'shortcut_path': str(lnk_file), 'target_path': target }) except Exception as e: print(f"Error searching for shortcuts: {e}") return results def get_window_handle(title:str): """根据窗口标题获取窗口句柄""" hwnd = win32gui.FindWindow(None, title) return hwnd if hwnd != 0 else None def is_window_alive(title:str): """检查指定标题的窗口是否存在""" return get_window_handle(title) is not None def find_lnk_files(desktop_path:str): """根据指定的桌面快捷方式名称寻找其指向的文件""" shortcuts = find_shortcuts_by_name(desktop_path) if shortcuts: for shortcut in shortcuts: return f"{shortcut['target_path']}" else: return None def launch_application(exe_path:str): """启动应用程序""" if not exe_path or not os.path.exists(exe_path): print(f"错误: 应用程序路径不存在 - {exe_path}") return False try: subprocess.Popen(exe_path) print(f"已启动应用程序: {exe_path}") return True except Exception as e: print(f"启动应用程序失败: {e}") return False def get_process_id_from_hwnd(hwnd): """从窗口句柄获取进程ID""" if not hwnd: return None try: _, process_id = win32process.GetWindowThreadProcessId(hwnd) return process_id except: return None def get_process_exe(pid): """获取进程的可执行文件路径""" if not pid: return None try: process = psutil.Process(pid) return process.exe() except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): return None def main(w_title:str,d_title:str,lnk_title:str): """主函数""" print("窗口监控与重启工具已启动") print("=" * 50) check_interval = 1 # 检查间隔(秒) last_seen = datetime.now() restart_delay = 1 # 检测到窗口关闭后等待的时间(秒) consecutive_misses = 0 max_consecutive_misses = 1 # 连续检测不到窗口的最大次数 try: while True: current_time = datetime.now() if is_window_alive(w_title): last_seen = current_time consecutive_misses = 0 print(f"\r窗口状态: 正常运行中 - 上次检查: {current_time.strftime('%H:%M:%S')}", end="") else: consecutive_misses += 1 time_since_close = (current_time - last_seen).total_seconds() print( f"\r窗口状态: 已关闭 {time_since_close:.1f} 秒 - 连续检测失败: {consecutive_misses}/{max_consecutive_misses}", end="") if consecutive_misses >= max_consecutive_misses: print(f"\n检测到窗口已关闭超过 {restart_delay} 秒，正在尝试重启...") exe_path=find_lnk_files(lnk_title) #若不存在通过桌面快捷方式运行指定软件 launch_application(exe_path) #启动软件 time.sleep(3) # click_login(d_title,r'C:\Users\Administrator\Desktop\rjsh\logo.png',r'C:\Users\Administrator\Desktop\rjsh\dl.png') #使用opencv处理登录 if is_window_alive(d_title): hwnd = get_window_handle(d_title) print(f"检测到登录窗口: {d_title} (句柄: {hwnd})") screenshot = Capture_Win32_Gpu(hwnd) # 使用win32api截图指定窗口，模式3 while True: if screenshot is not None: match_pos = find_template(screenshot, r'C:\Users\Administrator\Desktop\rjsh\logo.png', 0.8) # 检测登录页面是否加载完成 if match_pos is not None: # 登录页面加载完成 res = find_template(screenshot, r'C:\Users\Administrator\Desktop\rjsh\dl.png', 0.8) print(f"找到登录按钮，位置: {res[:2]}，匹配度: {res[2]:.2f}") click_in_window(hwnd, res[0], res[1]) # 使用win32api点击登录 break else: # 登录页面没加载完成 print(f"等待软件启动中") time.sleep(2) consecutive_misses = 0 last_seen = current_time time.sleep(check_interval) except KeyboardInterrupt: print("\n接收到退出信号") print("监控已停止") except Exception as e: print(f"\n发生错误: {e}") print("监控已停止") if __name__ == '__main__': main('','','')
# -*- coding: utf-8 -*- import ctypes import sys import threading import time class _MyThread(threading.Thread): def __init__(self, target=None, args=(), kwargs=None, daemon=True): super(_MyThread, self).__init__(target=target, args=args, kwargs=kwargs, daemon=daemon) self.flag = threading.Event() # 虽然能停止线,会有不可预知的问题 def stop(self): print(f"停止线程 {self.name}") if not self.is_alive(): return exc = ctypes.py_object(SystemExit) res = ctypes.pythonapi.PyThreadState_SetAsyncExc( ctypes.c_long(self.ident), exc) if res == 0: raise ValueError("找不到线程ID") elif res > 1: ctypes.pythonapi.PyThreadState_SetAsyncExc(self.ident, None) raise SystemError("线程已停止") class MyThread(threading.Thread): dict_ = [{"kill": False} for i in range(1000)] def __init__(self, group=None, target=None, name=None, args=(), kwargs=None, daemon=True, dict_={}): threading.Thread.__init__(self, group=group, target=target, name=name, args=args, kwargs=kwargs, daemon=daemon) if dict_: self.killed = dict_ else: self.killed = {"kill": False} def start(self): print(f"开启线程{self}") self.__run_backup = self.run self.run = self.__run threading.Thread.start(self) def __run(self): sys.settrace(self.globaltrace) self.__run_backup() self.run = self.__run_backup def globaltrace(self, frame, event, arg): if event == 'call': return self.localtrace else: return None def localtrace(self, frame, event, arg): if self.killed["kill"]: if event == 'line': if self.killed["kill"] == 1: raise SystemExit() elif self.killed["kill"] == 2: while self.killed["kill"]: time.sleep(0.1) return self.localtrace def stop(self): print(f"停止线程{self}") self.killed["kill"] = 1 def pause(self): print(f"暂停线程{self}") self.killed["kill"] = 2 def resume(self): print(f"恢复线程{self}") self.killed["kill"] = False def test_func(num): for i in range(num): time.sleep(1) print(i) if __name__ == '__main__': # 第一种线程 # t = _MyThread(target=test_func,args=(100,)) # t.start() # time.sleep(5) # t.stop() # input() # # 第二种线程 t = MyThread(target=test_func, args=(100,)) t.start() time.sleep(3) t.pause() time.sleep(3) t.resume() time.sleep(3) t.stop() input()
from email.parser import Parser from email.header import decode_header from email.utils import parseaddr from lxml import etree import requests import poplib class PopMail: def __init__(self, email, password): # 输入邮件地址, 口令和POP3服务器地址: self.email = email self.password = password self.pop3_server = "mail.recolte-cn.cn" self.header = { 'User-Agent': 'Apipost client Runtime/+https://www.apipost.cn/', 'Content-Type': 'application/x-www-form-urlencoded', } def connect(self): # 连接到POP3服务器: self.server = poplib.POP3(self.pop3_server) # 可以打开或关闭调试信息: self.server.set_debuglevel(1) # 可选:打印POP3服务器的欢迎文字: print(self.server.getwelcome().decode('utf-8')) # 身份认证: self.server.user(self.email) self.server.pass_(self.password) def get_msg(self): # stat()返回邮件数量和占用空间: print('Messages: %s. Size: %s' % self.server.stat()) # list()返回所有邮件的编号: resp, mails, octets = self.server.list() # 可以查看返回的列表类似[b'1 82923', b'2 2184', ...] print(mails) # 获取最新一封邮件, 注意索引号从1开始: index = len(mails) resp, lines, octets = self.server.retr(index) # lines存储了邮件的原始文本的每一行, # 可以获得整个邮件的原始文本: msg_content = b'\r\n'.join(lines).decode('utf-8') # 稍后解析出邮件: self.msg = Parser().parsestr(msg_content) # 可以根据邮件索引号直接从服务器删除邮件: # server.dele(index) # 关闭连接: self.server.quit() # indent用于缩进显示: def print_info(self, msg, indent=0): if indent == 0: for header in ['From', 'To', 'Subject']: value = msg.get(header, '') if value: if header == 'Subject': value = self.decode_str(value) else: hdr, addr = parseaddr(value) name = self.decode_str(hdr) value = u'%s <%s>' % (name, addr) print('%s%s: %s' % (' ' * indent, header, value)) if (msg.is_multipart()): parts = msg.get_payload() for n, part in enumerate(parts): print('%spart %s' % (' ' * indent, n)) print('%s--------------------' % (' ' * indent)) self.print_info(part, indent + 1) else: content_type = msg.get_content_type() if content_type == 'text/plain' or content_type == 'text/html': content = msg.get_payload(decode=True) charset = self.guess_charset(msg) if charset: content = content.decode(charset) print('%sText: %s' % (' ' * indent, content + '...')) self.content = content else: print('%sAttachment: %s' % (' ' * indent, content_type)) def decode_str(self, s): value, charset = decode_header(s)[0] if charset: value = value.decode(charset) return value def guess_charset(self, msg): charset = msg.get_charset() if charset is None: content_type = msg.get('Content-Type', '').lower() pos = content_type.find('charset=') if pos >= 0: charset = content_type[pos + 8:].strip() return charset def parse_xml(self,html,xpath_bds): parse_html = etree.HTML(html) r_list = parse_html.xpath(xpath_bds) return r_list def get_html(self): self.connect() # 连接邮箱 self.get_msg() # 获取邮件 self.print_info(self.msg) # 邮件内容转成html格式 return self.content # 激活邮箱 def activate(self): html = self.get_html() xpath_bds = "//a/@href" urls = self.parse_xml(html,xpath_bds) # 解析出激活链接 for url in urls: if "https://member-m.withhive.com" in url: # 请求链接 if url: response = requests.get(url,headers=self.header) print(response.text) return True # 获取邮箱验证码 def get_code(self): html = self.get_html() xpath_bds = "//a/@href" # pass code = self.parse_xml(html, xpath_bds) # 解析出激活链接 return code class CreateMail: def __init__(self, num): self.num = num # 创建邮箱数量 def set_cookie(self, cookie): cookies = {} for line in cookie.split(";"): if line.find("=") != -1: name, value = line.strip().split("=") cookies[name] = value print("cookies", cookies) return cookies def add_mail(self, cookie, user, __pass, username, ): """ :param cookie: 登录后在获取浏览器cookie :param user: 邮箱账号 :param __pass: 密码 :param username: 角色名称 :return: """ url = "http://mail.recolte-cn.cn:8010/Users/edit" cookies = self.set_cookie(cookie) header = { 'User-Agent': 'Apipost client Runtime/+https://www.apipost.cn/', 'Content-Type': 'application/x-www-form-urlencoded', } data = { "email": user, "uname": username, "tel": '', "active": 1, "password": __pass, "password2": __pass, "_method": "put", "_forward": "2FUsers", } r = requests.post(url, data=data, headers=header, cookies=cookies) with open("test.html", "w", encoding="utf-8") as fp: fp.write(r.text) if __name__ == '__main__': PopMail("0001@-cn.cn", "").activate()
# -*- coding: utf-8 -*- import os from comtypes.client import CreateObject class LW: def __init__(self): self.lw = None try: self.lw = CreateObject('lw.lwsoft3') print('本机系统中已经安装乐玩插件，版本为:', self.Version()) except: print('本机并未安装乐玩，正在注册调用') cmd = "regsvr32 -s lw.dll" os.system(cmd) self.lw = CreateObject('lw.lwsoft3') print('本机系统中已经安装乐玩插件，版本为:', self.Version()) def Version(self): return self.lw.Ver() if __name__ == '__main__': lw = LW()