164 lines
5.7 KiB
Python
164 lines
5.7 KiB
Python
"""
|
|
Audio generation module using OpenAI TTS
|
|
|
|
This module converts script text to audio using OpenAI's TTS API
|
|
"""
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Dict, Any, List, Optional
|
|
|
|
try:
|
|
from openai import OpenAI
|
|
except ImportError:
|
|
OpenAI = None
|
|
|
|
from backend.video_processor.utils import get_video_duration
|
|
|
|
|
|
class AudioGenerator:
|
|
"""Generate narration audio using OpenAI TTS"""
|
|
|
|
def __init__(self, api_key: Optional[str] = None):
|
|
self.api_key = api_key or os.getenv("OPENAI_API_KEY")
|
|
if not self.api_key:
|
|
raise ValueError("OPENAI_API_KEY not found in environment")
|
|
|
|
if OpenAI is None:
|
|
raise ImportError(
|
|
"openai package not installed. "
|
|
"Run: pip install openai"
|
|
)
|
|
|
|
self.client = OpenAI(api_key=self.api_key)
|
|
# CRITICAL: Output directly to public/ for Remotion to find them via staticFile
|
|
self.output_dir = Path("public")
|
|
self.output_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
def generate_audio(
|
|
self,
|
|
script_segments: List[Dict[str, Any]],
|
|
voice: str = "nova",
|
|
model: str = "tts-1"
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Generate audio files for each script segment
|
|
|
|
Args:
|
|
script_segments: List of script segments with text
|
|
voice: Voice to use (alloy, echo, fable, onyx, nova, shimmer)
|
|
model: TTS model (tts-1 or tts-1-hd)
|
|
|
|
Returns:
|
|
Updated script segments with audioPath field
|
|
"""
|
|
print(f" Generating audio with OpenAI TTS (voice: {voice})...")
|
|
|
|
updated_segments = []
|
|
|
|
for segment in script_segments:
|
|
segment_id = segment["id"]
|
|
text = segment["text"]
|
|
is_video_clip = segment.get("type") == "video_clip" or segment.get("visualType") == "video"
|
|
|
|
# Check if this segment should use existing audio (e.g. dialogue_clip)
|
|
current_audio_path = segment.get("audioPath", "")
|
|
is_manual_audio = "dialogue_clip" in current_audio_path
|
|
|
|
# Skip TTS for video clips or segments with manual audio
|
|
if is_video_clip or is_manual_audio:
|
|
if is_manual_audio:
|
|
print(f" ⏭️ Keeping manual audio for segment {segment_id}: {current_audio_path}")
|
|
else:
|
|
print(f" ⏭️ Keeping original audio for video segment {segment_id}")
|
|
|
|
# Still need to update duration if not present
|
|
if "duration" not in segment or segment["duration"] == 0:
|
|
try:
|
|
actual_duration = get_video_duration(str(current_audio_path))
|
|
segment["duration"] = actual_duration
|
|
except:
|
|
pass
|
|
|
|
updated_segments.append(segment)
|
|
continue
|
|
|
|
try:
|
|
# Generate audio
|
|
audio_filename = f"segment_{segment_id:03d}.mp3"
|
|
audio_path = self.output_dir / audio_filename
|
|
|
|
response = self.client.audio.speech.create(
|
|
model=model,
|
|
voice=voice,
|
|
input=text
|
|
)
|
|
|
|
# Save audio file
|
|
response.stream_to_file(str(audio_path))
|
|
|
|
# CRITICAL: Get actual duration of the generated audio file
|
|
try:
|
|
actual_duration = get_video_duration(str(audio_path))
|
|
segment["duration"] = actual_duration
|
|
print(f" ✓ Generated {audio_filename} (Actual duration: {actual_duration:.2f}s)")
|
|
except Exception as de:
|
|
print(f" ⚠️ Could not get actual duration for {audio_filename}: {de}")
|
|
# Keep existing duration as fallback
|
|
|
|
# Update segment with audio path
|
|
segment["audioPath"] = str(audio_path)
|
|
updated_segments.append(segment)
|
|
|
|
except Exception as e:
|
|
print(f" ⚠️ Failed to generate audio for segment {segment_id}: {e}")
|
|
# Keep segment without audio
|
|
updated_segments.append(segment)
|
|
|
|
print(f" ✓ Generated {len(updated_segments)} audio files")
|
|
return updated_segments
|
|
|
|
def estimate_audio_duration(self, text: str, rate: float = 150) -> float:
|
|
"""
|
|
Estimate audio duration based on text length
|
|
|
|
Args:
|
|
text: Text to estimate
|
|
rate: Characters per minute (Japanese: ~150-200)
|
|
|
|
Returns:
|
|
Estimated duration in seconds
|
|
"""
|
|
char_count = len(text)
|
|
duration = (char_count / rate) * 60
|
|
return round(duration, 1)
|
|
|
|
|
|
def generate_audio_for_script(
|
|
script_data: Dict[str, Any],
|
|
voice: str = "nova"
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Convenience function to generate audio for a script
|
|
|
|
Args:
|
|
script_data: Script data with segments
|
|
voice: Voice to use for TTS
|
|
|
|
Returns:
|
|
Updated script data with audio paths
|
|
"""
|
|
try:
|
|
generator = AudioGenerator()
|
|
segments = script_data["segments"]
|
|
updated_segments = generator.generate_audio(segments, voice=voice)
|
|
|
|
return {
|
|
**script_data,
|
|
"segments": updated_segments,
|
|
}
|
|
except (ValueError, ImportError) as e:
|
|
print(f" ⚠️ Cannot generate audio: {e}")
|
|
print(" Skipping audio generation...")
|
|
return script_data
|