본문 바로가기

업무_인공지능 활용 예시

[chatGPT/API] 영어 단어 또는 문장의 음원(mp3) 파일을 TTS 모델을 활용해서 제작하기

 


  원어민이 녹음한 단어의 음원과 TTS 머신으로 뽑아낸 음원은 당연히 발음의 차이가 크다.

  인공지능 세대의 학습자들 사이에서 영어 콘텐츠를 만드는 입장으로서, 실시간 번역을 지원하는 시대에 영어콘텐츠가 과연 무슨 의미가 있을까 싶기도 하지만... 그럼에도 불구하고 간단한 길이의 예문이나 3개 이하의 음절을 가진 단어들은 TTS로 제작해도 그 결과도가 충분히 유의미하다. 

 

  GPT-api 만 있다면, 또는 현재 무료로 활용할 수 있는 파이썬 라이브러리만 활용하면 크게 문제없이 제작이 가능하다.

 

활용 전제조건은 아래와 같다.

 

1. 엑셀 또는 csv 파일에서 '단어' 또는 '예문'을 행별로 불러와서 음원을 제작한다.

2. 음원이 제작될 때, api 모델이 제공하는 '성우 종류'와 '음성 속도' 등을 변경할 수 있다.

 

위 내용을 기준으로 먼저 OPENAI_API_KEY 가 필요하다.

유료버전으로 플랜을 구매하고, 그 후에 API 발급이 되면, 그 후로는 일사천리다.

 

0. 사용 라이브러리 소개

import os
import asyncio
import pandas as pd
import httpx

######
import re
from dotenv import load_dotenv

# 환경 변수 불러오기
load_dotenv()

 

  • os : 운영체제 상호작용 라이브러리 >> 파일 경로 조작, 디렉토리 생성 등에 사용
  • asyncio : 파이썬 비동기 인풋/아웃풋 지원 라이브러리 >> 워드 읽기 - 음원 제작까지 기다리는 용도
  • httpx : 파이썬 비동기 HTTP 요청 지원 라이브러리 >> TTS API 에 요청 보낼 때 사용

  • re : 파일명 이슈로 인해 사용 (사용하지 않아도 됨)
  • dotenv : api 키 별도 저장 위한 용도 (사용하지 않아도 됨)

 

1. API 설정 

OPENAI_TTS_API_ENDPOINT = "https://api.openai.com/v1/audio/speech"  # 실제 엔드포인트
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
if not OPENAI_API_KEY:
    raise ValueError("OPENAI_API_KEY 오휴")

 

위 코드에서 우선 발급받은 OPENAI_API 키를 입력(대체)한다.

 

 

2. 파일 경로 설정

# 엑셀 파일 경로
EXCEL_FILE_PATH = "words_tts_sample.xlsx"  # 엑셀 파일 이름으로 교체

# 저장할 폴더
OUTPUT_DIR = "generated_mp3_files"
os.makedirs(OUTPUT_DIR, exist_ok=True)

 

다음으로 단어 또는 예문 정보가 적힌 엑셀파일을 불러온다.

엑셀 파일의 컬럼은 'word'로 일단 지정해 놓았다. 아래에 구체적인 메인함수 있음.

 

 

3. TTS 변환 함수

async def text_to_speech(text: str, voice: str = "alloy") -> bytes:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            OPENAI_TTS_API_ENDPOINT,
            headers={
                "Authorization": f"Bearer {OPENAI_API_KEY}",
                "Content-Type": "application/json"
            },
            json={
                "model": "tts-1-hd",
                "input": text,
                "voice": voice
            }
        )
        response.raise_for_status()
        return response.content

 

 

model을 'tts-1-hd'로 설정했는데, tts-1으로 설정해도 괜찮다. 

위에서 'alloy' 라고 설정한 부분이 성우의 종류이며, 

https://platform.openai.com/docs/api-reference/audio/createSpeech

 

 

이름에서 특성이 느껴진다. 개인적으로 단어와 같이 짧은 길이의 음원을 쓸 때는 Alloy, Nova를 활용한다.

반대로 영어 지문을 읽을 때는 Fable을 주로 활용했다.

 

4. 단어 처리 함수

1) 단어 내 특수문자 등 제거 함수

def sanitize_filename(name):
    return re.sub(r'[^\w\-_. ]', '', name)

 간혹 단어에 '-' 하이픈/대시나 '_' 언더바 또는 ''' 어퍼스트로피가 있는 경우가 있을 수 있다.

예문을 만들 때는 문제가 없지만, 실제로 파일명을 만들 때 이슈가 생길 수 있다. 

따라서 이와 같은 1차 정제 작업을 해야한다.

 

 단, 예문을 만들 때 하이픈, 언더바, 어퍼스트로피 등이 악센트나 인토네이션에 영향을 미칠 수 있다.

이런 경우는 해당 함수를 사용하지 말고, 파일명 저장을 하기 위한 별도 열을 설정해야 한다.

 

 

2) 단어 비동기 처리 함수

 

async def process_row(index, word):
    try:
        audio_content = await text_to_speech(word)
        
        ###위에서 sanitize_filename 뺐으면 아래 행 제거하고,
        safe_word = sanitize_filename(word)
        
        ###여기서도 safe_word 대신 word 만 기입해야함
        file_name = f"{index}_{safe_word}.mp3"
        file_path = os.path.join(OUTPUT_DIR, file_name)
        with open(file_path, "wb") as f:
            f.write(audio_content)
        print(f"Saved: {file_path}")
    except Exception as e:
        print(f"Error processing '{word}': {e}")​

 

 위 코드는 단어 처리 함수로서 각 단어를 순차적으로 처리하는 '비동기 함수'다.

단어의 명칭은 불러온 단어의 수대로 'index'가 파일 제일 앞에 붙는다. 

그 뒤에 해당되는 단어의 원형이 따라붙는 형태로 음원이 만들어진다.

 

 

지금까지 설정한 단계를 살펴보면,

1. 단어 불러오기 방식 : 엑셀 파일 사용

2. 성우 목소리 결정 : alloy, nova, fable 

3. TTS 모델 결정 : tts-1 or tts-1-hd

4. 단어 처리 방식 결정 : 단어 내 특수문자 제거 / 보존 

 

 

남은 단계는 해당 과정을 순차적으로 루프 돌리는 것이다.

엑셀 파일에는 'word' 컬럼이 있고 아래에 행 단위로 단어가 리스트업 되어 있다고 가정한다.

필요한 경우 컬럼을 추가해서 '파일명'으로 활용할 수 있다. 문장이 긴 경우에는, 데이터 전처리를 사전에 진행해서 깨끗한 음원을 얻어낼 수도 있다.

 

5. 메인 함수 

async def main():
    # 엑셀 파일 읽기
    df = pd.read_excel(EXCEL_FILE_PATH)
    
    # 'Word' 또는 'word' 컬럼 확인
    if 'Word' not in df.columns and 'word' not in df.columns:
        print("엑셀 파일에 'Word' 또는 'word' 컬럼이 없습니다.")
        return
    
    word_column = 'Word' if 'Word' in df.columns else 'word'
    
    tasks = []
    for index, row in df.iterrows():
        word = str(row[word_column]).strip()
        if word:
            tasks.append(process_row(index, word))
    
    await asyncio.gather(*tasks)
    
    if __name__ == "__main__":
    asyncio.run(main())

 

한 가지 주의사항은 '주피터 노트북'과 같은 환경에서 해당 파일을 실행하면 오류를 일으킬 수 있다. 따라서 별도의 py 파일로 저장 후 터미널에서 별도 환경설정을 통해 실행하는 것이 안전하다. 

 

전체 코드를 아래에 넣는다.

 

 

***전체 코드, 내용 확인용***

import os
import asyncio
import pandas as pd
import httpx

# 환경 변수 또는 직접 설정
OPENAI_TTS_API_ENDPOINT = "https://api.openai.com/v1/audio/speech"  
OPENAI_API_KEY = "개인 API용 키 발급"

# 엑셀 파일 경로
EXCEL_FILE_PATH = "words_tts_sample.xlsx"  # 엑셀 파일 이름으로 교체

# 저장할 폴더
OUTPUT_DIR = "generated_mp3_files"
os.makedirs(OUTPUT_DIR, exist_ok=True)


async def text_to_speech(text: str, voice: str = "nova") -> bytes:  #alloy
    async with httpx.AsyncClient() as client:
        response = await client.post(
            OPENAI_TTS_API_ENDPOINT,
            headers={
                "Authorization": f"Bearer {OPENAI_API_KEY}",
                "Content-Type": "application/json"
            },
            json={
                "model": "tts-1-hd",
                "input": text,
                "voice": voice
            }
        )
        response.raise_for_status()
        return response.content

async def process_row(index, word):
    try:
        audio_content = await text_to_speech(word)
        
        
        #index 
        #P1_
        file_name = f"{index}_{word}.mp3"
        
        
        file_path = os.path.join(OUTPUT_DIR, file_name)
        with open(file_path, "wb") as f:
            f.write(audio_content)
        print(f"Saved: {file_path}")
    except Exception as e:
        print(f"Error processing '{word}': {e}")

async def main():
    # 엑셀 파일 읽기
    df = pd.read_excel(EXCEL_FILE_PATH)
    
    # 'word' 컬럼이 있다고 가정
    if 'word' not in df.columns:
        print("엑셀 파일에 'word' 컬럼이 없습니다.")
        return
    
    tasks = []
    for index, row in df.iterrows():
        word = str(row['word']).strip()
        if word:
            tasks.append(process_row(index, word))
    
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    asyncio.run(main())

 

 

생각보다 짧은 몇 줄의 코드로 함수를 만들 수 있는 것을 볼 수 있다.

 

추가적으로 더 나은 세팅을 위해서 고려할 수 있는 장치들이 있다.

 

+1. 동시 요청 수 제한

# 동시 요청 수 제한
SEM = asyncio.Semaphore(10)  # 동시에 최대 10개의 요청

async def process_row(index, word):
    async with SEM:
        try:
            audio_content = await text_to_speech(word)
            safe_word = sanitize_filename(word)
            file_name = f"{index}_{safe_word}.mp3"
            file_path = os.path.join(OUTPUT_DIR, file_name)
            with open(file_path, "wb") as f:
                f.write(audio_content)
            print(f"Saved: {file_path}")
        except Exception as e:
            print(f"Error processing '{word}': {e}")

 

많은 수의 단어를 동시에 처리할 경우, API 서버에서 요청을 누락할 수 있다. 따라서 동시 요청 수를 제한하는 것도 좋다.

 

 

+2. 진행 상황 표시

from tqdm.asyncio import tqdm

async def main():
    # 엑셀 파일 읽기
    df = pd.read_excel(EXCEL_FILE_PATH)
    
    # 'Word' 또는 'word' 컬럼 확인
    if 'Word' not in df.columns and 'word' not in df.columns:
        print("엑셀 파일에 'Word' 또는 'word' 컬럼이 없습니다.")
        return
    
    word_column = 'Word' if 'Word' in df.columns else 'word'
    
    tasks = []
    for index, row in df.iterrows():
        word = str(row[word_column]).strip()
        if word:
            tasks.append(process_row(index, word))
    
    for f in tqdm(asyncio.as_completed(tasks), total=len(tasks)):
        await f

 

TQDM은 전체 처리 과정에서 단어 별 데이터가 진행되는 상황을 볼 수 있다.

10개 이상의 데이터를 만드는 과정을 준비 중이라면, 고려해 볼 만한 옵션이다.

 

 

+3. 로그 파일 생성

import logging

# 로깅 설정
logging.basicConfig(
    filename='tts_errors.log',
    level=logging.ERROR,
    format='%(asctime)s:%(levelname)s:%(message)s'
)

async def process_row(index, word):
    try:
        audio_content = await text_to_speech(word)
        safe_word = sanitize_filename(word)
        file_name = f"{index}_{safe_word}.mp3"
        file_path = os.path.join(OUTPUT_DIR, file_name)
        with open(file_path, "wb") as f:
            f.write(audio_content)
        print(f"Saved: {file_path}")
    except Exception as e:
        logging.error(f"Error processing '{word}': {e}")
        print(f"Error processing '{word}': {e}")

 

 

TQDM 이 단순히 시각적으로 진행상황을 보여주는 라이브러리라면,

logging 기능은 단어 별 (제작되는 파일명 기준)로 파일이 완성될 때마다 위 포맷에 맞게 메시지를 띄워준다.

 

 

 

위 세 가지 제안사항을 모두 합쳐서 전체 코드를 개선하면,

 

*fin. 전체 개선된 코드 내용

import os
import asyncio
import pandas as pd
import httpx
import re
from dotenv import load_dotenv
import logging
from tqdm.asyncio import tqdm

# 환경 변수 불러오기
load_dotenv()

# 로깅 설정
logging.basicConfig(
    filename='tts_errors.log',
    level=logging.ERROR,
    format='%(asctime)s:%(levelname)s:%(message)s'
)

OPENAI_TTS_API_ENDPOINT = "https://api.openai.com/v1/audio/speech"  
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

# 엑셀 파일 경로
EXCEL_FILE_PATH = "words_tts_sample.xlsx"  # 엑셀 파일 이름으로 교체

# 저장할 폴더
OUTPUT_DIR = "generated_mp3_files"
os.makedirs(OUTPUT_DIR, exist_ok=True)

def sanitize_filename(name):
    return re.sub(r'[^\w\-_. ]', '', name)

# 동시 요청 수 제한
SEM = asyncio.Semaphore(10)  # 동시에 최대 10개의 요청

async def text_to_speech(text: str, voice: str = "alloy") -> bytes:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            OPENAI_TTS_API_ENDPOINT,
            headers={
                "Authorization": f"Bearer {OPENAI_API_KEY}",
                "Content-Type": "application/json"
            },
            json={
                "model": "tts-1-hd",
                "input": text,
                "voice": voice
            }
        )
        response.raise_for_status()
        return response.content

async def process_row(index, word):
    async with SEM:
        try:
            audio_content = await text_to_speech(word)
            safe_word = sanitize_filename(word)
            file_name = f"{index}_{safe_word}.mp3"
            file_path = os.path.join(OUTPUT_DIR, file_name)
            with open(file_path, "wb") as f:
                f.write(audio_content)
            print(f"Saved: {file_path}")
        except Exception as e:
            logging.error(f"Error processing '{word}': {e}")
            print(f"Error processing '{word}': {e}")

async def main():
    # 엑셀 파일 읽기
    df = pd.read_excel(EXCEL_FILE_PATH)
    
    # 'Word' 또는 'word' 컬럼 확인
    if 'Word' not in df.columns and 'word' not in df.columns:
        print("엑셀 파일에 'Word' 또는 'word' 컬럼이 없습니다.")
        return
    
    word_column = 'Word' if 'Word' in df.columns else 'word'
    
    tasks = []
    for index, row in df.iterrows():
        word = str(row[word_column]).strip()
        if word:
            tasks.append(process_row(index, word))
    
    # 진행 상황 표시
    for f in tqdm(asyncio.as_completed(tasks), total=len(tasks)):
        await f

if __name__ == "__main__":
    asyncio.run(main())

 

 

이외에도 저작권 해당사항이 없는 라이브러리가 다양하게 있다.

하지만 비동기 작업을 진행하지 않거나 혹은 이상하게 불안정한 라이브러리들이 많았다.

 

gTT, Coqui TTS, pyttsx3 등 다양한 라이브러리가 있었지만... 아래와 같은 비교 요소가 있다.

 

무료라는 측면에서는 gTTS가 가장 활용도 측면에서 좋았지만,

업무에서 콘텐츠 저작측면으로 활용해야 하기 때문에 openai의 TTS 모델을 활용했다.

 

기회가 되면 gTTS 활용 코드도 업로드할 수 있도록 해야겠다.