用于批量配置交换机的小脚本

## 一个简单的批量刷入配置exe的脚本

## 图形界面如图所示：

![](/media/202502/2025-02-20_150120_1261410.6902833415449089.png)
### 附件如下：

[【附件】交换机批量配置程序.zip](/media/attachment/2025/02/%E4%BA%A4%E6%8D%A2%E6%9C%BA%E6%89%B9%E9%87%8F%E9%85%8D%E7%BD%AE%E7%A8%8B%E5%BA%8F.zip)
```python
import sys
import os
import ipaddress
from PyQt6.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, 
                            QHBoxLayout, QLabel, QLineEdit, QPushButton, 
                            QTextEdit, QFileDialog, QComboBox, QMessageBox, 
                            QSpinBox, QProgressBar)
from PyQt6.QtCore import Qt, QThread, pyqtSignal
import paramiko
import telnetlib
import time
from concurrent.futures import ThreadPoolExecutor
from queue import Queue
import threading

def resource_path(relative_path):
    """ 获取资源的绝对路径 """
    try:
        # PyInstaller创建临时文件夹，将路径存储在_MEIPASS中
        base_path = sys._MEIPASS
    except Exception:
        base_path = os.path.abspath(".")
    return os.path.join(base_path, relative_path)

class CommandExecutor(QThread):
    output_received = pyqtSignal(str)
    progress_updated = pyqtSignal(int)  # 添加进度信号
    
    def __init__(self, start_ip, end_ip, username, password, protocol, commands, ip_list=None, max_threads=10):
        super().__init__()
        self.start_ip = start_ip
        self.end_ip = end_ip
        self.username = username
        self.password = password
        self.protocol = protocol
        self.commands = commands
        self.running = True
        self.ip_list = ip_list
        self.max_threads = max_threads  # 最大线程数
        self.output_queue = Queue()  # 用于存储输出信息
        self.lock = threading.Lock()  # 线程锁
        self.completed_ips = 0  # 已完成的IP数量

def run(self):
        try:
            # 确定要处理的IP列表
            if self.ip_list:
                ips_to_process = self.ip_list
            else:
                start = ipaddress.IPv4Address(self.start_ip)
                end = ipaddress.IPv4Address(self.end_ip)
                if start > end:
                    self.output_received.emit("错误：起始IP必须小于或等于结束IP\n")
                    return
                ips_to_process = [str(ipaddress.IPv4Address(ip)) 
                                for ip in range(int(start), int(end) + 1)]

total_ips = len(ips_to_process)
            self.output_received.emit(f"总共需要处理 {total_ips} 个IP地址\n")

# 创建线程池
            with ThreadPoolExecutor(max_workers=self.max_threads) as executor:
                # 提交所有任务
                future_to_ip = {
                    executor.submit(
                        self.process_single_ip, 
                        ip
                    ): ip for ip in ips_to_process
                }

# 处理输出队列的线程
                output_thread = threading.Thread(target=self.process_output_queue)
                output_thread.daemon = True
                output_thread.start()

# 等待所有任务完成
                for future in future_to_ip:
                    if not self.running:
                        executor.shutdown(wait=False)
                        break
                    try:
                        future.result()
                    except Exception as e:
                        ip = future_to_ip[future]
                        self.output_queue.put(f"处理IP {ip} 时发生错误: {str(e)}\n")

# 等待输出队列处理完成
            self.output_queue.put(None)  # 发送结束信号
            output_thread.join()

except Exception as e:
            self.output_received.emit(f"执行过程中发生错误: {str(e)}\n")

def process_single_ip(self, ip):
        """处理单个IP的方法"""
        try:
            self.output_queue.put(f"\n开始处理IP: {ip}\n")
            if self.protocol == "SSH":
                self.execute_ssh(ip)
            else:
                self.execute_telnet(ip)

with self.lock:
                self.completed_ips += 1
                progress = (self.completed_ips / len(self.ip_list or 
                    range(int(ipaddress.IPv4Address(self.start_ip)), 
                         int(ipaddress.IPv4Address(self.end_ip)) + 1))) * 100
                self.progress_updated.emit(int(progress))

except Exception as e:
            self.output_queue.put(f"[{ip}] 处理失败: {str(e)}\n")

def process_output_queue(self):
        """处理输出队列的方法"""
        while True:
            message = self.output_queue.get()
            if message is None:  # 结束信号
                break
            self.output_received.emit(message)
            self.output_queue.task_done()

def execute_ssh(self, ip):
        try:
            ssh = paramiko.SSHClient()
            ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            
            # 连接重试机制
            max_retries = 3
            retry_count = 0
            
            while retry_count < max_retries:
                try:
                    ssh.connect(
                        ip,
                        port=22,
                        username=self.username,
                        password=self.password,
                        timeout=10,
                        compress=True,
                        allow_agent=False,
                        look_for_keys=False
                    )
                    self.output_queue.put(f"[{ip}] SSH连接成功!\n")
                    break
                except paramiko.SSHException as ssh_error:
                    retry_count += 1
                    self.output_queue.put(f"SSH连接尝试 {retry_count}/{max_retries}: {str(ssh_error)}\n")
                    if retry_count == max_retries:
                        raise
                    time.sleep(2)
            
            channel = ssh.invoke_shell()
            time.sleep(1)
            
            for command in self.commands:
                if not self.running:
                    break
                
                self.output_queue.put(f"[{ip}] 执行命令: {command}\n")
                channel.send(command + '\n')
                time.sleep(2)  # 减少等待时间
                
                output = ''
                while channel.recv_ready():
                    output += channel.recv(4096).decode('utf-8', errors='ignore')
                self.output_queue.put(f"[{ip}] {output}\n")
            
            channel.close()
            ssh.close()
            
        except Exception as e:
            self.output_queue.put(f"[{ip}] 连接错误: {str(e)}\n")

def execute_telnet(self, ip):
        try:
            tn = telnetlib.Telnet(ip, timeout=10)
            
            # 根据是否有用户名来处理不同的登录流程
            if self.username:  # 如果有用户名
                tn.read_until(b"login: ", timeout=5)
                tn.write(self.username.encode('ascii') + b"\n")
                
            # 等待密码提示
            tn.read_until(b"Password: ", timeout=5)
            tn.write(self.password.encode('ascii') + b"\n")
            
            self.output_queue.put(f"[{ip}] Telnet连接成功!\n")

for command in self.commands:
                if not self.running:
                    break
                    
                self.output_queue.put(f"[{ip}] 执行命令: {command}\n")
                tn.write(command.encode('ascii') + b"\n")
                output = tn.read_until(b"#", timeout=5).decode('ascii')
                self.output_queue.put(f"[{ip}] {output}\n")

tn.close()
        except Exception as e:
            self.output_queue.put(f"[{ip}] 连接错误: {str(e)}\n")

def stop(self):
        self.running = False

class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("交换机批量命令执行程序")
        self.setMinimumSize(1000, 700)
        self.executor = None
        self.ip_list = []  # 存储从文件导入的IP列表
        
        # 创建主窗口部件
        main_widget = QWidget()
        self.setCentralWidget(main_widget)
        layout = QVBoxLayout(main_widget)
        
        # IP选择模式
        ip_mode_layout = QHBoxLayout()
        self.ip_mode_group = QComboBox()
        self.ip_mode_group.addItems(["IP范围模式", "IP列表模式"])
        self.ip_mode_group.currentTextChanged.connect(self.switch_ip_mode)
        ip_mode_layout.addWidget(QLabel("IP输入模式:"))
        ip_mode_layout.addWidget(self.ip_mode_group)
        ip_mode_layout.addStretch()
        layout.addLayout(ip_mode_layout)
        
        # IP范围配置区域
        self.ip_range_widget = QWidget()
        ip_range_layout = QHBoxLayout(self.ip_range_widget)
        self.start_ip_input = QLineEdit()
        self.start_ip_input.setPlaceholderText("起始IP")
        self.end_ip_input = QLineEdit()
        self.end_ip_input.setPlaceholderText("结束IP")
        
        ip_range_layout.addWidget(QLabel("IP范围:"))
        ip_range_layout.addWidget(self.start_ip_input)
        ip_range_layout.addWidget(QLabel("到"))
        ip_range_layout.addWidget(self.end_ip_input)
        layout.addWidget(self.ip_range_widget)
        
        # IP列表配置区域
        self.ip_list_widget = QWidget()
        ip_list_layout = QHBoxLayout(self.ip_list_widget)
        self.ip_file_path = QLineEdit()
        self.ip_file_path.setReadOnly(True)
        self.ip_file_path.setPlaceholderText("请选择包含IP地址的文件(支持txt/csv)")
        self.import_ip_button = QPushButton("导入IP")
        self.import_ip_button.clicked.connect(self.import_ip_list)
        
        ip_list_layout.addWidget(self.ip_file_path)
        ip_list_layout.addWidget(self.import_ip_button)
        layout.addWidget(self.ip_list_widget)
        self.ip_list_widget.hide()  # 默认隐藏IP列表模式
        
        # 添加线程数配置
        thread_config_layout = QHBoxLayout()
        self.thread_count_input = QSpinBox()
        self.thread_count_input.setRange(1, 50)  # 限制最大线程数
        self.thread_count_input.setValue(10)  # 默认10个线程
        thread_config_layout.addWidget(QLabel("并发线程数:"))
        thread_config_layout.addWidget(self.thread_count_input)
        thread_config_layout.addStretch()
        layout.addLayout(thread_config_layout)
        
        # 添加进度条
        self.progress_bar = QProgressBar()
        layout.addWidget(self.progress_bar)
        
        # 连接配置区域
        connection_layout = QHBoxLayout()
        self.username_input = QLineEdit()
        self.username_input.setPlaceholderText("用户名")
        self.password_input = QLineEdit()
        self.password_input.setPlaceholderText("密码")
        self.password_input.setEchoMode(QLineEdit.EchoMode.Password)
        
        self.protocol_select = QComboBox()
        self.protocol_select.addItems(["SSH", "Telnet"])
        
        connection_layout.addWidget(QLabel("连接类型:"))
        connection_layout.addWidget(self.protocol_select)
        connection_layout.addWidget(self.username_input)
        connection_layout.addWidget(self.password_input)
        layout.addLayout(connection_layout)
        
        # 命令显示和按钮区域
        self.command_display = QTextEdit()
        self.command_display.setPlaceholderText("导入的命令将显示在这里")
        self.command_display.setReadOnly(True)
        layout.addWidget(self.command_display)
        
        # 按钮区域
        button_layout = QHBoxLayout()
        self.import_button = QPushButton("导入脚本")
        self.import_button.clicked.connect(self.import_script)
        
        self.execute_button = QPushButton("执行")
        self.execute_button.clicked.connect(self.execute_commands)
        
        self.stop_button = QPushButton("停止")
        self.stop_button.clicked.connect(self.stop_execution)
        self.stop_button.setEnabled(False)
        
        button_layout.addWidget(self.import_button)
        button_layout.addWidget(self.execute_button)
        button_layout.addWidget(self.stop_button)
        layout.addLayout(button_layout)
        
        # 输出显示区域
        self.output_display = QTextEdit()
        self.output_display.setPlaceholderText("命令执行结果将显示在这里")
        self.output_display.setReadOnly(True)
        layout.addWidget(self.output_display)
        
        self.commands = []

# 添加错误处理
        try:
            # ... 现有的初始化代码 ...
            pass
        except Exception as e:
            QMessageBox.critical(self, "错误", f"程序初始化失败: {str(e)}")
            sys.exit(1)

def switch_ip_mode(self, mode):
        if mode == "IP范围模式":
            self.ip_range_widget.show()
            self.ip_list_widget.hide()
        else:
            self.ip_range_widget.hide()
            self.ip_list_widget.show()

def import_ip_list(self):
        try:
            file_name, _ = QFileDialog.getOpenFileName(
                self,
                "选择IP列表文件",
                os.path.expanduser("~"),  # 从用户目录开始
                "文本文件 (*.txt);;CSV文件 (*.csv)"
            )
            if file_name:
                self.ip_list = []
                with open(file_name, 'r', encoding='utf-8') as file:  # 指定编码
                    if file_name.endswith('.csv'):
                        import csv
                        reader = csv.reader(file)
                        for row in reader:
                            if row:
                                ip = row[0].strip()
                                if self.is_valid_ip(ip):
                                    self.ip_list.append(ip)
                    else:
                        for line in file:
                            ip = line.strip()
                            if self.is_valid_ip(ip):
                                self.ip_list.append(ip)

self.ip_file_path.setText(file_name)
                QMessageBox.information(self, "成功", f"成功导入 {len(self.ip_list)} 个IP地址")
        except Exception as e:
            QMessageBox.critical(self, "错误", f"无法读取文件: {str(e)}")

def is_valid_ip(self, ip):
        try:
            ipaddress.ip_address(ip)
            return True
        except ValueError:
            return False

def import_script(self):
        try:
            file_name, _ = QFileDialog.getOpenFileName(
                self,
                "选择脚本文件",
                os.path.expanduser("~"),  # 从用户目录开始
                "文本文件 (*.txt)"
            )
            if file_name:
                with open(file_name, 'r', encoding='utf-8') as file:  # 指定编码
                    self.commands = [line.strip() for line in file if line.strip()]
                    self.command_display.setText("\n".join(self.commands))
        except Exception as e:
            QMessageBox.critical(self, "错误", f"无法读取文件: {str(e)}")

def execute_commands(self):
        if not self.commands:
            QMessageBox.warning(self, "警告", "请先导入命令脚本")
            return
            
        if not self.validate_inputs():
            return
        
        # 重置进度条
        self.progress_bar.setValue(0)
        
        # 创建执行器时传入线程数
        if self.ip_mode_group.currentText() == "IP范围模式":
            self.executor = CommandExecutor(
                self.start_ip_input.text(),
                self.end_ip_input.text(),
                self.username_input.text(),
                self.password_input.text(),
                self.protocol_select.currentText(),
                self.commands,
                max_threads=self.thread_count_input.value()
            )
        else:
            if not self.ip_list:
                QMessageBox.warning(self, "警告", "请先导入IP列表")
                return
            self.executor = CommandExecutor(
                self.ip_list[0],
                self.ip_list[-1],
                self.username_input.text(),
                self.password_input.text(),
                self.protocol_select.currentText(),
                self.commands,
                self.ip_list,
                max_threads=self.thread_count_input.value()
            )
        
        self.executor.output_received.connect(self.update_output)
        self.executor.progress_updated.connect(self.progress_bar.setValue)
        self.executor.start()
        
        self.execute_button.setEnabled(False)
        self.stop_button.setEnabled(True)

def validate_inputs(self):
        if self.ip_mode_group.currentText() == "IP范围模式":
            if not self.start_ip_input.text() or not self.end_ip_input.text():
                QMessageBox.warning(self, "警告", "请输入IP范围")
                return False
        else:
            if not self.ip_list:
                QMessageBox.warning(self, "警告", "请导入IP列表")
                return False
                
        # 如果是Telnet模式，用户名可以为空
        if self.protocol_select.currentText() == "SSH" and not self.username_input.text():
            QMessageBox.warning(self, "警告", "SSH模式下必须输入用户名")
            return False
        
        if not self.password_input.text():
            QMessageBox.warning(self, "警告", "请输入密码")
            return False
        return True

def stop_execution(self):
        if self.executor:
            self.executor.stop()
            self.executor = None
        self.execute_button.setEnabled(True)
        self.stop_button.setEnabled(False)

def update_output(self, text):
        self.output_display.append(text)
        # 自动滚动到底部
        self.output_display.verticalScrollBar().setValue(
            self.output_display.verticalScrollBar().maximum()
        )

if __name__ == '__main__':
    try:
        app = QApplication(sys.argv)
        window = MainWindow()
        window.show()
        sys.exit(app.exec())
    except Exception as e:
        QMessageBox.critical(None, "错误", f"程序启动失败: {str(e)}") 
```