网站制度建设存在的问题,php 个人网站 源码,美食杰网站的建设目的,安徽建设工程网站1. 功能描述
1.1 系统具有的功能描述
#xff08;1#xff09;连接服务器#xff1a;用户可以通过系统连接到远程服务器#xff0c;系统支持多个服务器配置#xff0c;并且可以方便地管理这些配置。
#xff08;2#xff09;执行命令#xff1a;用户可以在连接成功后…1. 功能描述
1.1 系统具有的功能描述
1连接服务器用户可以通过系统连接到远程服务器系统支持多个服务器配置并且可以方便地管理这些配置。
2执行命令用户可以在连接成功后在界面中输入命令并执行在远程服务器上执行。执行结果会显示在界面中包括标准输出和标准错误。
3上传和下载文件用户可以通过系统上传本地文件到远程服务器或者下载远程服务器上的文件到本地。
4服务器监控系统可以监控远程服务器的 CPU 使用率、内存使用率和进程列表实时显示在界面上。
5查看日志文件用户可以通过系统查看远程服务器上的日志文件并且可以选择下载到本地进行查看。
6Vim编辑文件用户可以在界面上编辑远程服务器上的文件编辑后可以保存到服务器上。
1.2 所用技术与模块描述
1JSON模块json用于存储和加载配置信息、与服务器交换数据等。
操作系统路径模块os.path主要用于处理文件路径比如检查文件是否存在、获取文件或目录的详细信息等。
2正则表达式模块rePython的正则表达式模块用于字符串匹配和替换。
3套接字模块socket提供了网络通信的底层接口允许应用程序在网络上发送和接收数据。
4子进程模块subprocess允许从Python脚本中启动新的应用程序连接到它们的输入/输出/错误管道并获取它们的返回码。 5线程模块threading提供了对线程的基本支持线程是轻量级的进程允许在Python程序中并发执行代码。
时间模块time提供了与时间相关的各种功能如获取当前时间、执行时间延迟等。
6Tkinter简单对话框模块tkinter.simpledialogTkinter GUI库的一部分提供了简单的对话框功能如输入对话框、消息框等。
7Paramiko模块paramiko一个实现SSHv2协议的Python库支持连接和操作SSH服务器。
8Watchdog事件模块watchdog.eventswatchdog库的一部分提供了文件系统事件如文件创建、修改、删除等的接口。
9Watchdog观察者模块watchdog.observerswatchdog库的一部分提供了跨平台的文件系统事件观察者实现。
1.3 ssh暗影坤手概述
一个简单易用的SSH管理工具允许用户通过图形化界面实现对虚拟机VM的基本操作。通过集成SSH协议该工具能够提供远程登录、命令执行、文件传输等核心功能为用户提供便捷、安全的虚拟机管理体验。 2. 系统功能实现
2.1 服务器信息管理的功能设计与实现
当前目录下创建一个server.json的文件
将所有服务器信息保存在json文件中调用时借助函数加载文件读取数据。
代码展示如下
def load_server(self):with open(server.json, r, encodingutf-8) as fp:self.servers json.load(fp)print(f获取到的服务器信息{self.servers})def save_server(self):with open(server.json, w, encodingutf-8) as fp:fp.write(json.dumps(self.servers)) 2.2 连接服务器的功能设计与实现
1读取json文件中的数据将ip、端口、用户名等信息传到connect中连接服务器。
代码展示如下
def connecting(self):self.connected Noneself.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy)try:# 连接到服务器self.ssh.connect(hostnameself.server[ip], portself.server[port], usernameself.server[username],passwordself.server[password])print(f【success】连接到服务器【{self.server[ip]}】成功)self.connected Trueself.get_pwd()return Trueexcept Exception as e:print(f【errors】连接到服务器【{self.server[ip]}】失败)self.connected Falsereturn False
2创建连接指定服务器函数并且创建线程拥堵单独执行连接操作不影响其他操作的执行创建显示函数将连接成功或失败信息展示到界面以便用户清楚服务器是否已经连接成功。
代码展示如下
# 连接到指定的服务器
def connect(self, server):print(连接到指定的服务器)self.mssh.server serverprint(f服务器信息{self.mssh.server})socket.setdefaulttimeout(5)# 在单独的线程中执行连接操作thread_connect threading.Thread(targetself.connect_thread, args(server,))thread_connect.start()self.pop_win.destroy()def connect_thread(self, server):result self.mssh.connecting()if result:self.is_connected Trueself.update_connection_status(server, True)else:self.is_connected Falseself.update_connection_status(server, False)
#服务器连接展示
def update_connection_status(self, server, connected):if connected:self.label_server.config(textf服务器{server[ip]}连接成功, foregroundgreen)else:self.label_server.config(textf服务器{server[ip]}连接失败, foregroundred)
2.3 添加服务器的功能设计与实现
1首先创建函数用于添加服务器的操作然后输入服务器的IP地址账户名和密码添加服务器最后将服务器信息保存到json文件中。
代码展示如下
def add_server(self, ip, port, username, password, description):self.servers.append({description: description,ip: ip,port: port,username: username,password: password})self.save_server()
2创建新建函数用于与用户交互的图形化界面让用户输入服务器的IP地址账户名和密码添加服务器信息。
代码展示如下
# 新建连接
def new_connect_ui(self):self.pop_win tk.Toplevel(self.root)# 设置标题和大小self.pop_win.title(新建连接)self.pop_win.geometry(300x200600250)frame tk.Frame(self.pop_win)frame.grid()# IPip_label tk.Label(frame, text服务器IP地址:)ip_label.grid(row0, column0)self.ip_entry tk.Entry(frame)self.ip_entry.grid(row0, column1)# Portport_label tk.Label(frame, text服务器端口号:)port_label.grid(row1, column0)self.port_entry tk.Entry(frame)self.port_entry.grid(row1, column1)# Usernameusername_label tk.Label(frame, text用户名:)username_label.grid(row2, column0)self.username_entry tk.Entry(frame)self.username_entry.grid(row2, column1)# Passwordpassword_label tk.Label(frame, text密码:)password_label.grid(row3, column0)self.password_entry tk.Entry(frame, show*)self.password_entry.grid(row3, column1)# Descriptiondescription_label tk.Label(frame, text备注信息:)description_label.grid(row4, column0)self.description_entry tk.Entry(frame)self.description_entry.grid(row4, column1)# Buttonsubmit_button tk.Button(frame, text保存, commandself.save_server_info)submit_button.grid(row5, columnspan2)
3获取用户所输入的信息将服务器信息保存到文件。
代码展示如下
def save_server_info(self):ip self.ip_entry.get()port self.port_entry.get()username self.username_entry.get()password self.password_entry.get()description self.description_entry.get()# 调用 add_server 函数将获取到的信息添加到服务器列表中ssh.add_server(ipip, portint(port), usernameusername, passwordpassword, descriptiondescription)messagebox.showinfo(成功,服务器信息已保存)# print(服务器信息已保存) 2.4 删除服务器的功能设计与实现
创建删除函数当用户点击确定时调用remove移除服务器信息并且返回删除成功提示信息。
代码展示如下
def del_connect(self, server):print(删除指定的服务器)# 输出服务器信息print(f服务器信息{server})# 确认是否删除服务器confirm messagebox.askyesno(确认删除, f确认删除服务器 {server[ip]} 吗)if confirm:# 从服务器列表中删除指定的服务器if server in self.mssh.servers:self.mssh.servers.remove(server)self.pop_win.destroy()self.open_connect_ui()messagebox.showinfo(删除成功, f服务器 {server[ip]} 已成功删除)else:messagebox.showerror(错误, 服务器不存在无法删除)# 保存更新后的服务器列表到配置文件中self.mssh.save_server() 2.5 修改服务器的功能设计与实现
1创建修改函数用于与用户交互的图形化界面让用户修改服务器的IP地址、账户名和密码等服务器信息。
代码展示如下
def modify_server(self, server):print(修改服务器信息)self.mssh.server server.copy() # 添加这一行来复制服务器信息self.pop_win_modify tk.Toplevel(self.root)frame tk.Frame(self.pop_win_modify)frame.grid()# IPip_label tk.Label(frame, text服务器IP地址:)ip_label.grid(row0, column0)self.ip_entry tk.Entry(frame, textvariabletk.StringVar(valueserver[ip]))self.ip_entry.grid(row0, column1)# Portport_label tk.Label(frame, text服务器端口号:)port_label.grid(row1, column0)self.port_entry tk.Entry(frame, textvariabletk.StringVar(valuestr(server[port])))self.port_entry.grid(row1, column1)# Usernameusername_label tk.Label(frame, text用户名:)username_label.grid(row2, column0)self.username_entry tk.Entry(frame, textvariabletk.StringVar(valueserver[username]))self.username_entry.grid(row2, column1)# Passwordpassword_label tk.Label(frame, text密码:)password_label.grid(row3, column0)self.password_entry tk.Entry(frame, show*, textvariabletk.StringVar(valueserver[password]))self.password_entry.grid(row3, column1)# Descriptiondescription_label tk.Label(frame, text备注信息:)description_label.grid(row4, column0)self.description_entry tk.Entry(frame, textvariabletk.StringVar(valueserver[description]))self.description_entry.grid(row4, column1)# Buttonsubmit_button tk.Button(frame, text保存, commandself.save_modified_server)submit_button.grid(row5, columnspan2)
2获取修改后的服务器信息保存到json文件中。
代码展示如下
def save_modified_server(self):modified_server {ip: self.ip_entry.get(),port: int(self.port_entry.get()),username: self.username_entry.get(),password: self.mssh.server[password], # 假设密码不变保持原值description: self.description_entry.get(),}index self.mssh.servers.index(self.mssh.server)self.mssh.servers[index] modified_servermessagebox.showinfo(成功, 服务器信息已成功修改)self.pop_win_modify.destroy()self.open_connect_ui()self.mssh.save_server() 2.6 执行交互式命令的功能设计与实现
主要调用paramiko库的exec_command方法执行命令同时获取标准输入、标准输出和标准错误。
1基本命令cd的实现
创建exec方法主要用于执行远程命令并处理用户输入的路径确保命令相对于当前路径执行。方法接收用户输入的命令执行命令若是cd命令则更新当前路径获取命令执行结果并返回执行结果。
代码展示如下
def exec(self, command):if command.startswith(cd):tmp_path command.split( )[-1]if tmp_path.startswith(/):# 绝对路径处理self.path tmp_pathelse:# todo 如果用户输入的是一个相对路径呢print(用户输入的是相对路径)if tmp_path.startswith(..):self.path self.path / tmp_pathelse:self.path self.path tmp_pathelse:# 根据当前的路径去执行指令command fcd {self.path} {command}print(f-输入指令({self.path}){command})stdin, stdout, stderr self.ssh.exec_command(command)result stdout.read().decode(utf-8)print(f-返回{result})# 用户输入的路径里包含.. 通过pwd刷新当前路径if command.find(..) 0:self.get_pwd()return result 2创建函数用于获取当前路径
代码展示如下
# 使用pwd更新当前路径
def get_pwd(self):stdin, stdout, stderr self.ssh.exec_command(fcd {self.path} pwd)result stdout.read().decode(utf-8)self.path result.strip() 交互式命令实现
通过SSH与远程服务器进行交互,用户可以输入命令并发送到远程服务器同时应用会读取并显示来自远程服务器的输出。所有的输入/输出处理都在单独的线程中进行以保持GUI的响应性。
代码展示如下
def _read_channel(self, channel,output_callback):在独立线程中读取通道输出并调用回调函数处理while True:if channel.recv_ready():data channel.recv(1024)output_callback(data)def handle_output(self, data):# 先解码数据decoded_data data.decode(utf-8)# 去除颜色转义序列cleaned_data re.sub(r\x1B\[([0-?]*[ -/]*[-~]), , decoded_data)# 插入到文本框的末尾self.txt_result.insert(tk.END, cleaned_data)# 确保滚动条跟随到最后self.txt_result.see(tk.END)def okk(self,event):self.command.set(q)self.ok(self)def ok(self, event):command self.command.get()if not self.mssh.connected:self.txt_result.insert(tk.END, 未连接到服务器\n)return# 如果ssh_channel尚未创建创建一个新的if self.mssh_channel is None:self.mssh_channel self.mssh.ssh.invoke_shell()self.mssh_channel.set_combine_stderr(True)try:# 使用ssh_channel发送命令self.mssh_channel.send(f{command}\n)# 创建线程监听输出并调用UI提供的回调函数t threading.Thread(targetself.mssh._read_channel, args(self.mssh_channel, self.handle_output))t.daemon Truet.start()# 设置延时清空输入框self.root.after(2000, lambda: self.command.set())except Exception as e:self.txt_result.insert(tk.END, f执行错误: {str(e)}\n)# 清空输入框self.command.set()
2.7 vim编辑器功能设计与实现
利用watchdog模块监控文件系统的更改如文件的创建、修改、移动或删除等并在这些事件发生时执行自定义操作。
1定义watch_and_upload方法用于监视本地文件的修改并在远程服务器上进行相应的上传操作。
代码展示如下
def watch_and_upload(self, local_file_path, remote_full_path):class FileModifiedHandler(FileSystemEventHandler):def __init__(self, ssh_handler, remote_full_path):self.ssh_handler ssh_handlerself.remote_full_path remote_full_pathdef on_modified(self, event):if event.is_directory or not event.src_path local_file_path:return# messagebox.showinfo(文件被修改, f文件被修改{event.src_path})try:self.ssh_handler.connecting() # 重新连接SSH如果需要sftp self.ssh_handler.ssh.open_sftp()sftp.put(local_file_path, self.remote_full_path)sftp.close()# messagebox.showinfo(文件上传成功, fvim修改成功: {local_file_path} - {self.remote_full_path})except Exception as e:messagebox.showerror(文件上传失败, f文件上传失败: {e})event_handler FileModifiedHandler(self, remote_full_path)observer Observer()observer.schedule(event_handler, pathos.path.dirname(local_file_path), recursiveFalse)observer.start()try:# 等待用户操作或其他信号来停止监视# 这里使用time.sleep作为示例但建议使用更好的机制如事件或信号while True:time.sleep(1)except KeyboardInterrupt:observer.stop()observer.join()
2创建vim_edit 函数用于在vscode中编辑文件。首先提示用户输入要编辑的文件名并获取远程文件的路径用户选择本地目录并构造本地文件的完整路径连接到远程服务器并创建 SFTP 客户端检查远程文件是否存在如果不存在则在远程服务器上创建空文件。下载文件到本地并启动文件监视线程 watch_and_upload然后调用子进程打开文件编辑器。 def vim_edit(self):remote_file simpledialog.askstring(vim, 请输入要编辑的完整远程文件路径:)if remote_file is None: # 用户点击“取消”或关闭对话框returnelif not remote_file: # 用户输入了空字符串messagebox.showerror(错误, 请输入文件路径)return# 检查用户是否输入了以/开头的绝对路径if not remote_file.startswith(/):messagebox.showerror(错误, 请输入绝对路径例如/home/user/file.txt)return# 临时文件目录tmp filedialog.askdirectory(title选择下载到的本地目录)if not tmp: # 用户取消了选择目录returnlocal_file_path os.path.join(tmp, os.path.basename(remote_file))try:self.connecting()sftp self.ssh.open_sftp()# 检查远程文件是否存在try:sftp.stat(remote_file)except FileNotFoundError:# 如果不存在则在远程服务器上创建空文件with sftp.open(remote_file, wb) as f:pass # 写入空内容或者模板内容# 下载文件到本地sftp.get(remote_file, local_file_path)sftp.close()# 启动文件监视线程watch_thread threading.Thread(targetself.watch_and_upload, args(local_file_path, remote_file))watch_thread.start()# 打开文件编辑器subprocess.run([Code, local_file_path], shellTrue)except Exception as e:messagebox.showerror(文件下载失败, f文件下载失败{e})print(f文件下载失败{e})
2.8 文件远程上传下载的功能设计与实现
1文件上传功能实现
首先检查是否已连接到服务器。如果未连接到服务器则显示错误消息并返回。如果已连接到服务器则调用filedialog.askopenfilename()方法请求用户选择要上传的文件。接下来如果用户选择了文件则提取文件名并创建一个SFTP客户端。然后计算远程路径并在控制台打印文件和远程路径信息。最后使用SFTP客户端将本地文件上传到远程路径并显示上传成功的消息框。如果用户未选择文件则输出“请选择要上传的文件”。如果出现任何异常则显示文件上传失败的错误消息。
代码展示如下
# 上传文件
def upload_file(self):if not self.mssh.connected:messagebox.showerror(错误, 请先连接到服务器!)returnelse:self.file_path filedialog.askopenfilename()try:if self.file_path:# 获取文件名file_name self.file_path.split(/)[-1]# 创建 SFTP 客户端sftp_client self.mssh.ssh.open_sftp()remote_path f{self.mssh.path}/{file_name}print(f文件{self.file_path},上传到{remote_path})# 上传文件sftp_client.put(self.file_path, remote_path)messagebox.showinfo(成功, 上传成功)sftp_client.close()else:print(请选择要上传的文件)except Exception as e:messagebox.showerror(错误, f文件上传失败: {e})#判断是否连接服务器def vim_edit(self):if not self.mssh.connected:messagebox.showerror(错误, 请先连接到服务器!)returnself.mssh.vim_edit() 2文件下载功能实现
使用tk中的simpledialog.askstring 获取用户输入的要下载文件名filedialog.askdirectory获取用户所选目录的完整路径创建一个SFTP连接并调用get方法来下载文件调用完关闭SFTP连接弹窗提示用户下载成功。
代码展示如下 def download_file(self):#tk.simpledialog.askstring 的主要用途是获取用户输入的文本信息remote_file tk.simpledialog.askstring(输入文件名, 请输入要下载的文件名:)if remote_file is None: # 用户点击“取消”或关闭对话框returnelif not remote_file: # 用户输入了空字符串messagebox.showerror(错误, 请输入文件名)return#filedialog.askdirectory返回的是用户所选目录的完整路径而不是文件名local_path filedialog.askdirectory(title选择下载到的本地目录)if local_path:try:sftp self.ssh.open_sftp()local_file_path os.path.join(local_path, remote_file)sftp.get(remote_file, local_file_path)sftp.close()messagebox.showinfo(下载成功, f文件下载成功: {remote_file} - {local_path})except Exception as e:messagebox.showerror(下载失败, f文件下载失败: {e})#下载文件
def downloader(self):if not self.mssh.connected:messagebox.showerror(错误, 请先连接到服务器!)returnself.mssh.download_file() 2.9 服务器监控功能设计与实现
通过SSH连接到远程服务器并在界面上展示服务器的进程信息、系统资源利用情况和磁盘使用情况同时处理连接状态的变化。
1进程展示的实现
创建get_process方法通过执行ps -ef命令获取进程信息并展示在界面上的表格中。如果连接失败或执行命令出错会通过弹出窗口显示错误信息。 代码展示如下
def get_process(self):if not self.mssh.connected:messagebox.showerror(错误, 请先连接到服务器!)returnself.pop_win tk.Toplevel(self.root)self.pop_win.title(进程)self.pop_win.geometry(800x600600250)table_frame ttk.Frame(self.pop_win)table_frame.pack(padx10, pady10, fillboth, expandTrue)server self.mssh.serverip server[ip]username server[username]password server[password]try:client paramiko.SSHClient()client.set_missing_host_key_policy(paramiko.AutoAddPolicy())client.connect(ip, 22, username, password)stdin, stdout, stderr client.exec_command(ps -ef)result stdout.read().decode(utf-8).strip().split(\n)# 创建表格table ttk.Treeview(table_frame, columns(user, pid, cpu, mem, vsz, rss, tty, stat, start, time, command), showheadings)table.heading(user, text用户)table.heading(pid, text进程ID)table.heading(cpu, textCPU%)table.heading(mem, text内存%)table.heading(vsz, text虚拟内存)table.heading(rss, text常驻内存)table.heading(tty, text终端)table.heading(stat, text状态)table.heading(start, text启动时间)table.heading(time, textCPU时间)table.heading(command, text命令)table.pack(sideleft, fillboth, expandTrue)# 添加滚动条scrollbar ttk.Scrollbar(table_frame, orientvertical, commandtable.yview)scrollbar.pack(sideright, filly)table.configure(yscrollcommandscrollbar.set)# 填充表格数据for line in result[1:]:parts line.split()if len(parts) 10:user, pid, cpu, mem, vsz, rss, tty, stat, start, time, *command partscommand .join(command)table.insert(, end, values(user, pid, cpu, mem, vsz, rss, tty, stat, start, time, command))# 调整列宽度table.column(#0, width0, stretchno) # 隐藏第一列table.column(user, width80, anchorw)table.column(pid, width60, anchore)table.column(cpu, width60, anchore)table.column(mem, width60, anchore)table.column(vsz, width100, anchore)table.column(rss, width100, anchore)table.column(tty, width60, anchorw)table.column(stat, width60, anchorw)table.column(start, width120, anchorw)table.column(time, width80, anchore)table.column(command, width300, anchorw)# 自动调整行高def fixed_map(option):return [elm for elm in style.map(Treeview, query_optoption) if elm[:2] ! (!disabled, !selected)]style ttk.Style()style.map(Treeview, foregroundfixed_map(foreground), backgroundfixed_map(background))except paramiko.AuthenticationException:messagebox.showerror(错误, 认证失败,请验证您的凭据)except paramiko.SSHException as sshException:messagebox.showerror(错误, f无法建立SSH连接: {sshException})except paramiko.BadHostKeyException as badHostKeyException:messagebox.showerror(错误, f无法验证服务器的主机密钥: {badHostKeyException})except Exception as e:messagebox.showerror(错误, f发生错误: {e})finally:if client in locals() and client:client.close() # 确保连接被关闭# 当窗口关闭时,确保没有遗留的引用或连接self.pop_win.protocol(WM_DELETE_WINDOW, self.on_closing)
2CPU、内存、网络等信息展示的实现
创建create_widgets方法制作界面上的标签、进度条和按钮。用于展示CPU、内存、网络等信息并提供查看磁盘使用情况的按钮。使用update_info方法用于在界面上更新系统信息包括CPU利用率、内存使用情况和网络速率。reset_system_monitoring方法重置系统监控状态将CPU信息、内存信息和网络信息重置为初始状态。show_disk_usage展示磁盘使用情况并创建一个新窗口展示磁盘信息。
代码展示如下
def create_widgets(self, frame):self.cpu_label ttk.Label(frame, textCPU信息)self.cpu_label.grid()self.cpu_progress ttk.Progressbar(frame, orienthorizontal, length200, modedeterminate)self.cpu_progress.grid(pady(0, 10))self.mem_label ttk.Label(frame, textMemory Usage: 0 MB / 0 MB)self.mem_label.grid()self.mem_progress ttk.Progressbar(frame, orienthorizontal, length200, modedeterminate)self.mem_progress.grid(pady(0, 5))self.net_label ttk.Label(frame, text网络信息)self.net_label.grid(pady5)self.disk_button tk.Button(frame, text查看磁盘使用情况, commandself.show_disk_usage)self.disk_button.grid(pady(5, 5))def update_info(self, frame):self.create_widgets(frame) # 创建标签# 定义进度条样式style ttk.Style()style.theme_use(default)style.configure(green.Horizontal.TProgressbar, foregroundgreen, backgroundgreen)style.configure(yellow.Horizontal.TProgressbar, foregroundyellow, backgroundyellow)style.configure(red.Horizontal.TProgressbar, foregroundred, backgroundred)prev_connected self.mssh.connected # 记录上一次的连接状态while True:if ip in self.mssh.server:if self.mssh.connected None:result_connect (f服务器{self.mssh.server[ip]}连接中, red)self.reset_system_monitoring() # 重置系统监控状态elif self.mssh.connected:result_connect (f服务器{self.mssh.server[ip]}连接成功, green)# -----------------------------系统信息------------------------------------------------server self.mssh.serverip server[ip]username server[username]password server[password]self.client paramiko.SSHClient()self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())self.client.connect(ip, 22, username, password)# 执行远程命令获取内存信息stdin, stdout, stderr self.client.exec_command(free -m)result stdout.read().decode(utf-8)# 获取系统内存信息lines result.split(\n)mem_info lines[1].split()total_mem int(mem_info[1])used_mem int(mem_info[2])mem_usage used_mem / total_mem * 100# 获取网络速率stdin1, stdout1, stderr1 self.client.exec_command(ip a)result stdout1.read().decode(utf-8)# 使用正则表达式提取数据包和字节数信息matches re.findall(rens160.*?RX packets (\d) bytes (\d).*?TX packets (\d) bytes (\d),result, re.DOTALL)if matches:rx_packets, rx_bytes, tx_packets, tx_bytes matches[0]else:rx_packets, rx_bytes, tx_packets, tx_bytes 0, 0, 0, 0time.sleep(1)stdin2, stdout2, stderr3 self.client.exec_command(ip a)result stdout2.read().decode(utf-8)matches re.findall(rens160.*?RX packets (\d) bytes (\d).*?TX packets (\d) bytes (\d),result, re.DOTALL)if matches:rx_packets, new_rx_bytes, tx_packets, new_tx_bytes matches[0]else:rx_packets, new_rx_bytes, tx_packets, new_tx_bytes 0, 0, 0, 0# 计算速率upload_speed ((int(new_tx_bytes) - int(tx_bytes)) / 1024) # MB/sdownload_speed ((int(new_rx_bytes) - int(rx_bytes)) / 1024)# 获取cpu利用率stdin4, stdout4, stderr4 self.client.exec_command(top -bn 1)result stdout4.read().decode(utf-8)cpu 0for line in result.split(\n):if line.startswith(%Cpu(s)):cpu_us re.findall(r\d\.\d, line.split(:)[1].split(,)[0])[0]cpu_sy re.findall(r\d\.\d, line.split(:)[1].split(,)[1])[0]cpu_ni re.findall(r\d\.\d, line.split(:)[1].split(,)[2])[0]cpu_id re.findall(r\d\.\d, line.split(:)[1].split(,)[3])[0]cpu_wa re.findall(r\d\.\d, line.split(:)[1].split(,)[4])[0]cpu_hi re.findall(r\d\.\d, line.split(:)[1].split(,)[5])[0]cpu_si re.findall(r\d\.\d, line.split(:)[1].split(,)[6])[0]cpu_st re.findall(r\d\.\d, line.split(:)[1].split(,)[7])[0]sum float(cpu_st) float(cpu_si) float(cpu_hi) float(cpu_wa) float(cpu_id) float(cpu_ni) float(cpu_sy) float(cpu_us)cpu sum - float(cpu_id)# 设置进度条样式if cpu 80:cpu_style red.Horizontal.TProgressbarelif cpu 50:cpu_style yellow.Horizontal.TProgressbarelse:cpu_style green.Horizontal.TProgressbarif mem_usage 80:mem_style red.Horizontal.TProgressbarelif mem_usage 50:mem_style yellow.Horizontal.TProgressbarelse:mem_style green.Horizontal.TProgressbar# 更新标签self.cpu_label.config(textfCPU信息{cpu:.2f}%)self.cpu_progress.config(stylecpu_style, valuecpu)self.mem_label.config(textfMemory Usage: {used_mem} MB / {total_mem} MB ({mem_usage:.2f}%))self.mem_progress.config(stylemem_style, valuemem_usage)self.net_label.config(textf网络信息↑{upload_speed:.2f}MB/s \t ↓{download_speed:.2f}MB/s)else:result_connect (f服务器{self.mssh.server[ip]}连接失败, red)self.reset_system_monitoring() # 重置系统监控状态else:result_connect (f服务器未选择, black)self.reset_system_monitoring() # 重置系统监控状态# 检查连接状态是否发生变化if self.mssh.connected ! prev_connected:self.reset_system_monitoring() # 重置系统监控状态prev_connected self.mssh.connectedself.label_server.config(textf{result_connect[0]}, foregroundresult_connect[1])time.sleep(1)def reset_system_monitoring(self):# 重置系统监控状态self.cpu_label.config(textCPU信息)self.cpu_progress.config(style, value0)self.mem_label.config(textMemory Usage: 0 MB / 0 MB)self.mem_progress.config(style, value0)self.net_label.config(text网络信息)def show_disk_usage(self):if not self.mssh.connected:messagebox.showerror(错误, 请先连接到服务器!)returntry:self.mssh.connecting()stdin, stdout, stderr self.mssh.ssh.exec_command(df -h)disk_info stdout.read().decode(utf-8).strip().split(\n)# 创建新窗口disk_window tk.Toplevel(self.root)disk_window.title(磁盘使用情况)# 创建表格table_frame ttk.Frame(disk_window)table_frame.pack(padx10, pady10, fillboth, expandTrue)# 创建表格标题table ttk.Treeview(table_frame, columns(filesystem, size, used, avail, use%, mounted_on),showheadings)table.heading(filesystem, text文件系统)table.heading(size, text总大小)table.heading(used, text已用空间)table.heading(avail, text可用空间)table.heading(use%, text已用百分比)table.heading(mounted_on, text挂载点)table.pack(sideleft, fillboth, expandTrue)# 添加滚动条scrollbar ttk.Scrollbar(table_frame, orientvertical, commandtable.yview)scrollbar.pack(sideright, filly)table.configure(yscrollcommandscrollbar.set)# 填充表格数据for line in disk_info[1:]:parts line.split()if len(parts) 6:filesystem, size, used, avail, use_percent, mounted_on partstable.insert(, end, values(filesystem, size, used, avail, use_percent, mounted_on))# 调整列宽度table.column(#0, width0, stretchno) # 隐藏第一列table.column(filesystem, width100, anchorw)table.column(size, width80, anchore)table.column(used, width80, anchore)table.column(avail, width80, anchore)table.column(use%, width80, anchore)table.column(mounted_on, width200, anchorw)# 自动调整行高def fixed_map(option):return [elm for elm in style.map(Treeview, query_optoption) if elm[:2] ! (!disabled, !selected)]style ttk.Style()style.map(Treeview, foregroundfixed_map(foreground), backgroundfixed_map(background))except Exception as e:messagebox.showerror(错误, f无法获取磁盘使用情况: {e})
2.10 批量下载日志文件和查看日志功能设计与实现
1批量下载日志文件的实现
首先检查是否已连接到远程服务器然后设置远程目录为 /var/log。接下来它通过SSH连接到服务器并打开SFTP客户端。然后它检查远程目录是否存在并获取远程目录中的文件列表。接着它要求用户选择下载到的本地目录并将选择的本地目录设置为类属性 local_log_dir。然后它遍历远程目录中的文件筛选以 .log 结尾的文件并将这些文件下载到本地目录。最后它关闭SFTP连接并显示成功下载的文件数量。 代码展示如下 def download_logs(self):if not self.mssh.connected:messagebox.showerror(错误, 请先连接到服务器!)return# 将远程目录设置为 /var/logremote_dir /var/logtry:self.mssh.connecting()sftp self.mssh.ssh.open_sftp()# 检查远程目录是否存在try:sftp.stat(remote_dir)except IOError as e:messagebox.showerror(错误, f远程目录 {remote_dir} 不存在)returnremote_files sftp.listdir(remote_dir)local_dir filedialog.askdirectory(title选择下载到的本地目录)if local_dir:self.local_log_dir local_direlse:returndownloaded_count 0for filename in remote_files:# 检查文件是否以 .log 结尾if filename.endswith(.log):remote_path f{remote_dir}/{filename}local_path f{local_dir}/{filename}sftp.get(remote_path, local_path)downloaded_count 1print(f下载成功: {remote_path} - {local_path})sftp.close()print(f共下载了 {downloaded_count} 个文件)if downloaded_count 0:messagebox.showinfo(成功, f成功下载了 {downloaded_count} 个日志文件)except Exception as e:print(f下载失败: {e})messagebox.showerror(错误, f下载日志文件失败: {e})2查看日志功能的实现 先检查是否已连接到远程服务器并下载了日志文件。接着它尝试从 local_log_dir 属性获取之前选择的本地目录如果该属性不存在则要求用户选择下载日志文件所在的目录。然后它要求用户选择要查看的日志文件并使用VSCode打开选定的日志文件。代码展示如下def view_log(self):if not self.mssh.connected:messagebox.showerror(错误, 请先连接服务器并下载日志文件!)return# 从 download_logs 函数中获取之前选择的本地目录try:local_dir self.local_log_direxcept AttributeError:local_dir filedialog.askdirectory(title选择下载日志文件所在的目录)if not local_dir:returnlog_file filedialog.askopenfilename(initialdirlocal_dir, title选择要查看的日志文件,filetypes((Log Files, *.log),))if log_file:try:# 使用 VSCode 打开日志文件subprocess.run([Code, log_file], shellTrue)except Exception as e:messagebox.showerror(错误, f无法打开文件: {e}) 2.11界面窗口的实现
创建一个具有菜单、标签、输入框和文本框的用户界面(UI)以及一个后台线程来更新系统信息。同时通过不同的菜单项和按钮用户可以执行连接、上传文件、下载文件、查看进程、扩展功能等操作。 代码展示如下
def __init__(self, ssh):print(初始化)self.is_connected Falseself.mssh ssh# 主应用程序窗口self.root tk.Tk()frame ttk.Frame(self.root)self.root.geometry(800x550)# 设置窗口标题可选self.root.title(ssh 暗影坤手)# 使用 grid 布局self.root.grid_rowconfigure(0, weight1)self.root.grid_columnconfigure(0, weight1)# 布局网格布局# 创建一个框架容器frame ttk.Frame(self.root)frame.grid(stickynsew)# 添加菜单项 连接新建连接 打开连接 操作-- 帮助--menu_top tk.Menu(self.root)# 连接菜单menu_link tk.Menu(menu_top)menu_top.add_cascade(label连接, menumenu_link)menu_link.add_command(label新建连接, commandself.new_connect_ui)menu_link.add_command(label打开连接, commandself.open_connect_ui)# 操作菜单menu_func tk.Menu(menu_top)menu_top.add_cascade(label操作, menumenu_func)menu_func.add_command(label上传文件, commandself.upload_file)menu_func.add_command(label下载文件, commandself.downloader)menu_func.add_command(labelvim编辑文件, commandself.vim_edit)# 帮助菜单menu_help tk.Menu(menu_top)menu_top.add_cascade(label帮助, menumenu_help)menu_help.add_command(label查看进程, commandself.get_process)# menu_help.add_command(label查看磁盘使用情况, commandself.view_disk_usage)# 扩展菜单menu_ext tk.Menu(menu_top)menu_top.add_cascade(label扩展, menumenu_ext)menu_ext.add_command(label批量下载日志文件, commandself.download_logs)menu_ext.add_command(label查看日志文件, commandself.view_log)# 展示信息的labelself.label_server ttk.Label(frame, text服务器信息)self.label_server.grid(row0, column0, stickyw, padx10, pady5)# 输入指令框self.command tk.StringVar()entry_command ttk.Entry(frame, width60, textvariableself.command)entry_command.grid(row1, column0, stickyew, padx10, pady5)# 执行指令的按钮entry_command.bind(Return, self.ok) # 当按下Enter键时执行命令# 展示结果self.txt_result tk.Text(frame)self.txt_result.grid(row2, column0, stickynsew, padx10, pady5)self.root.config(menumenu_top)# 文本框背景/前景颜色self.txt_result.config(bg#262626, fg#D9D9D9)self.root.config(menumenu_top)# 导入 ttk 后style ttk.Style()# 设置 ttk 主题style.theme_use(clam)# 定制按钮样式style.configure(TButton, foregroundwhite, background#404040)style.map(TButton,foreground[(active, #CCCCCC), (disabled, #666666)],background[(active, #404040), (disabled, #262626)])# 让文本框自动扩展以填充剩余空间frame.grid_rowconfigure(2, weight1)frame.grid_columnconfigure(0, weight1)frame.grid_rowconfigure(3, weight0) # 信息监控面板不需要扩展# thread_infothreading.Thread(targetself.update_info)thread_info threading.Thread(targetself.update_info, args(frame,))thread_info.start()# 将主应用程序挂起self.root.protocol(WM_DELETE_WINDOW, self.on_closing_main)self.root.mainloop()
2.12 打包成可执行文件的实现按需求来 PyInstaller将Python程序打包成独立的可执行文件包括所有必要的依赖项。
在终端执行以下命令
pip install pyinstaller
Pyinstaller -F -w -i 图标.ico -p D:\python\Lib\site-packages --onefile --add-data server.json;./ demotest.py 3. 系统完成效果
3.1 服务器信息结果展示 3.2 连接服务器结果展示 3.3 添加服务器信息结果展示 3.4 删除服务器信息结果展示 3.5 修改服务器信息结果展示 3.6 执行交互式交互命令结果展示 3.7 vim编辑器结果展示 3.8 文件上传下载结果展示
1上传文件 2下载文件 3.9 服务器监控功能结果展示
1cpu、内存、网络信息展示 2磁盘使用情况展示 3进程信息 3.10 批量下载日志文件和查看日志文件结果展示
1批量下载文件展示 2查看日志信息 3.11窗口界面结果展示 4. 最终代码
import json
import os.path
import re
import socket
import subprocess
import threading
import time
from tkinter import simpledialogimport paramikofrom watchdog.events import FileSystemEventHandler
from watchdog.observers import Observerclass MySsh:def __init__(self):# 实例化 ssh服务器的会话高级表现形式把通道/传输类/sftp进行了封装self.ssh paramiko.SSHClient()self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())self.path ~# 所有服务器的列表self.servers []# 当前正在使用的服务器信息self.server {}self.load_server()self.connected []def load_server(self):with open(server.json, r, encodingutf-8) as fp:self.servers json.load(fp)print(f获取到的服务器信息{self.servers})def save_server(self):with open(server.json, w, encodingutf-8) as fp:fp.write(json.dumps(self.servers))def add_server(self, ip, port, username, password, description):self.servers.append({description: description,ip: ip,port: port,username: username,password: password})self.save_server()# def find_server(self,ip):def connecting(self):self.connected None# 将信任的主机自动添加到know_hosts文件~/.ssh/中self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy)try:# 连接到服务器self.ssh.connect(hostnameself.server[ip], portself.server[port], usernameself.server[username],passwordself.server[password])# messagebox.showinfo(成功, f连接到服务器【{self.server[ip]}】成功)print(f连接到服务器【{self.server[ip]}】成功)self.connected Trueself.get_pwd()return Trueexcept Exception as e:messagebox.showerror(错误, f连接到服务器【{self.server[ip]}】失败)self.connected Falsereturn Falsedef exec(self, command):if command.startswith(cd):tmp_path command.split( )[-1]if tmp_path.startswith(/):# 绝对路径处理self.path tmp_pathelse:# todo 如果用户输入的是一个相对路径呢print(用户输入的是相对路径)if tmp_path.startswith(..):self.path self.path / tmp_pathelse:self.path self.path tmp_pathelse:# 根据当前的路径去执行指令command fcd {self.path} {command}print(f-输入指令({self.path}){command})stdin, stdout, stderr self.ssh.exec_command(command)result stdout.read().decode(utf-8)print(f-返回{result})# 用户输入的路径里包含.. 通过pwd刷新当前路径if command.find(..) 0:self.get_pwd()return resultdef _read_channel(self, channel, output_callback):在独立线程中读取通道输出并调用回调函数处理while True:if channel.recv_ready():data channel.recv(1024)output_callback(data)def download_file(self):remote_file tk.simpledialog.askstring(输入文件名, 请输入要下载的文件名:)if remote_file is None: # 用户点击“取消”或关闭对话框returnelif not remote_file: # 用户输入了空字符串messagebox.showerror(错误, 请输入文件名)returnlocal_path filedialog.askdirectory(title选择下载到的本地目录)if local_path:try:sftp self.ssh.open_sftp()local_file_path os.path.join(local_path, remote_file)sftp.get(remote_file, local_file_path)sftp.close()messagebox.showinfo(下载成功, f文件下载成功: {remote_file} - {local_path})except Exception as e:messagebox.showerror(下载失败, f文件下载失败: {e})def watch_and_upload(self, local_file_path, remote_full_path):class FileModifiedHandler(FileSystemEventHandler):def __init__(self, ssh_handler, remote_full_path):self.ssh_handler ssh_handlerself.remote_full_path remote_full_pathdef on_modified(self, event):if event.is_directory or not event.src_path local_file_path:return# messagebox.showinfo(文件被修改, f文件被修改{event.src_path})try:# self.ssh_handler.connecting() # 重新连接SSH如果需要sftp self.ssh_handler.ssh.open_sftp()sftp.put(local_file_path, self.remote_full_path)sftp.close()messagebox.showinfo(文件上传成功, fvim修改成功: {local_file_path} - {self.remote_full_path})except Exception as e:messagebox.showerror(文件上传失败, f文件上传失败: {e})event_handler FileModifiedHandler(self, remote_full_path)observer Observer()observer.schedule(event_handler, pathos.path.dirname(local_file_path), recursiveFalse)observer.start()try:# 等待用户操作或其他信号来停止监视# 这里使用time.sleep作为示例但建议使用更好的机制如事件或信号while True:time.sleep(1)except KeyboardInterrupt:observer.stop()observer.join()def vim_edit(self):remote_file simpledialog.askstring(vim, 请输入要编辑的完整远程文件路径:)if remote_file is None: # 用户点击“取消”或关闭对话框returnelif not remote_file: # 用户输入了空字符串messagebox.showerror(错误, 请输入文件路径)return# 检查用户是否输入了以/开头的绝对路径if not remote_file.startswith(/):messagebox.showerror(错误, 请输入绝对路径例如/home/user/file.txt)return# 临时文件目录tmp filedialog.askdirectory(title选择下载到的本地目录)if not tmp: # 用户取消了选择目录returnlocal_file_path os.path.join(tmp, os.path.basename(remote_file))try:self.connecting()sftp self.ssh.open_sftp()# 检查远程文件是否存在try:sftp.stat(remote_file)except FileNotFoundError:# 如果不存在则在远程服务器上创建空文件with sftp.open(remote_file, wb) as f:pass # 写入空内容或者模板内容# 下载文件到本地sftp.get(remote_file, local_file_path)sftp.close()# 启动文件监视线程watch_thread threading.Thread(targetself.watch_and_upload, args(local_file_path, remote_file))watch_thread.start()# 打开文件编辑器subprocess.run([Code, local_file_path], shellTrue)except Exception as e:messagebox.showerror(文件下载失败, f文件下载失败{e})print(f文件下载失败{e})# 使用pwd更新当前路径def get_pwd(self):stdin, stdout, stderr self.ssh.exec_command(fcd {self.path} pwd)result stdout.read().decode(utf-8)self.path result.strip()import tkinter as tk
from tkinter import ttk, filedialog
from tkinter import messageboxclass MyUI:def __init__(self, ssh):print(初始化)self.is_connected Falseself.mssh sshself.mssh_channel None# 主应用程序窗口self.root tk.Tk()frame ttk.Frame(self.root)self.root.geometry(800x550)# 设置窗口标题可选self.root.title(暗影坤手)# 使用 grid 布局self.root.grid_rowconfigure(0, weight1)self.root.grid_columnconfigure(0, weight1)# 布局网格布局# 创建一个框架容器frame ttk.Frame(self.root)frame.grid(stickynsew)# 添加菜单项 连接新建连接 打开连接 操作-- 帮助--menu_top tk.Menu(self.root)# 连接菜单menu_link tk.Menu(menu_top)menu_top.add_cascade(label连接, menumenu_link)menu_link.add_command(label新建连接, commandself.new_connect_ui)menu_link.add_command(label打开连接, commandself.open_connect_ui)# 操作菜单menu_func tk.Menu(menu_top)menu_top.add_cascade(label操作, menumenu_func)menu_func.add_command(label上传文件, commandself.upload_file)menu_func.add_command(label下载文件, commandself.downloader)menu_func.add_command(labelvim编辑文件, commandself.vim_edit)# 帮助菜单menu_help tk.Menu(menu_top)menu_top.add_cascade(label帮助, menumenu_help)menu_help.add_command(label查看进程, commandself.get_process)# menu_help.add_command(label查看磁盘使用情况, commandself.view_disk_usage)# 扩展菜单menu_ext tk.Menu(menu_top)menu_top.add_cascade(label扩展, menumenu_ext)menu_ext.add_command(label批量下载日志文件, commandself.download_logs)menu_ext.add_command(label查看日志文件, commandself.view_log)# 展示信息的labelself.label_server ttk.Label(frame, text服务器信息)self.label_server.grid(row0, column0, stickyw, padx10, pady5)# 输入指令框self.command tk.StringVar()entry_command ttk.Entry(frame, width60, textvariableself.command)entry_command.grid(row1, column0, stickyew, padx10, pady5)# 执行指令的按钮entry_command.bind(Return, self.ok) # 当按下Enter键时执行命令entry_command.bind(Control-c, self.okk)# 展示结果self.txt_result tk.Text(frame)self.txt_result.grid(row2, column0, stickynsew, padx10, pady5)self.root.config(menumenu_top)# 文本框背景/前景颜色self.txt_result.config(bg#262626, fg#D9D9D9)self.root.config(menumenu_top)# 导入 ttk 后style ttk.Style()# 设置 ttk 主题style.theme_use(clam)# 定制按钮样式style.configure(TButton, foregroundwhite, background#404040)style.map(TButton,foreground[(active, #CCCCCC), (disabled, #666666)],background[(active, #404040), (disabled, #262626)])# 让文本框自动扩展以填充剩余空间frame.grid_rowconfigure(2, weight1)frame.grid_columnconfigure(0, weight1)frame.grid_rowconfigure(3, weight0) # 信息监控面板不需要扩展# thread_infothreading.Thread(targetself.update_info)thread_info threading.Thread(targetself.update_info, args(frame,))thread_info.start()# 将主应用程序挂起self.root.protocol(WM_DELETE_WINDOW, self.on_closing_main)self.root.mainloop()def okk(self, event):self.command.set(q)self.ok(self)def handle_output(self, data):# 先解码数据decoded_data data.decode(utf-8)# 去除颜色转义序列cleaned_data re.sub(r\x1B\[([0-?]*[ -/]*[-~]), , decoded_data)# 插入到文本框的末尾self.txt_result.insert(tk.END, cleaned_data)# 确保滚动条跟随到最后self.txt_result.see(tk.END)# ------------------------------------------帮助---------------------------------------------------def get_process(self):if not self.mssh.connected:messagebox.showerror(错误, 请先连接到服务器!)returnself.pop_win tk.Toplevel(self.root)self.pop_win.title(进程)self.pop_win.geometry(800x600600250)table_frame ttk.Frame(self.pop_win)table_frame.pack(padx10, pady10, fillboth, expandTrue)server self.mssh.serverip server[ip]username server[username]password server[password]try:client paramiko.SSHClient()client.set_missing_host_key_policy(paramiko.AutoAddPolicy())client.connect(ip, 22, username, password)stdin, stdout, stderr client.exec_command(ps -ef)result stdout.read().decode(utf-8).strip().split(\n)# 创建表格table ttk.Treeview(table_frame, columns(user, pid, cpu, mem, vsz, rss, tty, stat, start, time, command), showheadings)table.heading(user, text用户)table.heading(pid, text进程ID)table.heading(cpu, textCPU%)table.heading(mem, text内存%)table.heading(vsz, text虚拟内存)table.heading(rss, text常驻内存)table.heading(tty, text终端)table.heading(stat, text状态)table.heading(start, text启动时间)table.heading(time, textCPU时间)table.heading(command, text命令)table.pack(sideleft, fillboth, expandTrue)# 添加滚动条scrollbar ttk.Scrollbar(table_frame, orientvertical, commandtable.yview)scrollbar.pack(sideright, filly)table.configure(yscrollcommandscrollbar.set)# 填充表格数据for line in result[1:]:parts line.split()if len(parts) 10:user, pid, cpu, mem, vsz, rss, tty, stat, start, time, *command partscommand .join(command)table.insert(, end, values(user, pid, cpu, mem, vsz, rss, tty, stat, start, time, command))# 调整列宽度table.column(#0, width0, stretchno) # 隐藏第一列table.column(user, width80, anchorw)table.column(pid, width60, anchore)table.column(cpu, width60, anchore)table.column(mem, width60, anchore)table.column(vsz, width100, anchore)table.column(rss, width100, anchore)table.column(tty, width60, anchorw)table.column(stat, width60, anchorw)table.column(start, width120, anchorw)table.column(time, width80, anchore)table.column(command, width300, anchorw)# 自动调整行高def fixed_map(option):return [elm for elm in style.map(Treeview, query_optoption) if elm[:2] ! (!disabled, !selected)]style ttk.Style()style.map(Treeview, foregroundfixed_map(foreground), backgroundfixed_map(background))except paramiko.AuthenticationException:messagebox.showerror(错误, 认证失败,请验证您的凭据)except paramiko.SSHException as sshException:messagebox.showerror(错误, f无法建立SSH连接: {sshException})except paramiko.BadHostKeyException as badHostKeyException:messagebox.showerror(错误, f无法验证服务器的主机密钥: {badHostKeyException})except Exception as e:messagebox.showerror(错误, f发生错误: {e})finally:if client in locals() and client:client.close() # 确保连接被关闭# 当窗口关闭时,确保没有遗留的引用或连接self.pop_win.protocol(WM_DELETE_WINDOW, self.on_closing)def create_widgets(self, frame):self.cpu_label ttk.Label(frame, textCPU信息)self.cpu_label.grid()self.cpu_progress ttk.Progressbar(frame, orienthorizontal, length200, modedeterminate)self.cpu_progress.grid(pady(0, 10))self.mem_label ttk.Label(frame, textMemory Usage: 0 MB / 0 MB)self.mem_label.grid()self.mem_progress ttk.Progressbar(frame, orienthorizontal, length200, modedeterminate)self.mem_progress.grid(pady(0, 5))self.net_label ttk.Label(frame, text网络信息)self.net_label.grid(pady5)self.disk_button tk.Button(frame, text查看磁盘使用情况, commandself.show_disk_usage)self.disk_button.grid(pady(5, 5))def update_info(self, frame):self.create_widgets(frame) # 创建标签# 定义进度条样式style ttk.Style()style.theme_use(default)style.configure(green.Horizontal.TProgressbar, foregroundgreen, backgroundgreen)style.configure(yellow.Horizontal.TProgressbar, foregroundyellow, backgroundyellow)style.configure(red.Horizontal.TProgressbar, foregroundred, backgroundred)prev_connected self.mssh.connected # 记录上一次的连接状态while True:if ip in self.mssh.server:if self.mssh.connected None:result_connect (f服务器{self.mssh.server[ip]}连接中, red)self.reset_system_monitoring() # 重置系统监控状态elif self.mssh.connected:result_connect (f服务器{self.mssh.server[ip]}连接成功, green)# -----------------------------系统信息------------------------------------------------server self.mssh.serverip server[ip]username server[username]password server[password]self.client paramiko.SSHClient()self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())self.client.connect(ip, 22, username, password)# 执行远程命令获取内存信息#是Paramiko库中SSHClient对象的一个方法可用于在远程服务器上执行命令。# 它接受一个字符串参数作为要执行的命令并返回输入、输出和错误流作为结果。stdin, stdout, stderr self.client.exec_command(free -m)result stdout.read().decode(utf-8)# 获取系统内存信息lines result.split(\n)mem_info lines[1].split()total_mem int(mem_info[1])used_mem int(mem_info[2])mem_usage used_mem / total_mem * 100# 获取网络速率stdin1, stdout1, stderr1 self.client.exec_command(ifconfig)result stdout1.read().decode(utf-8)# 使用正则表达式提取数据包和字节数信息matches re.findall(rens160.*?RX packets (\d) bytes (\d).*?TX packets (\d) bytes (\d),result, re.DOTALL)if matches:rx_packets, rx_bytes, tx_packets, tx_bytes matches[0]else:rx_packets, rx_bytes, tx_packets, tx_bytes 0, 0, 0, 0time.sleep(1)stdin2, stdout2, stderr3 self.client.exec_command(ifconfig)result stdout2.read().decode(utf-8)matches re.findall(rens160.*?RX packets (\d) bytes (\d).*?TX packets (\d) bytes (\d),result, re.DOTALL)if matches:rx_packets, new_rx_bytes, tx_packets, new_tx_bytes matches[0]else:rx_packets, new_rx_bytes, tx_packets, new_tx_bytes 0, 0, 0, 0# 计算速率upload_speed ((int(new_tx_bytes) - int(tx_bytes)) / 1024) # MB/sdownload_speed ((int(new_rx_bytes) - int(rx_bytes)) / 1024)# 获取cpu利用率stdin4, stdout4, stderr4 self.client.exec_command(top -bn 1)result stdout4.read().decode(utf-8)cpu 0for line in result.split(\n):if line.startswith(%Cpu(s)):cpu_us re.findall(r\d\.\d, line.split(:)[1].split(,)[0])[0]cpu_sy re.findall(r\d\.\d, line.split(:)[1].split(,)[1])[0]cpu_ni re.findall(r\d\.\d, line.split(:)[1].split(,)[2])[0]cpu_id re.findall(r\d\.\d, line.split(:)[1].split(,)[3])[0]cpu_wa re.findall(r\d\.\d, line.split(:)[1].split(,)[4])[0]cpu_hi re.findall(r\d\.\d, line.split(:)[1].split(,)[5])[0]cpu_si re.findall(r\d\.\d, line.split(:)[1].split(,)[6])[0]cpu_st re.findall(r\d\.\d, line.split(:)[1].split(,)[7])[0]sum float(cpu_st) float(cpu_si) float(cpu_hi) float(cpu_wa) float(cpu_id) float(cpu_ni) float(cpu_sy) float(cpu_us)cpu sum - float(cpu_id)# 设置进度条样式if cpu 80:cpu_style red.Horizontal.TProgressbarelif cpu 50:cpu_style yellow.Horizontal.TProgressbarelse:cpu_style green.Horizontal.TProgressbarif mem_usage 80:mem_style red.Horizontal.TProgressbarelif mem_usage 50:mem_style yellow.Horizontal.TProgressbarelse:mem_style green.Horizontal.TProgressbar# 更新标签self.cpu_label.config(textfCPU信息{cpu:.2f}%)self.cpu_progress.config(stylecpu_style, valuecpu)self.mem_label.config(textfMemory Usage: {used_mem} MB / {total_mem} MB ({mem_usage:.2f}%))self.mem_progress.config(stylemem_style, valuemem_usage)self.net_label.config(textf网络信息↑{upload_speed:.2f}MB/s \t ↓{download_speed:.2f}MB/s)else:result_connect (f服务器{self.mssh.server[ip]}连接失败, red)self.reset_system_monitoring() # 重置系统监控状态else:result_connect (f服务器未选择, black)self.reset_system_monitoring() # 重置系统监控状态# 检查连接状态是否发生变化if self.mssh.connected ! prev_connected:self.reset_system_monitoring() # 重置系统监控状态prev_connected self.mssh.connectedself.label_server.config(textf{result_connect[0]}, foregroundresult_connect[1])time.sleep(1)def reset_system_monitoring(self):# 重置系统监控状态self.cpu_label.config(textCPU信息)self.cpu_progress.config(style, value0)self.mem_label.config(textMemory Usage: 0 MB / 0 MB)self.mem_progress.config(style, value0)self.net_label.config(text网络信息)def show_disk_usage(self):if not self.mssh.connected:messagebox.showerror(错误, 请先连接到服务器!)returntry:self.mssh.connecting()stdin, stdout, stderr self.mssh.ssh.exec_command(df -h)disk_info stdout.read().decode(utf-8).strip().split(\n)# 创建新窗口disk_window tk.Toplevel(self.root)disk_window.title(磁盘使用情况)# 创建表格table_frame ttk.Frame(disk_window)table_frame.pack(padx10, pady10, fillboth, expandTrue)# 创建表格标题table ttk.Treeview(table_frame, columns(filesystem, size, used, avail, use%, mounted_on),showheadings)table.heading(filesystem, text文件系统)table.heading(size, text总大小)table.heading(used, text已用空间)table.heading(avail, text可用空间)table.heading(use%, text已用百分比)table.heading(mounted_on, text挂载点)table.pack(sideleft, fillboth, expandTrue)# 添加滚动条scrollbar ttk.Scrollbar(table_frame, orientvertical, commandtable.yview)scrollbar.pack(sideright, filly)table.configure(yscrollcommandscrollbar.set)# 填充表格数据for line in disk_info[1:]:parts line.split()if len(parts) 6:filesystem, size, used, avail, use_percent, mounted_on partstable.insert(, end, values(filesystem, size, used, avail, use_percent, mounted_on))# 调整列宽度table.column(#0, width0, stretchno) # 隐藏第一列table.column(filesystem, width100, anchorw)table.column(size, width80, anchore)table.column(used, width80, anchore)table.column(avail, width80, anchore)table.column(use%, width80, anchore)table.column(mounted_on, width200, anchorw)# 自动调整行高def fixed_map(option):return [elm for elm in style.map(Treeview, query_optoption) if elm[:2] ! (!disabled, !selected)]style ttk.Style()style.map(Treeview, foregroundfixed_map(foreground), backgroundfixed_map(background))except Exception as e:messagebox.showerror(错误, f无法获取磁盘使用情况: {e})def ok(self, event):command self.command.get()if not self.mssh.connected:self.txt_result.insert(tk.END, 未连接到服务器\n)return# 如果ssh_channel尚未创建创建一个新的if self.mssh_channel is None:self.mssh_channel self.mssh.ssh.invoke_shell()self.mssh_channel.set_combine_stderr(True)try:# 使用ssh_channel发送命令self.mssh_channel.send(f{command}\n)# 创建线程监听输出并调用UI提供的回调函数t threading.Thread(targetself.mssh._read_channel, args(self.mssh_channel, self.handle_output))t.daemon Truet.start()# 设置延时清空输入框self.root.after(2000, lambda: self.command.set())except Exception as e:self.txt_result.insert(tk.END, f执行错误: {str(e)}\n)# 清空输入框self.command.set()# --------------------------------------------连接-----------------------------------------------# 新建连接def new_connect_ui(self):self.pop_win tk.Toplevel(self.root)# 设置标题和大小self.pop_win.title(新建连接)self.pop_win.geometry(300x200600250)frame tk.Frame(self.pop_win)frame.grid()# IPip_label tk.Label(frame, text服务器IP地址:)ip_label.grid(row0, column0)self.ip_entry tk.Entry(frame)self.ip_entry.grid(row0, column1)# Portport_label tk.Label(frame, text服务器端口号:)port_label.grid(row1, column0)self.port_entry tk.Entry(frame)self.port_entry.grid(row1, column1)# Usernameusername_label tk.Label(frame, text用户名:)username_label.grid(row2, column0)self.username_entry tk.Entry(frame)self.username_entry.grid(row2, column1)# Passwordpassword_label tk.Label(frame, text密码:)password_label.grid(row3, column0)self.password_entry tk.Entry(frame, show*)self.password_entry.grid(row3, column1)# Descriptiondescription_label tk.Label(frame, text备注信息:)description_label.grid(row4, column0)self.description_entry tk.Entry(frame)self.description_entry.grid(row4, column1)# Buttonsubmit_button tk.Button(frame, text保存, commandself.save_server_info)submit_button.grid(row5, columnspan2)# 保存新建连接的服务器信息def save_server_info(self):if self.ip_entry.get() or self.port_entry.get() or self.username_entry.get() or self.password_entry.get() :messagebox.showerror(错误, 请填写完整的服务器信息)returnelse:ip self.ip_entry.get()port self.port_entry.get()username self.username_entry.get()password self.password_entry.get()description self.description_entry.get()# 这里应该调用你的 add_server 函数将获取到的信息添加到服务器列表中ssh.add_server(ipip, portint(port), usernameusername, passwordpassword, descriptiondescription)messagebox.showinfo(成功, 服务器信息已保存)# 打开连接def open_connect_ui(self):print(打开连接)self.pop_win tk.Toplevel(self.root)# 设置标题和大小self.pop_win.title(打开连接)self.pop_win.geometry(480230)frame tk.Frame(self.pop_win)frame.grid()# 初始化索引变量 index 为 0用于控制部件的位置index 0# 遍历服务器列表 self.mssh.servers 中的每个服务器信息for item in self.mssh.servers:# 删除列表信息中的password这里假设item是一个字典#用于遍历 item.items() 中的键值对并使用它们创建一个新的字典# 创建一个新的字典其中排除了原始字典item中键为password的键值对。item_without_password {k: v for k, v in item.items() if k ! password}# 显示不包含密码的服务器信息# 创建一个标签显示当前服务器的信息并将其放置在连接界面中的指定位置# padx5 和 pady5 是用来设置周围的水平和垂直填充# 在部件的左右/上下两侧各增加 5 个像素的水平填充。tk.Label(self.pop_win, textstr(item_without_password)).grid(rowindex, column0, padx5, pady5)# 添加连接、删除、修改按钮...# lambda 表达式用于创建一个匿名函数# commandlambda itemitem# 将当前循环中的 item 值作为参数传递给 lambda 表达式这样在按钮被点击时# lambda 表达式将会调用相应的方法并将当前循环中的 item 值作为参数传递给这个方法# 使用 lambda itemitem 的形式可以确保每个按钮的回调函数都引用了当前循环中的正确值tk.Button(self.pop_win, text连接, commandlambda itemitem: self.connect(item)).grid(rowindex, column2)tk.Button(self.pop_win, text删除, commandlambda itemitem: self.del_connect(item)).grid(rowindex,column4)tk.Button(self.pop_win, text修改, commandlambda itemitem: self.modify_server(item)).grid(rowindex,column6)# 更新索引变量 index以便将下一个服务器的部件放置在下一行index index 1# 连接到指定的服务器def connect(self, server):print(连接到指定的服务器)# 将 self.mssh 对象的 server 属性设置为传入的 server 参数以便记录当前连接的服务器信息。self.mssh.server serverprint(f服务器信息{self.mssh.server})#如果在连接到远程服务器时遇到超时系统将等待 5 秒然后会引发socket.timeout异常。socket.setdefaulttimeout(5)# 在单独的线程中执行连接操# 在单独的线程中执行连接操作# 创建一个新的线程 thread_connect目标是调用 self.connect_thread 方法并传入 server 参数。# 然后启动该线程以便在单独的线程中执行连接操作以避免阻塞主线程。thread_connect threading.Thread(targetself.connect_thread, args(server,))thread_connect.start()# 销毁打开连接窗口以便在连接操作进行时不再显示连接界面self.pop_win.destroy()def connect_thread(self, server):# 调用 self.mssh 对象的 connecting 方法该方法用于执行连接操作并将连接结果存储在 result 变量中result self.mssh.connecting()if result:# 将对象的 is_connected 属性设置为 True表示连接状态为已连接self.is_connected True# 调用 update_connection_status 方法将连接状态更新为已连接并传入当前连接的服务器和连接状态self.update_connection_status(server, True)else:self.is_connected Falseself.update_connection_status(server, False)def update_connection_status(self, server, connected):if connected:# 如果连接成功通过 self.label_server 的 config 方法更新标签的文本内容为连接成功的消息# 使用了格式化字符串来显示服务器的 IP 地址并设置文本颜色为绿色self.label_server.config(textf服务器{server[ip]}连接成功, foregroundgreen)else:self.label_server.config(textf服务器{server[ip]}连接失败, foregroundred)def del_connect(self, server):print(删除指定的服务器)# 输出服务器信息# 输出要删除的服务器信息包括服务器的详细信息。print(f服务器信息{server})# 确认是否删除服务器confirm messagebox.askyesno(确认删除, f确认删除服务器 {server[ip]} 吗)if confirm:# 从服务器列表中删除指定的服务器# # 检查要删除的服务器是否存在于服务器列表中。if server in self.mssh.servers:self.mssh.servers.remove(server)self.pop_win.destroy()self.open_connect_ui()messagebox.showinfo(删除成功, f服务器 {server[ip]} 已成功删除)self.mssh.save_server()else:messagebox.showerror(错误, 服务器不存在无法删除)# 保存更新后的服务器列表到配置文件中def modify_server(self, server):self.pop_win.destroy()print(修改服务器信息)# 将传入的服务器信息进行复制并存储在 self.mssh.server 中。self.mssh.server server.copy() # 添加这一行来复制服务器信息self.pop_win_modify tk.Toplevel(self.root)frame tk.Frame(self.pop_win_modify)frame.grid()# IPip_label tk.Label(frame, text服务器IP地址:)ip_label.grid(row0, column0)# 创建了一个文本框Entry并将其父级设置为 frame 框架。此文本框用于接收用户输入的 IP 地址。# textvariable 参数是用来关联文本框的值的变量# 这里使用了 tk.StringVar 类来创建一个与文本框关联的字符串变量并将该变量的初始值设置为 server[ip]即当前服务器的IP地址self.ip_entry tk.Entry(frame, textvariabletk.StringVar(valueserver[ip]))self.ip_entry.grid(row0, column1)# Portport_label tk.Label(frame, text服务器端口号:)port_label.grid(row1, column0)self.port_entry tk.Entry(frame, textvariabletk.StringVar(valuestr(server[port])))self.port_entry.grid(row1, column1)# Usernameusername_label tk.Label(frame, text用户名:)username_label.grid(row2, column0)# 创建了一个文本框Entry并将其父级设置为 frame 框架。此文本框用于接收用户输入的 username。# textvariable 参数是用来关联文本框的值的变量# 这里使用了 tk.StringVar 类来创建一个与文本框关联的字符串变量并将该变量的初始值设置为 server[username]即当前服务器的用户名self.username_entry tk.Entry(frame, textvariabletk.StringVar(valueserver[username]))self.username_entry.grid(row2, column1)# Passwordpassword_label tk.Label(frame, text密码:)password_label.grid(row3, column0)self.password_entry tk.Entry(frame, show*, textvariabletk.StringVar(valueserver[password]))self.password_entry.grid(row3, column1)# Descriptiondescription_label tk.Label(frame, text备注信息:)description_label.grid(row4, column0)# 创建了一个文本框Entry并将其父级设置为 frame 框架。此文本框用于接收用户输入的 备注信息。# textvariable 参数是用来关联文本框的值的变量# 这里使用了 tk.StringVar 类来创建一个与文本框关联的字符串变量并将该变量的初始值设置为 server[description]即当前服务器的备注信息self.description_entry tk.Entry(frame, textvariabletk.StringVar(valueserver[description]))self.description_entry.grid(row4, column1)# Buttonsubmit_button tk.Button(frame, text保存, commandself.save_modified_server)submit_button.grid(row5, columnspan2)# 保存修改后的服务器信息def save_modified_server(self):if self.ip_entry.get() or self.port_entry.get() or self.username_entry.get() or self.password_entry.get() :messagebox.showerror(错误, 请填写完整的服务器信息)returnelse:modified_server {ip: self.ip_entry.get(),port: int(self.port_entry.get()),username: self.username_entry.get(),password: self.password_entry.get(), # 假设密码不变保持原值description: self.description_entry.get(),}# 找到了待修改服务器在服务器列表中的索引# 使用了 index 方法来查找 self.mssh.server 在 self.mssh.servers 中的位置。index self.mssh.servers.index(self.mssh.server)# 将修改后的服务器信息替换掉原来的服务器信息以完成对服务器信息的更新。self.mssh.servers[index] modified_server# 弹出一个消息框显示成功修改服务器信息的提示。messagebox.showinfo(成功, 服务器信息已成功修改)# 关闭了修改服务器信息的窗口self.pop_win_modify.destroy()# 调用了 open_connect_ui 方法打开连接用户界面让用户可以选择连接到修改后的服务器。self.open_connect_ui()# 调用了 save_server 方法用于保存修改后的服务器信息到文件中以便下次程序运行时能够读取到最新的服务器信息。self.mssh.save_server()################################## 操作 ################################################## 上传文件def upload_file(self):if not self.mssh.connected:messagebox.showerror(错误, 请先连接到服务器!)returnelse:self.file_path filedialog.askopenfilename()try:if self.file_path:# 获取文件名file_name self.file_path.split(/)[-1]# 创建 SFTP 客户端sftp_client self.mssh.ssh.open_sftp()remote_path f{self.mssh.path}/{file_name}print(f文件{self.file_path},上传到{remote_path})# 上传文件sftp_client.put(self.file_path, remote_path)messagebox.showinfo(成功, 上传成功)sftp_client.close()else:print(请选择要上传的文件)except Exception as e:messagebox.showerror(错误, f文件上传失败: {e})# #下载文件def downloader(self):if not self.mssh.connected:messagebox.showerror(错误, 请先连接到服务器!)returnself.mssh.download_file()def vim_edit(self):if not self.mssh.connected:messagebox.showerror(错误, 请先连接到服务器!)returnself.mssh.vim_edit()def download_logs(self):if not self.mssh.connected:messagebox.showerror(错误, 请先连接到服务器!)return# 将远程目录设置为 /var/logremote_dir /var/logtry:self.mssh.connecting()sftp self.mssh.ssh.open_sftp()# 检查远程目录是否存在try:sftp.stat(remote_dir)except IOError as e:messagebox.showerror(错误, f远程目录 {remote_dir} 不存在)returnremote_files sftp.listdir(remote_dir)local_dir filedialog.askdirectory(title选择下载到的本地目录)if local_dir:self.local_log_dir local_direlse:returndownloaded_count 0for filename in remote_files:# 检查文件是否以 .log 结尾if filename.endswith(.log):remote_path f{remote_dir}/{filename}local_path f{local_dir}/{filename}sftp.get(remote_path, local_path)downloaded_count 1print(f下载成功: {remote_path} - {local_path})sftp.close()print(f共下载了 {downloaded_count} 个文件)if downloaded_count 0:messagebox.showinfo(成功, f成功下载了 {downloaded_count} 个日志文件)except Exception as e:print(f下载失败: {e})messagebox.showerror(错误, f下载日志文件失败: {e})def view_log(self):if not self.mssh.connected:messagebox.showerror(错误, 请先连接服务器并下载日志文件!)return# 从 download_logs 函数中获取之前选择的本地目录try:local_dir self.local_log_direxcept AttributeError:local_dir filedialog.askdirectory(title选择下载日志文件所在的目录)if not local_dir:returnlog_file filedialog.askopenfilename(initialdirlocal_dir, title选择要查看的日志文件,filetypes((Log Files, *.log),))if log_file:try:# 使用 VSCode 打开日志文件subprocess.run([Code, log_file], shellTrue)except Exception as e:messagebox.showerror(错误, f无法打开文件: {e})def on_closing_main(self):if messagebox.askokcancel(Quit, 确认退出吗?):self.root.destroy()def on_closing(self):if messagebox.askokcancel(Quit, 确认退出吗?):self.pop_win.destroy()if __name__ __main__:ssh MySsh()ui MyUI(ssh)5. 总结与分析 使用 Python 的 Tkinter 模块编写 SSH 工具需要综合考虑界面设计、SSH 连接的实现、用户体验和安全等方面。通过 Tkinter 提供的丰富小部件可以创建直观且易于操作的界面让用户输入主机名、用户名、密码等信息并提供连接按钮来触发 SSH 连接。同时使用 Paramiko 或其他类似的库处理 SSH 连接和命令执行的细节考虑使用异步处理以避免界面冻结同时要注意处理连接错误和认证失败等异常情况可以使用多线程来解决。在设计中需确保安全处理用户凭据和敏感信息避免在界面中明文显示密码。最终在测试和调试过程中不断完善功能确保工具的稳定性和可靠性。通过充分利用 Python 和 Tkinter 的功能可以创建一个功能强大且易于使用的 SSH 工具。 注此项目是大学期间与搭子 霍格沃茨的纸飞鹤-CSDN博客 共同完成的现在分享给大家参考。
