def info(remote):
try:
sfmstat = remote.readlines(os.path.join(remote.root, '.sfmstat'))
except ftplib.error_perm, err:
if not err[0].startswith('550'):
log(err, abort=True)
sfmstat = None
print
if sfmstat:
last_updated, mirror_path = sfmstat[0].split(None, 1)
last_updated = datetime.datetime.fromtimestamp(float(last_updated))
print 'Mirror of', mirror_path
print last_updated.strftime('Last updated on %A, %d. %B %Y at %H:%M:%S')
else:
print 'No mirror recognized'
print
print 'Content of %s%s:' % (remote.host, remote.root)
remote.ftp.dir(remote.root)
print
python类error_perm()的实例源码
def _get_files_in_dir(ftp_session, source, dest, status):
filelist = ftp_session.nlst()
for path in filelist:
local_path = os.path.join(dest, path)
remote_path = source + path
try:
# See if it's a directory
new_remote_path = source + path + "/"
ftp_session.cwd(source + path + "/")
# It is! So we should create the directory on dest
try:
os.mkdir(local_path)
except OSError:
logger.debug("Dir: {0} already exists".format(local_path))
logger.debug("Get Folder: {0}".format(new_remote_path))
_get_files_in_dir(ftp_session, new_remote_path, local_path, status)
except ftplib.error_perm:
logger.debug("Get file: {0}".format(path))
with open(local_path, "wb") as file_handle:
ftp_session.retrbinary(
"RETR " + remote_path, file_handle.write)
def download_files(ftp_session, source, dest, status):
# dest must already exist
try:
ftp_session.cwd(source)
except ftplib.error_perm:
# invalid entry (ensure input form: "/dir/folder/something/")
msg = "Could not open the ftp directory: '{0}'".format(source)
logger.error(msg)
status.update({"error": msg})
return
if not os.path.isdir(dest):
msg = "Local path does not exist: '{0}'".format(dest)
logger.error(msg)
status.update({"error": msg})
return
_get_files_in_dir(ftp_session, source, dest, status)
def FtpListCommand(self, directory = '/'):
ls = []
if directory[-1] == '/':
try:
self.ftpDirectory += directory
self.ftpDirectory.replace('//','/')
self.ftp.retrlines('LIST %s' % self.ftpDirectory, ls.append)
except error_perm, msg:
errorListingMessage.text = ""
parsedErrorMessage = (str(msg)).split(" ")[1:]
for item in parsedErrorMessage:
errorListingMessage.text = errorListingMessage.text + str(item) + " "
listingErrorPopup.open()
else:
self.ftp.retrlines('LIST', ls.append)
return self.ListParser(ls)
def test_manager(self):
mem_fs = open_fs('mem://')
with self.assertRaises(errors.ResourceError):
with ftp_errors(mem_fs, path='foo'):
raise error_temp
with self.assertRaises(errors.OperationFailed):
with ftp_errors(mem_fs):
raise error_temp
with self.assertRaises(errors.InsufficientStorage):
with ftp_errors(mem_fs):
raise error_perm('552 foo')
with self.assertRaises(errors.ResourceNotFound):
with ftp_errors(mem_fs):
raise error_perm('501 foo')
with self.assertRaises(errors.PermissionDenied):
with ftp_errors(mem_fs):
raise error_perm('999 foo')
def makedir(self, path, permissions=None, recreate=False):
_path = self.validatepath(path)
with ftp_errors(self, path=path):
if _path == '/':
if recreate:
return self.opendir(path)
else:
raise errors.DirectoryExists(path)
if not (recreate and self.isdir(path)):
try:
self.ftp.mkd(_encode(_path, self.ftp.encoding))
except error_perm as error:
code, _ = _parse_ftp_error(error)
if code == 550:
if self.isdir(path):
raise errors.DirectoryExists(path)
else:
if self.exists(path):
raise errors.DirectoryExists(path)
raise errors.ResourceNotFound(path)
return self.opendir(path)
def removedir(self, path):
_path = self.validatepath(path)
if _path == '/':
raise errors.RemoveRootError()
with ftp_errors(self, path):
try:
self.ftp.rmd(_encode(_path, self.ftp.encoding))
except error_perm as error:
code, _ = _parse_ftp_error(error)
if code == 550:
if self.isfile(path):
raise errors.DirectoryExpected(path)
if not self.isempty(path):
raise errors.DirectoryNotEmpty(path)
raise # pragma: no cover
def getbytes(self, path):
_path = self.validatepath(path)
data = io.BytesIO()
with ftp_errors(self, path):
with self._manage_ftp() as ftp:
try:
ftp.retrbinary(
str("RETR ") + _encode(_path, self.ftp.encoding),
data.write
)
except error_perm as error:
code, _ = _parse_ftp_error(error)
if code == 550:
if self.isdir(path):
raise errors.FileExpected(path)
raise
data_bytes = data.getvalue()
return data_bytes
def get_mtime(self, name):
""" Return modification time of the file in ftp server.
If the file does not exist, return None.
If the name refers to a directory, return 0.
To get the local modification time, use
``get_mtime()-clock_offset``.
"""
filename = self.abspath(name)
try:
resp = self.ftp.sendcmd('MDTM ' + filename.replace('\\', '/'))
except error_perm, msg:
s = str(msg)
if s.startswith('550 I can only retrieve regular files'):
# filename is directory
return 0
if s.startswith('550 Can\'t check for file existence'):
# file with filename does not exist
return None
raise
assert resp[:3]=='213', `resp, filename`
modtime = int(time.mktime(time.strptime(resp[3:].strip(), '%Y%m%d%H%M%S')))
return modtime
def ftp_walk(ftpconn: FTP, rootpath=''):
"""Recursively traverse an ftp directory to discovery directory listing."""
current_directory = rootpath
try:
directories, files = directory_listing(ftpconn, current_directory)
except ftplib.error_perm:
return
# Yield before recursion
yield current_directory, directories, files
# Recurse into sub-directories
for name in directories:
new_path = os.path.join(current_directory, name)
for entry in ftp_walk(ftpconn, rootpath=new_path):
yield entry
else:
return
def placeFiles(ftp, path):
"""Upload the built files to FTP"""
for name in os.listdir(path):
if name != "config.py" and name != "config.pyc" and name != "templates" and name != "content":
localpath = os.path.join(path, name)
if os.path.isfile(localpath):
print("STOR", name, localpath)
ftp.storbinary('STOR ' + name, open(localpath, 'rb'))
elif os.path.isdir(localpath):
print("MKD", name)
try:
ftp.mkd(name)
# ignore "directory already exists"
except error_perm as e:
if not e.args[0].startswith('550'):
raise
print("CWD", name)
ftp.cwd(name)
placeFiles(ftp, localpath)
print("CWD", "..")
ftp.cwd("..")
def test_rnfr_rnto(self):
# rename file
tempname = os.path.basename(tempfile.mktemp(dir=HOME))
self.client.rename(self.tempfile, tempname)
self.client.rename(tempname, self.tempfile)
# rename dir
tempname = os.path.basename(tempfile.mktemp(dir=HOME))
self.client.rename(self.tempdir, tempname)
self.client.rename(tempname, self.tempdir)
# rnfr/rnto over non-existing paths
bogus = os.path.basename(tempfile.mktemp(dir=HOME))
self.assertRaises(ftplib.error_perm, self.client.rename, bogus, '/x')
self.assertRaises(
ftplib.error_perm, self.client.rename, self.tempfile, u('/'))
# rnto sent without first specifying the source
self.assertRaises(ftplib.error_perm, self.client.sendcmd,
'rnto ' + self.tempfile)
# make sure we can't rename root directory
self.assertRaisesRegex(ftplib.error_perm,
"Can't rename home directory",
self.client.rename, '/', '/x')
def test_unforeseen_mdtm_event(self):
# Emulate a case where the file last modification time is prior
# to year 1900. This most likely will never happen unless
# someone specifically force the last modification time of a
# file in some way.
# To do so we temporarily override os.path.getmtime so that it
# returns a negative value referring to a year prior to 1900.
# It causes time.localtime/gmtime to raise a ValueError exception
# which is supposed to be handled by server.
# On python 3 it seems that the trick of replacing the original
# method with the lambda doesn't work.
if not PY3:
_getmtime = AbstractedFS.getmtime
try:
AbstractedFS.getmtime = lambda x, y: -9000000000
self.assertRaisesRegex(
ftplib.error_perm,
"550 Can't determine file's last modification time",
self.client.sendcmd, 'mdtm ' + self.tempfile)
# make sure client hasn't been disconnected
self.client.sendcmd('noop')
finally:
AbstractedFS.getmtime = _getmtime
def test_stou_orphaned_file(self):
# Check that no orphaned file gets left behind when STOU fails.
# Even if STOU fails the file is first created and then erased.
# Since we can't know the name of the file the best way that
# we have to test this case is comparing the content of the
# directory before and after STOU has been issued.
# Assuming that TESTFN is supposed to be a "reserved" file
# name we shouldn't get false positives.
safe_remove(TESTFN)
# login as a limited user in order to make STOU fail
self.client.login('anonymous', '@nopasswd')
before = os.listdir(HOME)
self.assertRaises(ftplib.error_perm, self.client.sendcmd,
'stou ' + TESTFN)
after = os.listdir(HOME)
if before != after:
for file in after:
self.assertFalse(file.startswith(TESTFN))
def test_epsv(self):
# test wrong proto
try:
self.client.sendcmd('epsv ' + self.other_proto)
except ftplib.error_perm as err:
self.assertEqual(str(err)[0:3], "522")
else:
self.fail("Exception not raised")
# proto > 2
self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'epsv 3')
# test connection
for cmd in ('EPSV', 'EPSV ' + self.proto):
host, port = ftplib.parse229(self.client.sendcmd(cmd),
self.client.sock.getpeername())
with contextlib.closing(
socket.socket(self.client.af, socket.SOCK_STREAM)) as s:
s.settimeout(TIMEOUT)
s.connect((host, port))
self.client.sendcmd('abor')
def test_mlst(self):
# utility function for extracting the line of interest
def mlstline(cmd):
return self.client.voidcmd(cmd).split('\n')[1]
if self.utf8fs:
self.assertTrue('type=dir' in
mlstline('mlst ' + TESTFN_UNICODE))
self.assertTrue('/' + TESTFN_UNICODE in
mlstline('mlst ' + TESTFN_UNICODE))
self.assertTrue('type=file' in
mlstline('mlst ' + TESTFN_UNICODE_2))
self.assertTrue('/' + TESTFN_UNICODE_2 in
mlstline('mlst ' + TESTFN_UNICODE_2))
else:
self.assertRaises(ftplib.error_perm,
mlstline, 'mlst ' + TESTFN_UNICODE)
# --- file transfer
def test_stor(self):
if self.utf8fs:
data = b'abcde12345' * 500
os.remove(TESTFN_UNICODE_2)
dummy = BytesIO()
dummy.write(data)
dummy.seek(0)
self.client.storbinary('stor ' + TESTFN_UNICODE_2, dummy)
dummy_recv = BytesIO()
self.client.retrbinary('retr ' + TESTFN_UNICODE_2,
dummy_recv.write)
dummy_recv.seek(0)
self.assertEqual(dummy_recv.read(), data)
else:
dummy = BytesIO()
self.assertRaises(ftplib.error_perm, self.client.storbinary,
'stor ' + TESTFN_UNICODE_2, dummy)
def test_prot(self):
self.client.login(secure=False)
msg = "503 PROT not allowed on insecure control connection."
self.assertRaisesWithMsg(ftplib.error_perm, msg,
self.client.sendcmd, 'prot p')
self.client.login(secure=True)
# secured
self.client.prot_p()
sock = self.client.transfercmd('list')
with contextlib.closing(sock):
while True:
if not sock.recv(1024):
self.client.voidresp()
break
self.assertTrue(isinstance(sock, ssl.SSLSocket))
# unsecured
self.client.prot_c()
sock = self.client.transfercmd('list')
with contextlib.closing(sock):
while True:
if not sock.recv(1024):
self.client.voidresp()
break
self.assertFalse(isinstance(sock, ssl.SSLSocket))
def ls_ftp(dir):
from urlparse import urlparse
from ftplib import FTP, error_perm
o = urlparse(dir)
ftp = FTP(o.netloc)
ftp.login()
ftp.cwd(o.path)
files = []
try:
files = ftp.nlst()
except error_perm, resp:
if str(resp) == "550 No files found":
print "no files in this directory"
else:
raise
return files
def cwd(self,path):
#-----------------------------------------------------------------------------
# Name: cwd fuction
# param: path :cd to the path,please write absolute path
# explain: cd to this path
# Author: gongke
#
# Created: 2013/01/13
#-----------------------------------------------------------------------------
try:
self.mftp.cwd(path)
except ftplib.error_perm:
#print "cannot cd to this path %s" %path
print_mes = "cannot cd to this path " +path
print print_mes
info_public(print_mes)
self.mftp.quit()
return False
#print "cd %s" %path
print_mes = "cd " +path
print print_mes
info_public(print_mes)
return True
def open(self, flags):
if flags in ('r',):
self._socket = self._conn.transfercmd('RETR %s' % self.url.path)
elif flags in ('w', 'w+'):
try:
self._socket = self._conn.transfercmd('STOR %s' % self.url.path)
except ftplib.error_perm:
raise FileExistsError(str(self.url))
def size(self):
try:
return self._conn.size(self.url.path)
except ftplib.error_perm as err:
if 'not a regular file' in str(err):
return None
raise FileNotFoundError(self.url)
except ValueError:
return None
def isdir(self):
try:
self._conn.size(self.url.path)
except ftplib.error_perm as err:
if "regular" in str(err):
return True
return False
def delete(self):
if self.isfile():
self._conn.delete(self.url.path)
elif self.isdir():
try:
self._conn.rmd(self.url.path)
except ftplib.error_perm as err:
raise OSError(err)
else:
raise FileNotFoundError(str(self.url))
def mkdir(self, name=None):
try:
if name is None:
if self.isdir():
return self
self._conn.mkd(self.url.path)
return self
else:
if self.join(name).isdir():
return self.join(name)
self._conn.mkd(name)
return self.join(name)
except ftplib.error_perm as err:
raise TransfertPermissionError(err)
def retrfile(self, file, type):
import ftplib
self.endtransfer()
if type in ('d', 'D'): cmd = 'TYPE A'; isdir = 1
else: cmd = 'TYPE ' + type; isdir = 0
try:
self.ftp.voidcmd(cmd)
except ftplib.all_errors:
self.init()
self.ftp.voidcmd(cmd)
conn = None
if file and not isdir:
# Try to retrieve as a file
try:
cmd = 'RETR ' + file
conn = self.ftp.ntransfercmd(cmd)
except ftplib.error_perm, reason:
if str(reason)[:3] != '550':
raise IOError, ('ftp error', reason), sys.exc_info()[2]
if not conn:
# Set transfer mode to ASCII!
self.ftp.voidcmd('TYPE A')
# Try a directory listing. Verify that directory exists.
if file:
pwd = self.ftp.pwd()
try:
try:
self.ftp.cwd(file)
except ftplib.error_perm, reason:
raise IOError, ('ftp error', reason), sys.exc_info()[2]
finally:
self.ftp.cwd(pwd)
cmd = 'LIST ' + file
else:
cmd = 'LIST'
conn = self.ftp.ntransfercmd(cmd)
self.busy = 1
# Pass back both a suitably decorated object and a retrieval length
return (addclosehook(conn[0].makefile('rb'),
self.endtransfer), conn[1])
def get_ftpfile_lastmodified(self, file_path):
'''return lastmodified for a given file on ftp server.'''
try:
response = self.client.sendcmd('MDTM ' + file_path)
except error_perm as e:
self.logger.debug("Skip %s: %s" % (file_path,e))
return None
code, lastmodified = response.split()
# an example: 'last-modified': '20121128150000'
lastmodified = int(time.mktime(datetime.strptime(lastmodified, '%Y%m%d%H%M%S').timetuple()))
return lastmodified
def main():
try:
f = ftplib.FTP(HOST)
except (socket.error, socket.gaierror) as e:
print('ERROR: connot reach "%s"' % HOST)
exit(1)
print('*** Connected to host "%s"' % HOST)
try:
f.login()
except ftplib.error_perm:
print('ERROR: connot login anonymously')
f.quit()
exit(1)
print('*** Logged in as "anonymous"')
try:
f.cwd(DIRN)
except ftplib.error_perm:
print('ERROR: cannot cd to "%s"' % DIRN)
f.quit()
exit(1)
print('*** Changed to "%s" folder' % DIRN)
try:
locFile = open(FILE, 'wb')
f.retrbinary('RETR %s' % FILE, locFile.write)
except ftplib.error_perm:
print('ERROR: cannot read file "%s"' % FILE)
os.unlink(FILE)
else:
print('*** Downloaded "%s" to CWD' % FILE)
finally:
locFile.close()
f.quit()
return
def get_directory_listing(ftp_con):
try:
files = ftp_con.nlst()
if "." in files: files.remove(".")
if ".." in files: files.remove("..")
rnd_files = random.sample(set(files), 3)
return "{} files listed - sample of files: {}".format(len(files), rnd_files)
except ftplib.error_perm as resp:
if "550" in resp:
return "No files"
finally:
con.quit()
con.close()
def test_exceptions(self):
self.assertRaises(ftplib.error_temp, self.client.sendcmd, 'echo 400')
self.assertRaises(ftplib.error_temp, self.client.sendcmd, 'echo 499')
self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'echo 500')
self.assertRaises(ftplib.error_perm, self.client.sendcmd, 'echo 599')
self.assertRaises(ftplib.error_proto, self.client.sendcmd, 'echo 999')
