|
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)
|
Sadrok Rawrs