diff --git a/scrapy/contrib/pipeline/media.py b/scrapy/contrib/pipeline/media.py index ebaf7b672..2d0cd5053 100644 --- a/scrapy/contrib/pipeline/media.py +++ b/scrapy/contrib/pipeline/media.py @@ -1,3 +1,4 @@ +from collections import defaultdict from twisted.internet.defer import Deferred, DeferredList from scrapy.utils.defer import mustbe_deferred, defer_result @@ -8,20 +9,22 @@ from scrapy.utils.misc import arg_to_iter class MediaPipeline(object): - DOWNLOAD_PRIORITY = 1000 LOG_FAILED_RESULTS = True class SpiderInfo(object): def __init__(self, spider): self.spider = spider - self.downloading = {} + self.downloading = set() self.downloaded = {} - self.waiting = {} + self.waiting = defaultdict(list) - def __init__(self): + def __init__(self, download_func=None): self.spiderinfo = {} + self._download_func = download_func or self._default_download_func() + + def _default_download_func(self): from scrapy.project import crawler - self.crawler = crawler + return crawler.engine.download def open_spider(self, spider): self.spiderinfo[spider] = self.SpiderInfo(spider) @@ -32,63 +35,53 @@ class MediaPipeline(object): def process_item(self, item, spider): info = self.spiderinfo[spider] requests = arg_to_iter(self.get_media_requests(item, info)) - dlist = [self._enqueue(r, info) for r in requests] + dlist = [self._process_request(r, info) for r in requests] dfd = DeferredList(dlist, consumeErrors=1) return dfd.addCallback(self.item_completed, item, info) - def _enqueue(self, request, info): + def _process_request(self, request, info): fp = request_fingerprint(request) cb = request.callback or (lambda _: _) eb = request.errback - # if already downloaded, return cached result. + # Return cached result if request was already seen if fp in info.downloaded: return defer_result(info.downloaded[fp]).addCallbacks(cb, eb) + # Otherwise, wait for result wad = Deferred().addCallbacks(cb, eb) - # add to pending list for this request, and wait for result like the others. - info.waiting.setdefault(fp, []).append(wad) + info.waiting[fp].append(wad) - # if request is not downloading, download it. - if fp not in info.downloading: - self._download(request, info, fp) + # Check if request is downloading right now to avoid doing it twice + if fp in info.downloading: + return wad - return wad - - def _download(self, request, info, fp): - def _downloaded(result): - info.downloading.pop(fp) - info.downloaded[fp] = result - for wad in info.waiting.pop(fp): # pass result to each waiting client - defer_result(result).chainDeferred(wad) - - def _post_media_to_download(result): - if result is None: # continue with download - dwld = mustbe_deferred(self.download, request, info) - dwld.addCallbacks( - callback=self.media_downloaded, - callbackArgs=(request, info), - errback=self.media_failed, - errbackArgs=(request, info)) - else: # or use media_to_download return value as result - dwld = defer_result(result) - - info.downloading[fp] = (request, dwld) # fill downloading state data - dwld.addBoth(_downloaded) # append post-download hook - dwld.addErrback(log.err, spider=info.spider) - - # declare request in downloading state (None is used as place holder) - info.downloading[fp] = None - - # defer pre-download request processing + # Download request checking media_to_download hook output first + info.downloading.add(fp) dfd = mustbe_deferred(self.media_to_download, request, info) - dfd.addCallback(_post_media_to_download) + dfd.addCallback(self._check_media_to_download, request, info) + dfd.addBoth(self._cache_result_and_execute_waiters, fp, info) + dfd.addErrback(log.err, spider=info.spider) + return dfd.addBoth(lambda _: wad) # it must return wad at last + + def _check_media_to_download(self, result, request, info): + if result is not None: + return result + # Download request and process its response + return mustbe_deferred(self.download, request, info).addCallbacks( + callback=self.media_downloaded, callbackArgs=(request, info), + errback=self.media_failed, errbackArgs=(request, info)) + + def _cache_result_and_execute_waiters(self, result, fp, info): + info.downloading.remove(fp) + info.downloaded[fp] = result # cache result + for wad in info.waiting.pop(fp): + defer_result(result).chainDeferred(wad) ### Overradiable Interface def download(self, request, info): """Defines how to download the media request""" - request.priority = self.DOWNLOAD_PRIORITY - return self.crawler.engine.download(request, info.spider) + return self._download_func(request, info.spider) def media_to_download(self, request, info): """Check request before starting download""" @@ -109,8 +102,8 @@ class MediaPipeline(object): def item_completed(self, results, item, info): """Called per item when all media requests has been processed""" if self.LOG_FAILED_RESULTS: - for success, result in results: - if not success: - log.err(result, '%s found errors proessing %s' % (self.__class__.__name__, item)) + msg = '%s found errors proessing %s' % (self.__class__.__name__, item) + for ok, value in results: + if not ok: + log.err(value, msg, spider=info.spider) return item - diff --git a/scrapy/tests/test_pipeline_media.py b/scrapy/tests/test_pipeline_media.py index a9731e6bd..64b506820 100644 --- a/scrapy/tests/test_pipeline_media.py +++ b/scrapy/tests/test_pipeline_media.py @@ -1,85 +1,252 @@ from twisted.trial import unittest -from twisted.python import failure -from twisted.internet import defer, reactor +from twisted.python.failure import Failure +from twisted.internet import reactor +from twisted.internet.defer import Deferred, inlineCallbacks +from twisted.python import log as txlog -from scrapy.settings import Settings -from scrapy.crawler import Crawler from scrapy.http import Request, Response from scrapy.spider import BaseSpider from scrapy.utils.request import request_fingerprint from scrapy.contrib.pipeline.media import MediaPipeline +from scrapy.utils.test import get_crawler +from scrapy import log -class _MockedMediaPipeline(MediaPipeline): - - def download(self, request, info): - delay = request.meta.get('delay') - response = request.meta.get('response') - if delay is None: - return response - else: - dfd = defer.Deferred() - reactor.callLater(delay, dfd.callback, None) - return dfd.addCallback(lambda _: response) - - def get_media_requests(self, item, info): - return item.get('requests') +def _mocked_download_func(request, info): + response = request.meta.get('response') + return response() if callable(response) else response -class MediaPipelineTestCase(unittest.TestCase): +class BaseMediaPipelineTestCase(unittest.TestCase): - pipeline_class = _MockedMediaPipeline + pipeline_class = MediaPipeline def setUp(self): - self.crawler = Crawler(Settings()) - self.crawler.install() self.spider = BaseSpider('media.com') - self.pipe = self.pipeline_class() + self.pipe = self.pipeline_class(download_func=_mocked_download_func) self.pipe.open_spider(self.spider) + self.info = self.pipe.spiderinfo[self.spider] def tearDown(self): self.pipe.close_spider(self.spider) - self.crawler.uninstall() - @defer.inlineCallbacks - def test_return_item_by_default(self): - item = dict(name='sofa') + def test_default_media_to_download(self): + request = Request('url') + assert self.pipe.media_to_download(request, self.info) is None + + def test_default_get_media_requests(self): + item = dict(name='name') + assert self.pipe.get_media_requests(item, self.info) is None + + def test_default_media_downloaded(self): + request = Request('url') + response = Response('url', body='') + assert self.pipe.media_downloaded(response, request, self.info) is response + + def test_default_media_failed(self): + request = Request('url') + fail = Failure(Exception()) + assert self.pipe.media_failed(fail, request, self.info) is fail + + def test_default_item_completed(self): + item = dict(name='name') + assert self.pipe.item_completed([], item, self.info) is item + + # Check that failures are logged by default + fail = Failure(Exception()) + results = [(True, 1), (False, fail)] + + events = [] + txlog.addObserver(events.append) + new_item = self.pipe.item_completed(results, item, self.info) + txlog.removeObserver(events.append) + self.flushLoggedErrors() + + assert new_item is item + assert len(events) == 1 + assert events[0]['logLevel'] == log.ERROR + assert events[0]['failure'] is fail + + # disable failure logging and check again + self.pipe.LOG_FAILED_RESULTS = False + events = [] + txlog.addObserver(events.append) + new_item = self.pipe.item_completed(results, item, self.info) + txlog.removeObserver(events.append) + self.flushLoggedErrors() + assert new_item is item + assert len(events) == 0 + + @inlineCallbacks + def test_default_process_item(self): + item = dict(name='name') new_item = yield self.pipe.process_item(item, self.spider) assert new_item is item - @defer.inlineCallbacks + def test_default_download_func(self): + crawler = get_crawler() + crawler.install() + crawler.configure() + try: + pipe = MediaPipeline() + assert pipe._download_func == crawler.engine.download + finally: + crawler.uninstall() + + +class MockedMediaPipeline(MediaPipeline): + + def __init__(self, *args, **kwargs): + super(MockedMediaPipeline, self).__init__(*args, **kwargs) + self._mockcalled = [] + + def download(self, request, info): + self._mockcalled.append('download') + return super(MockedMediaPipeline, self).download(request, info) + + def media_to_download(self, request, info): + self._mockcalled.append('media_to_download') + return super(MockedMediaPipeline, self).media_to_download(request, info) + + def get_media_requests(self, item, info): + self._mockcalled.append('get_media_requests') + return item.get('requests') + + def media_downloaded(self, response, request, info): + self._mockcalled.append('media_downloaded') + return super(MockedMediaPipeline, self).media_downloaded(response, request, info) + + def media_failed(self, failure, request, info): + self._mockcalled.append('media_failed') + return super(MockedMediaPipeline, self).media_failed(failure, request, info) + + def item_completed(self, results, item, info): + self._mockcalled.append('item_completed') + item = super(MockedMediaPipeline, self).item_completed(results, item, info) + item['results'] = results + return item + + +class MediaPipelineTestCase(BaseMediaPipelineTestCase): + + pipeline_class = MockedMediaPipeline + + @inlineCallbacks + def test_result_succeed(self): + cb = lambda _: self.pipe._mockcalled.append('request_callback') or _ + eb = lambda _: self.pipe._mockcalled.append('request_errback') or _ + rsp = Response('url1') + req = Request('url1', meta=dict(response=rsp), callback=cb, errback=eb) + item = dict(requests=req) + new_item = yield self.pipe.process_item(item, self.spider) + self.assertEqual(new_item['results'], [(True, rsp)]) + self.assertEqual(self.pipe._mockcalled, + ['get_media_requests', 'media_to_download', 'download', + 'media_downloaded', 'request_callback', 'item_completed']) + + @inlineCallbacks + def test_result_failure(self): + self.pipe.LOG_FAILED_RESULTS = False + cb = lambda _: self.pipe._mockcalled.append('request_callback') or _ + eb = lambda _: self.pipe._mockcalled.append('request_errback') or _ + fail = Failure(Exception()) + req = Request('url1', meta=dict(response=fail), callback=cb, errback=eb) + item = dict(requests=req) + new_item = yield self.pipe.process_item(item, self.spider) + self.assertEqual(new_item['results'], [(False, fail)]) + self.assertEqual(self.pipe._mockcalled, + ['get_media_requests', 'media_to_download', 'download', + 'media_failed', 'request_errback', 'item_completed']) + + @inlineCallbacks + def test_mix_of_success_and_failure(self): + self.pipe.LOG_FAILED_RESULTS = False + rsp1 = Response('url1') + req1 = Request('url1', meta=dict(response=rsp1)) + fail = Failure(Exception()) + req2 = Request('url2', meta=dict(response=fail)) + item = dict(requests=[req1, req2]) + new_item = yield self.pipe.process_item(item, self.spider) + self.assertEqual(new_item['results'], [(True, rsp1), (False, fail)]) + m = self.pipe._mockcalled + # only once + self.assertEqual(m[0], 'get_media_requests') # first hook called + self.assertEqual(m.count('get_media_requests'), 1) + self.assertEqual(m.count('item_completed'), 1) + self.assertEqual(m[-1], 'item_completed') # last hook called + # twice, one per request + self.assertEqual(m.count('media_to_download'), 2) + self.assertEqual(m.count('download'), 2) + # one to handle success and other for failure + self.assertEqual(m.count('media_downloaded'), 1) + self.assertEqual(m.count('media_failed'), 1) + + @inlineCallbacks def test_get_media_requests(self): # returns single Request (without callback) - info = self.pipe.spiderinfo[self.spider] - req = Request('http://media.com/2.gif') + req = Request('url') item = dict(requests=req) # pass a single item new_item = yield self.pipe.process_item(item, self.spider) assert new_item is item - assert request_fingerprint(req) in info.downloaded + assert request_fingerprint(req) in self.info.downloaded # returns iterable of Requests - req1 = Request('http://media.com/1.gif') - req2 = Request('http://media.com/1.jpg') + req1 = Request('url1') + req2 = Request('url2') item = dict(requests=iter([req1, req2])) new_item = yield self.pipe.process_item(item, self.spider) assert new_item is item - assert info.downloaded.get(request_fingerprint(req1)) is None - assert info.downloaded.get(request_fingerprint(req2)) is None + assert request_fingerprint(req1) in self.info.downloaded + assert request_fingerprint(req2) in self.info.downloaded - @defer.inlineCallbacks - def test_requets_callback_is_called(self): - collected = [] - response = Response('http://media.com/2.gif') - request = Request('http://media.com/2.gif', meta=dict(response=response), callback=collected.append) - item = dict(requests=request) # pass a single item - yield self.pipe.process_item(item, self.spider) - assert collected == [response] + @inlineCallbacks + def test_results_are_cached_across_multiple_items(self): + rsp1 = Response('url1') + req1 = Request('url1', meta=dict(response=rsp1)) + item = dict(requests=req1) + new_item = yield self.pipe.process_item(item, self.spider) + self.assertTrue(new_item is item) + self.assertEqual(new_item['results'], [(True, rsp1)]) - @defer.inlineCallbacks - def test_requets_errback_is_called(self): - collected = [] - fail = failure.Failure(Exception()) - req = Request('http://media.com/2.gif', meta=dict(response=fail), callback=lambda _:_, errback=collected.append) - item = dict(requests=req) # pass a single item - yield self.pipe.process_item(item, self.spider) - assert collected == [fail] + # rsp2 is ignored, rsp1 must be in results because request fingerprints are the same + req2 = Request(req1.url, meta=dict(response=Response('http://donot.download.me'))) + item = dict(requests=req2) + new_item = yield self.pipe.process_item(item, self.spider) + self.assertTrue(new_item is item) + self.assertEqual(request_fingerprint(req1), request_fingerprint(req2)) + self.assertEqual(new_item['results'], [(True, rsp1)]) + + @inlineCallbacks + def test_results_are_cached_for_requests_of_single_item(self): + rsp1 = Response('url1') + req1 = Request('url1', meta=dict(response=rsp1)) + req2 = Request(req1.url, meta=dict(response=Response('http://donot.download.me'))) + item = dict(requests=[req1, req2]) + new_item = yield self.pipe.process_item(item, self.spider) + self.assertTrue(new_item is item) + self.assertEqual(new_item['results'], [(True, rsp1), (True, rsp1)]) + + @inlineCallbacks + def test_wait_if_request_is_downloading(self): + def _check_downloading(response): + fp = request_fingerprint(req1) + assert fp in self.info.downloading, self.info.downloading + assert fp in self.info.waiting, self.info.waiting + assert len(self.info.waiting[fp]) == 2, self.info.waiting + assert fp not in self.info.downloaded, self.info.downloaded + return response + + rsp1 = Response('url') + def rsp1_func(): + dfd = Deferred().addCallback(_check_downloading) + reactor.callLater(.1, dfd.callback, rsp1) + return dfd + + def rsp2_func(): + assert False, 'This can not be called' + + req1 = Request('url', meta=dict(response=rsp1_func)) + req2 = Request(req1.url, meta=dict(response=rsp2_func)) + item = dict(requests=[req1, req2]) + new_item = yield self.pipe.process_item(item, self.spider) + self.assertEqual(new_item['results'], [(True, rsp1), (True, rsp1)])