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)