Python3 通過FTP 文件、文件夾上傳因痛、下載
# -*- encoding: utf-8 -*-
"""
@Project : app
@File : ftpHelper.py
@Contact : networkaccounts@163.com
@Modify Time : 2020-08-26 15:46
@Author : Joker
@Version : 1.0
@Description : None
# 設(shè)置FTP當(dāng)前操作的路徑，同linux中的cd
ftp.cwd(path)
# 顯示目錄下所有信息
ftp.dir()
# 獲取目錄下的文件岸更，顯示的是文件名列表
ftp.nlst()
# 新建遠(yuǎn)程目錄
ftp.mkd(directory)
# 刪除遠(yuǎn)程目錄
ftp.rmd(directory)
# 將遠(yuǎn)程文件old重命名為new
ftp.rename(old, new)
# 刪除遠(yuǎn)程文件
ftp.delete(file_name)
# 上傳文件鸵膏，cmd是一個(gè)存儲(chǔ)命令，可以為"STOR filename.txt"怎炊， fp為類文件對(duì)象（有read方法）谭企，bufsize設(shè)置緩沖大小
ftp.storbinary(cmd, fp, bufsize)
# 下載文件，cmd是一個(gè)獲取命令评肆，可以為"RETR filename.txt"债查， callback是一個(gè)回調(diào)函數(shù)，用于讀取獲取到的數(shù)據(jù)塊
ftp.retrbinary(cmd, callback, bufsize)
"""
from ftplib import FTP, error_perm
import os
class ftpHelper(object):
conn = FTP()
def __init__(self, host, port=21):
"""
創(chuàng)建連接
:param host:
:param port:
"""
self.conn.connect(host, port)
def login(self, username, password):
"""
登錄
:param username:
:param password:
:return:
"""
self.conn.login(username, password)
self.conn.encoding = "GB2312"
# 打開調(diào)試級(jí)別2瓜挽，顯示詳細(xì)信息
self.conn.set_debuglevel(2)
self.conn.set_pasv(True)
# 0 主動(dòng)模式 1 #被動(dòng)模式
print(self.conn.welcome)
def _ftp_list(self, line):
li = line.split(' ')
if self.ftp_dir_name == li[-1] and "<DIR>" in li:
self._is_dir = True
def _is_ftp_file(self, ftp_path):
try:
if ftp_path in self.conn.nlst(os.path.dirname(ftp_path)):
return True
else:
return False
except error_perm as e:
return False
def _is_ftp_dir(self, ftp_path):
"""
用來判斷所給的路徑是文件還是文件夾
"""
ftp_path = ftp_path.rstrip('/')
ftp_parent_path = os.path.dirname(ftp_path)
self.ftp_dir_name = os.path.basename(ftp_path)
self._is_dir = False
if ftp_path == '.' or ftp_path == './' or ftp_path == '':
self._is_dir = True
else:
try:
self.conn.cwd(ftp_path)
self._is_dir = True
# self.conn.retrlines('LIST %s' %ftp_parent_path,self._ftp_list)
except error_perm as e:
return self._is_dir
return self._is_dir
def get_file(self, ftp_path="", local_path="."):
"""
下載文件
:param ftp_path:
:param local_path:
:return:
"""
print(ftp_path)
ftp_path = ftp_path.rstrip('/')
file_name = os.path.basename(ftp_path)
# 如果本地路徑是目錄盹廷，下載文件到該目錄
if os.path.isdir(local_path):
file_handler = open(os.path.join(local_path, file_name), 'wb')
self.conn.retrbinary("RETR %s" % ftp_path, file_handler.write)
file_handler.close()
# 如果本地路徑不是目錄，但上層目錄存在久橙，則按照本地路徑的文件名作為下載的文件名稱
elif os.path.isdir(os.path.dirname(local_path)):
file_handler = open(local_path, 'wb')
self.conn.retrbinary("RETR %s" % ftp_path, file_handler.write)
file_handler.close()
# 如果本地路徑不是目錄俄占，且上層目錄不存在，則退出
else:
print('EROOR:The dir:%s is not exist' % os.path.dirname(local_path))
def get_dir(self, ftp_path, local_path=".", begin=True):
"""
下載目錄下的所有文件
:param ftp_path:
:param local_path:
:param begin:
:return:
"""
if not self._is_ftp_dir(ftp_path):
self.get_file(ftp_path=ftp_path, local_path=local_path)
return
if begin:
local_path = os.path.join(local_path, os.path.basename(ftp_path))
# 如果本地目錄不存在淆衷，則創(chuàng)建目錄
if not os.path.isdir(local_path):
os.mkdir(local_path)
# 進(jìn)入ftp目錄缸榄，開始遞歸查詢
self.conn.cwd(ftp_path)
ftp_files = self.conn.nlst()
for file in ftp_files:
local_file = os.path.join(local_path, file)
# 如果file ftp路徑是目錄則遞歸上傳目錄（不需要再進(jìn)行初始化begin的標(biāo)志修改為False）
# 如果file ftp路徑是文件則直接上傳文件
if self._is_ftp_dir(file):
self.get_dir(file, local_file, False)
elif "idea" in file:
pass
else:
self.get_file(ftp_path=file, local_path=local_file)
# 如果當(dāng)前ftp目錄文件已經(jīng)遍歷完畢返回上一層目錄
self.conn.cwd("..")
def get_all_dir(self):
ftp_files = self.conn.nlst()
for file in ftp_files:
self.get_dir(file, "D:\\ftp", True)
def put_file(self, local_path, ftp_path="."):
"""
上傳文件
:param local_path:
:param ftp_path:
:return:
"""
ftp_path = ftp_path.rstrip('/')
if os.path.isfile(local_path):
file_handler = open(local_path, 'rb')
local_file_name = os.path.basename(local_path)
# 如果遠(yuǎn)程路徑是個(gè)目錄，則上傳文件到這個(gè)目錄祝拯，文件名不變
if self._is_ftp_dir(ftp_path):
self.conn.storbinary('STOR %s' % os.path.join(ftp_path, local_file_name), file_handler)
# 如果遠(yuǎn)程路徑的上層是個(gè)目錄甚带，則上傳文件，文件名按照給定命名
elif self._is_ftp_dir(os.path.dirname(ftp_path)):
print('STOP %s' % ftp_path)
self.conn.storbinary('STOR %s' % ftp_path, file_handler)
# 如果遠(yuǎn)程路徑不是目錄，且上一層的目錄也不存在鹰贵，則提示給定遠(yuǎn)程路徑錯(cuò)誤
else:
print('STOR %s' % ftp_path, file_handler)
def put_dir(self, local_path, ftp_path=".", begin=True):
"""
上傳文件夾
:param local_path:
:param ftp_path:
:param begin:
:return:
"""
ftp_path = ftp_path.rstrip('/')
if not os.path.isdir(local_path):
print('ERROR:The dir:%s is not exist' % local_path)
return
# 當(dāng)本地目錄存在時(shí)上傳
# 上傳初始化：如果給定的ftp路徑不存在需要?jiǎng)?chuàng)建晴氨，同時(shí)將本地的目錄存放在給定的ftp目錄下。
# 本地目錄下文件存放的路徑為ftp_path = ftp_path + os.path.basename(local_path)
# 例如碉输，將本地的文件夾a上傳到ftp的a/b目錄下瑞筐，則本地a目錄下的文件將上傳的ftp的a/b/a目錄下
if begin:
if not self._is_ftp_dir(ftp_path):
try:
self.conn.mkd(ftp_path)
except Exception as e:
print(e)
ftp_path = os.path.join(ftp_path, os.path.basename(local_path))
# 如果上傳路徑是文件夾，則創(chuàng)建目錄
if not self._is_ftp_dir(ftp_path):
try:
self.conn.mkd(ftp_path)
except Exception as e:
print(e)
# 進(jìn)入本地目錄腊瑟，開始遞歸查詢
os.chdir(local_path)
local_files = os.listdir('.')
for file in local_files:
ftp_file = os.path.join(ftp_path, file)
# 如果file本地路徑是目錄則遞歸上傳文件（不需要再進(jìn)行初始化begin的標(biāo)志修改為False）
# 如果file本地路徑是文件則直接上傳文件
if os.path.isdir(file):
self.put_dir(file, ftp_file, False)
elif "idea" in file:
pass
else:
self.put_file(file, ftp_path)
# 如果當(dāng)前本地目錄文件已經(jīng)遍歷完畢返回上一層目錄
os.chdir('..')
# todo
if __name__ == "__main__":
# local = '/home/app/ftp.py'
local = '/home/app'
ftp = ftpHelper('192.168.43.26')
ftp.login('test', "test")
# ftp.get_file("ftppath")
# ftp.get_dir("ftppath","localpath",True)
# ftp.get_dir("","",True)
# ftp.get_all_dir("")
# ftp.put_file(local)
ftp.put_dir(local, "/temp", begin=True)
