From 00b49752ceae6b1a8529c8fff48341f762ae48a8 Mon Sep 17 00:00:00 2001 From: Daniel Grana Date: Sat, 20 Jun 2009 18:15:02 -0300 Subject: [PATCH] Multiples changes to core scheduling and duplicates filtering * removed starters from engine * moved schedulermiddleware to scheduler * raise RuntimeError when trying to open/close a scheduler domain twice * removed dupesfilter singleton and spidermw dupefilter middleware --HG-- extra : rebase_source : e4c3ad4b970cbc8f532bc751ba9d8a944ca16be5 --- .../schedulermiddleware/duplicatesfilter.py | 20 ++++++-- .../spidermiddleware/duplicatesfilter.py | 23 ---------- scrapy/core/engine.py | 39 +++------------- scrapy/core/scheduler/middleware.py | 46 +++++++++++-------- scrapy/core/scheduler/schedulers.py | 28 +++++++---- scrapy/dupefilter/__init__.py | 17 ------- .../test_downloadermiddleware_redirect.py | 5 -- ...st_schedulermiddleware_duplicatesfilter.py | 19 ++++---- .../test_spidermiddleware_duplicatesfilter.py | 39 ---------------- 9 files changed, 79 insertions(+), 157 deletions(-) delete mode 100644 scrapy/contrib/spidermiddleware/duplicatesfilter.py delete mode 100644 scrapy/tests/test_spidermiddleware_duplicatesfilter.py diff --git a/scrapy/contrib/schedulermiddleware/duplicatesfilter.py b/scrapy/contrib/schedulermiddleware/duplicatesfilter.py index 2753d4fae..dff52cae4 100644 --- a/scrapy/contrib/schedulermiddleware/duplicatesfilter.py +++ b/scrapy/contrib/schedulermiddleware/duplicatesfilter.py @@ -2,14 +2,26 @@ DuplicatesFilterMiddleware: Filter out already visited urls """ -from scrapy.core.exceptions import IgnoreRequest -from scrapy.dupefilter import dupefilter - +from scrapy.core.exceptions import IgnoreRequest, NotConfigured +from scrapy.utils.misc import load_object +from scrapy.conf import settings class DuplicatesFilterMiddleware(object): """Filter out already seen requests to avoid visiting pages more than once.""" + def __init__(self): + clspath = settings.get('DUPEFILTER_FILTERCLASS') + if not clspath: + raise NotConfigured + + self.dupefilter = load_object(clspath)() def enqueue_request(self, domain, request): - added = dupefilter.add(domain, request) + added = self.dupefilter.add(domain, request) if not (added or request.dont_filter): raise IgnoreRequest('Skipped (already seen request)') + + def open_domain(self, domain): + self.dupefilter.open(domain) + + def close_domain(self, domain): + self.dupefilter.close(domain) diff --git a/scrapy/contrib/spidermiddleware/duplicatesfilter.py b/scrapy/contrib/spidermiddleware/duplicatesfilter.py deleted file mode 100644 index 51e99d211..000000000 --- a/scrapy/contrib/spidermiddleware/duplicatesfilter.py +++ /dev/null @@ -1,23 +0,0 @@ -""" -DuplicatesFilterMiddleware: Filter out already visited urls -""" - -from scrapy.http import Request -from scrapy.dupefilter import dupefilter -from scrapy import log - - -class DuplicatesFilterMiddleware(object): - """Filter out already seen requests to avoid visiting pages more than once.""" - - def process_spider_output(self, response, result, spider): - domain = spider.domain_name - for req in result: - if isinstance(req, Request): - has = dupefilter.has(domain, req) - if has and not req.dont_filter: - log.msg('Skipped (already processed): %s' % req, log.TRACE, domain=domain) - continue - yield req - - diff --git a/scrapy/core/engine.py b/scrapy/core/engine.py index e4ad7fab4..e7dbe3db2 100644 --- a/scrapy/core/engine.py +++ b/scrapy/core/engine.py @@ -15,7 +15,7 @@ from scrapy import log from scrapy.stats import stats from scrapy.conf import settings from scrapy.core import signals -from scrapy.core.scheduler import Scheduler, SchedulerMiddlewareManager +from scrapy.core.scheduler import Scheduler from scrapy.core.downloader import Downloader from scrapy.core.exceptions import IgnoreRequest, DontCloseDomain from scrapy.http import Response, Request @@ -64,14 +64,11 @@ class ExecutionEngine(object): Configure execution engine with the given scheduling policy and downloader. """ self.scheduler = scheduler or Scheduler() - self.schedulermiddleware = SchedulerMiddlewareManager(self.scheduler) self.domain_scheduler = load_object(settings['DOMAIN_SCHEDULER'])() self.downloader = downloader or Downloader(self) self.spidermiddleware = SpiderMiddlewareManager() self._scraping = {} self.pipeline = ItemPipelineManager() - # key dictionary of per domain lists of initial requests to scrape - self.starters = {} self.configured = True @@ -291,14 +288,11 @@ class ExecutionEngine(object): def schedule(self, request, spider): domain = spider.domain_name if not self.scheduler.domain_is_open(domain): - if self.debug_mode: - log.msg('Scheduling %s (delayed)' % request_info(request), log.DEBUG) - return self._add_starter(request, spider) - if self.debug_mode: - log.msg('Scheduling %s (now)' % request_info(request), log.DEBUG) - schd = self.schedulermiddleware.enqueue_request(domain, request) + self.scheduler.open_domain(domain) + if not self.downloader.domain_is_open(domain): + self.domain_scheduler.add_domain(domain) self.next_request(spider) - return schd + return self.scheduler.enqueue_request(domain, request) def _mainloop(self): """Add more domains to be scraped if the downloader has the capacity. @@ -313,23 +307,6 @@ class ExecutionEngine(object): if not self.next_domain(): return self._stop_if_idle() - def _add_starter(self, request, spider): - domain = spider.domain_name - if not self.domain_scheduler.has_pending_domain(domain): - self.domain_scheduler.add_domain(domain) - self.starters[domain] = [] - deferred = defer.Deferred() - self.starters[domain] += [(request, deferred)] - return deferred - - def _run_starters(self, spider): - domain = spider.domain_name - starters = self.starters.pop(domain, []) - while starters: - request, deferred = starters.pop(0) - schd = self.schedule(request, spider) - chain_deferred(schd, deferred) - def download(self, request, spider): if self.debug_mode: log.msg('Downloading %s' % request_info(request), log.DEBUG) @@ -372,16 +349,15 @@ class ExecutionEngine(object): def open_domain(self, domain, spider=None): log.msg("Domain opened", domain=domain) spider = spider or spiders.fromdomain(domain) + self.next_request(spider) self.cancelled.discard(domain) - self.scheduler.open_domain(domain) self.downloader.open_domain(domain) self.pipeline.open_domain(domain) self._scraping[domain] = set() - signals.send_catch_log(signals.domain_open, sender=self.__class__, domain=domain, spider=spider) + signals.send_catch_log(signals.domain_open, sender=self.__class__, domain=domain, spider=spider) signals.send_catch_log(signals.domain_opened, sender=self.__class__, domain=domain, spider=spider) - self._run_starters(spider) def _domain_idle(self, domain): """Called when a domain gets idle. This function is called when there are no @@ -433,7 +409,6 @@ class ExecutionEngine(object): signals.send_catch_log(signal=signals.domain_closed, sender=self.__class__, domain=domain, spider=spider, status=status) log.msg("Domain closed (%s)" % status, domain=domain) self.cancelled.discard(domain) - self.starters.pop(domain, None) self._mainloop() def getstatus(self): diff --git a/scrapy/core/scheduler/middleware.py b/scrapy/core/scheduler/middleware.py index 6972e859c..aa5254af4 100644 --- a/scrapy/core/scheduler/middleware.py +++ b/scrapy/core/scheduler/middleware.py @@ -5,6 +5,7 @@ For more information see the Scheduler Middleware doc in: docs/topics/scheduler-middleware.rst """ +from collections import defaultdict from twisted.internet.defer import Deferred from scrapy import log @@ -17,19 +18,13 @@ from scrapy.conf import settings class SchedulerMiddlewareManager(object): - def __init__(self, scheduler): - self.loaded = False - self.scheduler = scheduler - self.mw_enqueue_request = [] + def __init__(self): + self.mw_cbs = defaultdict(list) self.load() - def _add_middleware(self, mw): - if hasattr(mw, 'enqueue_request'): - self.mw_enqueue_request.append(mw.enqueue_request) - def load(self): """Load middleware defined in settings module""" - mws = [] + self.mws = [] mwlist = build_middleware_list(settings['SCHEDULER_MIDDLEWARES_BASE'], settings['SCHEDULER_MIDDLEWARES']) for mwpath in mwlist: @@ -37,23 +32,38 @@ class SchedulerMiddlewareManager(object): if cls: try: mw = cls() - self._add_middleware(mw) - mws.append(mw) except NotConfigured: pass - log.msg("Enabled scheduler middlewares: %s" % ", ".join([type(m).__name__ for m in mws])) - self.loaded = True + else: + self._add_middleware(mw) + self.mws.append(mw) - def enqueue_request(self, domain, request): + log.msg("Enabled scheduler middlewares: %s" % ", ".join([type(m).__name__ for m in self.mws])) + + def _add_middleware(self, mw): + for name in ('enqueue_request', 'open_domain', 'close_domain'): + mwfunc = getattr(mw, name, None) + if mwfunc: + self.mw_cbs[name].append(mwfunc) + + def enqueue_request(self, wrappedfunc, domain, request): def _enqueue_request(request): - for method in self.mw_enqueue_request: - result = method(domain=domain, request=request) + for mwfunc in self.mw_cbs['enqueue_request']: + result = mwfunc(domain=domain, request=request) assert result is None or isinstance(result, (Response, Deferred)), \ 'Middleware %s.enqueue_request must return None, Response or Deferred, got %s' % \ - (method.im_self.__class__.__name__, result.__class__.__name__) + (mwfunc.im_self.__class__.__name__, result.__class__.__name__) if result: return result - return self.scheduler.enqueue_request(domain=domain, request=request) + return wrappedfunc(domain=domain, request=request) deferred = mustbe_deferred(_enqueue_request, request) return deferred + + def open_domain(self, domain): + for mwfunc in self.mw_cbs['open_domain']: + mwfunc(domain) + + def close_domain(self, domain): + for mwfunc in self.mw_cbs['close_domain']: + mwfunc(domain) diff --git a/scrapy/core/scheduler/schedulers.py b/scrapy/core/scheduler/schedulers.py index 7428cd38b..8facf88e6 100644 --- a/scrapy/core/scheduler/schedulers.py +++ b/scrapy/core/scheduler/schedulers.py @@ -6,6 +6,7 @@ from twisted.internet import defer from scrapy.utils.datatypes import PriorityQueue, PriorityStack from scrapy.conf import settings +from .middleware import SchedulerMiddlewareManager class Scheduler(object): """The scheduler decides what to scrape next. In other words, it defines the @@ -18,6 +19,7 @@ class Scheduler(object): def __init__(self): self.pending_requests = {} self.dfo = settings['SCHEDULER_ORDER'].upper() == 'DFO' + self.middleware = SchedulerMiddlewareManager() def domain_is_open(self, domain): """Check if scheduler's resources were allocated for a domain""" @@ -30,13 +32,27 @@ class Scheduler(object): def open_domain(self, domain): """Allocates scheduling resources for the given domain""" + if domain in self.pending_requests: + raise RuntimeError('Scheduler domain already opened: %s' % domain) + Priority = PriorityStack if self.dfo else PriorityQueue self.pending_requests[domain] = Priority() + self.middleware.open_domain(domain) + + def close_domain(self, domain): + """Called when a spider has finished scraping to free any resources + associated with the domain. + """ + if domain not in self.pending_requests: + raise RuntimeError('Scheduler domain is not open: %s' % domain) + self.middleware.close_domain(domain) + self.pending_requests.pop(domain, None) def enqueue_request(self, domain, request): - """Enqueue a request to be downloaded for a domain that is currently - being scraped. - """ + """Enqueue a request to be downloaded for a domain that is currently being scraped.""" + return self.middleware.enqueue_request(self._enqueue_request, domain, request) + + def _enqueue_request(self, domain, request): dfd = defer.Deferred() self.pending_requests[domain].push((request, dfd), request.priority) return dfd @@ -55,12 +71,6 @@ class Scheduler(object): except (KeyError, IndexError): return (None, None) - def close_domain(self, domain): - """Called when a spider has finished scraping to free any resources - associated with the domain. - """ - self.pending_requests.pop(domain, None) - def is_idle(self): """Checks if the schedulers has any request pendings""" return not self.pending_requests diff --git a/scrapy/dupefilter/__init__.py b/scrapy/dupefilter/__init__.py index 54383dfff..c9763e645 100644 --- a/scrapy/dupefilter/__init__.py +++ b/scrapy/dupefilter/__init__.py @@ -1,19 +1,9 @@ -from pydispatch import dispatcher - -from scrapy.core import signals from scrapy.utils.request import request_fingerprint -from scrapy.utils.misc import load_object -from scrapy.conf import settings class BaseFilter(dict): """Base class defining the duplicates requests filtering api""" - def __init__(self, *args, **kwargs): - super(BaseFilter, self).__init__(*args, **kwargs) - dispatcher.connect(self.open, signals.domain_open) - dispatcher.connect(self.close, signals.domain_closed) - def open(self, domain): """Called when a domain starts""" raise NotImplementedError() @@ -74,10 +64,3 @@ class NullFilter(dict): return None -try: - dupefilter -except NameError: - clspath = settings.get('DUPEFILTER_FILTERCLASS') - cls = load_object(clspath) if clspath else NullFilter - dupefilter = cls() - diff --git a/scrapy/tests/test_downloadermiddleware_redirect.py b/scrapy/tests/test_downloadermiddleware_redirect.py index bba0d78a7..400c1feeb 100644 --- a/scrapy/tests/test_downloadermiddleware_redirect.py +++ b/scrapy/tests/test_downloadermiddleware_redirect.py @@ -1,7 +1,6 @@ import unittest from scrapy.contrib.downloadermiddleware.redirect import RedirectMiddleware -from scrapy.dupefilter import dupefilter from scrapy.spider import spiders from scrapy.http import Request, Response, Headers @@ -11,12 +10,8 @@ class RedirectMiddlewareTest(unittest.TestCase): spiders.spider_modules = ['scrapy.tests.test_spiders'] spiders.reload() self.spider = spiders.fromdomain('scrapytest.org') - dupefilter.open('scrapytest.org') self.mw = RedirectMiddleware() - def tearDown(self): - dupefilter.close('scrapytest.org') - def test_redirect_301(self): url = 'http://www.example.com/301' url2 = 'http://www.example.com/redirected' diff --git a/scrapy/tests/test_schedulermiddleware_duplicatesfilter.py b/scrapy/tests/test_schedulermiddleware_duplicatesfilter.py index 029d3fe08..cfb9c4e2d 100644 --- a/scrapy/tests/test_schedulermiddleware_duplicatesfilter.py +++ b/scrapy/tests/test_schedulermiddleware_duplicatesfilter.py @@ -3,28 +3,27 @@ import unittest from scrapy.http import Request from scrapy.core.exceptions import IgnoreRequest from scrapy.contrib.schedulermiddleware.duplicatesfilter import DuplicatesFilterMiddleware -from scrapy.dupefilter import dupefilter +DOMAIN = 'scrapytest.org' + class DuplicatesFilterMiddlewareTest(unittest.TestCase): def setUp(self): - dupefilter.open('scrapytest.org') + self.mw = DuplicatesFilterMiddleware() + self.mw.open_domain(DOMAIN) def tearDown(self): - dupefilter.close('scrapytest.org') + self.mw.close_domain(DOMAIN) def test_process_spider_output(self): - domain = 'scrapytest.org' - - mw = DuplicatesFilterMiddleware() r1 = Request('http://scrapytest.org/1') r2 = Request('http://scrapytest.org/2') r3 = Request('http://scrapytest.org/2') r4 = Request('http://scrapytest.org/1') - assert not mw.enqueue_request(domain, r1) - assert not mw.enqueue_request(domain, r2) - self.assertRaises(IgnoreRequest, mw.enqueue_request, domain, r3) - self.assertRaises(IgnoreRequest, mw.enqueue_request, domain, r4) + assert not self.mw.enqueue_request(DOMAIN, r1) + assert not self.mw.enqueue_request(DOMAIN, r2) + self.assertRaises(IgnoreRequest, self.mw.enqueue_request, DOMAIN, r3) + self.assertRaises(IgnoreRequest, self.mw.enqueue_request, DOMAIN, r4) diff --git a/scrapy/tests/test_spidermiddleware_duplicatesfilter.py b/scrapy/tests/test_spidermiddleware_duplicatesfilter.py deleted file mode 100644 index 3ae448068..000000000 --- a/scrapy/tests/test_spidermiddleware_duplicatesfilter.py +++ /dev/null @@ -1,39 +0,0 @@ -import unittest - -from scrapy.spider import spiders -from scrapy.http import Request, Response -from scrapy.core.exceptions import IgnoreRequest -from scrapy.contrib.spidermiddleware.duplicatesfilter import DuplicatesFilterMiddleware -from scrapy.dupefilter import dupefilter - -class DuplicatesFilterMiddlewareTest(unittest.TestCase): - - def setUp(self): - spiders.spider_modules = ['scrapy.tests.test_spiders'] - spiders.reload() - self.spider = spiders.fromdomain('scrapytest.org') - dupefilter.open('scrapytest.org') - - def tearDown(self): - dupefilter.close('scrapytest.org') - - def test_process_spider_output(self): - mw = DuplicatesFilterMiddleware() - - response = Response('http://scrapytest.org/') - response.request = Request('http://scrapytest.org/') - - r0 = Request('http://scrapytest.org/') - r1 = Request('http://scrapytest.org/1') - r2 = Request('http://scrapytest.org/2') - r3 = Request('http://scrapytest.org/2') - - dupefilter.add('scrapytest.org', r0) - dupefilter.add('scrapytest.org', r2) - - filtered = list(mw.process_spider_output(response, [r0, r1, r2, r3], self.spider)) - - assert r0 not in filtered - assert r1 in filtered - assert r2 not in filtered - assert r3 not in filtered