#!/usr/bin/env python3 """translation.json을 Gemini 멀티스피커 TTS(en-US)로 합성해 에피소드 mp3와 메타를 만든다. 슬라이드 단위로 합성하고 사이에 bell을 넣어 한국어판과 같은 진행감을 유지한다. 비밀키는 환경변수 GEMINI_API_KEY로만 받고 절대 출력하지 않는다. """ import argparse import json import os import subprocess import sys import time import wave from pathlib import Path EN_DIALOGUE_PROMPT = """TTS the following English radio conversation between Host and Guest. Host is a warm female radio presenter. She sounds calm, intelligent, and lightly amused. Guest is a friendly male questioner. He has dry humor, but he does not exaggerate. Read the English transcript exactly. Do not add lines. Do not skip lines. Keep a natural radio pace with short pauses between turns. Let the conversation breathe.""" def safe_error(exc: Exception) -> str: text = f"{type(exc).__name__}: {exc}" secret = os.environ.get("GEMINI_API_KEY", "") if secret: text = text.replace(secret, "[redacted]") return text[:1200] def run_cmd(cmd: list[str]) -> None: proc = subprocess.run(cmd, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) if proc.returncode != 0: raise RuntimeError(proc.stdout[-1500:]) def write_wave(path: Path, pcm: bytes, channels: int = 1, rate: int = 24000, width: int = 2) -> None: with wave.open(str(path), "wb") as wf: wf.setnchannels(channels) wf.setsampwidth(width) wf.setframerate(rate) wf.writeframes(pcm) def gemini_tts(path: Path, transcript: str, model: str, host_voice: str, guest_voice: str) -> None: api_key = os.environ.get("GEMINI_API_KEY", "").strip() if not api_key: raise RuntimeError("GEMINI_API_KEY is required") from google import genai from google.genai import types client = genai.Client(api_key=api_key) response = client.models.generate_content( model=model, contents=f"{EN_DIALOGUE_PROMPT}\n\n{transcript}", config=types.GenerateContentConfig( response_modalities=["AUDIO"], speech_config=types.SpeechConfig( language_code="en-US", multi_speaker_voice_config=types.MultiSpeakerVoiceConfig( speaker_voice_configs=[ types.SpeakerVoiceConfig( speaker="Host", voice_config=types.VoiceConfig( prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name=host_voice) ), ), types.SpeakerVoiceConfig( speaker="Guest", voice_config=types.VoiceConfig( prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name=guest_voice) ), ), ] ), ), temperature=1.2, ), ) data = response.candidates[0].content.parts[0].inline_data.data raw = path.with_name(path.stem + "_raw.wav") write_wave(raw, data) run_cmd(["ffmpeg", "-y", "-i", str(raw), "-ar", "44100", "-ac", "2", str(path)]) def tts_with_retries(path: Path, transcript: str, slide: int, primary: str, candidate: str, host_voice: str, guest_voice: str, events: list[str]) -> None: for attempt in range(1, 4): try: gemini_tts(path, transcript, primary, host_voice, guest_voice) return except Exception as exc: events.append(f"slide {slide}: {primary} attempt {attempt} failed: {safe_error(exc)}") time.sleep(10 * attempt) gemini_tts(path, transcript, candidate, host_voice, guest_voice) events.append(f"slide {slide}: used candidate {candidate}") def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--translation", required=True) parser.add_argument("--out-mp3", required=True) parser.add_argument("--bell", default="/home/ukimkj/Documents/DailyAINewsVideoAutomation/assets/bell.mp3") args = parser.parse_args() translation = json.loads(Path(args.translation).read_text(encoding="utf-8")) out_mp3 = Path(args.out_mp3) work = Path(args.translation).parent audio_dir = work / "audio" audio_dir.mkdir(parents=True, exist_ok=True) primary = os.environ.get("TTS_MODEL_PRIMARY", "gemini-2.5-pro-preview-tts") candidate = os.environ.get("TTS_MODEL_CANDIDATE", "gemini-3.1-flash-tts-preview") host_voice = os.environ.get("GEMINI_TTS_HOST_VOICE", "Sulafat") guest_voice = os.environ.get("GEMINI_TTS_GUEST_VOICE", "Achird") speed = float(os.environ.get("TTS_SPEED", "1.0")) if not 0.5 <= speed <= 2.0: raise RuntimeError(f"TTS_SPEED must be between 0.5 and 2.0: {speed}") slides: dict[int, list[dict]] = {} for seg in translation["segments"]: slides.setdefault(int(seg["slide"]), []).append(seg) events: list[str] = [] slide_wavs = [] for idx in sorted(slides): transcript = "\n".join(f'{s["speaker"]}: {s["text"]}' for s in slides[idx]) wav = audio_dir / f"slide_{idx:02d}.wav" if not wav.exists(): tts_with_retries(wav, transcript, idx, primary, candidate, host_voice, guest_voice, events) slide_wavs.append(wav) bell_wav = audio_dir / "bell.wav" has_bell = Path(args.bell).exists() if has_bell and not bell_wav.exists(): run_cmd(["ffmpeg", "-y", "-i", args.bell, "-ar", "44100", "-ac", "2", str(bell_wav)]) concat_list = audio_dir / "concat.txt" lines = [] for i, wav in enumerate(slide_wavs): lines.append(f"file '{wav}'") if has_bell and i < len(slide_wavs) - 1: lines.append(f"file '{bell_wav}'") concat_list.write_text("\n".join(lines) + "\n", encoding="utf-8") out_mp3.parent.mkdir(parents=True, exist_ok=True) # 속도 보정은 최종 합본에서 한 번만 건다. 캐시된 슬라이드 wav를 재사용하는 재실행에도 적용되게. concat_cmd = ["ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", str(concat_list)] if abs(speed - 1.0) >= 0.0001: concat_cmd += ["-filter:a", f"atempo={speed:.4f}"] concat_cmd += ["-ar", "44100", "-ac", "2", "-b:a", "96k", str(out_mp3)] run_cmd(concat_cmd) probe = subprocess.run( ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "csv=p=0", str(out_mp3)], text=True, stdout=subprocess.PIPE, ) duration = int(float(probe.stdout.strip().split("\n")[0])) meta = {"duration": duration, "title": translation["title"], "description": translation["description"]} if translation.get("link"): meta["link"] = translation["link"] out_mp3.with_suffix(".json").write_text(json.dumps(meta, ensure_ascii=False, indent=1), encoding="utf-8") for line in events: print("EVENT:", line) print(f"OK mp3: {out_mp3} ({duration}s, {len(slide_wavs)} slides, speed {speed:.4f})") return 0 if __name__ == "__main__": sys.exit(main())