AWS S3트리거를 사용하여 동영상 번역 및 자막 삽입 하기

Published in

Cloud Villains

17 min readOct 17, 2023

개요

이번에 AWS S3 트리거를 사용하여 동영상 번역을 진행 해 보았습니다.

지난 번 사내에서 본사(한국)로부터 회사 관련 동영상이 공유 되었지만, 그 내용이 한국어로 되어 있어 자막이 필요하다는 팀원이 있었습니다.
저는 이미 AWS에서 영상이나 언어 관련 서비스를 제공하고 있는 것을 알고 있어서 부담없이 자막 관련 작업을 진행 해 보았고, 그 내용을 본 게시글로 작성 하게 되었습니다.

Infrastructure

이번에 생성할 Infrastructure는 다음과 같습니다.

User가 동영상을 S3에 업로드하면 Lambda가 움직이게 됩니다.
Lambda는 동영상을 Amazon Transcribe로 SRT를 추출하고 File을 Amazon Translate로 번역합니다. 마지막으로 번역된 SRT를 동영상에 넣어 S3에 업로드하는 메커니즘입니다.

목표로 하는 Output은 아래의 왼쪽 Before를 오른쪽 After처럼 만드는 것입니다.

본 블로그는

구축 절차
좋았던 포인트
감상

순서로 써 보겠습니다.

1. 구축 절차

이번 구축은 기본 Console상에서 실시했습니다. 다음 순서로 설명드리겠습니다.

1) Lambda생성
2) S3생성
3) Lambda functions생성
4) 동작 확인

1) Lambda 생성

Lambda에서 생성 합니다.
값은 아래와 같이 Name과 Python의 선택 정도만 진행 하면 됩니다.

만든 Lambda 페이지는 다음과 같습니다.

function은 S3 트리거 설정 후에 설명 드리겠습니다.

2) S3 생성

S3 Bucket을 만듭니다. 이름을 입력하고 다른 Options는 Default로 두겠습니다.

S3 Bucket이 완료되면 Create folder에서 video, srt, substitle을 만듭니다.
각 역할은 다음과 같습니다.

YourBucket/video/ : 동영상을 업로드 할 트리거 용
YourBucket/srt/ : srt file보관용
YourBucket/substitle/ : 동영상 자막 버전 보관용

그럼 트리거 folder를 설정 해 보겠습니다.
Bucket의 Properties에 태그를 바꿉니다.

아래로 스크롤하면 Event notifications(트리거)가 있으므로 Event를 생성합니다.

Event는 다음과 같이 설정합니다.
Video folder에 Upload(PUT)하면, ‘1) Lambda생성’에서 생성한 Lambda function을 호출 할 수 있습니다.

위의 설정으로 만들면 다음과 같이 보입니다.

그리고 Lambda의 내용을 보면 트리거로 나타납니다.

3) Lambda functions 생성

여기까지 진행하여 S3트리거를 사용하여 Lambda를 호출하게 되었으므로, Lambda에 S3 권한을 부여하는 부분부터 설명 드리겠습니다.

IAM의 Roles로 이동하면, ‘1) Lambda생성’으로 생성했을 때에 만들어진 Role이 있을겁니다. 해당 역할 이름을 클릭합니다.

Add permissions 드롭다운에서 Attach policies를 클릭합니다.

S3에서 검색하면 여러 가지가 표시되지만 S3FullAccess를 선택하고, Add permissions를 클릭하여 Attach를 수행합니다.

이제 S3의 권한을 부여할 수 있게 되었습니다.

Lambda functions는 다음과 같이 만들었습니다.

간단하게 설명 드리자면, Amazon Transcribe에서 S3(video/)에 업로드 한 .mp4 언어를 식별하고 srt 파일을 S3(srt/)에 생성합니다.
해당 srt 파일을 Amazon Translate에서 번역 버전으로 만듭니다.
번역판의 srt file을 S3(video/)에 있는 .mp4에 넣고 S3(substitle/)에 Upload합니다.

import json
import re
import boto3
import random
import string
import subprocess

s3_client = boto3.client('s3')
transcribe_client = boto3.client('transcribe')
translate_client = boto3.client('translate')

TRANSCRIBE_LANG_CODE = {
    'en' : 'en-US' ,
    'ja' : 'ja-JP' ,
    'ko' : 'ko-KR' ,
    'zh' : 'zh-CN'
}
FFMPEG_LANG_CODE = {
    'en' : 'eng' ,
    'ja' : 'jpn' ,
    'ko' : 'kor' ,
    'zh' : 'zho'
}
SOURCE_LANG = 'en'
TARGET_LANG = 'ja'

LOCAL_SRT_PATH = '/tmp/jp_sub.srt'
LOCAL_VIDEO_PATH = '/tmp/video.mp4'
LOCAL_SUBTITLE_VIDEO_PATH = '/tmp/subtitle_video.mp4'

SRT_KEY_FOLDER = 'srt/'
SUBSTITLE_KEY_FOLDER = 'substitle/'

# TranscriptionJobName to avoid duplication
def generate_random_string(length):
    letters = string.ascii_letters
    return ''.join(random.choice(letters) for _ in range(length))

# Create Transcribe and Save to S3
def create_transcription_job_save_s3(bucket, key):
    random_string = generate_random_string(3) # make a TranscriptionJobName
    transcribe_job_name = key.split('/')[1].split('.')[0] + random_string
    transcribe_response = transcribe_client.start_transcription_job(
        TranscriptionJobName=transcribe_job_name,
        Subtitles={'Formats': ['srt']},
        Media={'MediaFileUri': f's3://{bucket}/{key}'},
        LanguageCode=TRANSCRIBE_LANG_CODE[SOURCE_LANG],
        OutputBucketName = bucket,
        OutputKey = SRT_KEY_FOLDER #save to srt folder
    )
    return transcribe_job_name

# Get S3 and Translate Target Languege
def get_s3_translate_text(bucket, key_srt):
    srt_body = s3_client.get_object(Bucket=bucket,Key=key_srt)['Body'].read().decode('utf-8')
    translate_response = translate_client.translate_text(
        Text= srt_body,
        SourceLanguageCode=SOURCE_LANG,
        TargetLanguageCode=TARGET_LANG
    )
    return translate_response['TranslatedText']

# When translated by AWS translate, the timeline format changes, so it is transformed to fit the srt format.
def format_subtitles(translated_text):
    pattern = r"(\d{2}:\d{2}:\d{2},\d{3})--> (\d{2}:\d{2}:\d{2},\d{3})"
    replacement = r"\1 --> \2"
    return re.sub(pattern, replacement, translated_text)

def lambda_handler(event, context):
    
    #Get our bucket and file name
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key']

    transcribe_job_name = create_transcription_job_save_s3(bucket, key)
    
    # looping until the work is finished
    while True:
        status = transcribe_client.get_transcription_job(TranscriptionJobName=transcribe_job_name)['TranscriptionJob']['TranscriptionJobStatus']
        if status in ['COMPLETED', 'FAILED']:
            break
    
    key_srt = SRT_KEY_FOLDER + transcribe_job_name + '.srt'
    translated_text = get_s3_translate_text(bucket, key_srt)
    formatted_subtitle = format_subtitles(translated_text)
    
    
    # srt save to local
    with open(LOCAL_SRT_PATH, "a") as f:
        f.write(formatted_subtitle)

    s3_client.download_file(bucket, key, LOCAL_VIDEO_PATH)
    
    # Add subtitles to videos using FFmpeg
    command = [
        "/opt/bin/ffmpeg",
        "-y",
        "-i", LOCAL_VIDEO_PATH,
        "-i", LOCAL_SRT_PATH,
        "-c", "copy",
        "-c:s", "mov_text",
        "-metadata:s:s:0", "language="+ FFMPEG_LANG_CODE[TARGET_LANG], 
        LOCAL_SUBTITLE_VIDEO_PATH
    ]
    subprocess.run(command, check=True)
    
    # Upload Subtitle Video
    output_key_name = SUBSTITLE_KEY_FOLDER + key.split('/')[1].split('.')[0] + f'-subtitle.mp4' # S3 uplpad key name e.g. substitle/sample-subtitle.mp4
    s3_client.upload_file(LOCAL_SUBTITLE_VIDEO_PATH, bucket, output_key_name)
    
    # Delete Transcription job
    response = transcribe_client.delete_transcription_job(TranscriptionJobName=transcribe_job_name)

    return {
        'statusCode': 200,
        'body': json.dumps('Saved video subtitles!')
    }

Code를 생성 및 수정할 때마다 Deploy 해 둡니다.

다음은 ffmpeg를 Layers에 등록합니다.
먼저 자신의 환경에서 다음 Command를 실행합니다. ffmpeg가 포함된 Zip을 만듭니다.

$ wget https://johnvansickle.com/ffmpeg/releases/ffmpeg-release-amd64-static.tar.xz
$ tar xvf ffmpeg-release-amd64-static.tar.xz
$ mkdir -p ffmpeg/bin
$ cp ffmpeg-6.0-amd64-static/ffmpeg ffmpeg/bin/
$ cd ffmpeg
$ zip -r ../ffmpeg.zip .

이 Zip을 ‘2) S3생성’으로 생성한 Bucket에 Upload 해 둡니다.

Lambda의 Layers로 이동하여 Layer를 만듭니다.
방금 전에 ffmpeg Zip을 업로드 한 Object URL을 붙여 넣습니다.

Layer 생성이 끝나면 세부 정보 페이지에 ARN이 표시됩니다. (아래 빨간색 상자) 이를 복사 해 주세요. (image에 있는 Version은 신경 쓰지 말아 주세요. 처음으로 작성되면 1로 되어 있을 것 입니다.)

Lambda function으로 돌아가서 생성한 Layer를 등록해 둡니다.
Add Layer에서 Specify an ARN을 선택하고 Copy한 ARN을 붙여넣고 Verify까지 진행합니다.
그 후 Add를 클릭하여 Lambda function에 Layer를 등록합니다.

Lambda function의 OverView에는 S3 트리거와 Layer(1)가 표시됩니다.

마지막으로 functions의 timeout을 설정합니다.
1분으로 설정 해 두었습니다.

4) 동작 확인

S3 Video에 샘플 동영상을 업로드합니다.

이 트리거를 사용하면 동일한 타이밍에 CloudWatch Logs가 생성되는지 확인할 수 있습니다.

Logs의 내용을 확인하여 끝까지 실행 된 것이 확인되면 S3의 srt/와 substitle/를 확인합니다.
시간을 보게되면 실행 완료까지 1분 정도 걸리는 것을 알 수 있습니다.

동영상에 자막이 있는 파일을 확인한 후 sample-substitle.mp4를 다운로드하여 확인 해 보았습니다.

시험 삼아 한국어로도 해 보았습니다.

추가로 Code의 Debug나 동작 확인을 위해, Configure test event를 설정해 두었습니다.
템플릿은 S3Put을 선택하고 bucket > name에는 Bucket 이름을 object > key 에는 Video / {~.mp4}로 설정했습니다.

좋았던 포인트

ffmpeg layer에 대해

Lambda에서 라이브러리를 사용할 때 import 이외의 방법은 별로 경험이 없었습니다. 보통 ffmpeg를 import하여 진행했는데, not found 에러에 빠져 버렸습니다. 그렇지만 나중에 패키징과 레이어를 사용하여 문제를 해결했습니다.
구체적으로 설명 드리면 ffmpeg를 포함한 사용자 정의 Layer를 만들고 Lambda 함수에서 해당 Layer를 사용하여 필요한 라이브러리와 실행 파일을 제공했습니다. 이렇게 하면 라이브러리를 찾을 수 없다는 오류를 피하면서 순조롭게 ffmpeg를 사용하여 작동 시킬 수 있었습니다.
이 경험을 통해 Lambda에서 외부 라이브러리를 사용할 때 패키징과 Layer의 활용이 중요하다는 것을 배웠습니다.

AWS translate 및 srt에 대해

AWS translate로 srt 파일을 번역하면 srt format이 무너지는 것을 알았습니다.
예를 들어, 맞는 것 : 00:01:23,150 → 00:01:27,880
틀린 것 : 00:01:23,150 → 00:01:27,880
상기의 틀린 것에서 나타낸 것처럼 화살표 모양이 변경되어 ffmpeg로 동영상에 자막을 넣을 때 invalid data 오류가 발생했습니다.
따라서 화살표 형식을 올바르게 수정하는 format_subtitles() 함수를 만들어 해결했습니다.
또한 AWS translate를 이용한 번역이어서, 필요에 따라 의역이 필요하다고 생각했습니다.

감상

오랜만에 AWS를 사용하여 즐거운 시간을 보냈습니다. 실제로 해 보면서 향후 어떻게 활용할 수 있을 지에 대해 공부 가능한 점이나 얻을 수 있는 것이 많았습니다.

이번에는 Lambda를 사용하여 동영상 파일을 기록하고 번역을 하여 자막을 붙여 보았지만, 한편으로는 Lambda 함수의 스케일링에 관한 부분이 부족했다고 느꼈습니다.

특히, TransscribeService의 start_transcription_job이라는 동영상 기록과 관련된 처리는 기록(이 경우에서는 SRT 형식의 생성)에 시간이 걸리는 사양이었습니다. Lambda내에서 실행하기 위해 작업의 완료 상태를 get_transcription_job을 사용하여 검색하고 완료 할 때까지 while 루프에서 대기했습니다. 이 방법으로 문제는 없었지만, 이러한 실시간 실행이 적절한지 여부를 계속 고민했습니다.
덧붙여서, 실시간으로 실행 가능한 start_stream_transcription 이라는 이벤트도 존재했지만, 이번과 같은 자막 대응에는 적용할 수 없었습니다.

Lambda 함수의 스케일링에 관한 것도 아직 많이 있어, 앞으로 실제로 경험하면서 깊이 이해 해 나가고 싶다고 생각 했습니다.