"""
Images Pipeline
See documentation in topics/media-pipeline.rst
"""
from __future__ import annotations
import functools
import hashlib
import warnings
from contextlib import suppress
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Self
from itemadapter import ItemAdapter
from scrapy.exceptions import NotConfigured, ScrapyDeprecationWarning
from scrapy.http import Request, Response
from scrapy.http.request import NO_CALLBACK
from scrapy.pipelines.files import FilesPipeline, _checksum_algorithm
from scrapy.pipelines.media import FileException
from scrapy.utils.defer import ensure_awaitable
if TYPE_CHECKING:
from collections.abc import Iterator
from os import PathLike
from PIL import Image
from scrapy.crawler import Crawler
from scrapy.pipelines.media import FileInfoOrError, MediaPipeline
class ImageException(FileException):
"""General image error exception"""
[docs]
class ImagesPipeline(FilesPipeline):
"""Pipeline that implements the handling logic specific to images."""
MEDIA_NAME: str = "image"
# Uppercase attributes kept for backward compatibility with code that subclasses
# ImagesPipeline. They may be overridden by settings.
MIN_WIDTH: int = 0
MIN_HEIGHT: int = 0
EXPIRES: int = 90
THUMBS: ClassVar[dict[str, tuple[int, int]]] = {}
PRESERVE_FORMAT: bool = False
DEFAULT_IMAGES_URLS_FIELD = "image_urls"
DEFAULT_IMAGES_RESULT_FIELD = "images"
def __init__(
self,
store_uri: str | PathLike[str],
download_func: None = None,
*,
crawler: Crawler,
):
if download_func is not None: # pragma: no cover
warnings.warn(
"The download_func argument of ImagesPipeline.__init__() is ignored"
" and will be removed in a future Scrapy version.",
category=ScrapyDeprecationWarning,
stacklevel=2,
)
try:
from PIL import Image, ImageOps # noqa: PLC0415
self._Image = Image
self._ImageOps = ImageOps
except ImportError:
raise NotConfigured(
"ImagesPipeline requires the Scrapy[images] extra to be installed"
) from None
super().__init__(store_uri, crawler=crawler)
settings = crawler.settings
resolve = functools.partial(
self._key_for_pipe,
base_class_name="ImagesPipeline",
settings=settings,
)
self.store.checksum_algorithm = _checksum_algorithm(
settings, resolve("IMAGES_CHECKSUM_ALGORITHM")
)
self.expires: float = settings.getfloat(resolve("IMAGES_EXPIRES"), self.EXPIRES)
if not hasattr(self, "IMAGES_RESULT_FIELD"):
self.IMAGES_RESULT_FIELD: str = self.DEFAULT_IMAGES_RESULT_FIELD
if not hasattr(self, "IMAGES_URLS_FIELD"):
self.IMAGES_URLS_FIELD: str = self.DEFAULT_IMAGES_URLS_FIELD
self.images_urls_field: str = settings.get(
resolve("IMAGES_URLS_FIELD"), self.IMAGES_URLS_FIELD
)
self.images_result_field: str = settings.get(
resolve("IMAGES_RESULT_FIELD"), self.IMAGES_RESULT_FIELD
)
self.min_width: int = settings.getint(
resolve("IMAGES_MIN_WIDTH"), self.MIN_WIDTH
)
self.min_height: int = settings.getint(
resolve("IMAGES_MIN_HEIGHT"), self.MIN_HEIGHT
)
self.thumbs: dict[str, tuple[int, int]] = settings.get(
resolve("IMAGES_THUMBS"), self.THUMBS
)
self.preserve_format: bool = settings.getbool(
resolve("IMAGES_PRESERVE_FORMAT"), self.PRESERVE_FORMAT
)
@classmethod
def from_crawler(cls, crawler: Crawler) -> Self:
return cls(crawler.settings["IMAGES_STORE"], crawler=crawler)
def _get_store_setting_name(self, name: str) -> str:
return self._key_for_pipe(
f"IMAGES_STORE_{name}",
"ImagesPipeline",
self.crawler.settings,
)
async def file_downloaded(
self,
response: Response,
request: Request,
info: MediaPipeline.SpiderInfo,
*,
item: Any = None,
) -> str:
return await self.image_downloaded(response, request, info, item=item)
async def image_downloaded(
self,
response: Response,
request: Request,
info: MediaPipeline.SpiderInfo,
*,
item: Any = None,
) -> str:
checksum: str | None = None
for path, image, buf in self.get_images(response, request, info, item=item):
if checksum is None:
buf.seek(0)
checksum = hashlib.file_digest(
buf, self.store.checksum_algorithm
).hexdigest()
width, height = image.size
content_type = self._Image.MIME.get(image.format or "JPEG", "image/jpeg")
await ensure_awaitable(
self.store.persist_file(
path,
buf,
info,
meta={"width": width, "height": height},
headers={"Content-Type": content_type},
)
)
assert checksum is not None
return checksum
def get_images(
self,
response: Response,
request: Request,
info: MediaPipeline.SpiderInfo,
*,
item: Any = None,
) -> Iterator[tuple[str, Image.Image, BytesIO]]:
path = self.file_path(request, response=response, info=info, item=item)
orig_image = self._Image.open(BytesIO(response.body))
assert orig_image.format is not None
image_format = orig_image.format
transposed_image = self._ImageOps.exif_transpose(orig_image)
width, height = transposed_image.size
if width < self.min_width or height < self.min_height:
raise ImageException(
"Image too small "
f"({width}x{height} < "
f"{self.min_width}x{self.min_height})"
)
image, buf = self.convert_image(
transposed_image,
image_format=image_format,
response_body=BytesIO(response.body),
)
yield path, image, buf
for thumb_id, size in self.thumbs.items():
thumb_path = self.thumb_path(
request, thumb_id, response=response, info=info, item=item
)
thumb_image, thumb_buf = self.convert_image(
image, size, image_format=image_format, response_body=buf
)
yield thumb_path, thumb_image, thumb_buf
def convert_image(
self,
image: Image.Image,
size: tuple[int, int] | None = None,
*,
image_format: str,
response_body: BytesIO,
) -> tuple[Image.Image, BytesIO]:
target_format = image_format if self.preserve_format else "JPEG"
if target_format == "JPEG":
if image_format in {"PNG", "WEBP"} and image.mode == "RGBA":
background = self._Image.new("RGBA", image.size, (255, 255, 255))
background.paste(image, image)
image = background.convert("RGB")
elif image.mode == "P":
image = image.convert("RGBA")
background = self._Image.new("RGBA", image.size, (255, 255, 255))
background.paste(image, image)
image = background.convert("RGB")
elif image.mode != "RGB":
image = image.convert("RGB")
if size:
image = image.copy()
resampling_filter = self._Image.Resampling.LANCZOS
image.thumbnail(size, resampling_filter)
elif target_format == image_format:
image.format = target_format
return image, response_body
buf = BytesIO()
image.save(buf, target_format)
image.format = target_format
return image, buf
[docs]
def item_completed(
self, results: list[FileInfoOrError], item: Any, info: MediaPipeline.SpiderInfo
) -> Any:
with suppress(KeyError):
ItemAdapter(item)[self.images_result_field] = [x for ok, x in results if ok]
return item
[docs]
def file_path(
self,
request: Request,
response: Response | None = None,
info: MediaPipeline.SpiderInfo | None = None,
*,
item: Any = None,
) -> str:
image_guid = hashlib.sha1(request.url.encode()).hexdigest() # noqa: S324
ext = "" if self.preserve_format else ".jpg"
return f"full/{image_guid}{ext}"
[docs]
def thumb_path(
self,
request: Request,
thumb_id: str,
response: Response | None = None,
info: MediaPipeline.SpiderInfo | None = None,
*,
item: Any = None,
) -> str:
thumb_guid = hashlib.sha1(request.url.encode()).hexdigest() # noqa: S324
ext = "" if self.preserve_format else ".jpg"
return f"thumbs/{thumb_id}/{thumb_guid}{ext}"