Merge pull request #3961 from OmarFarrag/ftp_files#3928

Add FTPFileStore to FilesPipeline

docs/topics/media-pipeline.rst

@@ -147,6 +147,25 @@ Where:
 * ``full`` is a sub-directory to separate full images from thumbnails (if
   used). For more info see :ref:`topics-images-thumbnails`.
 
+FTP server storage
+------------------
+
+:setting:`FILES_STORE` and :setting:`IMAGES_STORE` can point to an FTP server.
+Scrapy will automatically upload the files to the server.
+
+:setting:`FILES_STORE` and :setting:`IMAGES_STORE` should be written in one of the
+following forms::
+
+    ftp://username:password@address:port/path
+    ftp://address:port/path
+    
+If ``username`` and ``password`` are not provided, they are taken from the :setting:`FTP_USER` and
+:setting:`FTP_PASSWORD` settings respectively.
+
+FTP supports two different connection modes: active or passive. Scrapy uses
+the passive connection mode by default. To use the active connection mode instead,
+set the :setting:`FEED_STORAGE_FTP_ACTIVE` setting to ``True``.
+
 Amazon S3 storage
 -----------------
 

scrapy/extensions/feedexport.py

@@ -7,18 +7,16 @@ See documentation in docs/topics/feed-exports.rst
 import os
 import sys
 import logging
-import posixpath
 from tempfile import NamedTemporaryFile
 from datetime import datetime
 from urllib.parse import urlparse, unquote
-from ftplib import FTP
 
 from zope.interface import Interface, implementer
 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_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
@@ -173,16 +171,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(
+            path=self.path, file=file, host=self.host,
+            port=self.port, username=self.username,
+            password=self.password, use_active_mode=self.use_active_mode
+        )
 
 
 class SpiderSlot(object):

scrapy/pipelines/files.py

@@ -11,6 +11,7 @@ import os
 import time
 from collections import defaultdict
 from email.utils import parsedate_tz, mktime_tz
+from ftplib import FTP
 from io import BytesIO
 from urllib.parse import urlparse
 
@@ -26,6 +27,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_store_file
 
 
 logger = logging.getLogger(__name__)
@@ -257,6 +259,49 @@ class GCSFilesStore(object):
         )
 
 
+class FTPFilesStore(object):
+
+    FTP_USERNAME = None
+    FTP_PASSWORD = None
+    USE_ACTIVE_MODE = None
+
+    def __init__(self, uri):
+        assert uri.startswith('ftp://')
+        u = urlparse(uri)
+        self.port = u.port
+        self.host = u.hostname
+        self.port = int(u.port or 21)
+        self.username = u.username or self.FTP_USERNAME
+        self.password = u.password or self.FTP_PASSWORD
+        self.basedir = u.path.rstrip('/')
+
+    def persist_file(self, path, buf, info, meta=None, headers=None):
+        path = '%s/%s' % (self.basedir, path)
+        return threads.deferToThread(
+            ftp_store_file, path=path, file=buf,
+            host=self.host, port=self.port, username=self.username,
+            password=self.password, use_active_mode=self.USE_ACTIVE_MODE
+        )
+
+    def stat_file(self, path, info):
+        def _stat_file(path):
+            try:
+                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()
+                ftp.retrbinary('RETR %s' % file_path, m.update)
+                return {'last_modified': last_modified, 'checksum': m.hexdigest()}
+            # The file doesn't exist
+            except Exception:
+                return {}
+        return threads.deferToThread(_stat_file, path)
+
+
 class FilesPipeline(MediaPipeline):
     """Abstract pipeline that implement the file downloading
 
@@ -283,6 +328,7 @@ class FilesPipeline(MediaPipeline):
         'file': FSFilesStore,
         's3': S3FilesStore,
         'gs': GCSFilesStore,
+        'ftp': FTPFilesStore
     }
     DEFAULT_FILES_URLS_FIELD = 'file_urls'
     DEFAULT_FILES_RESULT_FIELD = 'files'
@@ -330,6 +376,11 @@ class FilesPipeline(MediaPipeline):
         gcs_store.GCS_PROJECT_ID = settings['GCS_PROJECT_ID']
         gcs_store.POLICY = settings['FILES_STORE_GCS_ACL'] or None
 
+        ftp_store = cls.STORE_SCHEMES['ftp']
+        ftp_store.FTP_USERNAME = settings['FTP_USER']
+        ftp_store.FTP_PASSWORD = settings['FTP_PASSWORD']
+        ftp_store.USE_ACTIVE_MODE = settings.getbool('FEED_STORAGE_FTP_ACTIVE')
+
         store_uri = settings['FILES_STORE']
         return cls(store_uri, settings=settings)
 

scrapy/pipelines/images.py

@@ -94,6 +94,11 @@ class ImagesPipeline(FilesPipeline):
         gcs_store.GCS_PROJECT_ID = settings['GCS_PROJECT_ID']
         gcs_store.POLICY = settings['IMAGES_STORE_GCS_ACL'] or None
 
+        ftp_store = cls.STORE_SCHEMES['ftp']
+        ftp_store.FTP_USERNAME = settings['FTP_USER']
+        ftp_store.FTP_PASSWORD = settings['FTP_PASSWORD']
+        ftp_store.USE_ACTIVE_MODE = settings.getbool('FEED_STORAGE_FTP_ACTIVE')
+
         store_uri = settings['IMAGES_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
 
 
@@ -14,3 +16,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
+    """
+    with FTP() as 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)

scrapy/utils/test.py

@@ -1,6 +1,9 @@
 """
 This module contains some assorted functions used in tests
 """
+
+from __future__ import absolute_import
+from posixpath import split
 import asyncio
 import os
 
@@ -63,6 +66,26 @@ def get_gcs_content_and_delete(bucket, path):
     return content, acl, blob
 
 
+def get_ftp_content_and_delete(
+        path, host, port, username,
+        password, use_active_mode=False):
+    from ftplib import FTP
+    ftp = FTP()
+    ftp.connect(host, port)
+    ftp.login(username, password)
+    if use_active_mode:
+        ftp.set_pasv(False)
+    ftp_data = []
+
+    def buffer_data(data):
+        ftp_data.append(data)
+    ftp.retrbinary('RETR %s' % path, buffer_data)
+    dirname, filename = split(path)
+    ftp.cwd(dirname)
+    ftp.delete(filename)
+    return "".join(ftp_data)
+
+
 def get_crawler(spidercls=None, settings_dict=None):
     """Return an unconfigured Crawler object. If settings_dict is given, it
     will be used to populate the crawler settings with a project level

tests/test_pipeline_files.py

@@ -10,12 +10,13 @@ from urllib.parse import urlparse
 from twisted.trial import unittest
 from twisted.internet import defer
 
-from scrapy.pipelines.files import FilesPipeline, FSFilesStore, S3FilesStore, GCSFilesStore
+from scrapy.pipelines.files import FilesPipeline, FSFilesStore, S3FilesStore, GCSFilesStore, FTPFilesStore
 from scrapy.item import Item, Field
 from scrapy.http import Request, Response
 from scrapy.settings import Settings
 from scrapy.utils.test import assert_aws_environ, get_s3_content_and_delete
 from scrapy.utils.test import assert_gcs_environ, get_gcs_content_and_delete
+from scrapy.utils.test import get_ftp_content_and_delete
 from scrapy.utils.boto import is_botocore
 
 
@@ -367,6 +368,31 @@ class TestGCSFilesStore(unittest.TestCase):
         self.assertIn(expected_policy, acl)
 
 
+class TestFTPFileStore(unittest.TestCase):
+    @defer.inlineCallbacks
+    def test_persist(self):
+        uri = os.environ.get('FTP_TEST_FILE_URI')
+        if not uri:
+            raise unittest.SkipTest("No FTP URI available for testing")
+        data = b"TestFTPFilesStore: \xe2\x98\x83"
+        buf = BytesIO(data)
+        meta = {'foo': 'bar'}
+        path = 'full/filename'
+        store = FTPFilesStore(uri)
+        empty_dict = yield store.stat_file(path, info=None)
+        self.assertEqual(empty_dict, {})
+        yield store.persist_file(path, buf, info=None, meta=meta, headers=None)
+        stat = yield store.stat_file(path, info=None)
+        self.assertIn('last_modified', stat)
+        self.assertIn('checksum', stat)
+        self.assertEqual(stat['checksum'], 'd113d66b2ec7258724a268bd88eef6b6')
+        path = '%s/%s' % (store.basedir, path)
+        content = get_ftp_content_and_delete(
+            path, store.host, store.port,
+            store.username, store.password, store.USE_ACTIVE_MODE)
+        self.assertEqual(data.decode(), content)
+
+
 class ItemWithFiles(Item):
     file_urls = Field()
     files = Field()