并行文件从FTP下载

Question

当我尝试从ftp下载文件时，它可以完美地工作

import ftplib
import os
import logging
class pyFTPDownload(object):
    def __init__(self,
                 remote_host=None,
                 port=None,
                 username=None,
                 passwd=None,
                 input_folder=None,
                 output_folder=None,
                 ftp_conn_id=None,
                 timeout=10
                 ):
        super(pyFTPDownload, self).__init__()
        self.remote_host = remote_host
        self.port = port
        self.username = username
        self.passwd = passwd
        self.input_folder = input_folder
        self.output_folder = output_folder
        self.ftp_conn_id = ftp_conn_id
        self.timeout = timeout
        self.client = None
    def get_conn(self):
        if not self.client:
            logging.info('creating ftp client for conn_id: {0}'.format(self.ftp_conn_id))
            if not self.username:
                raise Exception("Missing required param: username")
            if not self.passwd:
                raise Exception("Missing required param: passwd")
            if not self.remote_host:
                raise Exception("Missing required param: remote_host")
            if not self.ftp_conn_id:
                self.ftp_conn_id = str(self.username) + '@' + str(self.remote_host) + ":" + (
                    str(self.port) if self.port else "")
            try:
                client = ftplib.FTP()
                client.connect(self.remote_host, (self.port if not self.port else None))
                client.login(self.username, self.passwd)
                self.client = client
            except ftplib.all_errors as remote_host_error:
                logging.error("Auth failed while connecting to host: {0}, error: {1}"
                              .format(self.remote_host, remote_host_error))
            except Exception as error:
                logging.error("Error connecting to host: {0}, error: {1}"
                              .format(self.remote_host, error))
        return self.client

    def get_file(self, input_folder, output_folder=None,file_ext=None,thread_nbr=1):
        #os.chdir(output_folder)
        ftp = self.get_conn()
        #print ftp.dir(input_folder)
        logging.debug(input_folder)
        if not os.path.exists(input_folder):
            try:
                os.makedirs(output_folder, 0755)
                logging.debug("{0} is created".format(output_folder))
            except OSError, e:
                logging.error("ERROR: {0}".format(e))
        logging.info(output_folder)
        ftp.cwd(input_folder)
        for filename in ftp.nlst(file_ext):  # Loop - looking for matching files
            try:
                logging.debug("filename {0}".format(filename))
                local_file=os.path.join(output_folder,filename)
                logging.debug("local_file {0}".format(local_file))
                with open(local_file, 'wb') as fhandle:
                    logging.debug('Getting ' + filename)  # for confort sake, shows the file that's being retrieved
                    ftp.retrbinary('RETR ' + filename, fhandle.write)
                    fhandle.close()
            except Exception as e:
                logging.error("could not download file:{0}, terminated with error {1}".format(filename,e))

但是当我尝试并行做同样的事情时，我会得到

[Errno 9] Bad file descriptor

或当我尝试击退以下两行

时

xftp.connect(self.remote_host,self.port)
xftp.login(self.username,self.passwd)

我得到此错误：需要一个浮点。但是没有堆栈，所以我可以调试

我的完整代码如下：

    # coding=utf-8
from itertools import izip, repeat
import ftplib
import os
import multiprocessing
from pathos.multiprocessing import ProcessingPool as Pool
import logging
import traceback

class pyFTPDownload(object):
    def __init__(self,
                 remote_host,
                 port,
                 username,
                 passwd,
                 ftp_conn_id=None
                 ):
        super(pyFTPDownload, self).__init__()
        self.remote_host = remote_host
        self.port = port
        self.username = username
        self.passwd = passwd
        self.ftp_conn_id = ftp_conn_id
        self.client = None
        if not self.client:
            logging.info('creating ftp client for conn_id: {0}'.format(self.ftp_conn_id))
            if not self.username:
                raise Exception("Missing required param: username")
            if not self.passwd:
                raise Exception("Missing required param: passwd")
            if not self.remote_host:
                raise Exception("Missing required param: remote_host")
            if not self.ftp_conn_id:
                self.ftp_conn_id = str(self.username) + '@' + str(self.remote_host) + ":" + (
                    str(self.port) if self.port else "")
            try:
                client = ftplib.FTP()
                client.connect(self.remote_host, (self.port if self.port else None))
                client.login(self.username, self.passwd)
                self.client = client
            except ftplib.all_errors as remote_host_error:
                logging.error("Auth failed while connecting to host: {0}, error: {1}"
                              .format(self.remote_host, remote_host_error))
            except Exception as error:
                logging.error("Error connecting to host: {0}, error: {1}"
                              .format(self.remote_host, error))

    def get_conn(self):
        return self.client

class loadData(pyFTPDownload):
    def __init__(self,
                 remote_host,
                 port,
                 username,
                 passwd,
                 input_folder,
                 output_folder,
                 file_ext=None,
                 nbr_processes=None,
                 ftp_conn_id = None):
        super(loadData, self).__init__(remote_host,port,username,passwd)
        self.input_folder=input_folder
        self.output_folder=output_folder
        self.file_ext=file_ext
        self.nbr_processes=nbr_processes

        if not input_folder:
            raise Exception("Missing required params: input_folder")
        if not output_folder:
            raise Exception("Missing required params: output_folder")
        if not file_ext:
            logging.warn("All the existing files in {0} will be considered".format(input_folder))
        if not nbr_processes:
            logging.warn("The number of processes to be started will be set to {0}".format(Pool.ncpus))
            self.nbr_processes=multiprocessing.cpu_count()

    def downloadfunc(self,a):
        return self.downloadf(*a)
    def downloadf(self, inputf, filename, outputf):
        global xftp
        global local_file
        global fhandle
        print filename
        try:
            xftp = self.get_conn()
            xftp.connect(self.remote_host,self.port)
            xftp.login(self.username,self.passwd)
            print xftp
        except ftplib.all_errors as remote_host_error:
            logging.error("Auth failed while connecting to host: {0}, error: {1}"
                          .format(self.remote_host, remote_host_error))
        except Exception as error:
            logging.error("Error connecting to host: {0}, error: {1}"
                          .format(self.remote_host, error))
        try:
            logging.debug("filename {0}".format(filename))
            local_file = os.path.join(outputf, filename)
            logging.debug("local_file {0}".format(local_file))
        except Exception as sd:
            logging.error("Unkmown error: {}".format(sd))
        xftp.cwd(inputf)
        try:
           with open(local_file, 'wb') as fhandle:
                logging.debug('Getting ' + filename)  # for confort sake, shows the file that's being retrieved
                xftp.retrbinary('RETR ' + filename, fhandle.write)
                fhandle.close()
        except Exception as k:
          logging.error("Could not download {0} : {1}".format(local_file,k))
        finally:
            xftp.quit()
        print traceback

    def get_file(self):
        print "PREPARING FILE DOWNLOAD"
        print self.output_folder
        if not os.path.exists(self.output_folder):
            try:
                logging.debug("{} does not exists".format(self.output_folder))
                os.makedirs(self.output_folder,0755)
                logging.debug("{0} is created".format(self.output_folder))
            except OSError, e:
                logging.error("ERROR:{0} could not be created {1}, {2}".format(self.output_folder,e,OSError))
            except Exception as d:
                logging.error(d)
        ftpObj=self.get_conn()
        ftpObj.cwd(self.input_folder)
        files_to_dl=ftpObj.nlst(self.file_ext)
        p = Pool(self.nbr_processes) 
        try:
            p.map(self.downloadfunc, izip(repeat(self.input_folder),files_to_dl,repeat(self.output_folder)))
            p.close()
            p.join()
        except Exception as f:
            logging.error(f)

我对Python没有很多经验，因此检查我的代码非常好。我还有一些问题在这种情况下实施多处理的最佳方法是什么？

Answer 1

我发现了我的代码问题。在下载函数中，恰好在此代码部分

try:
        xftp = self.get_conn()
        xftp.connect(self.remote_host,self.port)
        xftp.login(self.username,self.passwd)
        print xftp
    except ftplib.all_errors as remote_host_error:
        logging.error("Auth failed while connecting to host: {0}, error: {1}"
                      .format(self.remote_host, remote_host_error))
    except Exception as error:
        logging.error("Error connecting to host: {0}, error: {1}"
                      .format(self.remote_host, error))

问题在于我通过xftp = self.get_conn()使用了相同的FTP连接实例，这就是为什么我发现的错误似乎没有描述源问题，我对代码进行了什么更改来解决该问题。解决方案是为每个过程实例化新的FTP连接，因此，我现在要进行folloy而不是上述代码行：xftp = ftplib.FTP()

也许有一个善良的灵魂可以为这个问题带来更多的" Pythonic"解释。我会被抓住