process result's items using generators to give pipeline a chance to consume while parsing new items

--HG--
extra : convert_revision : svn%3Ab85faa78-f9eb-468e-a121-7cced6da292c%40135
This commit is contained in:
Daniel Grana 2008-07-29 12:27:55 +00:00
parent 640e8b9131
commit 28bb53fa22
8 changed files with 46 additions and 44 deletions

View File

@ -33,4 +33,4 @@ class DepthMiddleware(object):
if self.stats and response.request.depth == 0: # otherwise we loose stats for depth=0 if self.stats and response.request.depth == 0: # otherwise we loose stats for depth=0
stats.incpath('%s/request_depth_count/0' % spider.domain_name) stats.incpath('%s/request_depth_count/0' % spider.domain_name)
return [r for r in result or () if _filter(r)] return (r for r in result or () if _filter(r))

View File

@ -9,5 +9,5 @@ class OffsiteMiddleware(object):
log.msg("Ignoring link (offsite): %s " % r.url, level=log.DEBUG, domain=spider.domain_name) log.msg("Ignoring link (offsite): %s " % r.url, level=log.DEBUG, domain=spider.domain_name)
return False return False
return True return True
return [r for r in result or () if _filter(r)] return (r for r in result or () if _filter(r))

View File

@ -8,5 +8,5 @@ class CrawlMiddleware(object):
if isinstance(r, Request): if isinstance(r, Request):
r.headers.setdefault('Referer', response.url) r.headers.setdefault('Referer', response.url)
return r return r
return [_set_referer(r) for r in result or ()] return (_set_referer(r) for r in result or ())

View File

@ -17,5 +17,5 @@ class RestrictMiddleware(object):
if isinstance(r, Request) and r.url not in self.allowed_urls: if isinstance(r, Request) and r.url not in self.allowed_urls:
return False return False
return True return True
return [r for r in result or () if _filter(r)] return (r for r in result or () if _filter(r))

View File

@ -19,4 +19,4 @@ class UrlLengthMiddleware(object):
else: else:
return True return True
return [r for r in result or () if _filter(r)] return (r for r in result or () if _filter(r))

View File

@ -17,7 +17,7 @@ from scrapy.item import ScrapedItem
from scrapy.item.pipeline import ItemPipeline from scrapy.item.pipeline import ItemPipeline
from scrapy.spider import spiders from scrapy.spider import spiders
from scrapy.spider.middleware import SpiderMiddlewareManager from scrapy.spider.middleware import SpiderMiddlewareManager
from scrapy.utils.misc import chain_deferred, defer_succeed, mustbe_deferred from scrapy.utils.misc import chain_deferred, defer_succeed, mustbe_deferred, deferred_degenerate
from scrapy.conf import settings from scrapy.conf import settings
@ -238,18 +238,24 @@ class ExecutionEngine(object):
signals.send_catch_log(signal=signals.item_passed, sender=self.__class__, item=item, spider=spider, response=response, pipe_output=pipe_result) signals.send_catch_log(signal=signals.item_passed, sender=self.__class__, item=item, spider=spider, response=response, pipe_output=pipe_result)
self.next_request(spider) self.next_request(spider)
def _onsuccess_per_item(item):
if isinstance(item, ScrapedItem):
log.msg("Scraped %s in <%s>" % (item, request.url), log.DEBUG, domain=domain)
signals.send_catch_log(signal=signals.item_scraped, sender=self.__class__, item=item, spider=spider, response=response)
piped = self.pipeline.pipe(item, spider, response) # TODO: remove response
piped.addBoth(_onpipelinefinish, item)
elif isinstance(item, Request):
signals.send_catch_log(signal=signals.request_received, sender=self.__class__, request=item, spider=spider, response=response)
self.crawl(request=item, spider=spider, priority=priority)
else:
log.msg('Garbage found in spider output while processing %s, got type %s' % (request, type(item)), log.TRACE, domain=domain)
class _ResultContainer(object):
def append(self, item):
_onsuccess_per_item(item)
def _onsuccess(result): def _onsuccess(result):
for item in result or (): return deferred_degenerate(result, _ResultContainer())
if isinstance(item, ScrapedItem):
log.msg("Scraped %s in <%s>" % (item, request.url), log.DEBUG, domain=domain)
signals.send_catch_log(signal=signals.item_scraped, sender=self.__class__, item=item, spider=spider, response=response)
piped = self.pipeline.pipe(item, spider, response) # TODO: remove response
piped.addBoth(_onpipelinefinish, item)
elif isinstance(item, Request):
signals.send_catch_log(signal=signals.request_received, sender=self.__class__, request=item, spider=spider, response=response)
self.crawl(request=item, spider=spider, priority=priority)
else:
log.msg('Garbage found in spider output while processing %s, got type %s' % (request, type(item)), log.TRACE, domain=domain)
def _onerror(_failure): def _onerror(_failure):
if not isinstance(_failure.value, IgnoreRequest): if not isinstance(_failure.value, IgnoreRequest):
@ -259,10 +265,13 @@ class ExecutionEngine(object):
def _bugtrap(_failure): def _bugtrap(_failure):
log.msg('FRAMEWORK BUG processing %s: %s' % (request, _failure), log.ERROR, domain=domain) log.msg('FRAMEWORK BUG processing %s: %s' % (request, _failure), log.ERROR, domain=domain)
scd = self.scrape(request, response, spider) scd = self.scrape(request, response, spider)
scd.addCallbacks(_onsuccess, _onerror) scd.addCallbacks(_onsuccess, _onerror)
scd.addErrback(_bugtrap) scd.addErrback(_bugtrap)
scd.addBoth(lambda _:response)
self._scraping[domain].add(response)
scd.addBoth(lambda _: self._scraping[domain].remove(response))
return scd return scd
def _cleanfailure(_failure): def _cleanfailure(_failure):
@ -277,13 +286,7 @@ class ExecutionEngine(object):
return schd return schd
def scrape(self, request, response, spider): def scrape(self, request, response, spider):
domain = spider.domain_name return self.spidermiddleware.scrape(request, response, spider)
scd = self.spidermiddleware.scrape(request, response, spider)
self._scraping[domain].add(response)
def _remove(_):
self._scraping[domain].remove(response)
return _
return scd.addBoth(_remove)
def schedule(self, request, spider, priority=1, domain_priority=1): def schedule(self, request, spider, priority=1, domain_priority=1):
domain = spider.domain_name domain = spider.domain_name

View File

@ -2,15 +2,20 @@
Spider middleware manager Spider middleware manager
""" """
import types
from twisted.python.failure import Failure from twisted.python.failure import Failure
from scrapy.core import log from scrapy.core import log
from scrapy.core.exceptions import NotConfigured from scrapy.core.exceptions import NotConfigured
from scrapy.utils.misc import load_class, mustbe_deferred, deferred_degenerate from scrapy.utils.misc import load_class, mustbe_deferred
from scrapy.conf import settings from scrapy.conf import settings
def _isiterable(possible_iterator):
try:
return iter(possible_iterator)
except TypeError:
return None
class SpiderMiddlewareManager(object): class SpiderMiddlewareManager(object):
def __init__(self, callback=None, errback=None): def __init__(self, callback=None, errback=None):
self.loaded = False self.loaded = False
@ -55,8 +60,8 @@ class SpiderMiddlewareManager(object):
def process_scrape(response): def process_scrape(response):
for method in self.spider_middleware: for method in self.spider_middleware:
result = method(response=response, spider=spider) result = method(response=response, spider=spider)
assert result is None or isinstance(result, (list, tuple)), \ assert result is None or _isiterable(result), \
'Middleware %s must returns None, list or tuple, got %s ' % \ 'Middleware %s must returns None or an iterable object, got %s ' % \
(fname(method), type(result)) (fname(method), type(result))
if result is not None: if result is not None:
return result return result
@ -66,8 +71,8 @@ class SpiderMiddlewareManager(object):
def process_result(result): def process_result(result):
for method in self.result_middleware: for method in self.result_middleware:
result = method(response=response, result=result, spider=spider) result = method(response=response, result=result, spider=spider)
assert isinstance(result, (list, tuple)), \ assert _isiterable(result), \
'Middleware %s must returns list or tuple, got %s ' % \ 'Middleware %s must returns an iterable object, got %s ' % \
(fname(method), type(result)) (fname(method), type(result))
return result return result
@ -75,20 +80,14 @@ class SpiderMiddlewareManager(object):
exception = _failure.value exception = _failure.value
for method in self.exception_middleware: for method in self.exception_middleware:
result = method(response=response, exception=exception, spider=spider) result = method(response=response, exception=exception, spider=spider)
assert result is None or isinstance(result, (list, tuple)), \ assert result is None or _isiterable(result), \
'Middleware %s must returns None, list or tuple, got %s ' % \ 'Middleware %s must returns None, or an iterable object, got %s ' % \
(fname(method), type(result)) (fname(method), type(result))
if result is not None: if result is not None:
return result return result
return _failure return _failure
def _degenerate(gen):
if isinstance(gen, types.GeneratorType):
return deferred_degenerate(gen)
return gen
dfd = mustbe_deferred(process_scrape, response) dfd = mustbe_deferred(process_scrape, response)
dfd.addCallback(_degenerate)
dfd.addErrback(process_exception) dfd.addErrback(process_exception)
dfd.addCallback(process_result) dfd.addCallback(process_result)
return dfd return dfd

View File

@ -111,19 +111,19 @@ def memoize(cache, hash):
return decorator return decorator
def deferred_degenerate(generator): def deferred_degenerate(generator, container=None, next_delay=0):
generator = iter(generator) generator = iter(generator)
deferred = defer.Deferred() deferred = defer.Deferred()
result = [] container = container or []
def _next(): def _next():
try: try:
result.append(generator.next()) container.append(generator.next())
except StopIteration: except StopIteration:
reactor.callLater(0, deferred.callback, result) reactor.callLater(0, deferred.callback, container)
except: except:
reactor.callLater(0, deferred.errback, failure.Failure()) reactor.callLater(0, deferred.errback, failure.Failure())
else: else:
reactor.callLater(0, _next) reactor.callLater(next_delay, _next)
_next() _next()
return deferred return deferred