1. ftp工具类

编写ftp工具类，我这里取名为 ftp_util.py

import os
from ftplib import FTPclass FtpUtil:def __init__(self, ip, username, password, port=21):self.ip = ipself.username = usernameself.password = passwordself.port = portself.ftp = Nonedef connect(self):''''建立ftp远程连接'''try:self.ftp = FTP()self.ftp.connect(host=self.ip, port=self.port)self.ftp.login(user=self.username, passwd=self.password)except Exception as e:print(e)def disconnect(self):''''断开远程连接'''if self.ftp:self.ftp.quit()def upload_file(self, localPath, remotePath, mode='binary'):'''上传文件'''try:with open(localPath, 'rb') as f:if mode == 'binary': # 上传二进制文件self.ftp.storbinary(f'STOR {remotePath}', f)elif mode == 'text': # 上传文本文件self.ftp.storlines(f'STOR {remotePath}', f)print(f'文件已上传，本地文件路径：{localPath}，远端文件路径：{remotePath}')except Exception as e:print(f'文件上传失败：{e}')def download_file(self, localPath, remotePath, mode='binary'):'''下载文件'''try:with open(localPath, 'wb') as f:if mode == 'binary': # 下载二进制文件self.ftp.retrbinary(f'RETR {remotePath}', f.write)elif mode == 'text': # 下载文本文件self.ftp.retrlines(f'RETR {remotePath}', f.write)print(f'文件已下载，本地文件路径：{localPath}，远端文件路径：{remotePath}')except Exception as e:print(f'文件下载失败：{e}')def upload_files(self, localDirPath, remoteDirPath, mode='binary'):'''指定目录，上传本地目录下的所有文件到远端目录（不递归）'''files = os.listdir(localDirPath)for file in files:localPath = os.path.join(localDirPath, file)remotePath = f'{remoteDirPath}/{file}'self.upload_file(localPath, remotePath, mode)def download_files(self, localDirPath, remoteDirPath, mode='binary'):'''指定目录，下载远端目录下的所有文件到本地目录（不递归）'''self.ftp.cwd(remoteDirPath) # 切换到指定目录files = self.ftp.nlst() # 列出目录下的文件列表for file in files:localPath = os.path.join(localDirPath, file)remotePath = f'{remoteDirPath}/{file}'self.download_file(localPath, remotePath, mode)def rename_remote_file(self, remoteDirPath, oldFileName, newFileName):'''重命名远端文件'''try:self.ftp.cwd(remoteDirPath) # 切换到指定目录self.ftp.rename(oldFileName, newFileName)print(f'{remoteDirPath}目录下，文件重命名：{oldFileName} -> {newFileName}')except Exception as e:print(f'重命名文件失败：{e}')def delete_remote_file(self, remotePath):'''删除远端文件'''try:self.ftp.delete(remotePath)print(f'文件已删除：{remotePath}')except Exception as e:print(f'重命名文件失败：{e}')def delete_remote_files(self, remoteDirPath):'''删除远端目录下的所有文件（不递归）'''self.ftp.cwd(remoteDirPath) # 切换到指定目录files = self.ftp.nlst()  # 列出目录下的文件列表for file in files:remotePath = f'{remoteDirPath}/{file}'self.delete_remote_file(remotePath)if __name__ == '__main__':# 主机信息ip = '192.168.215.1'username = 'admin'password = '123456'# 创建对象ftp = FtpUtil(ip, username, password)ftp.connect()# 上传文件localPath = r'D:\测试文件\a.txt'remotePath = r'/usr/a.txt'ftp.upload_file(localPath, remotePath)# 下载文件localPath = r'D:\测试文件\b.txt'remotePath = r'/usr/b.txt'ftp.download_file(localPath, remotePath)ftp.disconnect()

2. sftp工具类

首先安装paramiko第三方库：

pip install paramiko

出现如下图，表示安装成功：

编写sftp工具类，我这里取名为 sftp_util.py

import os
import stat
import paramikoclass SftpUtil:def __init__(self, ip, username, password, port=22):self.ip = ipself.username = usernameself.password = passwordself.port = portself.transport = Noneself.sftp = Nonedef connect(self):''''建立sftp远程连接'''try:self.transport = paramiko.Transport(sock=(self.ip, self.port))self.transport.connect(username=self.username, password=self.password)self.sftp = paramiko.SFTPClient.from_transport(self.transport)except Exception as e:print(e)def disconnect(self):''''断开远程连接'''if self.sftp:self.sftp.close()if self.transport:self.transport.close()def upload_file(self, localPath, remotePath):'''上传文件'''try:self.sftp.put(localPath, remotePath)print(f'文件已上传，本地文件路径：{localPath}，远端文件路径：{remotePath}')except Exception as e:print(f'文件上传失败：{e}')def download_file(self, localPath, remotePath):'''下载文件'''try:self.sftp.get(remotePath, localPath)print(f'文件已下载，本地文件路径：{localPath}，远端文件路径：{remotePath}')except Exception as e:print(f'文件下载失败：{e}')def upload_files(self, localDirPath, remoteDirPath):'''指定目录，上传本地目录下的所有文件到远端目录（不递归）'''files = os.listdir(localDirPath)for file in files:localPath = os.path.join(localDirPath, file)remotePath = f'{remoteDirPath}/{file}'self.upload_file(localPath, remotePath)def download_files(self, localDirPath, remoteDirPath):'''指定目录，下载远端目录下的所有文件到本地目录（不递归）'''files = self.sftp.listdir(remoteDirPath)for file in files:localPath = os.path.join(localDirPath, file)remotePath = f'{remoteDirPath}/{file}'self.download_file(localPath, remotePath)def get_remote_files(self, remoteDirPath):'''获取远端目录下所有文件路径（递归），返回文件路径列表'''filePathList = []items = self.sftp.listdir_attr(remoteDirPath)for item in items:remotePath = f'{remoteDirPath}/{item.filename}'# 如果是目录，则递归调用if stat.S_ISDIR(item.st_mode):subDirList = self.get_remote_files(remotePath)filePathList.extend(subDirList)# 如果是文件，则直接添加到结果列表中elif stat.S_ISREG(item.st_mode):filePathList.append(remotePath)return filePathListdef recursive_download_files(self, localDirPath, remoteDirPath, baseDirPath=''):'''递归下载远端目录下的所有文件到本地目录（把整个远端目录下载到本地目录下）'''remotePathList = self.get_remote_files(remoteDirPath)for remotePath in remotePathList:if baseDirPath:stitchedPath = remotePath.split(f'{baseDirPath}/')[-1].replace('/', '\\')else:stitchedPath = remotePath.replace('/', '\\')localPath = os.path.join(localDirPath, stitchedPath)localDirPath = os.path.dirname(localPath)# 如果本地目录不存在，则创建if not os.path.exists(localDirPath):os.makedirs(localDirPath)self.sftp.get(remotePath, localPath)print(f'文件已下载，本地文件目录路径：{localDirPath}，远端文件目录路径：{remoteDirPath}')if __name__ == '__main__':# 主机信息ip = '192.168.215.1'username = 'admin'password = '123456'# 创建对象sftp = SftpUtil(ip, username, password)sftp.connect()# 上传文件localPath = r'D:\测试文件\a.txt'remotePath = r'/usr/a.txt'sftp.upload_file(localPath, remotePath)# 下载文件localPath = r'D:\测试文件\b.txt'remotePath = r'/usr/b.txt'sftp.download_file(localPath, remotePath)sftp.disconnect()