|
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