diff --git a/scrapy/trunk/scrapy/contrib/pipeline/images.py b/scrapy/trunk/scrapy/contrib/pipeline/images.py index b026514a0..4f12d64f2 100644 --- a/scrapy/trunk/scrapy/contrib/pipeline/images.py +++ b/scrapy/trunk/scrapy/contrib/pipeline/images.py @@ -1,23 +1,19 @@ -import re import os import time import hashlib -import urllib -import urlparse +import Image from cStringIO import StringIO -import Image +from twisted.internet import defer from scrapy import log from scrapy.stats import stats +from scrapy.utils.misc import md5sum from scrapy.core.exceptions import DropItem, NotConfigured from scrapy.core.exceptions import HttpException, IgnoreRequest from scrapy.conf import settings - from scrapy.contrib.pipeline.media import MediaPipeline -# the age at which we download images again -IMAGE_EXPIRES = settings.getint('IMAGES_EXPIRES', 90) class NoimagesDrop(DropItem): """Product with no images exception""" @@ -27,10 +23,48 @@ class ImageException(Exception): class BaseImagesPipeline(MediaPipeline): - MEDIA_TYPE = 'image' + """Abstract pipeline that implement the image downloading and thumbnail generation logic + + This pipeline tries to minimize network transfers and image processing, + doing stat of the images and determining if image is new, uptodate or + expired. + + `new` images are those that pipeline never processed and needs to be + downloaded from supplier site the first time. + + `uptodate` images are the ones that the pipeline processed and are still + valid images. + + `expired` images are those that pipeline already processed but the last + modification was made long time ago, so a reprocessing is recommended to + refresh it in case of change. + + IMAGES_EXPIRES setting controls the maximun days since an image was modified + to consider it uptodate. + + THUMBS is a tuple of tuples, each sub-tuple is a pair of thumb_id string + and a compatible python image library size (a tuple). + See thumbnail method at http://www.pythonware.com/library/pil/handbook/image.htm + + Downloaded images are skipped if sizes aren't greater than MIN_WIDTH and + MIN_HEIGHT limit. A proper log messages will be printed. + + """ + + MIN_WIDTH = settings.getint('IMAGES_MIN_WIDTH', 0) + MIN_HEIGHT = settings.getint('IMAGES_MIN_HEIGHT', 0) + IMAGES_EXPIRES = settings.getint('IMAGES_EXPIRES', 90) + MEDIA_NAME = 'image' + THUMBS = ( +# ('50', (50, 50)), +# ('110', (110, 110)), +# ('270', (270, 270)) + ) + + def media_downloaded(self, response, request, info): - mtype = self.MEDIA_TYPE + mtype = self.MEDIA_NAME referer = request.headers.get('Referer') if not response or not response.body: @@ -43,7 +77,18 @@ class BaseImagesPipeline(MediaPipeline): msg = 'Image (%s): Downloaded %s from %s referred in <%s>' % (status, mtype, request, referer) log.msg(msg, level=log.DEBUG, domain=info.domain) self.inc_stats(info.domain, status) - return self.image_downloaded(response, request, info) + + try: + key = self.image_key(request.url) + checksum = self.image_downloaded(response, request, info) + except ImageException, ex: + log.msg(str(ex), level=log.WARNING, domain=info.domain) + raise ex + except Exception, ex: + log.msg(str(ex), level=log.WARNING, domain=info.domain) + raise ex + + return '%s#%s' % (key, checksum) if checksum else key def media_failed(self, failure, request, info): referer = request.headers.get('Referer') @@ -53,31 +98,133 @@ class BaseImagesPipeline(MediaPipeline): errmsg = str(failure) msg = 'Image (http-error): Error downloading %s from %s referred in <%s>: %s' \ - % (self.MEDIA_TYPE, request, referer, errmsg) + % (self.MEDIA_NAME, request, referer, errmsg) log.msg(msg, level=log.WARNING, domain=info.domain) raise ImageException(msg) + def media_to_download(self, request, info): + def _onsuccess(result): + last_modified = result.get('last_modified', None) + if not last_modified: + return # returning None force download + + age_seconds = time.time() - last_modified + age_days = age_seconds / 60 / 60 / 24 + if age_days > self.IMAGES_EXPIRES: + return # returning None force download + + referer = request.headers.get('Referer') + log.msg('Image (uptodate) type=%s at <%s> referred from <%s>' % \ + (self.MEDIA_NAME, request.url, referer), level=log.DEBUG, domain=info.domain) + self.inc_stats(info.domain, 'uptodate') + + checksum = result.get('checksum', None) + return '%s#%s' % (key, checksum) if checksum else key + + key = self.image_key(request.url) + dfd = defer.maybeDeferred(self.stat_key, key, info) + dfd.addCallbacks(_onsuccess, lambda _:None) + dfd.addErrback(log.err, self.__class__.__name__ + '.stat_key') + return dfd + + def image_downloaded(self, response, request, info): + first_buf = None + for key, image, buf in self.get_images(response, request, info): + self.store_image(key, image, buf, info) + if first_buf is None: + first_buf = buf + return md5sum(first_buf) + + def get_images(self, response, request, info): + key = self.image_key(request.url) + orig_image = Image.open(StringIO(response.body)) + + width, height = orig_image.size + if width < self.MIN_WIDTH or height < self.MIN_HEIGHT: + raise ImageException("Image too small (%dx%d < %dx%d): %s" % \ + (width, height, self.MIN_WIDTH, self.MIN_HEIGHT, response.url)) + + image, buf = self.convert_image(orig_image) + yield key, image, buf + + for thumb_id, size in self.THUMBS or []: + thumb_key = self.thumb_key(request.url, thumb_id) + thumb_image, thumb_buf = self.convert_image(image, size) + yield thumb_key, thumb_image, thumb_buf + def inc_stats(self, domain, status): stats.incpath('%s/image_count' % domain) stats.incpath('%s/image_status_count/%s' % (domain, status)) - def image_downloaded(self, response, request, info): + def convert_image(self, image, size=None): + if image.mode != 'RGB': + image = image.convert('RGB') + + if size: + image = image.copy() + image.thumbnail(size, Image.ANTIALIAS) + + buf = StringIO() + try: + image.save(buf, 'JPEG') + except Exception, ex: + raise ImageException("Cannot process image. Error: %s" % ex) + + return image, buf + + def image_key(self, url): + image_guid = hashlib.sha1(url).hexdigest() + return 'full/%s.jpg' % (image_guid) + + def thumb_key(self, url, thumb_id): + image_guid = hashlib.sha1(url).hexdigest() + return 'thumbs/%s/%s.jpg' % (thumb_id, image_guid) + + # Required overradiable interface + def store_image(self, key, image, buf, info): + """Override this method with specific code to persist an image + + This method is used to persist the full image and any defined + thumbnail, one a time. + + Return value is ignored. + + """ + raise NotImplementedError + + def stat_key(self, key, info): + """Override this method with specific code to stat an image + + this method should return and dictionary with two parameters: + * last_modified: the last modification time in seconds since the epoch + * checksum: the md5sum of the content of the stored image if found + + If an exception is raised or last_modified is None, then the image + will be re-downloaded. + + If the difference in days between last_modified and now is greater than + IMAGES_EXPIRES settings, then the image will be re-downloaded + + The checksum value is appended to returned image path after a hash + sign (#), if checksum is None, then nothing is appended including the + hash sign. + + """ raise NotImplementedError class ImagesPipeline(BaseImagesPipeline): - MIN_WIDTH = 0 - MIN_HEIGHT = 0 - THUMBS = ( -# ('50', (50, 50)), -# ('110', (110, 110)), -# ('270', (270, 270)) - ) + """Images pipeline with filesystem support as image's store backend + + If IMAGES_DIR setting has a valid value, this pipeline is enabled and use + path defined at setting as dirname for storing images. + + """ class DomainInfo(BaseImagesPipeline.DomainInfo): def __init__(self, domain): - super(ImagesPipeline.DomainInfo, self).__init__(domain) self.created_directories = set() + super(ImagesPipeline.DomainInfo, self).__init__(domain) def __init__(self): if not settings['IMAGES_DIR']: @@ -85,44 +232,27 @@ class ImagesPipeline(BaseImagesPipeline): self.BASEDIRNAME = settings['IMAGES_DIR'] self.mkdir(self.BASEDIRNAME) - - self.MIN_WIDTH = settings.getint('IMAGES_MIN_WIDTH', 0) - self.MIN_HEIGHT = settings.getint('IMAGES_MIN_HEIGHT', 0) super(ImagesPipeline, self).__init__() - def media_to_download(self, request, info): - relative, absolute = self._get_paths(request) - if not should_download(absolute): - self.inc_stats(info.domain, 'uptodate') - referer = request.headers.get('Referer') - log.msg('Image (uptodate): Downloaded %s from %s referred in <%s>' % \ - (self.MEDIA_TYPE, request, referer), level=log.DEBUG, domain=info.domain) - return relative - - def image_downloaded(self, response, request, info): - mtype = self.MEDIA_TYPE - relpath, abspath = self._get_paths(request) - dirname = os.path.dirname(abspath) - self.mkdir(dirname, info) + def store_image(self, key, image, buf, info): + absolute_path = self.get_filesystem_path(key) + self.mkdir(os.path.dirname(absolute_path), info) + image.save(absolute_path) + def stat_key(self, key, info): + absolute_path = self.get_filesystem_path(key) try: - save_image_with_thumbnails(response, abspath, self.THUMBS, self.MIN_WIDTH, self.MIN_HEIGHT) - except ImageException, ex: - log.msg(str(ex), level=log.WARNING, domain=info.domain) - raise ex - except Exception, ex: - referer = request.headers.get('Referer') - msg = 'Image (processing-error): Error thumbnailing %s from %s referred in <%s>: %s' \ - % (mtype, request, referer, ex) - log.msg(msg, level=log.WARNING, domain=info.domain) - raise ImageException(msg) + last_modified = os.path.getmtime(absolute_path) + except: + return {} - return relpath # success value sent as input result for item_media_downloaded + with open(absolute_path, 'rb') as imagefile: + checksum = md5sum(imagefile) - def _get_paths(self, request): - relative = image_path(request.url) - absolute = os.path.join(self.BASEDIRNAME, relative) - return relative, absolute + return {'last_modified': last_modified, 'checksum': checksum} + + def get_filesystem_path(self, key): + return os.path.join(self.BASEDIRNAME, key) def mkdir(self, dirname, info=None): already_created = info.created_directories if info else set() @@ -130,66 +260,3 @@ class ImagesPipeline(BaseImagesPipeline): if not os.path.exists(dirname): os.makedirs(dirname) already_created.add(dirname) - - -def should_download(path): - """Should the image downloader download the image to the location specified - """ - try: - mtime = os.path.getmtime(path) - age_seconds = time.time() - mtime - age_days = age_seconds / 60 / 60 / 24 - return age_days > IMAGE_EXPIRES - except: - return True - -_MULTIPLE_SLASHES_REGEXP = re.compile(r"\/{2,}") -_FINAL_SLASH_REGEXP = re.compile(r"\/$") -def image_path(url): - """Return the relative path on the target filesystem for an image to be - downloaded to. - """ - _, netloc, urlpath, query, _ = urlparse.urlsplit(url) - urlpath = _MULTIPLE_SLASHES_REGEXP.sub('/', urlpath) - urlpath = _FINAL_SLASH_REGEXP.sub('.jpg', urlpath) - if os.sep != '/': - urlpath.replace('/', os.sep) - if query: - img_path = os.path.join(netloc, hashlib.sha1(url).hexdigest()) - else: - img_path = os.path.join(netloc, urlpath[1:]) - return urllib.unquote(img_path) - - -def thumbnail_name(image, sizestr): - """Get the name of a thumbnail image given the name of the original file. - - There will can be many types of thumbnails, so we will have a "name" for - each type. - """ - return os.path.splitext(image)[0] + '_' + sizestr + '.jpg' - -def save_scaled_image(image, img_path, name, size): - thumb = image.copy() if image.mode == 'RGB' else image.convert('RGB') - thumb.thumbnail(size, Image.ANTIALIAS) - filename = thumbnail_name(img_path, name) - thumb.save(filename, 'JPEG') - -def save_image_with_thumbnails(response, path, thumbsizes, min_width=0, min_height=0): - memoryfile = StringIO(response.body) - im = Image.open(memoryfile) - if im.mode != 'RGB': - log.msg("Found non-RGB image during scraping %s" % path, level=log.WARNING) - for name, size in thumbsizes or []: - save_scaled_image(im, path, name, size) - try: - im.save(path) - except Exception, ex: - log.msg("Image (processing-error): cannot process %s, so writing direct file: Error: %s" % (path, ex)) - f = open(path, 'wb') - f.write(response.body) - f.close() - width, height = im.size - if width < min_width or height < min_height: - raise ImageException("Image too small (%dx%d < %dx%d): %s" % (width, height, min_width, min_height, response.url)) - diff --git a/scrapy/trunk/scrapy/contrib/pipeline/s3images.py b/scrapy/trunk/scrapy/contrib/pipeline/s3images.py index fc746d74f..3adbd8066 100644 --- a/scrapy/trunk/scrapy/contrib/pipeline/s3images.py +++ b/scrapy/trunk/scrapy/contrib/pipeline/s3images.py @@ -1,150 +1,77 @@ -import time -import hashlib import rfc822 -from cStringIO import StringIO -import Image - -from scrapy import log from scrapy.http import Request from scrapy.core.engine import scrapyengine from scrapy.core.exceptions import NotConfigured -from scrapy.contrib.pipeline.media import MediaPipeline -from scrapy.contrib.aws import sign_request +from scrapy.contrib.pipeline.images import BaseImagesPipeline, md5sum from scrapy.conf import settings -from .images import BaseImagesPipeline, ImageException - - -def md5sum(buffer): - m = hashlib.md5() - buffer.seek(0) - while 1: - d = buffer.read(8096) - if not d: break - m.update(d) - return m.hexdigest() - class S3ImagesPipeline(BaseImagesPipeline): - MEDIA_TYPE = 'image' - THUMBS = ( -# ('50', (50, 50)), -# ('110', (110, 110)), -# ('270', (270, 270)) - ) + """Images pipeline with amazon S3 support as image's store backend - # Automatically sign requests with AWS authorization header, - # alternative we can do this using scrapy.contrib.aws.AWSMiddleware - sign_requests = True + This pipeline tries to minimize the PUT requests made to amazon doing a + HEAD per full image, if HEAD returns a successfully response, then the + Last-Modified header is compared to current timestamp and if the difference + in days are greater that IMAGE_EXPIRES setting, then the image is + downloaded, reprocessed and uploaded to S3 again including its thumbnails. - s3_custom_spider = None + It is recommended to add an spider with domain_name 's3.amazonaws.com', + doing that you will overcome the limit of request per spider. The following + is the minimal code for this spider: + + from scrapy.spider import BaseSpider + + class S3AmazonAWSSpider(BaseSpider): + domain_name = "s3.amazonaws.com" + max_concurrent_requests = 100 + start_urls = ('http://s3.amazonaws.com/',) + + SPIDER = S3AmazonAWSSpider() + + Commonly uploading images to S3 requires requests to be signed, the + recommended way is to enable scrapy.contrib.aws.AWSMiddleware downloader + middleware and configure AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY + settings + + More info about amazon S3 at http://docs.amazonwebservices.com/AmazonS3/2006-03-01/ + + """ + + # amazon s3 bucket name to put images + bucket_name = settings.get('S3_BUCKET') + + # prefix to prepend to image keys + key_prefix = settings.get('S3_PREFIX', '') + + # Optional spider to use for image uploading + AmazonS3Spider = None def __init__(self): if not settings['S3_IMAGES']: raise NotConfigured - - self.bucket_name = settings['S3_BUCKET'] - self.prefix = settings['S3_PREFIX'] - self.access_key = settings['AWS_ACCESS_KEY_ID'] - self.secret_key = settings['AWS_SECRET_ACCESS_KEY'] - self.image_refresh_days = settings.getint('IMAGES_EXPIRES', 90) - MediaPipeline.__init__(self) + super(S3ImagesPipeline, self).__init__() def s3_request(self, key, method, body=None, headers=None): - url = 'http://%s.s3.amazonaws.com/%s%s' % (self.bucket_name, self.prefix, key) + url = 'http://%s.s3.amazonaws.com/%s%s' % (self.bucket_name, self.key_prefix, key) req = Request(url, method=method, body=body, headers=headers) - if self.sign_requests: - sign_request(req, self.access_key, self.secret_key) return req - def image_downloaded(self, response, request, info): - try: - key = self.s3_image_key(request.url) - etag = self.s3_store_image(response, request.url, info) - except ImageException, ex: - log.msg(str(ex), level=log.WARNING, domain=info.domain) - raise ex - except Exception, ex: - log.msg(str(ex), level=log.WARNING, domain=info.domain) - raise ex - - return '%s#%s' % (key, etag) # success value sent as input result for item_media_downloaded - - def s3_image_key(self, url): - """Return the relative path on the target filesystem for an image to be - downloaded to. - """ - image_guid = hashlib.sha1(url).hexdigest() - return 'full/%s.jpg' % (image_guid) - - def s3_thumb_key(self, url, thumb_id): - """Return the relative path on the target filesystem for an image to be - downloaded to. - """ - image_guid = hashlib.sha1(url).hexdigest() - return 'thumbs/%s/%s.jpg' % (thumb_id, image_guid) - - def media_to_download(self, request, info): - """Return if the image should be downloaded by checking if it's already in - the S3 storage and not too old""" - + def stat_key(self, key, info): def _onsuccess(response): - if 'Last-Modified' not in response.headers: - return # returning None force download - - # check if last modified date did not expires + checksum = response.headers['Etag'].strip('"') last_modified = response.headers['Last-Modified'] modified_tuple = rfc822.parsedate_tz(last_modified) modified_stamp = int(rfc822.mktime_tz(modified_tuple)) - age_seconds = time.time() - modified_stamp - age_days = age_seconds / 60 / 60 / 24 + return {'checksum': checksum, 'last_modified': modified_stamp} - if age_days > self.image_refresh_days: - return # returning None force download - - etag = response.headers['Etag'].strip('"') - referer = request.headers.get('Referer') - log.msg('Image (uptodate) type=%s at <%s> referred from <%s>' % \ - (self.MEDIA_TYPE, request.url, referer), level=log.DEBUG, domain=info.domain) - - self.inc_stats(info.domain, 'uptodate') - return '%s#%s' % (key, etag) - - key = self.s3_image_key(request.url) req = self.s3_request(key, method='HEAD') dfd = self.s3_download(req, info) - dfd.addCallbacks(_onsuccess, lambda _:None) - dfd.addErrback(log.err, 'S3ImagesPipeline.media_to_download') + dfd.addCallback(_onsuccess) return dfd - def s3_store_image(self, response, url, info): + def store_image(self, key, image, buf, info): """Upload image to S3 storage""" - buf = StringIO(response.body) - image = Image.open(buf) - key = self.s3_image_key(url) - _, jpegbuf = self._s3_put_image(image, key, info) - self.s3_store_thumbnails(image, url, info) - return md5sum(jpegbuf) # Etag - - def s3_store_thumbnails(self, image, url, info): - """Upload image thumbnails to S3 storage""" - for thumb_id, size in self.THUMBS or []: - thumb = image.copy() if image.mode == 'RGB' else image.convert('RGB') - thumb.thumbnail(size, Image.ANTIALIAS) - key = self.s3_thumb_key(url, thumb_id) - self._s3_put_image(thumb, key, info) - - def _s3_put_image(self, image, key, info): - if image.mode != 'RGB': - image = image.convert('RGB') - - buf = StringIO() - try: - image.save(buf, 'JPEG') - except Exception, ex: - raise ImageException("Cannot process image. Error: %s" % ex) - width, height = image.size headers = { 'Content-Type': 'image/jpeg', @@ -156,7 +83,7 @@ class S3ImagesPipeline(BaseImagesPipeline): buf.seek(0) req = self.s3_request(key, method='PUT', body=buf.read(), headers=headers) - return self.s3_download(req, info), buf + self.s3_download(req, info) def s3_download(self, request, info): """This method is used for HEAD and PUT requests sent to amazon S3 @@ -165,8 +92,8 @@ class S3ImagesPipeline(BaseImagesPipeline): to current domain spider. """ - if self.s3_custom_spider: - return scrapyengine.schedule(request, self.s3_custom_spider) + if self.AmazonS3Spider: + return scrapyengine.schedule(request, self.AmazonS3Spider) return self.download(request, info) diff --git a/scrapy/trunk/scrapy/utils/misc.py b/scrapy/trunk/scrapy/utils/misc.py index c6e5881eb..7538961cc 100644 --- a/scrapy/trunk/scrapy/utils/misc.py +++ b/scrapy/trunk/scrapy/utils/misc.py @@ -185,3 +185,20 @@ def string_camelcase(string): return CAMELCASE_INVALID_CHARS.sub('', string.title()) +def md5sum(buffer): + """Calculate the md5 checksum of a file + + >>> from StringIO import StringIO + >>> md5sum(StringIO('file content to hash')) + '784406af91dd5a54fbb9c84c2236595a' + + """ + m = hashlib.md5() + buffer.seek(0) + while 1: + d = buffer.read(8096) + if not d: break + m.update(d) + return m.hexdigest() + +