python FTPS下载文件

python FTPS下载文件

关于FTP大家应该不陌生，那么FTPS是什么东西呢？ 简单一句话概括：类似于http 和https的关系。也就是说FTPS就是加密过的FTP。

公司项目需要，需要从使用FTPS来下载装置保存的各种文件，所以研究了下FTPS。

首先介绍一下，python有专门ftp相关的第三方库：ftplib 关于ftplib的信息网上一搜一大把，百度一下就知道其包含哪些方法，这里不做赘述。

ftplib中有两个大类：FTP 和 FTP_TLS,其中FTP_TLS的类，是专门用于FTPS的；该类其实也是继承了FTP类的；也就是FTP的子类，这里大家可以看下源码即可了解。

下面直接上代码吧，边看边解释：

在使用之前需要先导入FTP_TLS

from ftplib import FTP_TLS

 由于连接时报错，所以需要在源码中connect方法中添加一行：

self.af = self.sock.family
self.sock = ssl.wrap_socket(self.sock, self.keyfile, self.certfile, ssl_version=ssl.PROTOCOL_TLSv1) #这行
self.file = self.sock.makefile('r', encoding=self.encoding)

为了便于项目使用就在项目代码中将该方法重写，代码如下：

class _FTPS(FTP_TLS):
	def __init__(self, *args):
		super().__init__(*args)

	def connect(self, host='', port=0, timeout=-999, source_address=None):
		'''The original connect function could not successfully connect our device, so we     
        reconstructed this function'''
		if host != '':
			self.host = host
		if port > 0:
			self.port = port
		if timeout != -999:
			self.timeout = timeout
		if source_address is not None:
			self.source_address = source_address
		self.sock = socket.create_connection((self.host, self.port), self.timeout, source_address=self.source_address)
		self.af = self.sock.family
		self.sock = ssl.wrap_socket(self.sock, self.keyfile, self.certfile, ssl_version=ssl.PROTOCOL_TLSv1)
		self.file = self.sock.makefile('r', encoding=self.encoding)
		self.welcome = self.getresp()

		return self.welcome

	def nlst(self, *args):
		'''The original nlst method in ftplib does not list complete files, so the nlst method is reconstructed here'''
		cmd = 'LIST'
		templist = []
		func = None

		if args[-1:] and type(args[-1]) != type(''):
			args, func = args[:-1], args[-1]
		for arg in args:
			if arg:
				cmd = cmd + (' ' + arg)
		self.retrlines(cmd, templist.append)

		return templist

在项目代码中重写了_FTPS这个类，继承了FTP_TLS；同时呢重写的connect方法和nlst方法。

重写connect方法是为了加之前提到的那一行代码。

基础工作完成之后就可以封装FTPS相关的函数给项目使用了，大概有以下几个函数：

 

第一个是建立ftps连接的函数：

很简单，没啥可介绍的，其中添加的一个self._acess类属性是为了其他函数使用的，目的是为了验证该账号是否有ftps的登录权限（因为装置根据不同角色做了权限限制）

	def open_ftps_connection(self, host=None, username=None, password=None, port=990):
		host = host if host is not None else self._device.active_device().ipAddress
		username = username if username is not None else self._device.active_device().rbacUsername
		password = password if password is not None else self._device.active_device().rbacPassword

		self.ftps = _FTPS()
		self.ftps.connect(host=host, port=port)

		try:
			self.ftps.login(username, password)
			self.ftps.prot_p()
			self._access = True
		except Exception as err:
			if 'Not logged in' in str(err):
				self._access = False
			logger.error('Login ftp server error :{}'.format(err))
			self.close_ftps_connection()

		return self.ftps

然后是关闭ftps连接的函数：

更简单，就是调用了ftp已有的方法（其实就是发送了一条‘QUIT’命令）

	def close_ftps_connection(self):
		self.ftps.quit()

再接下来就是获取文件列表的函数：

这里就是跟大家各自项目相关的了，大家参考下即可

	def get_file_list_from_ftps(self, remoteFolder):
		self.ftps.cwd(remoteFolder)

		if _DeviceKeywords.active_device().deviceType in [SupportedDevice.Q100]:
			files = self.ftps.nlst()
			fileNames = [file[-12:] for file in files]
		elif _DeviceKeywords.active_device().deviceType in [SupportedDevice.Q200]:
			pass # TODO: The structure of Q200 file has not been determined yet, this part needs to be changed

		return fileNames

再接下来就是开始下载文件了：

很简单不多做介绍，就是传入文件名称，下载到什么路径，然后调用retrbinary方法即可

	def download_file_from_ftps(self, filePath, destFilePath=None, bufsize=1024):
		if destFilePath is None:
			targetFolder = self._common.get_output_folder()
			filename = pathlib.PurePath(filePath).name
			destFilePath = targetFolder / filename
		else:
			destFilePath = pathlib.Path(destFilePath)

		if destFilePath.exists():
			raise IOError('File {:s} already exists in test case file folder'.format(destFilePath.name))

		try:
			with open(destFilePath, 'wb') as f:
				self.ftps.retrbinary('RETR {}'.format(filePath), f.write, bufsize)
		except Exception as err:
			logger.error('Download file error :{}'.format(err))

		return destFilePath

 最后呢，就是验证用户是否有权限登录ftps；因为我司项目中有多个角色的成员，admin，viewer，manager 等等，不同角色权限不同，有的角色是没有ftp权限的，所以这里写了两个函数，便于再robotframework中写case。

	def validate_user_can_access_ftps(self, username, password):
		logger.info('Validating user can access ftps with name:{} password:{}'.format(username, password))

		self.open_ftps_connection(username=username, password=password)

		if self._access:
			logger.info('User is able to access ftps server.')
			self.close_ftps_connection()
		else:
			raise AssertionError('User is unable to access ftps server.')

	def validate_user_can_not_access_ftps(self, username, password):
		logger.info('Validating user can not access ftps with name:{} password:{}'.format(username, password))

		self.open_ftps_connection(username=username, password=password)

		if self._access:
			self.close_ftps_connection()
			raise AssertionError('User is able to access ftps server.')
		else:
			logger.info('User is unable to access ftps server.')

总结：其实关于FTPS的 *** 作没有太复杂的东西，因为ftp的方法可以直接看源码，即可直接使用了，我这里无非是根据项目需要做了一层封装，然后便于写case。如果对大家有帮助，那就再好不过了。

哎，拖延症患者终于再2021年最后一天把几个月前做的一个小项目总结了一下。。