import ftplib
import mimetypes
import os
import sys
class FtpTools:
""" long-range ftp Interactive auxiliary class """
def __init__(self, ip, username, password, local_dir, remote_dir, is_clean: bool = True):
""" initialization :param ip: long-range ftp The server - Address :param username: long-range ftp The server - user name :param password: long-range ftp The server - password :param local_dir: Local directory :param remote_dir: Remote directory :param is_clean: Delete the files under the backup directory , By default, delete before backup """
self.ip = ip
self.username = username
self.password = password
self.local_dir = local_dir
self.remote_dir = remote_dir
self.is_clean = is_clean
self.file_count = 0
self.folder_count = 0
self.connection = None
self.non_passive = False
self.print_info()
def print_info(self):
msg = f"| The earth site ：ftp://{
self.ip}:21\n" \
f"| use Household ：{
self.username}\n" \
f"| The secret code ：{
self.password}\n" \
f"| Local directory ：{
self.local_dir}\n" \
f"| Remote directory ：{
self.remote_dir}\n" \
f"| Is it empty ：{
self.is_clean}\n" \
f"| How to connect ：{
' passive ' if self.non_passive else ' Take the initiative '}"
print('*' * 40 + '\n' + msg + '\n' + '*' * 40)
@staticmethod
def is_text_kind(rname, trace=True):
""" Judge whether the file is text File or binary file For files of unknown type , The result could be (None,None) type :param rname: long-range ftp Server file name :param trace: Whether to output prompt information """
mimetypes.add_type('text/x-python-win', '.pyw')
# strict=False Accept additional types
mimetype, encoding = mimetypes.guess_type(rname, strict=False)
mimetype = mimetype or '?/?'
maintype = mimetype.split('/')[0]
if trace:
print(maintype, encoding or '')
return maintype == 'text' and encoding is None
def connect_ftp(self):
""" Connect to the remote according to the configuration information ftp The server """
print(' Connecting ...')
connection = ftplib.FTP(self.ip)
connection.login(self.username, self.password)
connection.cwd(self.remote_dir)
connection.encoding = 'gbk'
if self.non_passive:
connection.set_pasv(False)
self.connection = connection
def clean_locals(self):
""" Try to empty the local folder according to the configuration file """
if self.is_clean:
for root, sdir, files in os.walk(self.local_dir, topdown=False, followlinks=False):
for directory in sdir:
path = os.path.join(root, directory)
print(f' Deleting local directory {
path}')
try:
os.rmdir(path)
except:
print(f' Cannot delete local directory {
path}')
for file in files:
path = os.path.join(root, file)
print(f' Deleting local files {
path}')
try:
os.remove(path)
except:
print(f' Cannot delete local file {
path}')
def clean_remotes(self):
""" Try to empty the remote according to the configuration file ftp Server folder """
if self.is_clean:
lines = []
self.connection.dir(lines.append)
for line in lines:
parsed = line.split()
permiss, fname = parsed[0], parsed[-1]
if fname in ('.', '..'):
continue
elif permiss[0] != 'd':
print(f' Deleting files {
fname}')
self.connection.delete(fname)
self.file_count += 1
else:
print(f' Deleting Directory {
fname}')
self.connection.cwd(fname)
self.clean_remotes()
self.connection.cwd('..')
self.connection.rmd(fname)
self.folder_count += 1
def download_one(self, rname, local_path):
""" adopt ftp Download a single file :param rname: long-range ftp file name :param local_path: The local path """
if self.is_text_kind(rname):
local_file = open(local_path, 'w', encoding=self.connection.encoding)
def callback(line):
local_file.write(line + '\n')
self.connection.retrlines('retr ' + rname, callback)
else:
local_file = open(local_path, 'wb')
self.connection.retrbinary('retr ' + rname, local_file.write)
local_file.close()
def upload_one(self, local_name, local_path, rname):
""" adopt ftp Upload a single file :param local_name: Local filename :param local_path: The local path :param rname: long-range ftp file name """
if self.is_text_kind(local_name):
local_file = open(local_path, 'rb')
self.connection.storlines('stor ' + rname, local_file)
else:
local_file = open(local_path, 'rb')
self.connection.storbinary('stor ' + rname, local_file)
def download_dir(self):
""" From remote ftp The server downloads all files in the specified directory """
lines = []
self.connection.dir(lines.append)
for line in lines:
parsed = line.split()
permiss, fname = parsed[0], parsed[-1]
if fname in ('.', '..'):
continue
elif permiss[0] != 'd':
print(f' Downloading files {
fname}', end=" ")
self.download_one(fname, os.path.join(self.local_dir, fname))
self.file_count += 1
else:
print(f' Creating a directory {
fname}')
self.local_dir = os.path.join(self.local_dir, fname)
try:
os.mkdir(self.local_dir)
except:
print(sys.exc_info()[0])
else:
self.folder_count += 1
self.connection.cwd(fname)
self.download_dir()
self.connection.cwd('..')
self.local_dir = os.path.split(self.local_dir)[0]
def callback(self):
print(f' End {
self.local_dir} There is an exception ')
def upload_dir(self):
""" Upload simple files for each directory in the entire directory tree , Recursion into subdirectories """
for root, sdirs, files in os.walk(self.local_dir, topdown=True, onerror=callable, followlinks=False):
try:
rdir = root.replace('.', self.remote_dir)
rdir, mdir = os.path.split(rdir)
print(rdir, mdir)
self.connection.cwd('~')
self.connection.cwd(rdir)
self.connection.mkd(mdir)
self.connection.cwd(mdir)
print(f' Catalog {
mdir} Created .')
except:
print(sys)
print(f' Directory not created {
mdir}')
else:
self.folder_count += 1
for name in files:
local_path = os.path.join(root, name)
print(' Upload ', local_path, ' to ', name, end=' ')
self.upload_one(name, local_path, name)
self.file_count += 1
def run(self, clean_target=lambda: None, transfer_act=lambda: None):
self.connect_ftp()
clean_target()
transfer_act()
self.connection.quit()
msg = f'| Statistics of this backup ：\n' \
f'| \t Number of folders ：{
self.folder_count}\n' \
f'| \t Number of files ：{
self.file_count}'
print('*' * 40 + '\n' + msg + '\n' + '*' * 40)
if __name__ == '__main__':
ip = '192.168.0.156'
username = 'ftpuser'
password = 'ftpuser_pwd'
local_dir = '/Users/Downloads/data'
remote_dir = '/data'
ftp = FtpTools(ip, username, password, local_dir, remote_dir)
# ftp.run(clean_target=ftp.clean_remotes,transfer_act=ftp.upload_dir) # Backup local files to FTP
ftp.run(clean_target=ftp.clean_locals, transfer_act=ftp.download_dir) # FTP File backup to local
Running results
****************************************
| The earth site ：ftp://192.168.0.156:21
| use Household ：ftpuser
| The secret code ：ftpuser_pwd
| Local directory ：/Users/Downloads/data
| Remote directory ：/data
| Is it empty ：True
| How to connect ： Take the initiative
****************************************
Connecting ...
Deleting local files /Users/Downloads/data/c/d.txt
Deleting local directory /Users/Downloads/data/c
Deleting local files /Users/Downloads/data/b.txt
Deleting local files /Users/Downloads/data/a.txt
Downloading files a.txt text
Downloading files b.txt text
Creating a directory c
Downloading files d.txt text
Downloading files e ?
****************************************
| Statistics of this backup ：
| Number of folders ：1
| Number of files ：4
****************************************
