SFTP文件本地及远程创建，上传，下载，删除，获取脚本

import os
import time
import paramiko

class SftpLocal:
    """
    SFTP操作
    支持配置协议，主机，账号，密码
    支持生成文件
    支持下载
    支持删除
    支持上传
    """

    def __init__(self,
                 sftp_route: str,
                 local_route: str):
        """
        :param sftp_route: SFTP路径
        :param local_route: 本地路径
        """
        self.sftp_route = sftp_route
        self.local_route = local_route
        # SFTP配置协议登录信息
        self.protocol = 'sftp'  # SFTP协议
        self.host = '10.100.***.**'  # 服务器主机名或IP地址
        self.port = 9999  # SFTP服务器端口号，默认为22
        self.username = '******'  # 服务器用户名
        self.password = ''******''  # 服务器密码
        # 创建SSH客户端并连接到SFTP服务器
        self.ssh = paramiko.SSHClient()
        self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        self.ssh.connect(self.host, self.port, self.username, self.password)
        # 创建SFTP对象
        self.sftp = self.ssh.open_sftp()

    def upload(self):
        """
        样例：
        local_route = '/path/to/local/file.txt'
        sftp_route = '/path/to/remote/file.txt'
        :param sftp_route: sftp路径
        :param local_route: 本地路径
        :return:
        """
        # 上传本地文件到指定路径
        self.sftp.put(self.local_route, self.sftp_route)
        return True

    def download(self):
        """
        :param sftp_route: sftp路径
        :param local_route: 本地路径
        :return:
        """
        files = self.sftp.listdir(self.sftp_route)
        for filename in files:
            remote_file = os.path.join(self.sftp_route, filename)
            local_file = os.path.join(self.local_route, filename)
            self.sftp.get(remote_file, local_file)
        return True

    def delete(self):
        """
        删除sftp路径下的文件
        :param sftp_route:
        :return:
        """
        self.sftp.remove(self.sftp_route)
        return True

    def write_txt(self, route: str, filename: str, header: list, data: list, filename_success: bool = False):
        """
        # 定义表头和数据
        header = ['Name', 'Age', 'Gender']
        data = [
            ['Alice', 25, 'Female'],
            ['Bob', 30, 'Male']
        ]
        :param route: 文件路径
        :param filename: 文件名称
        :param filename_success: 是否要生成一个success文件
        :param args: 随机表头参数和内容以及其他随机参数
        :return:
        """
        if not filename_success:
            # 打开文件并写入表头和数据
            with open(route + filename + ".txt", 'w') as f:
                # 写入表头
                f.write(','.join(header) + '\n')
                # 写入数据
                for row in data:
                    f.write(','.join(row) + '\n')
        else:
            # 打开文件并写入表头和数据
            with open(route + filename + ".txt", 'w') as f:
                # 写入表头
                f.write(','.join(header) + '\n')
                # 写入数据
                for row in data:
                    f.write(','.join(row) + '\n')
            self.create_txt_success(route + filename)
            return True

    def create_txt_success(self, filename: str):
        open(filename + ".txt" + ".success", "a")
        return True


# 调试
# 上传
# SftpLocals = SftpLocal(sftp_route='/****/marketing/7433800/20231129/123.txt',
# local_route=r'G:\pycharmDl\job\MarkeScriptTest\sftp_local_txt\123.txt')
# print(SftpLocals.upload())
# 下载
# SftpLocals = SftpLocal(sftp_route='/****/marketing/7433800/20231129/',
#     local_route = r'G:\pycharmDl\job\MarkeScriptTest\sftp_local_txt')
# print(SftpLocals.download())
# 创建
# SftpLocals = SftpLocal(sftp_route='/****/marketing/7433800/20231129/',
#     local_route=r'G:\pycharmDl\job\MarkeScriptTest\sftp_local_txt')
# local_route = r'G:\pycharmDl\job\MarkeScriptTest\sftp_local_txt'
# today = time.strftime('%Y%m%d', time.localtime(time.time()))
# header = ['custNum', 'cell']
# data = [['asdfqwert', '13112341234'], ['asdfqwert', '13112341234']]
# print(SftpLocals.write_txt(route=local_route,
#     filename=today,
#     header=header,
#     data=data,
#     filename_success=True))