import ftplib
import mimetypes
import os
import sys
class FtpTools:
""" 遠程ftp交互輔助類 """
def __init__(self, ip, username, password, local_dir, remote_dir, is_clean: bool = True):
""" 初始化 :param ip: 遠程ftp服務器 - 地址 :param username: 遠程ftp服務器 - 用戶名 :param password: 遠程ftp服務器 - 密碼 :param local_dir: 本地目錄 :param remote_dir: 遠程目錄 :param is_clean: 是否刪除備份目錄下的文件，默認先刪除再備份 """
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"| 地 址：ftp://{
self.ip}:21\n" \
f"| 用 戶：{
self.username}\n" \
f"| 密 碼：{
self.password}\n" \
f"| 本地目錄：{
self.local_dir}\n" \
f"| 遠程目錄：{
self.remote_dir}\n" \
f"| 是否清空：{
self.is_clean}\n" \
f"| 連接方式：{
'被動' if self.non_passive else '主動'}"
print('*' * 40 + '\n' + msg + '\n' + '*' * 40)
@staticmethod
def is_text_kind(rname, trace=True):
""" 判斷文件是text文件還是二進制文件 對於未知類型文件，結果可能是(None，None)類型 :param rname: 遠程ftp服務器文件名 :param trace: 是否輸出提示信息 """
mimetypes.add_type('text/x-python-win', '.pyw')
# strict=False 接受額外類型
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):
""" 根據配置信息連接遠程ftp服務器 """
print('連接中 ...')
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):
""" 根據配置文件嘗試清空本地文件夾 """
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'正在刪除本地目錄 {
path}')
try:
os.rmdir(path)
except:
print(f'無法刪除本地目錄 {
path}')
for file in files:
path = os.path.join(root, file)
print(f'正在刪除本地文件 {
path}')
try:
os.remove(path)
except:
print(f'無法刪除本地文件 {
path}')
def clean_remotes(self):
""" 根據配置文件嘗試清空遠程ftp服務器文件夾 """
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'正在刪除文件 {
fname}')
self.connection.delete(fname)
self.file_count += 1
else:
print(f'正在刪除目錄 {
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):
""" 通過ftp下載單個文件 :param rname: 遠程ftp文件名 :param 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):
""" 通過ftp上傳單個文件 :param local_name: 本地文件名 :param local_path: 本地路徑 :param rname: 遠程ftp文件名 """
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):
""" 從遠程ftp服務器指定目錄下載所有文件 """
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'正在下載文件 {
fname}', end=" ")
self.download_one(fname, os.path.join(self.local_dir, fname))
self.file_count += 1
else:
print(f'正在創建目錄 {
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'終止 {
self.local_dir} 出現一個例外')
def upload_dir(self):
""" 對於整個目錄樹裡的每個目錄上傳簡單的文件，遞歸進入子目錄 """
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'目錄 {
mdir} 已創建。')
except:
print(sys)
print(f'未創建目錄 {
mdir}')
else:
self.folder_count += 1
for name in files:
local_path = os.path.join(root, name)
print('上傳 ', local_path, ' 至 ', 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'| 此次備份統計：\n' \
f'| \t文件夾數：{
self.folder_count}\n' \
f'| \t文件數：{
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) # 本地文件備份到FTP
ftp.run(clean_target=ftp.clean_locals, transfer_act=ftp.download_dir) # FTP文件備份到本地
運行結果
****************************************
| 地 址：ftp://192.168.0.156:21
| 用 戶：ftpuser
| 密 碼：ftpuser_pwd
| 本地目錄：/Users/Downloads/data
| 遠程目錄：/data
| 是否清空：True
| 連接方式：主動
****************************************
連接中 ...
正在刪除本地文件 /Users/Downloads/data/c/d.txt
正在刪除本地目錄 /Users/Downloads/data/c
正在刪除本地文件 /Users/Downloads/data/b.txt
正在刪除本地文件 /Users/Downloads/data/a.txt
正在下載文件 a.txt text
正在下載文件 b.txt text
正在創建目錄 c
正在下載文件 d.txt text
正在下載文件 e ?
****************************************
| 此次備份統計：
| 文件夾數：1
| 文件數：4
****************************************
