From 31e497e5ebd13f1a376873be27028b6e8b592a64 Mon Sep 17 00:00:00 2001 From: Daniel Grana Date: Mon, 15 Sep 2008 04:33:40 +0000 Subject: [PATCH] add media_to_download hook support, and return cached result if available --HG-- extra : convert_revision : svn%3Ab85faa78-f9eb-468e-a121-7cced6da292c%40231 --- scrapy/trunk/scrapy/contrib/pipeline/media.py | 63 ++++++++++--------- scrapy/trunk/scrapy/utils/misc.py | 16 ++--- 2 files changed, 43 insertions(+), 36 deletions(-) diff --git a/scrapy/trunk/scrapy/contrib/pipeline/media.py b/scrapy/trunk/scrapy/contrib/pipeline/media.py index 6346ffae3..3214ae324 100644 --- a/scrapy/trunk/scrapy/contrib/pipeline/media.py +++ b/scrapy/trunk/scrapy/contrib/pipeline/media.py @@ -8,7 +8,7 @@ from scrapy.spider import spiders from scrapy.core.exceptions import DropItem, NotConfigured from scrapy.core.exceptions import HttpException from scrapy.stats import stats -from scrapy.utils.misc import chain_deferred, mustbe_deferred +from scrapy.utils.misc import chain_deferred, mustbe_deferred, defer_result from scrapy.conf import settings @@ -19,6 +19,7 @@ class DomainInfo(object): self.downloading = {} self.downloaded = {} self.waiting = {} + self.extra = {} class MediaPipeline(object): @@ -29,49 +30,55 @@ class MediaPipeline(object): urls = self.get_urls_from_item(item) info = self.cache[domain] - ulist = [] + lst = [] for url in urls or (): dfd = self._enqueue(url, info) dfd.addCallbacks( callback=self.new_item_media, - callbackArgs=(item,), + callbackArgs=(item, url), errback=self.failed_item_media, - errbackArgs=(item,), + errbackArgs=(item, url), ) - ulist.append(dfd) + lst.append(dfd) - dlist = defer.DeferredList(ulist, consumeErrors=False) - dlist.addBoth(self._item_completed, item) - return dlist + dlst = defer.DeferredList(lst, consumeErrors=False) + dlst.addBoth(lambda _: self.item_completed(item)) + return dlst def _enqueue(self, url, info): - wad = defer.Deferred() + request = self.media_to_download(url, info) + if not isinstance(request, Request): + return defer_result(request) - request = Request(url=url) fp = request.fingerprint() - waiting = info.waiting.setdefault(fp, []).append(wad) + if fp in info.downloaded: + return defer_result(info.downloaded[fp]) if fp not in info.downloading: - dwld = scrapyengine.schedule(request, info.spider, priority=0) - dwld.addCallbacks(self.media_downloaded, self.media_failure) - dwld.addBoth(self._download_finished, info, fp) - info.downloading[fp] = (request, dwld) + self._download(request, info, fp) + + wad = defer.Deferred() + waiting = info.waiting.setdefault(fp, []).append(wad) return wad + def _download(self, request, info, fp): + dwld = mustbe_deferred(self.request(request, info)) + dwld.addCallbacks(self.media_downloaded, self.media_failure) + dwld.addBoth(self._download_finished, info, fp) + info.downloading[fp] = (request, dwld) + def _download_finished(self, result, info, fp): - del info.downloading[fp] info.downloaded[fp] = result # cache result - waiting = info.waiting[fp] + del info.downloading[fp] + + waiting = info.waiting[fp] # client list del info.waiting[fp] - isfail = isinstance(result, failure.Failure) for wad in waiting: - tocall = wad.errback if isfail else wad.callback - tocall(result) - return result + defer_result(result).chainDeferred(wad) - def _item_completed(self, result, item): - return self.item_completed(item) #dummy as hell + def request(self, request, info): + return scrapyengine.schedule(request, info.spider, priority=0) def open_domain(self, domain): self.cache[domain] = DomainInfo(domain) @@ -80,12 +87,11 @@ class MediaPipeline(object): del self.cache[domain] ### Overradiable Interface - def get_urls_from_item(self, item): return item.image_urls - def media_to_download(self, url): - pass + def media_to_download(self, url, info): + return Request(url=url) def media_downloaded(self, response): pass @@ -93,10 +99,10 @@ class MediaPipeline(object): def media_failure(self, _failure): return _failure - def new_item_media(self, result, item): + def new_item_media(self, result, item, url): pass - def failed_item_media(self, _failure, item): + def failed_item_media(self, _failure, item, url): pass def item_completed(self, item): @@ -104,3 +110,4 @@ class MediaPipeline(object): + diff --git a/scrapy/trunk/scrapy/utils/misc.py b/scrapy/trunk/scrapy/utils/misc.py index 2b1514382..f7ae7953d 100644 --- a/scrapy/trunk/scrapy/utils/misc.py +++ b/scrapy/trunk/scrapy/utils/misc.py @@ -42,22 +42,22 @@ def defer_succeed(result): reactor.callLater(0, d.callback, result) return d +def defer_result(result): + if isinstance(result, defer.Deferred): + return result + elif isinstance(result, failure.Failure): + return defer_fail(result) + else: + return defer_succeed(result) def mustbe_deferred(f, *args, **kw): """same as twisted.internet.defer.maybeDeferred, but delay calling callback/errback""" - deferred = None try: result = f(*args, **kw) except: return defer_fail(failure.Failure()) else: - if isinstance(result, defer.Deferred): - return result - elif isinstance(result, failure.Failure): - return defer_fail(result) - else: - return defer_succeed(result) - return deferred + return defer_result(result) def chain_deferred(d1, d2):