포스트

[2부] 스토리북 자동화 처리 시스템 구축기: 동영상 생성 엔진과 엔드투엔드 통합

게시 업데이트
By SouthGlory
14 분읽는 시간

[영상]

들어가며 ** [1부]에서는 스토리북 자동화 시스템의 전체 구조와 안정성 확보 방법에 대해 알아보았습니다. https://blog.naver.com/devramyun/223811470692

[1부] 스토리북 자동화 처리 시스템 구축기: 파이프라인 설계와 안정성 확보 들어가며… AI와 자동화 기술이 콘텐츠 생산 방식을 혁신하고 있는 시대입니다. 저는 수천 개의 스토리… 들어가며… AI와 자동화 기술이 콘텐츠 생산 방식을 혁신하고 있는 시대입니다. 저는 수천 개의 스토리…

이번 포스트에서는 파이프라인의 핵심인 동영상 생성 엔진(movie_maker)의 상세 구현과 세 가지 핵심 컴포넌트가 어떻게 통합되어 완전한 엔드투엔드 솔루션을 구성하는지 살펴보겠습니다.

동영상 생성 엔진 설계 ** movie_maker 컴포넌트는 다운로드된 스토리북 콘텐츠(이미지, 텍스트)를 매력적인 동영상으로 변환하는 역할을 담당합니다.

이 컴포넌트는 단일 책임 원칙(Single Responsibility Principle)을 따라 모듈화된 구조로 설계되었습니다.

모듈화된 아키텍처

1
2
3
4
5
6

movie_maker/
├── maker/
│   ├── movie_maker_v2.py     # 실제 동영상 생성 로직
│   └── utils/                # 유틸리티 함수들
├── constants/                # 상수 및 설정
└── tester.py                 # 명령줄 인터페이스

이러한 모듈화된 구조는 코드의 유지보수성을 높이고, 개별 기능을 독립적으로 테스트할 수 있게 합니다.

강력한 명령줄 인터페이스 ** tester.py는 다양한 매개변수를 지원하는 명령줄 인터페이스를 제공합니다:

1
2
3
4
5
6
7
8
9
10
11
12
13

parser = argparse.ArgumentParser(description="동화책 이미지를 동영상으로 변환")
parser.add_argument("--book_path", type=str, help="책 폴더 경로")
parser.add_argument("--font_kr_path", type=str, help="한글 폰트 경로")
parser.add_argument("--intro_path", type=str, help="인트로 이미지 경로")
parser.add_argument("--voice_path", type=str, help="음성 파일 경로")
parser.add_argument("--voice_id", type=str, help="음성 아이디")
parser.add_argument("--font_size", type=int, help="폰트 크기")
parser.add_argument("--line_spacing", type=int, help="줄 간격")
parser.add_argument("--stroke_width", type=int, help="외곽선 두께")
parser.add_argument("--padding", type=int, help="여백")
parser.add_argument("--max_pixel_width", type=int, help="최대 픽셀 너비")
parser.add_argument("--lines_at_once", type=int, help="한 번에 표시할 줄 수")
parser.add_argument("--language_code", type=str, help="언어 코드")

이를 통해 사용자는 세밀한 설정으로 동영상 생성 과정을 제어할 수 있습니다:

1
2
3

python tester.py --book_path "./book_123" --font_kr_path "./fonts/korean.ttf" \
                 --font_size 60 --line_spacing 8 --stroke_width 4 \
                 --language_code "ko-KR"

FFmpeg 기반 고품질 렌더링 ** 동영상 생성의 핵심은 FFmpeg 라이브러리의 활용입니다. 이를 통해 이미지, 텍스트, 오디오 요소를 결합하여 전문적인 품질의 동영상을 생성합니다:

1
2
3
4
5
6
7
8
9

# 예시: FFmpeg 명령 구성 (단순화된 버전)
ffmpeg_cmd = [
    "ffmpeg", 
    "-i", f"{image_path}", 
    "-i", f"{audio_path}",
    "-c:v", "libx264", "-preset", "medium", 
    "-c:a", "aac", "-b:a", "192k",
    f"{output_path}"
]

자산 다운로더와의 연계 ** movie_maker는 asset_loader_program에서 다운로드한 자산을 활용합니다.

두 컴포넌트 간의 원활한 연동을 위해 표준화된 경로 구조를 사용합니다:

1
2
3
4
5
6
7
8
9

# StoryQueueProcessor.py에서 경로 구성
book_path = os.path.join(base_path, "output", f"book_{bookid}")

# movie_maker에서 동일한 경로 구조 활용
result = subprocess.run([
    "./AI스토리북동영상생성기.exe",
    "--book_path", book_path,
    # 기타 매개변수...
])

견고한 오류 처리 시스템 ** 동영상 생성 과정은 다양한 예외 상황(파일 접근 오류, 리소스 부족, 외부 서비스 장애 등)에 노출됩니다.

이러한 상황에 대응하기 위한 포괄적인 오류 처리 시스템을 구현했습니다.

main_program_decorator를 통한 계층적 예외 처리 ** tester.py의 핵심은 강력한 데코레이터 패턴을 통한 예외 처리입니다:

1
2
3
4
5

@main_program_decorator
def run_main(**kwargs):
    from maker.movie_maker_v2 import run
    run(book_path, voice_path, voice_id, intro_path)
    return True

이 데코레이터는 다양한 종류의 예외를 계층적으로 처리합니다:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

try:
    # 메인 로직 실행
except KeyboardInterrupt:
    myLogger.warning("사용자가 실행을 중단했습니다 (Ctrl+C)")
    myLogger.set_exit_code(EXIT_KEYBOARD_INTERRUPT)
except FileNotFoundError as e:
    myLogger.error(f"파일 또는 디렉토리를 찾을 수 없음: {e}")
    myLogger.set_exit_code(EXIT_PATH_NOT_FOUND)
except OSError as e:
    # 경로 관련 OS 오류 처리 (권한 문제 등)
    if "No such file or directory" in str(e):
        myLogger.error(f"파일 또는 디렉토리를 찾을 수 없음: {e}")
        myLogger.set_exit_code(EXIT_PATH_NOT_FOUND)
    elif "Permission denied" in str(e):
        myLogger.error(f"파일 또는 디렉토리 접근 권한 없음: {e}")
        myLogger.set_exit_code(EXIT_PATH_ACCESS_DENIED)
    else:
        myLogger.error(f"OS 오류 발생: {e}")
        myLogger.set_exit_code(EXIT_GENERAL_ERROR)
except Exception as e:
    # 기타 모든 예외 처리
    error_msg = str(e)
    stack_trace = traceback.format_exc()
    # 예외 유형별 처리...

세분화된 종료 코드 ** 오류 상황을 정확히 식별하고 적절히 대응하기 위해 9가지 상세한 종료 코드를 정의했습니다:

1
2
3
4
5
6
7
8
9

EXIT_SUCCESS = 0            # 성공적인 완료
EXIT_GENERAL_ERROR = 1      # 일반 오류
EXIT_KEYBOARD_INTERRUPT = 2 # 사용자 중단
EXIT_API_QUOTA_ERROR = 3    # API 할당량 초과
EXIT_AUTH_ERROR = 4         # 인증 오류
EXIT_CRITICAL_ERROR = 5     # 심각한 오류
EXIT_PATH_NOT_FOUND = 6     # 경로 찾을 수 없음
EXIT_PATH_INVALID = 7       # 유효하지 않은 경로
EXIT_PATH_ACCESS_DENIED = 8 # 경로 접근 거부

이러한 세분화된 종료 코드는 StoryQueueProcessor가 오류 원인을 정확히 파악하고 적절히 대응할 수 있게 해줍니다:

1
2
3
4
5
6
7
8
9
10
11

# StoryQueueProcessor.py에서 종료 코드 처리
result = subprocess.run([...], capture_output=True, text=True, check=False)

if result.returncode != EXIT_SUCCESS:
    logger.error(f"❌ 동영상 생성 실패 (코드: {result.returncode})")
    # 코드별 오류 메시지 출력
    if result.returncode == EXIT_PATH_NOT_FOUND:
        logger.error("경로를 찾을 수 없음")
    elif result.returncode == EXIT_PATH_INVALID:
        logger.error("경로 형식 오류")
    # 추가 오류 코드 처리...

사전 경로 검증 ** 동영상 생성 프로세스를 시작하기 전에 모든 필수 입력 경로의 유효성을 철저히 검증합니다:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

# 모든 필수 경로 검증
paths_to_check = {
    "책 경로": mySettings.book_path, 
    "폰트 경로": mySettings.font_kr_path,
    "인트로 이미지 경로": mySettings.intro_path, 
    "음성 파일 경로": mySettings.voice_path
}

for path_name, path_value in paths_to_check.items():
    if path_value:
        if os.path.exists(path_value):
            myLogger.info(f"{path_name} 확인 완료: {path_value}")
        else:
            myLogger.error(f"{path_name}가 존재하지 않습니다: {path_value}")
            myLogger.set_exit_code(EXIT_PATH_NOT_FOUND)
            return None

이러한 사전 검증은 프로세스 중간에 발생할 수 있는 예기치 않은 오류를 방지하고, 실패의 원인을 명확하게 식별할 수 있게 합니다.

엔드투엔드 콘텐츠 처리 파이프라인 ** 지금까지 살펴본 세 가지 핵심 컴포넌트

StoryQueueProcessor,

asset_loader_program,

movie_maker

가 어떻게 통합되어 완전한 엔드투엔드 파이프라인을 구성하는지 살펴보겠습니다.

  • 작업 큐 관리 (StoryQueueProcessor): API로부터 READY 상태의 책을 확인하고 작업 상태 추적
  • 콘텐츠 다운로드 (asset_loader_program): API에서 책 데이터와 이미지를 다운로드하여 구조화된 형태로 저장
  • 동영상 생성 (movie_maker): 다운로드된 자산을 사용하여 고품질 동영상 생성
  • 결과물 배포 (StoryQueueProcessor): 완성된 콘텐츠를 FTP 서버에 업로드하고 상태 업데이트
1

READY → PROCESSING(n) → COMPLETED(n) / ERROR(n)

컴포넌트 간 통합 ** 각 컴포넌트는 표준화된 인터페이스와 경로 구조를 통해 통합됩니다:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

# StoryQueueProcessor.py의 메인 처리 흐름
def process_book_queue():
    # 생략...
    
    # 1. 책 다운로드 (asset_loader_program 활용)
    if not download_book(book_id):
        logger.error(f"❌ 책 다운로드 실패: {book_id}")
        update_status(book_id, f"ERROR({COMP_INDEX})")
        continue
    
    # 2. 동영상 생성 (movie_maker 활용)
    if not generate_video(book_id):
        logger.error(f"❌ 동영상 생성 실패: {book_id}")
        continue
    
    # 3. FTP 업로드 및 상태 업데이트
    if not upload_to_ftp(book_id, sb):
        logger.error(f"❌ FTP 업로드 실패: {book_id}")
        continue

    # 생략...
  • 분산 처리: 여러 서버가 동일한 큐에서 작업을 가져와 병렬 처리
  • 서버 식별: 각 서버는 고유 COMP_INDEX로 식별되어 작업 충돌 방지
  • 독립적 컴포넌트: 각 단계는 독립적으로 스케일 업/다운 가능
  • 명령줄 옵션: 다양한 명령줄 옵션을 통한 유연한 구성
  • 처리 용량: 하루 평균 약 100-200권의 책 처리 (서버 당)
  • 성공률: 99% 이상의 높은 성공률 (네트워크 오류 제외)
  • 평균 처리 시간:책 다운로드: (책 크기에 따라 다름)
  • 동영상 생성: (페이지 수에 따라 다름)
  • FTP 업로드: (파일 크기에 따라 다름)
  • 오류 복구: 자동 재시도 메커니즘으로 대부분의 일시적 오류 해결

성능 모니터링 ** 운영 과정에서 중요한 부분은 성능 모니터링입니다. 각 단계의 처리 시간을 측정하여 병목 지점을 식별합니다:

1
2
3
4
5

# 동영상 생성 시간 측정
start_time = time.time()
result = subprocess.run([...])
end_time = time.time()
logger.info(f"✅ 동영상 생성 완료 (시간: {end_time - start_time:.2f}초)")

이러한 시간 측정 데이터는 시스템 최적화에 중요한 인사이트를 제공합니다.

  • 모듈화된 설계: 각 컴포넌트는 단일 책임을 가지며 독립적으로 개발, 테스트, 유지보수 가능
  • 견고한 오류 처리: 다양한 예외 상황에 대비한 포괄적인 오류 처리 시스템
  • 확장 가능한 아키텍처: 처리량 증가에 따라 수평적 확장 가능
  • 유연한 구성: 환경 변수, 설정 파일, 명령줄 옵션을 통한 다양한 환경 지원

sticker

감사합니다.

sticker

#python_그림책낭독_동영상자동생성_cli_클라이언트_빌드

개발, 서비스개발일지
개발일지 회고
이 기사는 저작권자의 CC BY 4.0 라이센스를 따릅니다.