Vb/image converter

helpers/webp_image_converter/samconfig.toml

@@ -0,0 +1,22 @@
+version = 0.1
+
+[default.deploy.parameters]
+stack_name = "dev-image-converter"
+s3_prefix = "dev-image-converter"
+parameter_overrides = "Stage=\"dev\", "
+confirm_changeset = false
+resolve_s3 = true
+region = "us-west-2"
+force-upload = true
+capabilities = "CAPABILITY_IAM"
+
+
+[prod.deploy.parameters]
+stack_name = "image-converter"
+s3_prefix = "image-converter"
+parameter_overrides = "Stage=\"prod\""
+confirm_changeset = true
+resolve_s3 = true
+region = "us-west-2"
+force-upload = true
+capabilities = "CAPABILITY_IAM"

helpers/webp_image_converter/src/app.py

@@ -0,0 +1,84 @@
+import io
+import os
+
+import boto3
+from PIL import Image
+from aws_lambda_powertools import Logger
+from aws_xray_sdk.core import patch_all
+from sosw.app import Processor as SoswProcessor, LambdaGlobals, get_lambda_handler
+
+__author__ = "Valentin Bakin"
+
+logger = Logger()
+patch_all()
+
+IS_PROD = os.getenv('env') == 'prod'
+
+
+class Processor(SoswProcessor):
+    DEFAULT_CONFIG = {
+        'init_clients': ['s3'],
+        's3_config':    {
+            'bucket_name': 'img-converter-test', # Change it with your actual s3 bucket
+            'path':        'img/', # Change it with your actual path
+
+            'image_types': ['jpg', 'jpeg', 'png', 'gif', 'bmp', 'tiff'],
+        },
+        'path_prefix':  '/prod',
+    }
+
+    s3_client: boto3.client = None
+
+
+    def get_config(self, name):
+        pass
+
+
+    def __call__(self, event):
+
+        bucket_name = self.config['s3_config']['bucket_name']
+        path = self.config['s3_config']['path']
+        objects = self.list_s3_objects(bucket_name, path)
+
+        for obj in objects:
+            key = obj['Key']
+            logger.info(obj)
+            file_format = key.split('.')[-1].lower()
+            if file_format not in self.config['s3_config']['image_types']:
+                continue
+
+            image_data = self.download_from_s3(bucket_name, key)
+            if file_format != 'webp':
+                webp_data = self.convert_to_webp(image_data)
+                new_key = key.rsplit('.', 1)[0] + '.webp'
+                self.upload_to_s3(bucket_name, new_key, webp_data)
+                # You can use it if you want to delete old images from s3 bucket,
+                # but it is not recommended cause not all browsers read webp format
+                # self.delete_from_s3(bucket_name,key)
+
+
+    def convert_to_webp(self, image_data):
+        image = Image.open(io.BytesIO(image_data))
+        output = io.BytesIO()
+        image.save(output, format="WEBP")
+        return output.getvalue()
+
+    def list_s3_objects(self, bucket_name, path):
+        response = self.s3_client.list_objects_v2(Bucket=bucket_name, Prefix=path)
+        logger.info(response.get('Contents', []))
+        return response.get('Contents', [])
+
+    def download_from_s3(self, bucket_name, key):
+        response = self.s3_client.get_object(Bucket=bucket_name, Key=key)
+        return response['Body'].read()
+
+    def upload_to_s3(self, bucket_name, key, data):
+        self.s3_client.put_object(Bucket=bucket_name, Key=key, Body=data)
+
+
+    def delete_from_s3(self, bucket_name, key):
+        self.s3_client.delete_object(Bucket=bucket_name, Key=key)
+
+
+global_vars = LambdaGlobals()
+lambda_handler = get_lambda_handler(Processor, global_vars)

helpers/webp_image_converter/src/requirements.txt

@@ -0,0 +1 @@
+Pillow==10.3.0

helpers/webp_image_converter/template.yaml

@@ -0,0 +1,43 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Transform: 'AWS::Serverless-2016-10-31'
+Description: Export any image to webp
+
+Parameters:
+  CustomFunctionName:
+    Type: String
+    Default: image_converter_to_webp
+  Stage:
+    Type: String
+    Default: dev
+  BucketName:
+    Type: String
+    Default: img-converter-test
+
+Conditions:
+  isProd: !Equals [ !Ref Stage, prod ]
+
+Resources:
+  ImageConverterToWEBP:
+    Type: AWS::Serverless::Function
+    Properties:
+      FunctionName: !If [ isProd, !Ref CustomFunctionName, !Sub "${Stage}_${CustomFunctionName}" ]
+      CodeUri: src/
+      Description: Image converter to webp
+      Handler: app.lambda_handler
+      MemorySize: 256
+      PackageType: Zip
+      Runtime: python3.11
+      Timeout: 60
+      Tracing: Active
+#      FunctionUrlConfig:
+##        AuthType: AWS_IAM #For future use
+#        AuthType: NONE
+#        InvokeMode: RESPONSE_STREAM
+
+      Policies:
+        - S3CrudPolicy:
+            BucketName: !Ref BucketName
+      Layers:
+        - !Sub 'arn:aws:lambda:${AWS::Region}:${AWS::AccountId}:layers:sosw:1'
+      Tags:
+        Env: !Ref Stage