#!/usr/bin/env python3 """ narrator.py ----------- Core VoiceoverNarrator class for the kokoro-voiceover package. Provides broadcast-grade documentary and long-form speech synthesis using Kokoro-ONNX and the tuned Grounded Documentary voice profile (55% bm_george + 45% bm_daniel). """ import os import re import json import time import subprocess from pathlib import Path from typing import Optional, Union, List, Tuple, Dict, Any import numpy as np import soundfile as sf from kokoro_onnx import Kokoro from .pronunciation import apply_pronunciation_lexicon # Gemini-Calibrated Broadcast Mastering Filter Chain (Warm sub-chest resonance + surgical de-essing) BROADCAST_AUDIO_FILTER = ( "highpass=f=75:p=2," "equalizer=f=180:t=q:w=1.2:g=3.2," "equalizer=f=480:t=q:w=2.0:g=-1.8," "equalizer=f=3200:t=q:w=1.5:g=1.2," "equalizer=f=7200:t=q:w=2.5:g=-3.0," "lowpass=f=11000," "deesser=i=0.7:m=0.5:f=0.65:s=o," "compand=attacks=0.02:decays=0.25:points=-70/-70|-45/-32|-20/-14|0/-6:soft-knee=6," "loudnorm=I=-16:TP=-1.5:LRA=9" ) def clean_text_base(raw_text: str, apply_lexicon: bool = True) -> str: """Sanitizes raw markdown script text for natural TTS delivery and applies phonetic lexicon.""" t = raw_text if apply_lexicon: t = apply_pronunciation_lexicon(t) t = re.sub(r"\*\*([^*]+)\*\*", r"\1", t) t = re.sub(r"\*([^*]+)\*", r"\1", t) t = re.sub(r"`([^`]+)`", r"\1", t) t = re.sub(r"#{1,6}\s*", "", t) t = re.sub(r"\[(?:PAUSE|BREATH|BEAT|SILENCE|MUSIC|SFX|STAGE DIRECTION)[^\]]*\]", "", t, flags=re.IGNORECASE) t = re.sub(r"\[[^\]]*\]", "", t) t = re.sub(r"\((?:pause|breath|beat|silence)[^)]*\)", "", t, flags=re.IGNORECASE) t = t.replace("\ufffd", " - ") t = t.replace("’", "'").replace("‘", "'").replace("“", '"').replace("”", '"') t = t.replace("—", " - ").replace("–", " - ") t = re.sub(r"\s*\(([^)]+)\)\s*", r" — \1 — ", t) t = t.replace(";", " — ") t = re.sub(r":\s+", " — ", t) t = re.sub(r"\.{3,}", "...", t) t = re.sub(r"\.\.\.(?=[A-Za-z])", "... ", t) t = re.sub(r"\s*—\s*", " — ", t) t = re.sub(r"\s+-\s+", " - ", t) t = re.sub(r"( —\s*)+", " — ", t) t = re.sub(r"[ \t]+", " ", t) return t.strip() class VoiceoverNarrator: """ Broadcast-grade documentary narrator using Kokoro-ONNX and the Grounded Documentary profile. """ def __init__( self, model_path: Optional[Union[str, Path]] = None, voices_path: Optional[Union[str, Path]] = None, profile_path: Optional[Union[str, Path]] = None, ): root_dir = Path(__file__).resolve().parents[4] # repo root default_model = root_dir / "models" / "kokoro" / "kokoro-v1.0.onnx" default_voices = root_dir / "models" / "kokoro" / "voices-v1.0.bin" default_profile = root_dir / "models" / "kokoro" / "grounded_documentary.npy" self.model_path = Path(model_path) if model_path else default_model self.voices_path = Path(voices_path) if voices_path else default_voices self.profile_path = Path(profile_path) if profile_path else default_profile if not self.model_path.exists(): raise FileNotFoundError(f"Kokoro model not found at: {self.model_path}") if not self.voices_path.exists(): raise FileNotFoundError(f"Voices library not found at: {self.voices_path}") self.kokoro = Kokoro(str(self.model_path), str(self.voices_path)) if self.profile_path.exists(): self.voice_style = np.load(str(self.profile_path)) self.voice_label = "grounded_documentary (loaded profile)" else: v_daniel = self.kokoro.get_voice_style("bm_daniel") v_george = self.kokoro.get_voice_style("bm_george") self.voice_style = 0.55 * v_george + 0.45 * v_daniel self.voice_label = "grounded_documentary (55% bm_george + 45% bm_daniel)" def synthesize_chunks( self, chunks: List[Tuple[str, float]], out_wav: Path, out_mp3: Path, out_json: Optional[Path] = None, section_name: str = "Voiceover", speed: float = 0.91, sentence_pause: float = 0.65, clause_pause: float = 0.30, lang: str = "en-gb", apply_mastering: bool = True, ) -> float: """ Synthesizes structured chunks into a broadcast-mastered MP3 audio track. Each chunk is (text, pause_after_seconds). """ out_wav.parent.mkdir(parents=True, exist_ok=True) out_mp3.parent.mkdir(parents=True, exist_ok=True) audio_segments = [] sample_rate = 24000 meta_segments = [] current_time = 0.0 t_start = time.time() total_words = sum(len(text.split()) for text, _ in chunks) print(f"\n--- Synthesizing: {section_name} ({len(chunks)} chunks, {total_words} words) ---") for idx, (chunk_text, pause_after) in enumerate(chunks, 1): cleaned = clean_text_base(chunk_text) if not cleaned: continue words = len(cleaned.split()) print(f" [{idx}/{len(chunks)}] ({words}w, pause {pause_after:.2f}s): {cleaned[:60]}...") seg_audio, sr = self.kokoro.create( cleaned, voice=self.voice_style, speed=speed, lang=lang, sentence_pause=sentence_pause, clause_pause=clause_pause, continuous=False ) sample_rate = sr seg_dur = len(seg_audio) / sr meta_segments.append({ "unit_index": idx, "text": cleaned, "word_count": words, "start_time": round(current_time, 3), "end_time": round(current_time + seg_dur, 3), "duration": round(seg_dur, 3), "pause_after": round(pause_after, 3) }) audio_segments.append(seg_audio) current_time += seg_dur if pause_after > 0: silence_samples = int(sample_rate * pause_after) audio_segments.append(np.zeros(silence_samples, dtype=np.float32)) current_time += pause_after full_audio = np.concatenate(audio_segments) total_dur = len(full_audio) / sample_rate elapsed = time.time() - t_start # 1. Save WAV sf.write(str(out_wav), full_audio, sample_rate) # 2. Master to MP3 cmd_mp3 = ["ffmpeg", "-y", "-i", str(out_wav)] if apply_mastering: cmd_mp3.extend(["-af", BROADCAST_AUDIO_FILTER]) cmd_mp3.extend([ "-c:a", "libmp3lame", "-b:a", "192k", str(out_mp3) ]) subprocess.run(cmd_mp3, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) # 3. Save JSON metadata if out_json: meta = { "section": section_name, "voice_profile": self.voice_label, "lang": lang, "speed": speed, "sentence_pause": sentence_pause, "clause_pause": clause_pause, "sample_rate": sample_rate, "duration_seconds": round(total_dur, 3), "word_count": total_words, "generation_time_seconds": round(elapsed, 2), "mastered": apply_mastering, "units": meta_segments } with open(out_json, "w", encoding="utf-8") as jf: json.dump(meta, jf, indent=2) print(f"-> Finished {section_name}: {total_dur:.2f}s (~{total_dur/60:.2f} min) in {elapsed:.1f}s") return total_dur def synthesize_text( self, text: str, output_path: Union[str, Path], speed: float = 0.91, sentence_pause: float = 0.65, clause_pause: float = 0.30, apply_mastering: bool = True ) -> float: """Single-string simple synthesis helper.""" out_p = Path(output_path) wav_path = out_p.with_suffix(".wav") mp3_path = out_p.with_suffix(".mp3") chunks = [(p.strip(), 0.85) for p in text.split("\n\n") if p.strip()] return self.synthesize_chunks( chunks=chunks, out_wav=wav_path, out_mp3=mp3_path, out_json=out_p.with_suffix(".json"), section_name=out_p.stem, speed=speed, sentence_pause=sentence_pause, clause_pause=clause_pause, apply_mastering=apply_mastering )