2、python3向ftp服务器上传和下载封装（二）（包括文件夹的上传和下载）
代码如下：
# -*- encoding: utf-8 -*-
import os
import re
import time
import ftplib
class MyFtp(object):
"""Ftp上传下载接口类"""
conn = ftplib.FTP()
def __init__(self, host, port=21):
self.conn.connect(host, port)
self.conn.encoding = "utf-8" # "GB2312"
# self.conn.set_debuglevel(2) # 打开调试级别2，显示详细信息
# self.conn.set_pasv(True) # 0 主动模式 1 #被动模式
def login(self, user, passwd):
"""登录
:param user: 用户名
:param passwd: 密码
:return: 返回self，支持链试调用
"""
self.conn.login(user, passwd)
print(f'{
self.conn.welcome} 登陆成功')
return self
def close(self):
self.conn.quit()
def _is_ftp_dir(self, dir_path):
try:
self.conn.dir(dir_path)
return True
except Exception as e:
print(f'e: {
e}')
return False
def download_file(self, local_file, remote_file):
"""下载单个文件，将 remote_file 下载到 local_file。
:param local_file: 本地文件地址
:param remote_file: 远程目录文件地址
:return: 是否成功标志
"""
ret = True
try:
file_handler = open(local_file, "wb")
self.conn.retrbinary("RETR " + remote_file, file_handler.write)
file_handler.close()
except Exception as err:
ret = False
print(f"file down load err: {
err}")
return ret
def download_dir(self, local_dir, remote_dir, filter=""): # 下载文件夹
"""下载单个文件，将 remote_dir 下载到 local_dir。
:param local_dir: 本地文件夹地址
:param remote_dir: 远程目录文件夹地址
:param filter: 正则过滤器
:return: 是否成功标志
"""
print("远程文件夹 remote_dir:", remote_dir)
if not os.path.exists(local_dir):
os.makedirs(local_dir)
self.conn.cwd(remote_dir)
remote_names = self.conn.nlst()
if 0 == len(remote_names):
print("远程文件目录：", remote_names)
return False
for file in remote_names:
local = os.path.join(local_dir, file)
print("正在下载", self.conn.nlst(file))
if file.find(".") == -1: # 子文件夹递归
if not os.path.exists(local):
os.makedirs(local)
self.download_dir(local, file)
else:
if filter and not re.search(filter, file):
continue
self.download_file(local, file)
self.conn.cwd("..")
return True
def list_dir(self, remote_dir, filter=None, real_path=False):
org_dir = self.conn.pwd()
self.conn.cwd(remote_dir)
file_names = self.conn.nlst()
if filter is not None:
file_names = [v for v in file_names if re.search(filter, v)]
if real_path:
file_names = [os.path.join(remote_dir, v) for v in file_names]
self.conn.cwd(org_dir)
return file_names
def put_file(self, local_file, remote_path="."):
"""上传文件
如果远程路径是个目录，则上传文件到这个目录，文件名不变；如果远程路径的上层是个目录，则上传文件，文件名按照给定命名。
:param local_file: 本地文件
:param remote_path: 远程目录或文件
:return: 是否成功标志
"""
ret = True
ftp_path = remote_path.rstrip('/')
if os.path.isfile(local_file):
file_handler = open(local_file, 'rb')
local_file_name = os.path.basename(local_file)
# 如果远程路径是个目录，则上传文件到这个目录，文件名不变
if self._is_ftp_dir(ftp_path):
self.conn.storbinary(f'STOR {
os.path.join(ftp_path, local_file_name)}', file_handler)
# 如果远程路径的上层是个目录，则上传文件，文件名按照给定命名
elif self._is_ftp_dir(os.path.dirname(ftp_path)):
print(f'STOP {
ftp_path}')
self.conn.storbinary(f'STOR {
ftp_path}', file_handler)
# 如果远程路径不是目录，且上一层的目录也不存在，则提示给定远程路径错误
else:
ret = False
print(f'STOR {
ftp_path}', file_handler)
return ret
def put_dir(self, local_dir, remote_dir=".", begin=True):
"""上传文件夹
:param local_dir: 本地文件夹
:param remote_dir: 远程文件夹
:param begin:
:return: 是否成功标志
"""
ftp_path = remote_dir.rstrip('/')
if not os.path.isdir(local_dir):
print(f'ERROR:The dir:{
local_dir} is not exist')
return False
# 当本地目录存在时上传
# 上传初始化：如果给定的ftp路径不存在需要创建，同时将本地的目录存放在给定的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_dir))
# 如果上传路径是文件夹，则创建目录
if not self._is_ftp_dir(ftp_path):
try:
self.conn.mkd(ftp_path)
except Exception as e:
print(e)
# 进入本地目录，开始递归查询
os.chdir(local_dir)
local_files = os.listdir('.')
for file in local_files:
ftp_file = os.path.join(ftp_path, file)
# 如果file本地路径是目录则递归上传文件（不需要再进行初始化begin的标志修改为False）
# 如果file本地路径是文件则直接上传文件
if os.path.isdir(file):
self.put_dir(file, ftp_file, False)
print(f' -- {
time.strftime("%Y-%m-%d %H:%M:%S")} success dir: {
file}')
elif "idea" in file:
pass
else:
if 'DS_Store' in file:
continue
self.put_file(file, ftp_path)
print(f' -- {
time.strftime("%Y-%m-%d %H:%M:%S")} success file: {
file}')
# 如果当前本地目录文件已经遍历完毕返回上一层目录
os.chdir('..')
print(f'finish')
if __name__ == '__main__':
my_ftp_info = MyFtp(host='192.168.38.40')
login_result = my_ftp_info.login(user='', passwd='')
print(login_result)