Python+Paramiko编写简单的SFTP文件上传下载工具类

pip install paramiko

from paramiko import Transport, SFTPClient
import time
import errno
import logging

# 配置日志，便于观察上传进度
logging.basicConfig(format='%(levelname)s : %(message)s',
                    level=logging.INFO)

class SftpClient:
    _connection = None  # 类级别共享的连接对象

    def __init__(self, host, port, username, password):
        self.host = host
        self.port = port
        self.username = username
        self.password = password
        self.create_connection(self.host, self.port,
                               self.username, self.password)

    @classmethod
    def create_connection(cls, host, port, username, password):
        """建立 SFTP 连接"""
        transport = Transport(sock=(host, port))
        transport.connect(username=username, password=password)
        cls._connection = SFTPClient.from_transport(transport)
        logging.info("SFTP 连接成功：{}@{}:{}".format(username, host, port))

    @staticmethod
    def uploading_info(uploaded_file_size, total_file_size):
        """上传进度回调函数"""
        percentage = (uploaded_file_size / total_file_size) * 100
        logging.info('已上传：{} / {} ({:.2f}%)'.format(
            uploaded_file_size, total_file_size, percentage))

    def upload(self, local_path, remote_path):
        """上传文件，支持进度显示"""
        self._connection.put(localpath=local_path,
                             remotepath=remote_path,
                             callback=self.uploading_info,
                             confirm=True)
        logging.info("文件上传成功：{} -> {}".format(local_path, remote_path))

    def file_exists(self, remote_path):
        """判断远程文件是否存在"""
        try:
            self._connection.stat(remote_path)
            return True
        except IOError as e:
            if e.errno == errno.ENOENT:
                return False
            raise  # 其他错误向上抛出

    def download(self, remote_path, local_path, retry=5):
        """
        下载文件
        如果文件不存在，会等待5秒后重试，最多重试5次（适合文件正在生成场景）
        """
        if self.file_exists(remote_path):
            self._connection.get(remote_path, local_path)
            logging.info("文件下载成功：{} -> {}".format(remote_path, local_path))
        elif retry > 0:
            logging.warning("远程文件不存在，{}秒后重试（剩余重试次数：{}）".format(5, retry))
            time.sleep(5)
            self.download(remote_path, local_path, retry=retry - 1)
        else:
            raise FileNotFoundError("下载失败：远程文件 {} 不存在，且重试次数已用完".format(remote_path))

    def close(self):
        """关闭连接"""
        if self._connection:
            self._connection.close()
            logging.info("SFTP 连接已关闭")

if __name__ == '__main__':
    # 请替换为真实的服务器信息
    host = 'example.com'
    port = 22
    username = 'your_username'
    password = 'your_password'

    download_remote_path = 'outgoing/download.txt'
    download_local_path = 'download.txt'

    upload_local_path = 'upload.txt'
    upload_remote_path = 'incoming/upload.txt'

    # 使用示例
    client = SftpClient(host, port, username, password)

    # 上传本地文件到远程服务器
    client.upload(upload_local_path, upload_remote_path)

    # 下载远程文件到本地（如果文件暂时不存在会自动重试）
    client.download(download_remote_path, download_local_path)

    # 记得关闭连接
    client.close()