import base64 import io import logging from collections import namedtuple import boto3 from PIL import Image # lambda function link: https://us-east-1.console.aws.amazon.com/lambda/home?region=us-east-1#/functions/image-resize-cloudfront?tab=aliases # # this lambda is to handle the image request from https://cdn2.suno.ai/.jpeg # # steps to update the functions: # 1. updating the code # 2. create a zip file with the following command in this folder # zip -r function.zip . # # 3. open the lambda and test the lambda is working # 4. publish the version in the lambda # 5. go to the aliases tab of lambda function and update the prod lambda to the version you published cloudwatchClient = boto3.client("cloudwatch") s3Client = boto3.client("s3") bucketName = "suno-data-uploads" prefix = "studio/uploads/" cacheLength = "2592000" # 30 days imageTypeSizeMap = {"small": 100, "medium": 256, "large": 360, "xlarge": 720} imageSizeTypeMap = {"100": "small", "256": "medium", "360": "large", "720": "xlarge"} logger = logging.getLogger(__name__) def send_metrics(eventName): try: cloudwatchClient.put_metric_data( Namespace="Cloudfront", MetricData=[ { "MetricName": "image-resize", "Dimensions": [{"Name": "EventName", "Value": eventName}], "Value": 1, "Unit": "Count", }, ], ) except Exception as e: logger.warning("nonfatal -- failed to send cloudwatch metrics: %s", e) def get_jpeg_file_key(type, filename): key = prefix + filename if type != "": key = key + "_" + type key = key + ".jpeg" return key def get_response(imageData, imageBase64): return { "statusCode": 200, "headers": { "Accept-Ranges": "bytes", "Content-Type": "image/jpeg", "Content-Length": str(len(imageData)), "Cache-Control": "public,max-age=" + cacheLength, }, "body": imageBase64, "isBase64Encoded": True, } Dimensions = namedtuple("Dimensions", ["width", "height"]) def get_thumbnail_dimensions( dimensions: Dimensions, to_size: int | None = None, to_width: int | None = None, to_height: int | None = None, ) -> Dimensions: width = dimensions.width height = dimensions.height aspect_ratio = width / height new_width = width new_height = height if to_size is not None: # Resize so shorter side is to_size px if width <= height: new_width = to_size new_height = int(new_width / aspect_ratio) else: new_height = to_size new_width = int(new_height * aspect_ratio) elif to_width is not None: # Resize so width is to_width px new_width = to_width new_height = int(new_width / aspect_ratio) elif to_height is not None: # Resize so height is to_height px new_height = to_height new_width = int(new_height * aspect_ratio) else: raise Exception("Missing axis measurement for resizing") return Dimensions(new_width, new_height) def lambda_handler(event, context): logger.info("event: %s", event) response = {"statusCode": 403} eventName = "bypass" pathParameters = event.get("pathParameters", {}) imageName = pathParameters.get("image_name", None) type = "" queryStringParameters = event.get("queryStringParameters", {}) if queryStringParameters: to_width = queryStringParameters.get("width", None) to_height = queryStringParameters.get("height", None) to_size = queryStringParameters.get("size", None) if to_width: if to_width not in imageSizeTypeMap: eventName = "bad_width" logger.info("bad width: %s", to_width) send_metrics(eventName) return response else: type = imageSizeTypeMap.get(to_width) elif to_height: if to_height not in imageSizeTypeMap: eventName = "bad_height" logger.info("bad height: %s", to_height) send_metrics(eventName) return response else: type = imageSizeTypeMap.get(to_height) elif to_size: if to_size not in imageSizeTypeMap: eventName = "bad_size" logger.info("bad size: %s", to_size) send_metrics(eventName) return response else: type = imageSizeTypeMap.get(to_size) try: if imageName.endswith((".jpeg")): filename = imageName[:-5] jpeg = get_jpeg_file_key(type, filename) logger.info("jpeg file key is: %s", jpeg) try: # if image already existing, return it s3Object = s3Client.get_object(Bucket=bucketName, Key=jpeg) imageData = s3Object["Body"].read() imageBase64 = base64.b64encode(imageData).decode("utf-8") response = get_response(imageData, imageBase64) eventName = "load_existing" logger.info("loading existing jpeg") except Exception as e: logger.info("could not find existing file, entering gen flow: %s", e) try: if type: logger.info("type is: %s", type) try: # if image not existing try to load the default jpeg and generate a customzied type jpeg s3Object = s3Client.get_object(Bucket=bucketName, Key=prefix + imageName) except Exception as e: logger.info("error getting exisging jpeg, trying to find png: %s", e) # if image not existing try to load the png png = prefix + filename + ".png" s3Object = s3Client.get_object(Bucket=bucketName, Key=png) else: png = prefix + filename + ".png" s3Object = s3Client.get_object(Bucket=bucketName, Key=png) s3Data = s3Object["Body"].read() image = Image.open(io.BytesIO(s3Data)) if type != "": width, height = image.size dimensions = Dimensions(width, height) size_by_type = imageTypeSizeMap.get(type) resized_dimensions = get_thumbnail_dimensions( dimensions, to_size=size_by_type if to_size else None, to_width=size_by_type if to_width else None, to_height=size_by_type if to_height else None, ) image = image.resize(resized_dimensions, Image.Resampling.LANCZOS) rgbIm = image.convert("RGB") buffer = io.BytesIO() rgbIm.save(buffer, format="JPEG", quality=95) buffer.seek(0) imageData = buffer.read() s3Client.put_object( Bucket=bucketName, Key=jpeg, Body=imageData, ContentType="image/jpeg", ) imageBase64 = base64.b64encode(imageData).decode("utf-8") response = get_response(imageData, imageBase64) logger.info("successfully created jpeg") eventName = "success_create" except Exception as e: logger.info("failed jpeg gen: %s", e) eventName = "failed_gen_jpeg" except Exception as e: logger.info("error in outer try statement: %s", e) eventName = "unknown_error" send_metrics(eventName) return response