python + ssh+ rich 升级和备份脚本

升级版本 （根据AI提供的脚本,修改后）

import os
import paramiko
from scp import SCPClient
from rich.progress import (BarColumn,DownloadColumn,Progress,TaskID,TextColumn,TimeRemainingColumn,TransferSpeedColumn,
)def get_file_size(file_path):return os.path.getsize(file_path)def progress_callback(filename, size, sent, progress_instance, task_id):progress_instance.update(task_id, completed=sent)# 创建 SSH 客户端并连接到远程服务器
def create_ssh_client(host, port, username, password):client = paramiko.SSHClient()client.set_missing_host_key_policy(paramiko.AutoAddPolicy())client.connect(host, port, username, password)return clientdef scp_transfer_with_progress(local_file, remote_path, host, port, username, password):os.chdir(local_file)ssh_client = create_ssh_client(host, port, username, password)with Progress(TextColumn("[bold blue] {task.fields[filename]}", justify="right"),BarColumn(),"[progress.percentage]{task.percentage:>3.1f}%","•",DownloadColumn(),"•",TransferSpeedColumn(),"•",TimeRemainingColumn(), ) as progress:task_id = progress.add_task(filename="upgrade ...", description="upgrade...")def userProgress(filename, size, sent):progress.update(task_id, total=size, advance=sent,filename="Upgrade: " + os.path.basename(filename).decode())with SCPClient(ssh_client.get_transport(), progress=userProgress, socket_timeout=15) as scp:scp.put("./", remote_path, recursive=True)ssh_client.close()import sys
import timeif __name__ == "__main__":host = sys.argv[1]port = 22username = sys.argv[2]password = sys.argv[3]local_file = "upgradeSoftware"remote_path = "/mnt/flash"scp_transfer_with_progress(local_file, remote_path, host, port, username, password)`在这里插入

备份版本 （根据AI提供的脚本,修改后）

import os
import paramiko
from scp import SCPClient
from rich.progress import Progress
from rich.progress import (BarColumn,DownloadColumn,Progress,TaskID,TextColumn,TimeRemainingColumn,TransferSpeedColumn,
)```python
def get_file_size(file_path):return os.path.getsize(file_path)def progress_callback(filename, size, sent, progress_instance, task_id):progress_instance.update(task_id, completed=sent)# 创建 SSH 客户端并连接到远程服务器
def create_ssh_client(host, port, username, password):client = paramiko.SSHClient()client.set_missing_host_key_policy(paramiko.AutoAddPolicy())client.connect(host, port, username, password)return clientdef scp_transfer_with_progress(local_file, remote_path, host, port, username, password):ssh_client = create_ssh_client(host, port, username, password)ssh_client.exec_command("rm  -rf /mnt/flash/log/message.*txt")with Progress(TextColumn("[bold blue] {task.fields[filename]}", justify="right"),BarColumn(),"[progress.percentage]{task.percentage:>3.1f}%","•",DownloadColumn(),"•",TransferSpeedColumn(),"•",TimeRemainingColumn(), ) as progress:task_id = progress.add_task(filename="Downloading...", description="Downloading...")def userProgress(filename, size, sent):progress.update(task_id, total=size, advance=sent, filename="Current Download:" + os.path.basename(filename))with SCPClient(ssh_client.get_transport(), progress=userProgress, socket_timeout=15) as scp:scp.get(remote_path, local_file, recursive=True)ssh_client.close()import sys
import timeif __name__ == "__main__":host = sys.argv[1]port = 22username = sys.argv[2]password = sys.argv[3]local_file = time.strftime("%Y%m%dT%H%M%S", time.localtime())os.mkdir(local_file)remote_path = sys.argv[4]scp_transfer_with_progress(local_file, remote_path, host, port, username, password)

可以提示很多的中间信息

感觉AI 没以前好用了

功能齐全