scripts/generate_ad_vo.py

"""Generate ElevenLabs premium voiceover WITH word-level timestamps for the
DocDoe ad reel, so on-screen captions land exactly with the spoken word.

Uses POST /v1/text-to-speech/{voice}/with-timestamps → returns audio + char
alignment. We request PCM 24k, wrap to WAV, and fold char timings into word
timings. Writes public/ad-vo/<id>.wav + manifest.json (with `words`).

  EL_KEY=sk_... backend/.venv/Scripts/python.exe backend/scripts/generate_ad_vo.py
"""
from __future__ import annotations

import base64
import json
import os
import sys
import time
import wave
from pathlib import Path

import requests

ROOT = Path(__file__).resolve().parents[2]
OUT_DIR = ROOT / "public" / "ad-vo"
OUT_DIR.mkdir(parents=True, exist_ok=True)

VOICE_ID = "nPczCjzI2devNBz1zQrb"  # Brian — deep, resonant, crystal-clear enunciation
MODEL = "eleven_multilingual_v2"   # reliable char timestamps
SAMPLE_RATE = 24000
OUTPUT_FORMAT = "pcm_24000"
# higher stability = cleaner, steadier delivery + crisp consonants
VOICE_SETTINGS = {"stability": 0.6, "similarity_boost": 0.9, "style": 0.1, "use_speaker_boost": True}

# "DocDoe" -> "Dock Doe" so TTS says it correctly ("dock-doh").
# Captions re-merge "Dock"+"Doe" -> "DocDoe" and "A"+"I" -> "AI" for display.
SEGMENTS: list[dict[str, str]] = [
    {"id": "hook", "text": "Studying all night, but still scoring low?"},
    {"id": "problem", "text": "Your textbook is huge. The exam tests only what matters."},
    {"id": "intro", "text": "Meet Dock Doe. Your A.I. exam coach."},
    {"id": "upload", "text": "Upload a PDF. Ask any question. Get an exam ready answer."},
    {"id": "features", "text": "Notes, quizzes, flashcards, and mark wise answers, in seconds."},
    {"id": "predict", "text": "Dock Doe predicts the important questions, and writes mark wise answers."},
    {"id": "video", "text": "Need a teacher? Dock Doe builds a full lesson, just for you."},
    {"id": "proof", "text": "Free to start. Built for every board and subject."},
    {"id": "cta", "text": "Start free today, at Dock Doe dot in."},
]


def synth(text: str, key: str) -> dict:
    url = f"https://api.elevenlabs.io/v1/text-to-speech/{VOICE_ID}/with-timestamps"
    params = {"output_format": OUTPUT_FORMAT}
    body = {"text": text, "model_id": MODEL, "voice_settings": VOICE_SETTINGS}
    headers = {"xi-api-key": key, "Content-Type": "application/json"}
    last = None
    for attempt in range(5):
        try:
            resp = requests.post(url, params=params, json=body, headers=headers, timeout=120)
            if resp.status_code != 200:
                raise SystemExit(f"ElevenLabs {resp.status_code}: {resp.text[:300]}")
            return resp.json()
        except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as exc:
            last = exc
            time.sleep(2 * (attempt + 1))
            print(f"   retry {attempt + 1}/5", flush=True)
    raise SystemExit(f"ElevenLabs unreachable: {last}")


def write_wav(pcm: bytes, path: Path) -> float:
    with wave.open(str(path), "wb") as w:
        w.setnchannels(1)
        w.setsampwidth(2)
        w.setframerate(SAMPLE_RATE)
        w.writeframes(pcm)
    with wave.open(str(path), "rb") as r:
        return round(r.getnframes() / float(r.getframerate()), 3)


def chars_to_words(text: str, chars: list[str], starts: list[float], ends: list[float]) -> list[dict]:
    """Fold per-character alignment into per-word [text,start,end]."""
    words: list[dict] = []
    cur = ""
    cur_start = None
    cur_end = 0.0
    for ch, st, en in zip(chars, starts, ends):
        if ch.strip() == "":
            if cur:
                words.append({"w": cur, "start": round(cur_start or 0, 3), "end": round(cur_end, 3)})
                cur, cur_start = "", None
            continue
        if cur_start is None:
            cur_start = st
        cur += ch
        cur_end = en
    if cur:
        words.append({"w": cur, "start": round(cur_start or 0, 3), "end": round(cur_end, 3)})
    return words


def main() -> int:
    key = os.environ.get("EL_KEY", "").strip()
    if not key:
        raise SystemExit("Set EL_KEY env var.")
    manifest, total = [], 0.0
    for seg in SEGMENTS:
        print(f"-> {seg['id']}: {seg['text'][:46]}...", flush=True)
        data = synth(seg["text"], key)
        pcm = base64.b64decode(data["audio_base64"])
        out = OUT_DIR / f"{seg['id']}.wav"
        dur = write_wav(pcm, out)
        align = data.get("alignment") or data.get("normalized_alignment") or {}
        words = chars_to_words(
            seg["text"],
            align.get("characters", []),
            align.get("character_start_times_seconds", []),
            align.get("character_end_times_seconds", []),
        )
        total += dur
        manifest.append({"id": seg["id"], "text": seg["text"], "duration": dur, "file": f"ad-vo/{seg['id']}.wav", "words": words})
        print(f"   {dur}s, {len(words)} words", flush=True)
    (OUT_DIR / "manifest.json").write_text(json.dumps(manifest, indent=2), encoding="utf-8")
    print(f"\n{MODEL} | total {round(total, 2)}s / {len(manifest)} segs (word-timed)", flush=True)
    return 0


if __name__ == "__main__":
    sys.exit(main())