From 8c0c605ea0d59355ece2e4f2755acb535ab7b90d Mon Sep 17 00:00:00 2001 From: François Schmidts Date: Thu, 5 Mar 2015 23:36:52 +0100 Subject: correcting wait counter and reactivating last_retrieved --- pyaggr3g470r/lib/crawler.py | 41 ++++++++++++++++++++++++++++++----------- 1 file changed, 30 insertions(+), 11 deletions(-) (limited to 'pyaggr3g470r/lib') diff --git a/pyaggr3g470r/lib/crawler.py b/pyaggr3g470r/lib/crawler.py index 64ef8b6d..97f14363 100644 --- a/pyaggr3g470r/lib/crawler.py +++ b/pyaggr3g470r/lib/crawler.py @@ -58,6 +58,7 @@ class AbstractCrawler: __counter__ = 0 def __init__(self, auth): + AbstractCrawler.__counter__ += 1 self.auth = auth self.session = self.get_session() self.url = conf.PLATFORM_URL @@ -84,6 +85,13 @@ class AbstractCrawler: return result return wrapper + @classmethod + def get_counter_callback(cls): + cls.__counter__ += 1 + def debump(*args, **kwargs): + cls.__counter__ -= 1 + return debump + def query_pyagg(self, method, urn, data=None): """A wrapper for internal call, method should be ones you can find on requests (header, post, get, options, ...), urn the distant @@ -95,13 +103,15 @@ class AbstractCrawler: return method("%s%s%s" % (self.url, API_ROOT, urn), auth=self.auth, data=json.dumps(data, default=default_handler), - headers={'Content-Type': 'application/json'}) + headers={'Content-Type': 'application/json', + 'User-Agent': 'pyaggr3g470r'}) @classmethod def wait(cls): "See count_on_me, that method will just wait for the counter to be 0" time.sleep(1) while cls.__counter__: + print('running %d' % cls.__counter__) time.sleep(1) @@ -144,6 +154,7 @@ class PyAggUpdater(AbstractCrawler): def callback(self, response): """Will process the result from the challenge, creating missing article and updating the feed""" + AbstractCrawler.__counter__ -= 1 results = response.result().json() logger.debug('%r %r - %d entries were not matched and will be created', self.feed['id'], self.feed['title'], len(results)) @@ -158,10 +169,12 @@ class PyAggUpdater(AbstractCrawler): self.feed['id'], self.feed['title'], self.headers.get('etag'), now) - self.query_pyagg('put', 'feed/%d' % self.feed['id'], {'error_count': 0, - 'etag': self.headers.get('etag', ''), - 'last_error': '', - 'last_modified': self.headers.get('last-modified', '')}) + dico = {'error_count': 0, 'last_error': '', + 'etag': self.headers.get('etag', ''), + 'last_modified': self.headers.get('last-modified', '')} + if any([dico[key] == self.feed.get(key) for key in dico]): + future = self.query_pyagg('put', 'feed/%d' % self.feed['id'], dico) + future.add_done_callback(self.get_counter_callback()) class FeedCrawler(AbstractCrawler): @@ -173,13 +186,15 @@ class FeedCrawler(AbstractCrawler): def clean_feed(self): """Will reset the errors counters on a feed that have known errors""" if self.feed.get('error_count') or self.feed.get('last_error'): - self.query_pyagg('put', 'feed/%d' % self.feed['id'], - {'error_count': 0, 'last_error': ''}) + future = self.query_pyagg('put', 'feed/%d' % self.feed['id'], + {'error_count': 0, 'last_error': ''}) + future.add_done_callback(self.get_counter_callback()) @AbstractCrawler.count_on_me def callback(self, response): """will fetch the feed and interprete results (304, etag) or will challenge pyagg to compare gotten entries with existing ones""" + AbstractCrawler.__counter__ -= 1 try: response = response.result() response.raise_for_status() @@ -188,9 +203,10 @@ class FeedCrawler(AbstractCrawler): logger.warn('%r %r - an error occured while fetching feed; bumping' ' error count to %r', self.feed['id'], self.feed['title'], error_count) - self.query_pyagg('put', 'feed/%d' % self.feed['id'], - {'error_count': error_count, - 'last_error': str(error)}) + future = self.query_pyagg('put', 'feed/%d' % self.feed['id'], + {'error_count': error_count, + 'last_error': str(error)}) + future.add_done_callback(self.get_counter_callback()) return if response.status_code == 304: @@ -222,10 +238,11 @@ class CrawlerScheduler(AbstractCrawler): def __init__(self, username, password): self.auth = (username, password) super(CrawlerScheduler, self).__init__(self.auth) + AbstractCrawler.__counter__ = 0 def prepare_headers(self, feed): """For a known feed, will construct some header dictionnary""" - headers = {} + headers = {'User-Agent': 'pyaggr3g470r/crawler'} if feed.get('etag', None): headers['If-None-Match'] = feed['etag'] if feed.get('last_modified'): @@ -237,6 +254,7 @@ class CrawlerScheduler(AbstractCrawler): @AbstractCrawler.count_on_me def callback(self, response): """processes feeds that need to be fetched""" + AbstractCrawler.__counter__ -= 1 response = response.result() response.raise_for_status() feeds = response.json() @@ -254,4 +272,5 @@ class CrawlerScheduler(AbstractCrawler): and launch the whole thing""" logger.debug('retreving fetchable feed') future = self.query_pyagg('get', 'feeds/fetchable', kwargs) + AbstractCrawler.__counter__ += 1 future.add_done_callback(self.callback) -- cgit From 822e59f043ba7b12962c5e65f59f2fd33a339f54 Mon Sep 17 00:00:00 2001 From: François Schmidts Date: Fri, 6 Mar 2015 11:07:43 +0100 Subject: better crawling crontrol --- pyaggr3g470r/lib/crawler.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) (limited to 'pyaggr3g470r/lib') diff --git a/pyaggr3g470r/lib/crawler.py b/pyaggr3g470r/lib/crawler.py index 97f14363..8e61b7cf 100644 --- a/pyaggr3g470r/lib/crawler.py +++ b/pyaggr3g470r/lib/crawler.py @@ -111,16 +111,16 @@ class AbstractCrawler: "See count_on_me, that method will just wait for the counter to be 0" time.sleep(1) while cls.__counter__: - print('running %d' % cls.__counter__) time.sleep(1) class PyAggUpdater(AbstractCrawler): - def __init__(self, feed, entries, headers, auth): + def __init__(self, feed, entries, headers, parsed_feed, auth): self.feed = feed self.entries = entries self.headers = headers + self.parsed_feed = parsed_feed.get('feed', {}) super(PyAggUpdater, self).__init__(auth) def to_article(self, entry): @@ -171,7 +171,10 @@ class PyAggUpdater(AbstractCrawler): dico = {'error_count': 0, 'last_error': '', 'etag': self.headers.get('etag', ''), - 'last_modified': self.headers.get('last-modified', '')} + 'last_modified': self.headers.get('last-modified', ''), + 'site_link': self.parsed_feed.get('link')} + if not self.feed.get('title'): + dico['title'] = self.parsed_feed.get('title', '') if any([dico[key] == self.feed.get(key) for key in dico]): future = self.query_pyagg('put', 'feed/%d' % self.feed['id'], dico) future.add_done_callback(self.get_counter_callback()) @@ -229,7 +232,8 @@ class FeedCrawler(AbstractCrawler): logger.debug('%r %r - found %d entries %r', self.feed['id'], self.feed['title'], len(ids), ids) future = self.query_pyagg('get', 'articles/challenge', {'ids': ids}) - updater = PyAggUpdater(self.feed, entries, response.headers, self.auth) + updater = PyAggUpdater(self.feed, entries, response.headers, + parsed_response, self.auth) future.add_done_callback(updater.callback) -- cgit From 0cc0e87d3f3bafba6a22c883cdf24e9962fafe37 Mon Sep 17 00:00:00 2001 From: François Schmidts Date: Tue, 10 Mar 2015 09:25:10 +0100 Subject: failover for bad counter manipulation --- pyaggr3g470r/lib/crawler.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) (limited to 'pyaggr3g470r/lib') diff --git a/pyaggr3g470r/lib/crawler.py b/pyaggr3g470r/lib/crawler.py index 8e61b7cf..9df37993 100644 --- a/pyaggr3g470r/lib/crawler.py +++ b/pyaggr3g470r/lib/crawler.py @@ -107,11 +107,17 @@ class AbstractCrawler: 'User-Agent': 'pyaggr3g470r'}) @classmethod - def wait(cls): + def wait(cls, max_wait=600): "See count_on_me, that method will just wait for the counter to be 0" time.sleep(1) + second_waited = 1 while cls.__counter__: + if second_waited > max_wait: + logger.warn('Exiting after %d seconds, counter at %d', + max_wait, cls.__counter__) + break time.sleep(1) + second_waited += 1 class PyAggUpdater(AbstractCrawler): -- cgit From 4142e50e512bf05bba7e47be232c3cc3ae4e32f5 Mon Sep 17 00:00:00 2001 From: François Schmidts Date: Tue, 10 Mar 2015 14:57:14 +0100 Subject: implementing cache construction on crawler side (limiting useless pushes) --- pyaggr3g470r/lib/crawler.py | 37 ++++++++++++++++++++++++++++--------- 1 file changed, 28 insertions(+), 9 deletions(-) (limited to 'pyaggr3g470r/lib') diff --git a/pyaggr3g470r/lib/crawler.py b/pyaggr3g470r/lib/crawler.py index 9df37993..99967671 100644 --- a/pyaggr3g470r/lib/crawler.py +++ b/pyaggr3g470r/lib/crawler.py @@ -19,6 +19,7 @@ import logging import requests import feedparser import dateutil.parser +from hashlib import md5 from functools import wraps from datetime import datetime from concurrent.futures import ThreadPoolExecutor @@ -29,6 +30,10 @@ logger = logging.getLogger(__name__) API_ROOT = "api/v2.0/" +def to_hash(text): + return md5(text.encode('utf8')).hexdigest() + + def extract_id(entry, keys=[('link', 'link'), ('published', 'retrieved_date'), ('updated', 'retrieved_date')], force_id=False): @@ -40,8 +45,8 @@ def extract_id(entry, keys=[('link', 'link'), if entry_id: return {'entry_id': entry_id} if not entry_id and force_id: - entry_id = hash("".join(entry[entry_key] for _, entry_key in keys - if entry_key in entry)) + entry_id = to_hash("".join(entry[entry_key] for _, entry_key in keys + if entry_key in entry).encode('utf8')) else: ids = {} for entry_key, pyagg_key in keys: @@ -218,18 +223,32 @@ class FeedCrawler(AbstractCrawler): future.add_done_callback(self.get_counter_callback()) return + etag_generated = False if response.status_code == 304: logger.info("%r %r - feed responded with 304", - self.feed['id'], self.feed['title']) + self.feed['id'], self.feed['title']) self.clean_feed() return - if self.feed['etag'] and response.headers.get('etag') \ - and response.headers.get('etag') == self.feed['etag']: - logger.info("%r %r - feed responded with same etag (%d)", - self.feed['id'], self.feed['title'], - response.status_code) + if not response.headers.get('etag'): + etag_generated = True + logger.debug('%r %r - manually generating etag', + self.feed['id'], self.feed['title']) + response.headers['etag'] = 'pyagg/"%s"' % to_hash(response.text) + if self.feed['etag'] and response.headers['etag'] == self.feed['etag']: + if etag_generated: + logger.info("%r %r - calculated hash matches (%d)", + self.feed['id'], self.feed['title'], + response.status_code) + else: + logger.info("%r %r - feed responded with same etag (%d)", + self.feed['id'], self.feed['title'], + response.status_code) self.clean_feed() return + else: + logger.info('%r %r - etag mismatch %r != %r', + self.feed['id'], self.feed['title'], + response.headers['etag'], self.feed['etag']) ids, entries = [], {} parsed_response = feedparser.parse(response.text) for entry in parsed_response['entries']: @@ -253,7 +272,7 @@ class CrawlerScheduler(AbstractCrawler): def prepare_headers(self, feed): """For a known feed, will construct some header dictionnary""" headers = {'User-Agent': 'pyaggr3g470r/crawler'} - if feed.get('etag', None): + if feed.get('etag') and 'pyagg' not in feed.get('etag', ''): headers['If-None-Match'] = feed['etag'] if feed.get('last_modified'): headers['If-Modified-Since'] = feed['last_modified'] -- cgit From f2463bc333cc207ffa9ab935b7edf59a9894720d Mon Sep 17 00:00:00 2001 From: François Schmidts Date: Mon, 6 Apr 2015 10:19:58 +0200 Subject: misc update updating the way we maintain feed up to date in the database fixing the counter bumping the minimum error count --- pyaggr3g470r/lib/crawler.py | 49 +++++++++++++++++++++++++++------------------ 1 file changed, 30 insertions(+), 19 deletions(-) (limited to 'pyaggr3g470r/lib') diff --git a/pyaggr3g470r/lib/crawler.py b/pyaggr3g470r/lib/crawler.py index 99967671..1ac6029a 100644 --- a/pyaggr3g470r/lib/crawler.py +++ b/pyaggr3g470r/lib/crawler.py @@ -85,9 +85,12 @@ class AbstractCrawler: @wraps(func) def wrapper(*args, **kwargs): cls.__counter__ += 1 - result = func(*args, **kwargs) - cls.__counter__ -= 1 - return result + try: + return func(*args, **kwargs) + except: + logger.exception('an error occured while %r', func) + finally: + cls.__counter__ -= 1 return wrapper @classmethod @@ -172,21 +175,27 @@ class PyAggUpdater(AbstractCrawler): for id_to_create in results: entry = self.to_article( self.entries[tuple(sorted(id_to_create.items()))]) - logger.info('creating %r - %r', entry['title'], id_to_create) + logger.warn('%r %r - creating %r - %r', self.feed['id'], + self.feed['title'], entry['title'], id_to_create) self.query_pyagg('post', 'article', entry) now = datetime.now() logger.debug('%r %r - updating feed etag %r last_mod %r', self.feed['id'], self.feed['title'], - self.headers.get('etag'), now) + self.headers.get('etag', ''), + self.headers.get('last-modified', '')) - dico = {'error_count': 0, 'last_error': '', + dico = {'error_count': 0, 'last_error': None, 'etag': self.headers.get('etag', ''), 'last_modified': self.headers.get('last-modified', ''), 'site_link': self.parsed_feed.get('link')} if not self.feed.get('title'): dico['title'] = self.parsed_feed.get('title', '') - if any([dico[key] == self.feed.get(key) for key in dico]): + logger.info('%r %r - pushing feed attrs %r', + self.feed['id'], self.feed['title'], + {key: "%s -> %s" % (dico[key], self.feed.get(key)) + for key in dico if dico[key] != self.feed.get(key)}) + if any([dico[key] != self.feed.get(key) for key in dico]): future = self.query_pyagg('put', 'feed/%d' % self.feed['id'], dico) future.add_done_callback(self.get_counter_callback()) @@ -223,19 +232,18 @@ class FeedCrawler(AbstractCrawler): future.add_done_callback(self.get_counter_callback()) return - etag_generated = False if response.status_code == 304: logger.info("%r %r - feed responded with 304", self.feed['id'], self.feed['title']) self.clean_feed() return - if not response.headers.get('etag'): - etag_generated = True + if 'etag' not in response.headers: logger.debug('%r %r - manually generating etag', self.feed['id'], self.feed['title']) response.headers['etag'] = 'pyagg/"%s"' % to_hash(response.text) - if self.feed['etag'] and response.headers['etag'] == self.feed['etag']: - if etag_generated: + if response.headers['etag'] and self.feed['etag'] \ + and response.headers['etag'] == self.feed['etag']: + if 'pyagg' in self.feed['etag']: logger.info("%r %r - calculated hash matches (%d)", self.feed['id'], self.feed['title'], response.status_code) @@ -246,9 +254,12 @@ class FeedCrawler(AbstractCrawler): self.clean_feed() return else: - logger.info('%r %r - etag mismatch %r != %r', - self.feed['id'], self.feed['title'], - response.headers['etag'], self.feed['etag']) + logger.debug('%r %r - etag mismatch %r != %r', + self.feed['id'], self.feed['title'], + response.headers['etag'], self.feed['etag']) + logger.info('%r %r - cache validation failed, challenging entries', + self.feed['id'], self.feed['title']) + ids, entries = [], {} parsed_response = feedparser.parse(response.text) for entry in parsed_response['entries']: @@ -272,10 +283,10 @@ class CrawlerScheduler(AbstractCrawler): def prepare_headers(self, feed): """For a known feed, will construct some header dictionnary""" headers = {'User-Agent': 'pyaggr3g470r/crawler'} - if feed.get('etag') and 'pyagg' not in feed.get('etag', ''): - headers['If-None-Match'] = feed['etag'] if feed.get('last_modified'): headers['If-Modified-Since'] = feed['last_modified'] + if feed.get('etag') and 'pyagg' not in feed['etag']: + headers['If-None-Match'] = feed['etag'] logger.debug('%r %r - calculated headers %r', feed['id'], feed['title'], headers) return headers @@ -289,8 +300,8 @@ class CrawlerScheduler(AbstractCrawler): feeds = response.json() logger.debug('%d to fetch %r', len(feeds), feeds) for feed in feeds: - logger.info('%r %r - fetching resources', - feed['id'], feed['title']) + logger.debug('%r %r - fetching resources', + feed['id'], feed['title']) future = self.session.get(feed['link'], headers=self.prepare_headers(feed)) future.add_done_callback(FeedCrawler(feed, self.auth).callback) -- cgit