Skip to content

Instantly share code, notes, and snippets.

@larry1001

larry1001/ftp_util.py

Created March 15, 2019 07:26
Show Gist options
  • Save larry1001/6c6936745611b6ac353a100515df676a to your computer and use it in GitHub Desktop.
Save larry1001/6c6936745611b6ac353a100515df676a to your computer and use it in GitHub Desktop.
Download ZIP
ftp客户端实现
Raw
ftp_util.py
from ftplib import FTP
import os
import sys
import time
import socket
# 本地日志存储路径
local_log_path = "log.txt"
class FtpUtil(object):
def __init__(self, host, port):
'''
初始化ftp参数
:param host:
:param port:
'''
self.host = host
self.port = port
self.ftp = FTP()
self.ftp.encoding = 'utf-8'
self.log_file = open(local_log_path, "a")
self.file_list = []
self.ftp.set_pasv(False)
def login(self, username, password):
'''
登录ftp
:param username:
:param password:
:return:
'''
try:
timeout = 60
socket.setdefaulttimeout(timeout)
# 0主动模式 1 #被动模式
self.ftp.set_pasv(1)
# 打开调试级别2，显示详细信息
self.ftp.set_debuglevel(2)
print('开始连接到 %s' % self.host)
self.ftp.connect(self.host, self.port)
print('成功连接到 %s' % self.host)
print('开始登录到 %s' % self.host)
self.ftp.login(username, password)
print('成功登录到 %s' % self.host)
print(self.ftp.welcome)
except Exception as err:
self.deal_error("ftp 连接或登录失败 ，错误描述为：%s " % err)
print("ftp 连接或登录失败 ，错误描述为：%s " % err)
pass
def _try_cwd(self, item):
'''
测试ftp文件夹是否存在
:param item:
:return:
'''
try:
self.ftp.cwd(item)
return True
except Exception:
return False
def download_file(self, local_file, remote_path, remote_file):
'''
从ftp下载文件
:param local_file:
:param remote_file:
:return:
'''
self.debug_print("download_file()---> local_path = %s ,remote_path = %s" % (local_file, remote_file))
try:
self.debug_print('>>>>>>>>>>>>下载文件 %s ... ...' % local_file)
self.ftp.cwd(remote_path)
with open(local_file, 'wb') as f:
self.ftp.retrbinary('RETR {0}'.format(remote_file), f.write)
except Exception as err:
self.debug_print('下载文件出错，出现异常：%s ' % err)
return
def download_file_tree(self, local_path, remote_path):
'''
从远程目录下载多个文件到本地目录
:param local_path:
:param remote_path:
:return:
'''
print("download_file_tree()---> local_path = %s ,remote_path = %s" % (local_path, remote_path))
try:
self.ftp.cwd(remote_path)
except Exception as err:
self.debug_print('远程目录%s不存在，继续...' % remote_path + " ,具体错误描述为：%s" % err)
return
if not os.path.isdir(local_path):
self.debug_print('本地目录%s不存在，先创建本地目录' % local_path)
os.makedirs(local_path)
self.debug_print('切换至目录: %s' % self.ftp.pwd())
self.file_list = []
# 方法回调
self.ftp.dir(self.get_file_list)
remote_names = self.file_list
self.debug_print('远程目录 列表: %s' % remote_names)
for item in remote_names:
file_type = item[0]
file_name = item[1]
local = os.path.join(local_path, file_name)
if file_type == 'd':
print("download_file_tree()---> 下载目录： %s" % file_name)
self.download_file_tree(local, file_name)
elif file_type == '-':
print("download_file()---> 下载文件： %s" % file_name)
self.download_file(local, file_name)
self.ftp.cwd("..")
self.debug_print('返回上层目录 %s' % self.ftp.pwd())
return True
def upload_file(self, local_file, remote_path):
'''
从本地上传文件到ftp
:param local_file: 本地文件
:param remote_path: 远程路径
:return:
'''
try:
if not self._try_cwd(remote_path):
self.ftp.mkd(remote_path)
self.ftp.cwd(remote_path)
if not os.path.isdir(local_file):
self.ftp.storbinary("STOR {0}".format(os.path.basename(local_file)), open(local_file, "rb"))
size = os.path.getsize(local_file)
self.debug_print('上传: %s' % local_file + " 成功!文件大小： " + str(size) + " Bytes.")
else:
print('%s 不是文件' % local_file)
return
except Exception as err:
self.debug_print('上传失败，继续...' % remote_path + " ,具体错误描述为：%s" % err)
return
def upload_All_file(self, local_path, remote_path):
'''
从本地上传指定文件夹下的所有文件到ftp
:param local_path:
:param remote_path:
:return:
'''
if len(os.listdir(local_path)) == 0:
self.debug_print("指定的目录不存在或为空：%s" % local_path)
print("指定的目录不存在或为空：%s" % local_path)
else:
for files in os.listdir(local_path):
self.upload_file(local_path + "/" + files, remote_path + '/')
def close(self):
'''
退出ftp
:return:
'''
print("close()---> FTP退出")
self.ftp.quit()
self.log_file.close()
def debug_print(self, s):
'''
打印日志
:param s:
:return:
'''
self.write_log(s)
def deal_error(self, e):
'''
处理错误异常
:param e:
:return:
'''
log_str = '发生错误: %s' % e
self.write_log(log_str)
sys.exit()
def write_log(self, log_str):
'''
记录日志
:param log_str:
:return:
'''
time_now = time.localtime()
date_now = time.strftime('%Y-%m-%d %H:%M:%S', time_now)
format_log_str = "%s ---> %s \r\n" % (date_now, log_str)
print(format_log_str)
self.log_file.write(format_log_str)
def get_file_list(self, line):
'''
获取文件列表
:param line:
:return:
'''
file_arr = self.get_file_name(line)
# 去除 . 和 ..
if file_arr[1] not in ['.', '..']:
self.file_list.append(file_arr)
def get_file_name(self, line):
'''
获取文件名
:param line:
:return:
'''
pos = line.rfind(':')
while (line[pos] != ' '):
pos += 1
while (line[pos] == ' '):
pos += 1
file_arr = [line[0], line[pos:]]
return file_arr
def ftp_dirs_exists(self, root_path, dir_name):
'''
判断ftp服务器上是否存在指定的文件夹,不存在则创建
:param root_path:
:param dir_name:
:return:
'''
ftplist = self.ftp.nlst(root_path)
real_path_dir = root_path + "/" + dir_name
# 判断该路径地址下是否存在文件夹，若不存在则创建指定格式的文件夹
if len(ftplist) == 0:
self.ftp.mkd(real_path_dir)
else:
# 遍历指定路径下非空文件夹下的所有文件信息
for dirs_name in ftplist:
# 如果指定路径下的文件夹信息和规定的文件夹信息一致，则不操作
if dirs_name == dir_name:
return
else:
# 否则创建规定的文件夹
self.ftp.mkd(real_path_dir)
if __name__ == '__main__':
my_ftp = FtpUtil('localhost', 21)
my_ftp.login("xxxx", "xxxx")
my_ftp.upload_file('E:\\test\\xxxx.png', '/media/test')
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment