110 lines
4.6 KiB
Python
110 lines
4.6 KiB
Python
"""④ tts — 나레이션 음성 + 실측 타임라인.
|
|
|
|
영상 비트를 먼저 못 박고 음성을 우겨넣으면 말이 잘리거나 비므로, 문장별 실측 길이를
|
|
먼저 재고 그 큐에 씬·카메라를 맞춘다. 트림·합성·인코딩까지 전부 메모리에서 처리한다.
|
|
"""
|
|
import asyncio
|
|
|
|
import httpx
|
|
import numpy as np
|
|
from openai import AsyncOpenAI
|
|
|
|
from models.tts import NarrationAudio, NarrationCue, NarrationTimeline, TypecastVoice
|
|
from settings import settings
|
|
from utils.audio import SAMPLE_RATE, decode_mp3, encode_mp3, trim_silence
|
|
from utils.prompt import load_prompt
|
|
|
|
TYPECAST_API = "https://api.typecast.ai/v1/text-to-speech"
|
|
OPENAI_TTS_MODEL = "gpt-4o-mini-tts"
|
|
TYPECAST_MODEL = "ssfm-v30"
|
|
|
|
# "천천히"라고 지시하면 TTS가 과하게 늘어진다. 톤은 지시문으로, 속도는 파라미터로 통제한다.
|
|
TTS_INSTRUCTIONS = load_prompt("tts_instructions")
|
|
DEFAULT_SPEED = 1.28
|
|
DEFAULT_PAUSES = (0.35, 0.35, 0.30)
|
|
DEFAULT_LEAD_IN = 0.35
|
|
DEFAULT_TAIL = 1.7
|
|
|
|
|
|
async def speak_openai(text: str, voice: str, speed: float, instructions: str) -> bytes:
|
|
response = await AsyncOpenAI(api_key=settings.chatgpt_api_key).audio.speech.create(
|
|
model=OPENAI_TTS_MODEL,
|
|
voice=voice,
|
|
input=text,
|
|
instructions=instructions,
|
|
speed=speed,
|
|
response_format="mp3",
|
|
)
|
|
return await response.aread()
|
|
|
|
|
|
async def speak_typecast(text: str, typecast: TypecastVoice) -> bytes:
|
|
if not settings.typecast_api_key:
|
|
raise RuntimeError("TYPECAST_API_KEY 없음 — typecast 엔진을 쓰려면 필요하다")
|
|
async with httpx.AsyncClient(timeout=90) as client:
|
|
response = await client.post(
|
|
TYPECAST_API,
|
|
headers={"X-API-KEY": settings.typecast_api_key},
|
|
json={
|
|
"text": text,
|
|
"model": TYPECAST_MODEL,
|
|
"voice_id": typecast.voice_id,
|
|
"language": "kor",
|
|
"prompt": {"emotion_type": "preset",
|
|
"emotion_preset": typecast.emotion,
|
|
"emotion_intensity": typecast.emotion_intensity},
|
|
"output": {"audio_format": "mp3", "volume": 100,
|
|
"audio_tempo": typecast.tempo},
|
|
})
|
|
response.raise_for_status()
|
|
return response.content
|
|
|
|
|
|
def merge_clips(clips: list[np.ndarray], starts: list[float], total: float) -> np.ndarray:
|
|
"""큐가 겹치지 않으므로 시작 시각에 그대로 덮어써 한 트랙으로 만든다."""
|
|
track = np.zeros(round(total * SAMPLE_RATE), dtype=np.int16)
|
|
for clip, start in zip(clips, starts, strict=True):
|
|
offset = round(start * SAMPLE_RATE)
|
|
track[offset:offset + len(clip)] = clip[:max(0, len(track) - offset)]
|
|
return track
|
|
|
|
|
|
async def synthesize(sentences: list[str], voice: str = "nova", *,
|
|
speed: float = DEFAULT_SPEED,
|
|
instructions: str = TTS_INSTRUCTIONS,
|
|
typecast: TypecastVoice | None = None,
|
|
pauses: tuple[float, ...] = DEFAULT_PAUSES,
|
|
lead_in: float = DEFAULT_LEAD_IN,
|
|
tail: float = DEFAULT_TAIL) -> NarrationAudio:
|
|
"""typecast를 주면 그 엔진으로 가므로, voice_id 없이 typecast를 고를 수 없다."""
|
|
if not sentences:
|
|
raise ValueError("나레이션 문장이 없다")
|
|
|
|
if typecast:
|
|
spoken = await asyncio.gather(*(speak_typecast(text, typecast) for text in sentences))
|
|
else:
|
|
spoken = await asyncio.gather(
|
|
*(speak_openai(text, voice, speed, instructions) for text in sentences))
|
|
# TTS가 문장 끝에 붙이는 최대 1.2초 공백이 그대로 영상 길이가 되면 훅 직후에 침묵이 생긴다
|
|
clips = [trim_silence(decode_mp3(mp3)) for mp3 in spoken]
|
|
|
|
cues, starts = [], []
|
|
cursor = lead_in
|
|
for index, (text, clip) in enumerate(zip(sentences, clips, strict=True)):
|
|
duration = len(clip) / SAMPLE_RATE
|
|
pause = pauses[index] if index < len(pauses) else pauses[-1]
|
|
starts.append(cursor)
|
|
cues.append(NarrationCue(index=index, text=text, start=round(cursor, 3),
|
|
duration=round(duration, 3), end=round(cursor + duration, 3),
|
|
pause_after=pause))
|
|
cursor += duration + pause
|
|
|
|
total = round(cues[-1].end + tail, 2)
|
|
return NarrationAudio(
|
|
timeline=NarrationTimeline(
|
|
engine="typecast" if typecast else "openai",
|
|
voice=typecast.voice_id if typecast else voice,
|
|
lead_in=lead_in, tail=tail, total=total, cues=cues),
|
|
mp3=encode_mp3(merge_clips(clips, starts, total)),
|
|
)
|