Skip to main content

generateThumbnails.py (Source)

import logging
import os
from io import BytesIO
from pathlib import PurePath

import azure.functions as func
from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobServiceClient, ContentSettings
from PIL import Image, UnidentifiedImageError

ACCOUNT_URL = os.environ["STORAGE_ACCOUNT_URL"]

credential = DefaultAzureCredential()

CONTAINER_NAME = "data"
THUMB_PREFIX = "thumb_"
THUMB_SIZE = 300


def get_crop(size):
    """
    Get the crop coordinates given the source image sizes

    Attributes:
        size (tuple)  - the (width, height) of the image
    """
    shortest_edge = min(size)
    centers = (size[0] // 2, size[1] // 2)
    return (
        centers[0] - shortest_edge // 2,
        centers[1] - shortest_edge // 2,
        centers[0] + shortest_edge // 2,
        centers[1] + shortest_edge // 2,
    )


def main(myblob: func.InputStream):
    """
    Generate thumbnails from incoming blob if it's an image and
    upload the thumbnail blob to the same location but with a
    file name prefix.
    """

    source_path = PurePath(myblob.name)
    source_name = source_path.name
    container_name = source_path.parts[0]
    source_path = source_path.relative_to(container_name)

    if container_name != CONTAINER_NAME:
        # Strange, we're suppose to only react to events in the data container
        return

    if source_name.startswith(THUMB_PREFIX):
        # This is already a thumbnail so we can just leave it
        return

    try:
        # Attempt to open the file as an image. If it's not, too bad so sad.
        source_image = Image.open(myblob)
    except UnidentifiedImageError:
        return

    logging.info(
        f"Resizing image blob: \n" f"Name: {myblob.name}\n" f"URI: {myblob.uri}\n"
    )

    # Crop a square in the center and resize to thumbnail size
    thumb_image = source_image.crop(get_crop(source_image.size))
    thumb_image = thumb_image.resize((THUMB_SIZE, THUMB_SIZE))

    thumb_path = source_path.parent / (THUMB_PREFIX + source_path.name)
    thumb_stream = BytesIO()
    thumb_image.save(thumb_stream, format=source_image.format)

    with BlobServiceClient(
        account_url=ACCOUNT_URL, credential=credential
    ) as storage, storage.get_container_client(CONTAINER_NAME) as data_container:
        logging.info(
            "Uploading image %s to %s. %s bytes",
            thumb_image,
            thumb_path,
            thumb_stream.tell(),
        )

        # Rewind to the start of the stream after .save wrote to it.
        thumb_stream.seek(0)
        with data_container.upload_blob(
            thumb_path.as_posix(),
            thumb_stream,
            overwrite=True,
            content_settings=ContentSettings(
                content_type=source_image.get_format_mimetype()
            ),
        ) as out_blob:
            logging.info("Created image thumbnail: %s", out_blob.url)