如何将paramiko执行SSH命令的实时输出同步显示到tkinter的Text组件中
解决思路
核心是拆分阻塞的SSH操作和tkinter主循环分离,子线程处理SSH收发,通过after()方法调度UI更新,保证线程安全,多会话之间互不干扰。
完整修改后代码
import paramiko import sys import time import tkinter from threading import Thread # 单独封装SSH会话窗口类,方便创建多窗口互不干扰 class SSHOutputWindow: def __init__(self, window_title, output_file_path): # 初始化窗口 self.window = tkinter.Tk() self.window.wm_title(window_title) self.window.geometry("600x400") # 文本组件和滚动条 self.text = tkinter.Text(self.window, height=600, width=400) scrollbar = tkinter.Scrollbar(self.window) scrollbar.pack(side=tkinter.RIGHT, fill=tkinter.Y) scrollbar.config(command=self.text.yview) self.text.config(yscrollcommand=scrollbar.set) self.text.pack(side=tkinter.LEFT, fill=tkinter.BOTH, expand=True) # 输出文件 self.output_file = open(output_file_path, "a+") self.channel = None self.ssh = None # 线程安全的UI更新方法 def append_output(self, content): self.text.insert(tkinter.END, content) # 自动滚动到底部 self.text.see(tkinter.END) self.output_file.write(content) self.output_file.flush() print(content) # SSH连接和命令执行逻辑,放子线程运行 def run_ssh_commands(self, hostname, username, password, cmd_list): # 连接SSH i = 0 while True: try: self.ssh = paramiko.SSHClient() self.ssh.load_system_host_keys() self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) self.ssh.connect(hostname, 22, username, password) break except paramiko.AuthenticationException: self.window.after(0, self.append_output, f"{hostname} 认证失败\n") sys.exit(1) except: self.window.after(0, self.append_output, f"无法连接到{hostname},等待重试\n") i += 1 time.sleep(2) if i == 30: self.window.after(0, self.append_output, f"连接{hostname}失败,放弃连接\n") sys.exit(1) self.window.after(0, self.append_output, "\n****** 连接成功 ******\n") self.channel = self.ssh.invoke_shell() # 执行命令 for command in cmd_list: self.channel.send(command + '\n') time.sleep(3) self.channel.send(chr(3)) # CTRL+C终止命令 outlines = self.channel.recv(10240).decode("ascii") # 通过after调度UI更新,线程安全 self.window.after(0, self.append_output, outlines) # 执行完关闭资源 self.channel.close() self.ssh.close() self.output_file.close() # 启动子线程执行SSH逻辑 def start(self, hostname, username, password, cmd_list): Thread(target=self.run_ssh_commands, args=(hostname, username, password, cmd_list), daemon=True).start() self.window.mainloop() def main(): # 第一个会话示例 cmd_list1 = [ "config vdom", "edit Test", "diag sniffer packet any 100 l", "end" ] window1 = SSHOutputWindow("会话1输出窗口", "session1_output.txt") # 要是开第二个并行会话就再实例化一个即可,输出完全隔离不会混淆 # cmd_list2 = [你的第二组命令] # window2 = SSHOutputWindow("会话2输出窗口", "session2_output.txt") # Thread(target=window2.start, args=("目标IP", "账号", "密码", cmd_list2), daemon=True).start() window1.start("1.1.1.1", "admin", "admin", cmd_list1) if __name__ == '__main__': main()
关键改动说明
- 封装
SSHOutputWindow类,每个SSH会话对应一个独立实例,窗口、输出文件、SSH连接完全隔离,多会话并行不会出现输出混淆 - SSH连接和命令执行逻辑放到独立子线程运行,不会阻塞tkinter主循环,窗口启动后立刻显示,收到输出立刻更新
- 所有UI更新操作通过
window.after(0, 待执行方法, 参数)调度到tkinter主线程执行,符合tkinter线程安全要求,不会出现界面卡死或更新失败问题 - 新增自动滚动到文本底部,方便查看最新输出,输出文件实时flush避免缓存延迟写入
内容的提问来源于stack exchange,提问作者user_PM
相关产品推荐
相关产品推荐

