Make `stat_file` thread safe .. Refactor file storing.. Support act/psv

scrapy/extensions/feedexport.py

@@ -19,7 +19,7 @@ from twisted.internet import defer, threads
 from w3lib.url import file_uri_to_path
 
 from scrapy import signals
-from scrapy.utils.ftp import ftp_makedirs_cwd
+from scrapy.utils.ftp import ftp_makedirs_cwd, ftp_store_file
 from scrapy.exceptions import NotConfigured
 from scrapy.utils.misc import create_instance, load_object
 from scrapy.utils.log import failure_to_exc_info
@@ -174,16 +174,11 @@ class FTPFeedStorage(BlockingFeedStorage):
         )
 
     def _store_in_thread(self, file):
-        file.seek(0)
-        ftp = FTP()
-        ftp.connect(self.host, self.port)
-        ftp.login(self.username, self.password)
-        if self.use_active_mode:
-            ftp.set_pasv(False)
-        dirname, filename = posixpath.split(self.path)
-        ftp_makedirs_cwd(ftp, dirname)
-        ftp.storbinary('STOR %s' % filename, file)
-        ftp.quit()
+        ftp_store_file(
+            self.path, file, self.host,
+            self.port, self.username, 
+            self.password, self.use_active_mode
+        )
 
 
 class SpiderSlot(object):

scrapy/pipelines/files.py

@@ -32,7 +32,7 @@ from scrapy.utils.python import to_bytes
 from scrapy.utils.request import referer_str
 from scrapy.utils.boto import is_botocore
 from scrapy.utils.datatypes import CaselessDict
-from scrapy.utils.ftp import ftp_makedirs_cwd
+from scrapy.utils.ftp import ftp_makedirs_cwd, ftp_store_file
 
 logger = logging.getLogger(__name__)
 
@@ -254,6 +254,7 @@ class FTPFilesStore(object):
 
     FTP_USERNAME = None
     FTP_PASSWORD = None
+    USE_ACTIVE_MODE = None
     
     def __init__(self, uri):
         assert uri.startswith('ftp://')
@@ -265,31 +266,26 @@ class FTPFilesStore(object):
         self.password = u.password or self.FTP_PASSWORD
         self.basedir = u.path.rstrip('/')
         
-    def persist_file(self, path, buf, info, meta=None, headers=None):
-        
-        def _persist_file(path, buf):
-            ftp = FTP()
-            ftp.connect(self.host, self.port)
-            ftp.login(self.username, self.password)
-            buf.seek(0)
-            # If the path is like 'x/y/z.ext' the 'x/y' is rel_path and 
-            # 'z.ext' is file name
-            # If path is only the file name 'z.ext', then rel_path is
-            # the empty string and filename is 'z.ext'
-            x = path.rsplit('/',1)
-            rel_path, filename = ('/' + x[0].lstrip('/'), x[1]) if len(x) > 1 else ('', x[0])
-            abs_path = self.basedir + rel_path
-            ftp_makedirs_cwd(ftp, abs_path)
-            ftp.storbinary('STOR %s' % filename, buf)
-
-        return threads.deferToThread(_persist_file, path, buf)
+    def persist_file(self, path, buf, info, meta=None, headers=None):     
+        path = '%s/%s' % (self.basedir, path)
+        return threads.deferToThread(
+            ftp_store_file, path,buf,
+            self.host, self.port,self.username,
+            self.password, self.USE_ACTIVE_MODE
+        )
             
     def stat_file(self, path, info):
         def _stat_file(path):
             try:
-                last_modified = float(self.ftp.voidcmd("MDTM " + self.basedir + '/' + path)[4:].strip())
+                ftp = FTP()
+                ftp.connect(self.host, self.port)
+                ftp.login(self.username, self.password)
+                if self.USE_ACTIVE_MODE:
+                    ftp.set_pasv(False)
+                file_path = "%s/%s" % (self.basedir, path)
+                last_modified = float(ftp.voidcmd("MDTM %s" % file_path)[4:].strip())
                 m = hashlib.md5()
-                self.ftp.retrbinary('RETR %s' % self.basedir + path, m.update)
+                ftp.retrbinary('RETR %s' % file_path, m.update)
                 return {'last_modified': last_modified, 'checksum': m.hexdigest()}
             # The file doesn't exist
             except Exception as e :
@@ -372,6 +368,7 @@ class FilesPipeline(MediaPipeline):
         ftp_store = cls.STORE_SCHEMES['ftp']
         ftp_store.FTP_USERNAME = settings['FTP_USER']           # Default is 'anonymous'
         ftp_store.FTP_PASSWORD = settings['FTP_PASSWORD']       # Default is `guest`
+        ftp_store.USE_ACTIVE_MODE = settings.getbool('FEED_STORAGE_FTP_ACTIVE')
         
         store_uri = settings['FILES_STORE']
         return cls(store_uri, settings=settings)

scrapy/utils/ftp.py

@@ -1,4 +1,6 @@
-from ftplib import error_perm
+import posixpath
+
+from ftplib import error_perm, FTP
 from posixpath import dirname
 
 def ftp_makedirs_cwd(ftp, path, first_call=True):
@@ -13,3 +15,20 @@ def ftp_makedirs_cwd(ftp, path, first_call=True):
         ftp.mkd(path)
         if first_call:
             ftp.cwd(path)
+
+def ftp_store_file( 
+        path, file, host ,port,
+        username, password, use_active_mode=False):
+    """Opens a FTP connection with passed credentials,sets current directory
+    to the directory extracted from given path, then uploads the file to server
+    """
+    ftp = FTP()
+    ftp.connect(host, port)
+    ftp.login(username, password)
+    if use_active_mode:
+        ftp.set_pasv(False)
+    file.seek(0)
+    dirname, filename = posixpath.split(path)
+    ftp_makedirs_cwd(ftp, dirname)
+    ftp.storbinary('STOR %s' % filename, file)
+    ftp.quit()