lambda_function.py

import boto3
import os
from PIL import Image, ImageOps
import subprocess

s3_client = boto3.client('s3')
thumbnail_bucket = os.environ.get('THUMBNAIL_BUCKET')

IMG_EXT_LIST = ["jpg","jpeg","png","jfif","bmp","webp"]
VDO_EXT_LIST = ["mp4","mov"]

FFMPEG_BIN = "/opt/bin/ffmpeg"

def create_video_thumbnail_image(download_path, resized_path, quality=70):
 """
 비디오에서 첫 번째 프레임을 추출하여 썸네일 생성
 예외 발생 시 상위 함수에서 처리
 """
 cmd = [
 FFMPEG_BIN,
 "-i", download_path,
 "-vframes", "1",
 "-q:v", str(quality),
 resized_path
 ]
 subprocess.run(cmd, check=True, capture_output=True, text=True)
 create_thumbnail_image(resized_path, resized_path)

def create_thumbnail_image(image_path, resized_path, size=256, quality=30):
 """
 썸네일용 이미지 리사이징 (비율 유지, 짧은 변 최대 길이 thumb_size 제한, 256 * 256 크기로 crop), WebP 포맷
 """
 with Image.open(image_path) as image:
 image = ImageOps.exif_transpose(image)

 # 긴 변 기준 리사이즈 (비율 유지)
 width, height = image.size
 scale = size / min(width, height)
 new_width = int(width * scale)
 new_height = int(height * scale)
 image = image.resize((new_width, new_height), Image.LANCZOS)

 # 중앙 기준 crop
 left = (new_width - size) // 2
 top = (new_height - size) // 2
 right = left + size
 bottom = top + size
 image = image.crop((left, top, right, bottom))

 # WebP 포맷으로 저장 (품질 30%, 최적화 옵션 적용)
 image.save(resized_path, format='WEBP', quality=quality, optimize=True)

def delete_existing_thumbnails(bucket, prefix):
 """prefix 경로에 있는 모든 객체 삭제"""
 response = s3_client.list_objects_v2(Bucket=bucket, Prefix=prefix)

 if 'Contents' in response:
 objects_to_delete = [{'Key': obj['Key']} for obj in response['Contents']]
 if objects_to_delete:
 s3_client.delete_objects(
 Bucket=bucket, 
 Delete={'Objects': objects_to_delete}
 )

def lambda_handler(event, context):
 """
 event 예시 구조:
 {
 "bucket": "원본 이미지가 저장된 버킷",
 "key": "원본 이미지의 키 (S3 경로)"
 }
 """
 download_path = None
 upload_path = None
 filename = None
 
 try:
 bucket = event['bucket']
 key = event['key']
 post_id = key.split('/')[1]
 if not post_id.isdigit():
 error_msg = f"Post id not in format: {key}"
 print(error_msg)
 return {
 "statusCode": 400,
 "body": error_msg
 }

 filename = os.path.basename(key)
 base_name = filename.rsplit('.', 1)[0]
 ext = filename.lower().rsplit('.', 1)[-1]
 thumb_filename = f"thumbnail-{base_name}.webp"

 download_path = f'/tmp/{filename}'
 upload_path = f'/tmp/{thumb_filename}'

 # 기존 썸네일 삭제
 thumbnail_prefix = f'post-thumbnail/{post_id}/'
 delete_existing_thumbnails(thumbnail_bucket, thumbnail_prefix)
 print(f"Deleted existing thumbnails under {thumbnail_prefix}")

 if filename == "":
 # 빈 파일명 - 썸네일 삭제 성공 (최적화 작업이 아니므로 SQS 전송 안함)
 return {
 "statusCode": 200,
 "body": "Thumbnail deleted successfully."
 }
 else:
 # 원본 이미지 다운로드
 s3_client.download_file(bucket, key, download_path)
 print(f"Downloaded original image {key} from bucket {bucket}")

 # 썸네일 생성
 if ext in IMG_EXT_LIST:
 create_thumbnail_image(download_path, upload_path)
 elif ext in VDO_EXT_LIST:
 create_video_thumbnail_image(download_path, upload_path)
 else:
 error_msg = f"Attachment extension {ext} unsupported: {filename}"
 return {
 "statusCode": 400,
 "body": error_msg
 }
 print(f"Thumbnail created at {upload_path}")

 # 썸네일 업로드
 thumb_key = f'post-thumbnail/{post_id}/{thumb_filename}'
 with open(upload_path, 'rb') as f:
 s3_client.upload_fileobj(
 f,
 thumbnail_bucket,
 thumb_key,
 ExtraArgs={
 'ContentType': 'image/webp',
 'ACL': 'public-read',
 'CacheControl': 'public, max-age=31536000'
 }
 )
 print(f"Uploaded thumbnail to {thumbnail_bucket}/{thumb_key}")

 return {
 "statusCode": 200,
 "body": f"Thumbnail created successfully: {thumb_key}"
 }

 except Exception as e:
 print(f"Error processing {key}: {e}")

 return {
 "statusCode": 500,
 "body": f"Failed to create thumbnail: {str(e)}"
 }
 finally:
 for path in [download_path, upload_path]:
 try:
 if path and os.path.exists(path):
 os.remove(path)
 except:
 # 파일 삭제 오류 무시
 pass