Lab3-S3ImageConvert_LambdaLayer+CDK/cdk/lambda/lambda_function.py [53:202]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - logger = logging.getLogger() logger.setLevel(logging.INFO) client = boto3.client('s3') # get img from s3 def load_s3(bucket, key): logger.info(f'load from s3://{bucket}/{key}') try: response = client.get_object( Bucket=bucket, Key=key ) logger.info(json.dumps(response, default=str)) body = response['Body'].read() except Exception as e: logging.error(json.dumps(e, default=str)) os._exit(0) return body # auto orientation base on exif info from image def auto_exif_orientation(image): try: exif = image._getexif() except AttributeError: exif = None if exif is None: logger.info('no exif_orientation') return image exif = {ExifTags.TAGS[k]: v for k, v in exif.items() if k in ExifTags.TAGS} orientation = exif.get('Orientation', None) logger.info(f'original orientation:{orientation} to:') if orientation == 1: logger.info('no change') return image elif orientation == 2: logger.info('left-to-right mirror') return ImageOps.mirror(image) elif orientation == 3: logger.info('rotate 180') return image.transpose(Image.ROTATE_180) elif orientation == 4: logger.info('top-to-bottom mirror') return ImageOps.flip(image) elif orientation == 5: logger.info('top-to-left mirror') return ImageOps.mirror(image.transpose(Image.ROTATE_270)) elif orientation == 6: logger.info('rotate 270') return image.transpose(Image.ROTATE_270) elif orientation == 7: logger.info('top-to-right mirror') return ImageOps.mirror(image.transpose(Image.ROTATE_90)) elif orientation == 8: logger.info('rotate 90') return image.transpose(Image.ROTATE_90) else: return image # convert def img_convert(body, convert_format): logger.info('convert ...') try: im = Image.open(io.BytesIO(body)) if preserv_original_format: convert_format = im.format # auto orientation if auto_orientation: im = auto_exif_orientation(im) # resize if resize_feature.lower() == 'percentile': logger.info('resizing percentile and keep ratio ...') w, h = im.size w, h = int(w*resize_Percentile_w), int(h*resize_Percentile_h) im.thumbnail((w,h)) if resize_feature.lower() == 'fixsize': logger.info('resizing fixsize and keep ratio ...') w, h = resize_FixSize_w, resize_FixSize_h im.thumbnail((w, h)) if resize_feature.lower() == 'percentilenoratio': logger.info('resizing percentile and ignore ratio ...') w, h = im.size w, h = int(w*resize_Percentile_w), int(h*resize_Percentile_h) im = im.resize((w, h), resample=Image.BICUBIC) if resize_feature.lower() == 'fixsizenoratio': logger.info('resizing fixsize and igore ratio ...') w, h = resize_FixSize_w, resize_FixSize_h im = im.resize((w, h), resample=Image.BICUBIC) logger.info(f'target size:{im.size}') # convert PNG RGBA mode to RGB for JPEG if im.mode != 'RGB' and convert_format.lower() == 'jpeg': im = im.convert(mode='RGB') #save to target format in_mem_img = io.BytesIO() im.save(in_mem_img, format=convert_format, lossless=True, quality=save_quality, progressive=jpeg_progressive) in_mem_img.seek(0) except Exception as e: logging.error(json.dumps(e, default=str)) os._exit(0) return in_mem_img # put img to s3 def save_s3(bucket, key, body): logger.info(f'save to s3://{bucket}/{key}') try: response = client.put_object( Bucket=bucket, Key=key, Body=body ) logger.info(json.dumps(response, default=str)) except Exception as e: logging.error(json.dumps(e, default=str)) os._exit(0) return # change prefix from input/ to output/ and change postfix def change_key(key, convert_postfix): key_pre = os.path.splitext(key)[0].replace('input', 'output', 1) if not preserv_original_format: key_new = key_pre+convert_postfix else: key_new = key_pre+os.path.splitext(key)[1] return key_new def lambda_handler(event, context): logger.info(json.dumps(event, default=str)) bucket = event['Records'][0]['s3']['bucket']['name'] key = urllib.parse.unquote_plus(event['Records'][0]['s3']['object']['key'], encoding='utf-8') img_org = load_s3(bucket, key) body = img_convert(img_org, convert_format) key_new = change_key(key, convert_postfix) save_s3(bucket, key_new, body) return { 'statusCode': 200, 'body': 'OK' } - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - Lab3-S3ImageConvert_LambdaLayer+CDK/lambda_handler.py [52:201]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - logger = logging.getLogger() logger.setLevel(logging.INFO) client = boto3.client('s3') # get img from s3 def load_s3(bucket, key): logger.info(f'load from s3://{bucket}/{key}') try: response = client.get_object( Bucket=bucket, Key=key ) logger.info(json.dumps(response, default=str)) body = response['Body'].read() except Exception as e: logging.error(json.dumps(e, default=str)) os._exit(0) return body # auto orientation base on exif info from image def auto_exif_orientation(image): try: exif = image._getexif() except AttributeError: exif = None if exif is None: logger.info('no exif_orientation') return image exif = {ExifTags.TAGS[k]: v for k, v in exif.items() if k in ExifTags.TAGS} orientation = exif.get('Orientation', None) logger.info(f'original orientation:{orientation} to:') if orientation == 1: logger.info('no change') return image elif orientation == 2: logger.info('left-to-right mirror') return ImageOps.mirror(image) elif orientation == 3: logger.info('rotate 180') return image.transpose(Image.ROTATE_180) elif orientation == 4: logger.info('top-to-bottom mirror') return ImageOps.flip(image) elif orientation == 5: logger.info('top-to-left mirror') return ImageOps.mirror(image.transpose(Image.ROTATE_270)) elif orientation == 6: logger.info('rotate 270') return image.transpose(Image.ROTATE_270) elif orientation == 7: logger.info('top-to-right mirror') return ImageOps.mirror(image.transpose(Image.ROTATE_90)) elif orientation == 8: logger.info('rotate 90') return image.transpose(Image.ROTATE_90) else: return image # convert def img_convert(body, convert_format): logger.info('convert ...') try: im = Image.open(io.BytesIO(body)) if preserv_original_format: convert_format = im.format # auto orientation if auto_orientation: im = auto_exif_orientation(im) # resize if resize_feature.lower() == 'percentile': logger.info('resizing percentile and keep ratio ...') w, h = im.size w, h = int(w*resize_Percentile_w), int(h*resize_Percentile_h) im.thumbnail((w,h)) if resize_feature.lower() == 'fixsize': logger.info('resizing fixsize and keep ratio ...') w, h = resize_FixSize_w, resize_FixSize_h im.thumbnail((w, h)) if resize_feature.lower() == 'percentilenoratio': logger.info('resizing percentile and ignore ratio ...') w, h = im.size w, h = int(w*resize_Percentile_w), int(h*resize_Percentile_h) im = im.resize((w, h), resample=Image.BICUBIC) if resize_feature.lower() == 'fixsizenoratio': logger.info('resizing fixsize and igore ratio ...') w, h = resize_FixSize_w, resize_FixSize_h im = im.resize((w, h), resample=Image.BICUBIC) logger.info(f'target size:{im.size}') # convert PNG RGBA mode to RGB for JPEG if im.mode != 'RGB' and convert_format.lower() == 'jpeg': im = im.convert(mode='RGB') #save to target format in_mem_img = io.BytesIO() im.save(in_mem_img, format=convert_format, lossless=True, quality=save_quality, progressive=jpeg_progressive) in_mem_img.seek(0) except Exception as e: logging.error(json.dumps(e, default=str)) os._exit(0) return in_mem_img # put img to s3 def save_s3(bucket, key, body): logger.info(f'save to s3://{bucket}/{key}') try: response = client.put_object( Bucket=bucket, Key=key, Body=body ) logger.info(json.dumps(response, default=str)) except Exception as e: logging.error(json.dumps(e, default=str)) os._exit(0) return # change prefix from input/ to output/ and change postfix def change_key(key, convert_postfix): key_pre = os.path.splitext(key)[0].replace('input', 'output', 1) if not preserv_original_format: key_new = key_pre+convert_postfix else: key_new = key_pre+os.path.splitext(key)[1] return key_new def lambda_handler(event, context): logger.info(json.dumps(event, default=str)) bucket = event['Records'][0]['s3']['bucket']['name'] key = urllib.parse.unquote_plus(event['Records'][0]['s3']['object']['key'], encoding='utf-8') img_org = load_s3(bucket, key) body = img_convert(img_org, convert_format) key_new = change_key(key, convert_postfix) save_s3(bucket, key_new, body) return { 'statusCode': 200, 'body': 'OK' } - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -