FTP, SFTP file transfer

ftp

#!/usr/bin/python
# -*- coding: UTF-8 -*-

from ftplib import FTP
import os


class MyFtp:
    def __init__(self, host=None, port=21, username=None, password=None):
        self.ftp = FTP()
        #  Reset the encoding 
        self.ftp.encoding = 'utf-8'
        try:
            # 0 Active mode  1 # Passive mode 
            self.ftp.set_pasv(True)
            self.ftp.connect(host, port)
            self.ftp.login(username, password)
            print(' Successfully connected to ：', host, " Login successful ...")
        except Exception as err:
            print("FTP  Connection or login failed , The error is described as ：", err)

    def __del__(self):
        print(" Close the connection ")
        self.ftp.quit()

    def is_same_size(self, local_file, remote_file):
        #  Determine whether the file size is equal 
        try:
            remote_file_size = self.ftp.size(remote_file)
        except Exception as err:
            # print("is_same_size()  The error is described as ：%s" % err)
            remote_file_size = -1
        try:
            local_file_size = os.path.getsize(local_file)
        except Exception as err:
            # print("is_same_size()  The error is described as ：%s" % err)
            local_file_size = -1
        # print(f'local_file_size:{local_file_size}, remote_file_size:{remote_file_size}')
        if remote_file_size == local_file_size:
            return 1
        else:
            return 0

    def download_file(self, remote_file, local_file):
        remote_file = remote_file if self.ftp.pwd() == "/" else self.ftp.pwd() + "/" + remote_file
        print(f" Download the file : {remote_file} ---> {local_file}")
        if self.is_same_size(local_file, remote_file):
            print(" Same file size , This file skipped ")
        else:
            try:
                buf_size = 1024
                file_handler = open(local_file, 'wb')
                self.ftp.retrbinary(f'RETR {remote_file}', file_handler.write, blocksize=buf_size)
                file_handler.close()
            except Exception as err:
                print(f' Error downloading file , Something unusual happened ：{err}')

    def download_dir(self, remote_dir, local_dir):
        remote_dir = remote_dir if self.ftp.pwd() == "/" else self.ftp.pwd() + "/" + remote_dir
        print(f" Download directory : {remote_dir} ---> {local_dir}")
        try:
            self.ftp.cwd(remote_dir)
            print(f' Switch to the directory : {self.ftp.pwd()}')
        except Exception as err:
            print(f' Remote directory {remote_dir} non-existent , continue ..., The specific error is described as ：{err}')
            return
        os.makedirs(local_dir, exist_ok=True)
        remote_dirs = []
        self.ftp.dir(".", remote_dirs.append)
        remote_list = list(map(lambda x: x.split()[-1], remote_dirs))
        print(f' Remote directory list : {remote_list}')
        for remote_detail in remote_dirs:
            file_type = remote_detail[0]
            file_name = remote_detail.split()[-1]
            local_path = os.path.join(local_dir, file_name)
            if file_type == "d":
                self.download_dir(file_name, local_path)
            elif file_type == '-':
                self.download_file(file_name, local_path)
        self.ftp.cwd("..")
        print(" Return to the upper directory ： ", self.ftp.pwd())

    def upload_file(self, local_file, remote_file):
        #  Upload files 
        remote_file = remote_file if self.ftp.pwd() == "/" else self.ftp.pwd() + "/" + remote_file
        print(f" Upload files ： {local_file} ---> {remote_file}")
        if not os.path.isfile(local_file):
            print(f'{local_file}  non-existent ')
            return
        if self.is_same_size(local_file, remote_file):
            print(' Same file size , This file skipped ')
            return
        try:
            buf_size = 1024
            file_handler = open(local_file, 'rb')
            self.ftp.storbinary(f'STOR {remote_file}', file_handler, buf_size)
            file_handler.close()
        except Exception as err:
            print(f' Error uploading file , Something unusual happened ：{err}')

    def upload_dir(self, local_dir, remote_dir):
        #  Upload directory 
        remote_dir = remote_dir if self.ftp.pwd() == "/" else self.ftp.pwd() + "/" + remote_dir
        print(f" Upload directory ： {local_dir} ---> {remote_dir}")
        if not os.path.isdir(local_dir):
            print(f' Local directory  {local_dir}  non-existent ')
            return
        try:
            self.ftp.mkd(remote_dir)
        except Exception as err:
            print(f" Create a remote directory {remote_dir} Failure , The error is described as ：{err}")
            return
        self.ftp.cwd(remote_dir)
        print(' Switch to remote directory : ', self.ftp.pwd())
        local_name_list = os.listdir(local_dir)
        print(" Local directory list ： ", local_name_list)
        for local_name in local_name_list:
            local_name_path = os.path.join(local_dir, local_name)
            if os.path.isdir(local_name_path):
                self.upload_dir(local_name_path, local_name)
            else:
                self.upload_file(local_name_path, local_name)
        self.ftp.cwd("..")
        print(" Return to the upper directory ： ", self.ftp.pwd())

sftp

#!/usr/bin/python
# coding=utf-8
import paramiko
import os
import stat


class MySftp:
    def __init__(self, host=None, port=22, username=None, password=None):
        self.sftp_connect = paramiko.Transport((host, port))
        self.sftp_connect.connect(username=username, password=password)
        self.sftp = paramiko.SFTPClient.from_transport(self.sftp_connect)
        print(' Successfully connected to ：', host, " Login successful ...")

    def __del__(self):
        print(" Close the connection ")
        self.sftp_connect.close()

    def is_same_size(self, local_file, remote_file):
        #  Determine whether the file size is equal 
        try:
            remote_file_size = self.sftp.stat(remote_file).st_size
        except Exception as err:
            # print("is_same_size()  The error is described as ：%s" % err)
            remote_file_size = -1
        try:
            local_file_size = os.path.getsize(local_file)
        except Exception as err:
            # print("is_same_size()  The error is described as ：%s" % err)
            local_file_size = -1
        # print(f'local_file_size:{local_file_size}, remote_file_size:{remote_file_size}')
        if remote_file_size == local_file_size:
            return 1
        else:
            return 0

    def download_file(self, remote_file, local_file):
        remote_file = self.sftp.getcwd() + "/" + remote_file if self.sftp.getcwd() else remote_file
        print(f" Download the file ：{remote_file} ---> {local_file}")
        if self.is_same_size(local_file, remote_file):
            print(" The file already exists and the file already exists and the file size is the same , This file skipped ")
            return
        try:
            self.sftp.get(remote_file, local_file)
        except Exception as err:
            print(f' Error downloading file , Something unusual happened ：{err}')

    def download_dir(self, remote_dir, local_dir):
        remote_dir = self.sftp.getcwd() + "/" + remote_dir if self.sftp.getcwd() else remote_dir
        print(f" Download directory : {remote_dir} ---> {local_dir}")
        try:
            self.sftp.chdir(remote_dir)
            print(f' Switch to the directory : {self.sftp.getcwd()}')
        except Exception as err:
            print(f' Remote directory {remote_dir} non-existent , continue ..., The specific error is described as ：{err}')
            return
        os.makedirs(local_dir, exist_ok=True)
        remote_dirs = self.sftp.listdir(remote_dir)
        print(f' Remote directory list : {remote_dirs}')
        for remote_path in remote_dirs:
            local_path = os.path.join(local_dir, remote_path)
            if str(self.sftp.stat(remote_path))[0] == "d":
                self.download_dir(remote_path, local_path)
            elif str(self.sftp.stat(remote_path))[0] == "-":
                self.download_file(remote_path, local_path)
        self.sftp.chdir("..")
        print(" Return to the upper directory : ", self.sftp.getcwd())
    
    def upload_file(self, local_file, remote_file):
        remote_file = self.sftp.getcwd() + "/" + remote_file if self.sftp.getcwd() else remote_file
        print(f" Upload files ： {local_file} ---> {remote_file}")
        if not os.path.isfile(local_file):
            print(f" Local files  {local_file}  non-existent ")
            return
        if self.is_same_size(local_file, remote_file):
            print(' The file already exists and the file size is the same , This file skipped ')
            return
        try:
            self.sftp.put(local_file, remote_file)
        except Exception as err:
            print(f' Error uploading file , Something unusual happened ：{err}') 

    def upload_dir(self, local_dir, remote_dir):
        remote_dir = self.sftp.getcwd() + "/" + remote_dir if self.sftp.getcwd() else remote_dir
        print(f" Upload directory ： {local_dir} ---> {remote_dir}")
        if not os.path.isdir(local_dir):
            print(f' Local directory  {local_dir}  non-existent ')
            return
        if not self.is_sftp_dir(remote_dir):
            self.sftp.mkdir(remote_dir)
        self.sftp.chdir(remote_dir)
        print(' Switch to remote directory : ', self.sftp.getcwd())
        local_name_list = os.listdir(local_dir)
        print(" Local directory list ： ", local_name_list)
        for local_name in local_name_list:
            local_name_path = os.path.join(local_dir, local_name)
            if os.path.isdir(local_name_path):
                self.upload_dir(local_name_path, local_name)
            else:
                self.upload_file(local_name_path, local_name)
        self.sftp.chdir("..")
        print(" Return to the upper directory : ", self.sftp.getcwd())

    def is_sftp_dir(self, remote_dir):
        try:
            if stat.S_ISDIR(self.sftp.stat(remote_dir).st_mode):  #  If remote_dir Exists and is a directory , Then return to True
                return True
        except:
            return False