Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- 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')
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement