Elemental MediaConvert 사용기 with FastAPI, S3 Presigned URL

 안녕하세요.

사이드 프로젝트로 챌린지 관련 프로젝트를 진행하고 있습니다.

 

사용자들은 챌린지를 수행했다는 인증 영상을 촬영해서 업로드하고, 관리자는 그 영상을 보고 승인 여부를 기록 합니다.

여기서 든 생각은 사용자들은 갤럭시, 아이폰 등을 사용하여 영상을 촬영할 것이고, 촬영한 영상을 업로드 한다면 보통 4K 30~60fps, HEVC(H.265)로 S3에 저장될 것 입니다. 스토리지 비용이 많이 발생할 것으로 생각되어 많은 고민을 한 결과 MediaConvert를 이용해서 트랜스코딩 해서 영상 화질과 비트레이트를 낮추는 걸로 결정했습니다.

Media Convert란?

• 클라우드 기반 비디오 트랜스코딩(Transcoding; 사이즈와 압축(Codec)) 서비스
• 즉 ,한 가지 포맷의 영상 → 다양한 포맷/코덱/해상도로 변환해주는 역할
• 완전 관리형(Managed Service)이라서 직접 인프라나 인코더 장비를 운영할 필요 없음

주요 기능

1. 포맷 변환 (Transcoding)
   1. 입력: MP4, MOV, MPEG-TS 등
   2. 출력: H.264, H.265(HEVC), AV1, MPEG-2, VP9 등 다양한 코뎅
   3. 해상도: 4K, 1080p, 720p, Adaptive Bitrate(ABR) 지원
2. 오디오 채널
   1. 오디오 채널 매핑, 다국어 트랙, Dolby Audio,  AAC 등 지원
3. 자막 처리
   1. 자막(Closed Caption, Subtitles) 삽입 가능
4. 자동 스케일링
   1. 워크로드에 따라 자동으로 확장/축소 → 인코딩 작업이 몰려도 안정적

비디오 파이프라인에서의 역할

사이드 프로젝트에서 진행하는 파이프라인은 다음과 같다

1. 사용자가 10초 이내 인증 영상을 S3(input_videos/<user_id>/<yymmdd>/<uuid>) 업로드
2. 트랜스코딩(MediaConvert) → 영상을 여러 해상도/코덱/포맷으로 변환
   1. 720p(720x480), bitrate 1500000, MPEG-4 AVC(H.264)
3. S3(output_videos/<user_id> /<yymmdd>/<uuid>) 저장

FastAPI 로 구현

1. S3 서비스 만들기

https://nueijeel.tistory.com/69

• Presigned URL 생성하는 S3Service를 생성
• 나중에 변환 작업을 요청할 때 필요한 object_key (S3 내 파일의 고유 주소) 를 함께 반환합니다

# app/services/s3_service.py
import boto3
import uuid
from typing import Dict
from app.config import config
from botocore.exceptions import NoCredentialsError


class S3Service:
	def __init(self, region_name: str = 'ap-northeast-2'):
    	self.s3_client = boto3.client('s3', region_name=region_name)
        self.bucket_name = config.STORAGE_BUCKET_NAME
        self.storage_domain = config.STORAGE_DOMAIN
        
def generate_presigned_url(
        self,
        dir: str,
        file_extension: str = "",
        expiration: int = 300,
        content_type: str = "application/octet-stream",
    ) -> dict[str, str]:
    object_key = f"{dir}/{uuid.uuid4()}.{file_extension}"

    try:
        presigned_url = self.s3_client.generate_presigned_url(
            "put_object",
            Params={
                "Bucket": self.bucket_name,
                "Key": object_key,
                "ContentType": content_type,
            },
            ExpiresIn=expiration,
        )
        return {
            "upload_url": presigned_url,
            "download_url": f"{self.storage_domain}/{object_key}",
            "object_key": object_key,
        }

    except NoCredentialsError:
        raise Exception("AWS credentials not available")

2. FastAPI 엔드포인트 구현

• 중요한 부분은   # * 로 표시했습니다.
• 서비스 리포지토리 패턴(Service-Repository Pattern) 계층형 구조
  • 요청 → FastAPI  (Schema로 Deserialization) → Controller → Service → Repository
  • Repository → Service → Controller → FastAPI (Schema로 Serialization) → 응답

# app/api/v1/endpoints/storage_controller.py
from fastapi import APIRouter, Depends, Query, HTTPException, status, Body
from pydantic import BaseModel, Field


router = APIRouter(prefix="/storages", tags=["파일 업로드"])

# 아래의 클래스는 schema에 위치해야 되지만 간단한 설명을 위해 여기 적음
class UploadURLResponse(BaseModel):
    upload_url: str = Field(..., description="파일을 업로드할 Presigned URL")
    object_key: str = Field(..., description="업로드 완료 후 파일의 다운로드 URL")
    download_url: str = Field(..., description="S3에 저장된 객체 키")
    

class ConversionRequest(BaseModel):
    object_key: str = Field(..., description="S3에 업로드가 완료된 비디오의 객체 키")
    
  
class ConversionResponse(BaseModel):
    message: str = Field(..., description="작업 상태에 대한 메시지")
    job_id: str = Field(..., description="생성된 AWS MediaConvert 작업의 고유 ID")
    url: str = Field(..., description="변환이 완료된 MP4 파일의 최종 URL")
    

import logging
from datetime import datetime
from pathlib import Path
from zoneinfo import ZoneInfo

import boto3
from botocore.exceptions import ClientError
from fastapi import APIRouter, Body, Depends, HTTPException, Query, status

from app.api.v1.schemas.storage_schema import (
    ConversionRequest,
    ConversionResponse,
    PresignedUrlResponse,
)
from app.config import config
from app.config.get_current_user import get_current_user
from app.services.s3_service import S3Service
from app.utils.response_helper import create_response

router = APIRouter()
logger = logging.getLogger(__name__)

mediaconvert_client = boto3.client(  # *
    "mediaconvert",
    region_name=config.AWS_REGION_NAME,
    endpoint_url=config.AWS_MEDIACONVERT_ENDPOINT_URL,
)


@router.get("/upload_url", summary="파일 upload를 위한 Presigned URL 생성")
def generate_presigned_url(
    file_extension: str = Query(..., description="파일 확장자 (. 제외) eg) mp4, png"),
    user_id: str = Depends(get_current_user),
):
    try:
        if not file_extension:
            return create_response(
                {"error": "param is not valid."}, status.HTTP_400_BAD_REQUEST, "failed"
            )

        if file_extension == "png":
            upload_dir = f"images/{user_id}"
            content_type = "image/png"
        elif file_extension == "mp4":
            upload_dir = f"videos/input_videos/{user_id}"
            content_type = "video/mp4"
        else:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=f"지원하지 않는 파일 확장자입니다: '{file_extension}'. ('png', 또는 'mp4'만 가능)",
            )

        s3_service = S3Service()
        presigned_url = s3_service.generate_presigned_url(
            dir=upload_dir, file_extension=file_extension, content_type=content_type
        )

        return create_response(
            PresignedUrlResponse(**presigned_url).model_dump(),
            status.HTTP_200_OK,
            "success",
        )
    except Exception as err:
        return create_response(
            {"error": str(err)}, status.HTTP_500_INTERNAL_SERVER_ERROR, "failed"
        )


@router.post(
    "videos/start-conversion",
    summary="S3 비디오 변환 작업 시작",
    response_model=ConversionResponse,
    status_code=status.HTTP_202_ACCEPTED,
)
def start_video_conversion(
    request: ConversionRequest = Body(...), user_id: str = Depends(get_current_user)
):
    """
    프론트앤드에서 S3로 비디오 업로드가 완료된 후, 이 엔드포인트를 호출하여 MediaConvert 변환 작업을 시작합니다.
    """
    object_key = request.object_key

    seoul_tz = ZoneInfo("Asia/Seoul")
    now = datetime.now(tz=seoul_tz).strftime("%Y%m%d")

    input_s3_path = f"s3://{config.STORAGE_BUCKET_NAME}/{object_key}"
    output_key_prefix = f"videos/output_videos/{user_id}/{now}/"

    try:
        job_settings = {
            "Role": config.AWS_MEDIACONVERT_ROLE_ARN,
            "Settings": {
                "Inputs": [
                    {
                        "FileInput": input_s3_path,
                        "AudioSelectors": {
                            "Audio Selector 1": {"DefaultSelection": "DEFAULT"}
                        },
                        "VideoSelector": {},
                    }
                ],
                "OutputGroups": [
                    {
                        "Name": "File Group",
                        "OutputGroupSettings": {
                            "Type": "FILE_GROUP_SETTINGS",
                            "FileGroupSettings": {
                                "Destination": f"s3://{config.STORAGE_BUCKET_NAME}/{output_key_prefix}"
                            },
                        },
                        "Outputs": [
                            {
                                "ContainerSettings": {"Container": "MP4"},
                                "VideoDescription": {
                                    "Width": 720,  # *
                                    "Height": 480,  # *
                                    "CodecSettings": {
                                        "Codec": "H_264",  # *
                                        "H264Settings": {
                                            "MaxBitrate": 1500000,  # *
                                            "RateControlMode": "QVBR",
                                            "QualityTuningLevel": "SINGLE_PASS_HQ",
                                            "FramerateControl": "SPECIFIED",
                                            "FramerateNumerator": 24,
                                            "FramerateDenominator": 1,
                                        },
                                    },
                                },
                                "AudioDescriptions": [
                                    {
                                        "AudioSourceName": "Audio Selector 1",
                                        "CodecSettings": {
                                            "Codec": "AAC",
                                            "AacSettings": {
                                                "Bitrate": 96000,
                                                "CodingMode": "CODING_MODE_2_0",
                                                "SampleRate": 48000,
                                            },
                                        },
                                    }
                                ],
                            }
                        ],
                    }
                ],
            },
        }

        job_response = mediaconvert_client.create_job(**job_settings)
        job_id = job_response["Job"]["Id"]

    except ClientError as e:
        logger.error(f"MediaConvert 작업 생성 실패: {e}")
        raise HTTPException(
            status_code=500, detail="영상 변환 작업 요청에 실패 했습니다."
        )

    filename = f"{Path(object_key).stem}.mp4"
    url = f"{config.STORAGE_DOMAIN}/{output_key_prefix}{filename}"

    # 사용자님의 스타일에 맞춰 create_response 헬퍼 사용
    response_data = {
        "message": "영상 변환 작업이 성공적으로 시작되었습니다.",
        "job_id": job_id,
        "url": url,
    }
    return create_response(
        ConversionResponse(**response_data).model_dump(),
        status.HTTP_202_ACCEPTED,
        "success",
    )

 

3. Postman 테스트

1. Postman 테스트에서 나온 응답에 upload_url 복사

• postman URL 에 붙여넣기
• Body → binary 선택 후 mp4 파일 업로드
• Headers → Content-Type=video/mp4 입력

영상 업로드

Content-Type 설정

성공

2. Postman 테스트에서 나온 응답에 object_key 복사

성공

3. S3 업로드 확인

마무리

성공

 FastAPI, Presigned URL, MediaConvert 를 사용하여 영상을 원하는 해상도 및 비트레이트를 수정하여 저장하는 방법에 대해 알아봤습니다. 궁금한 점이나 더 설명이 필요하시면 댓글로 남겨주시면 감사하겠습니다.