"""This module responds for image transcoding.""" import os from PIL import Image from src.common import util from src.constants import transcoding_configs def transcode_image(image, output_config, output_buffer, **params): """Open image and save in given format and size. Args: image (PIL.Image): Source image object. output_config (dict): Paths for tmp directory, s3 key and size. E.g.: { 'size': (2000, 2000), 'key': 'img/jpg/test_image_2000_2000.jpg', 'format': 'JPEG' } params (dict): extra parameters for PIL.Image.save. output_buffer (io.BytesIO): Buffer for output image. Returns: io.BytesIO: transcoded file object """ color_profile = image.info.get('icc_profile') if output_config['size']: resized_image = image.resize( output_config['size'], Image.LANCZOS) resized_image.save( output_buffer, icc_profile=color_profile, format=output_config['format'], **params) resized_image.close() else: image.save( output_buffer, format=output_config['format'], **params) output_buffer.seek(0) return output_buffer def generate_transcoding_configs(key, descriptors): """Generate output paths. Args: key (str): target S3 key. descriptors (list): Contains dictionaries of size and output format. Returns: list: Contains dictionaries describing configuration of output files. """ base_name = os.path.basename(key) file_name, ext = os.path.splitext(base_name) result_configs = [] for descriptor in descriptors: if descriptor['size']: size_mark = '{}_{}'.format(*descriptor['size']) else: size_mark = '' name_prefix = util.generate_unique_filename() s3_output_template = ( transcoding_configs.IMAGE_PREVIEW_OUTPUT_PATH_TEMPLATE if descriptor['preview'] else transcoding_configs.IMAGE_OUTPUT_PATH_TEMPLATE) s3_output_key = s3_output_template.format( base_path=transcoding_configs.BASE_IMAGE_OUTPUT_PATH, unique_prefix=name_prefix, file_name=file_name, size=size_mark, extension=descriptor['extension'], subtype=descriptor.get('asset_subtype') ) config = { 'size': descriptor['size'], 'key': s3_output_key, 'preview': descriptor['preview'], 'format': descriptor['format'], 'asset_type': descriptor['asset_type'], 'asset_subtype': descriptor.get('asset_subtype'), 'output_image_buffer': None } result_configs.append(config) return result_configs def convert(file_object, image_descriptors, output_buffers): """Convert image file to new images by received descriptors. Args: file_object (io.BytesIO): File object. image_descriptors (list): List of sizes and output formats. output_buffers (list): List of buffer for output files. Returns: list: image descriptors with file-like output objects. """ described_output = [] with Image.open(file_object) as image: if image.mode != 'RGB': image = image.convert(mode='RGB') for descriptor in zip(image_descriptors, output_buffers): transcoded_image = transcode_image( image, descriptor[0], descriptor[1], dpi=transcoding_configs.DEFAULT_DENSITY, quality=transcoding_configs.DEFAULT_QUALITY, ) updated_descriptor = dict.copy(descriptor[0]) updated_descriptor['output_image_buffer'] = transcoded_image described_output.append(updated_descriptor) return described_output