2

当我尝试按顺序从 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 没有很多经验，所以如果你检查我的代码会非常好。我还有一些问题在这种情况下实现多处理的最佳方法是什么？

4

1 回答 1

2

我发现我的代码有问题。在downloadf函数中，并且恰好在此代码部分中

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))

问题是我使用了相同的 ftp 连接实例，xftp = self.get_conn()这就是为什么我得到的错误似乎没有描述源问题，我对代码进行了哪些更改来解决这个问题。解决方案是为每个进程实例化一个新的 ftp 连接，所以我现在执行以下代码，而不是上面的代码行：xftp = ftplib.FTP()

也许有一个善良的灵魂可以为这个问题带来更“pythonic”的解释。我将不胜感激

于 2017-03-09T14:30:08.153 回答