Gemini TTS: text file → audiobook WAVs (sync + batch). Zero pre-install: GEMINI_API_KEY=xxx uv run https://gist.githubusercontent.com/drewbitt/25d570edd4b48be1f15a5a52d191d24b/raw/tts_book.py book.txt output/

batch_tts.py

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.12"
# dependencies = ["google-genai>=1.70"]
# ///
"""GEMINI_API_KEY=xxx uv run batch_tts.py book.txt output/ [--voice Kore] [--model gemini-3.1-flash-tts-preview] [--chunk-size 4000]

Submits TTS chunks to Gemini Batch API (50% cheaper), polls until done, saves WAV files.
"""
import argparse
import base64
import json
import logging
import re
import struct
import time
from dataclasses import dataclass
from pathlib import Path

from google import genai
from google.genai import types


@dataclass
class TTSConfig:
    voice: str = "Kore"
    model: str = "gemini-3.1-flash-tts-preview"
    chunk_size: int = 4000
    poll_interval: int = 30
    sample_rate: int = 24000
    output_format: str = "wav"
    sync: bool = False
    verbose: bool = False


def setup_logging(verbose: bool) -> None:
    # Default: INFO for normal output, DEBUG for verbose
    level = logging.DEBUG if verbose else logging.INFO
    root = logging.getLogger()
    root.setLevel(level)
    if root.handlers:
        for h in root.handlers[:]:
            root.remove_handler(h)
    handler = logging.StreamHandler()
    handler.setLevel(level)
    handler.setFormatter(logging.Formatter("%(message)s"))
    root.addHandler(handler)
    # Silence noisy http library logs unless verbose
    if not verbose:
        for logger_name in ["httpcore", "charset_normalizer", "httpx"]:
            logging.getLogger(logger_name).setLevel(logging.WARNING)


def chunk(text: str, n: int = 4000) -> list[str]:
    """Paragraph-aware text splitter."""
    parts, cur = [], ""
    for p in (p.strip() for p in re.split(r"\n\s*\n", text) if p.strip()):
        if cur and len(cur) + len(p) > n:
            parts.append(cur)
            cur = p
        else:
            cur = f"{cur}\n\n{p}" if cur else p
    return [*parts, cur] if cur else parts


def wav(pcm: bytes, sr: int = 24000) -> bytes:
    """Minimal WAV wrapper — mono 16-bit PCM."""
    n = len(pcm)
    return struct.pack(
        "<4sI4s4sIHHIIHH4sI",
        b"RIFF", 36 + n, b"WAVE", b"fmt ", 16, 1, 1, sr, sr * 2, 2, 16, b"data",
        n,
    ) + pcm


def make_speech_config(voice: str) -> types.SpeechConfig:
    return types.SpeechConfig(
        voice_config=types.VoiceConfig(
            prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name=voice)
        )
    )


def make_generate_config(voice: str) -> types.GenerateContentConfig:
    return types.GenerateContentConfig(
        response_modalities=["AUDIO"],
        speech_config=make_speech_config(voice),
    )


def create_batch_requests(chunks: list[str], voice: str, model: str) -> list[dict]:
    """Build request dicts for batch API inline submission."""
    config = make_generate_config(voice)
    return [
        {
            "model": model,
            "contents": [{"parts": [{"text": text}]}],
            "config": config.model_dump(by_alias=True, exclude_none=True),
        }
        for text in chunks
    ]


def extract_audio_data(result, index: int) -> tuple[bytes | None, str]:
    """Extract audio data from a batch result. Returns (data, error_msg)."""
    if isinstance(result, dict):
        resp = result.get("response") or {}
        candidates = resp.get("candidates", [])
        if not candidates:
            return None, f"Chunk {index}: no candidates"
        content = candidates[0].get("content") or {}
        parts = content.get("parts", [])
        if not parts:
            return None, f"Chunk {index}: no parts"
        inline_data = parts[0].get("inlineData")
        if not inline_data:
            return None, f"Chunk {index}: no inlineData"
        data = inline_data.get("data", "")
    else:
        resp = getattr(result, "response", None)
        if resp is None:
            return None, f"Chunk {index}: no response object"
        candidates = getattr(resp, "candidates", [])
        if not candidates:
            return None, f"Chunk {index}: no candidates"
        content = getattr(candidates[0], "content", None)
        if content is None:
            return None, f"Chunk {index}: no content"
        parts = getattr(content, "parts", [])
        if not parts:
            return None, f"Chunk {index}: no parts"
        part = parts[0]
        inline_data = getattr(part, "inline_data", None) if hasattr(part, "inline_data") else None
        if not inline_data:
            return None, f"Chunk {index}: no inline_data"
        data = getattr(inline_data, "data", "") or ""

    audio_bytes = data if isinstance(data, bytes) else base64.b64decode(data)
    return audio_bytes, ""


def process_results(results: list, output_dir: Path, sample_rate: int, output_format: str) -> int:
    """Parse inline batch results, write audio files. Returns count of files written."""
    log = logging.getLogger(__name__)
    count = 0
    for i, result in enumerate(results):
        audio, err = extract_audio_data(result, i)
        if audio is None:
            log.debug(err)
            continue
        filename = output_dir / f"{i:04d}.{output_format}"
        if output_format == "wav":
            filename.write_bytes(wav(audio, sample_rate))
        else:
            filename.write_bytes(audio)
        log.debug("Chunk %d: %d bytes", i, len(audio))
        count += 1
    return count


def process_jsonl_results(results: list, output_dir: Path, sample_rate: int, output_format: str) -> int:
    """Parse JSONL format batch results. Returns count of files written."""
    log = logging.getLogger(__name__)
    count = 0
    for result in results:
        key = result.get("key")
        resp = result.get("response", {})
        cands = resp.get("candidates", [{}])
        parts = (cands[0].get("content", {}) or {}).get("parts", []) if isinstance(cands[0], dict) else []
        for part in parts:
            if "inlineData" not in part:
                continue
            data = part["inlineData"].get("data", "")
            audio = base64.b64decode(data)
            filename = output_dir / f"{int(key):04d}.{output_format}"
            if output_format == "wav":
                filename.write_bytes(wav(audio, sample_rate))
            else:
                filename.write_bytes(audio)
            log.debug("Chunk %s: %d bytes", key, len(audio))
            count += 1
    return count


def sync_tts_chunk(client: genai.Client, text: str, voice: str, model: str) -> bytes | None:
    """Generate TTS for a single chunk using sync API. Returns PCM audio or None on failure."""
    log = logging.getLogger(__name__)
    try:
        response = client.models.generate_content(
            model=model,
            contents=[{"parts": [{"text": text}]}],
            config=make_generate_config(voice),
        )
        candidates = getattr(response, "candidates", [])
        if not candidates:
            return None
        content = getattr(candidates[0], "content", None)
        if content is None:
            return None
        parts = getattr(content, "parts", [])
        for part in parts:
            inline_data = getattr(part, "inline_data", None) if hasattr(part, "inline_data") else None
            if inline_data is None:
                continue
            data = getattr(inline_data, "data", b"") or b""
            if isinstance(data, str):
                data = base64.b64decode(data)
            return data
        return None
    except Exception as e:
        log.debug("Sync API error: %s", e)
        return None


def sync_tts(chunks: list[str], voice: str, model: str, output_dir: Path, sample_rate: int, output_format: str) -> int:
    """Generate TTS for all chunks using sync API. Returns count of files written."""
    log = logging.getLogger(__name__)
    client = genai.Client()
    count = 0
    for i, text in enumerate(chunks):
        log.debug("Chunk %d: submitting sync...", i)
        audio = sync_tts_chunk(client, text, voice, model)
        if audio is None:
            log.debug("Chunk %d: no audio", i)
            continue
        filename = output_dir / f"{i:04d}.{output_format}"
        if output_format == "wav":
            filename.write_bytes(wav(audio, sample_rate))
        else:
            filename.write_bytes(audio)
        log.debug("Chunk %d: %d bytes", i, len(audio))
        count += 1
    return count


def wait_for_job(client: genai.Client, job_name: str, poll_interval: int = 30) -> types.BatchJob:
    """Poll batch job until completion."""
    log = logging.getLogger(__name__)
    while True:
        job = client.batches.get(name=job_name)
        state = job.state.name
        log.debug("Job state: %s", state)
        if state in ("JOB_STATE_SUCCEEDED", "JOB_STATE_FAILED", "JOB_STATE_CANCELLED"):
            return job
        time.sleep(poll_interval)


def handle_existing_job(client: genai.Client, job_name: str, job_marker: Path, output_dir: Path, sample_rate: int, output_format: str) -> int | None:
    """Handle resuming an existing job. Returns count if completed, None if should continue."""
    log = logging.getLogger(__name__)
    log.debug("Resuming job %s", job_name)
    job = client.batches.get(name=job_name)
    log.debug("Job state: %s", job.state.name)

    if job.state.name != "JOB_STATE_SUCCEEDED":
        job = wait_for_job(client, job_name)
        if job.state.name != "JOB_STATE_SUCCEEDED":
            log.warning("Job %s failed: %s", job_name, job.state.name)
            job_marker.write_text("")
            return None

    if job.dest and hasattr(job.dest, "inlined_responses") and job.dest.inlined_responses:
        count = process_results(job.dest.inlined_responses, output_dir, sample_rate, output_format)
    elif job.dest and hasattr(job.dest, "file_name") and job.dest.file_name:
        result_file = client.files.download(file=job.dest.file_name)
        raw = result_file.decode("utf-8").strip()
        results = [json.loads(line) for line in raw.split("\n") if line]
        count = process_jsonl_results(results, output_dir, sample_rate, output_format)
    else:
        log.error("No results found in job destination")
        job_marker.write_text("")
        return 0

    job_marker.write_text("")
    return count


def submit_and_process_batch(
    client: genai.Client,
    chunks: list[str],
    voice: str,
    model: str,
    input_stem: str,
    output_dir: Path,
    job_marker: Path,
    poll_interval: int,
    sample_rate: int,
    output_format: str,
) -> int | None:
    """Submit batch job and process results. Returns count or None on failure."""
    log = logging.getLogger(__name__)
    requests = create_batch_requests(chunks, voice, model)
    batch_job = client.batches.create(
        model=model,
        src=requests,
        config={"display_name": f"tts-{input_stem}"},
    )
    job_name = batch_job.name
    job_marker.write_text(job_name)
    log.debug("Job: %s", job_name)

    job = wait_for_job(client, job_name, poll_interval)
    log.debug("Job done: %s", job.state.name)

    if job.state.name != "JOB_STATE_SUCCEEDED":
        raise RuntimeError(f"Batch job {job.state.name}")

    if job.dest and hasattr(job.dest, "inlined_responses") and job.dest.inlined_responses:
        return process_results(job.dest.inlined_responses, output_dir, sample_rate, output_format)
    elif job.dest and hasattr(job.dest, "file_name") and job.dest.file_name:
        result_file = client.files.download(file=job.dest.file_name)
        raw = result_file.decode("utf-8").strip()
        results = [json.loads(line) for line in raw.split("\n") if line]
        return process_jsonl_results(results, output_dir, sample_rate, output_format)
    else:
        raise RuntimeError("No results found in job destination")


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("input", type=Path)
    parser.add_argument("output", type=Path)
    parser.add_argument("--voice", default="Kore")
    parser.add_argument("--model", default="gemini-3.1-flash-tts-preview")
    parser.add_argument("--chunk-size", type=int, default=4000)
    parser.add_argument("--poll-interval", type=int, default=30, help="Seconds between status checks")
    parser.add_argument("--sample-rate", type=int, default=24000, help="Output sample rate (24000 or 48000)")
    parser.add_argument("--output-format", default="wav", choices=["wav", "pcm"], help="Output format")
    parser.add_argument("--sync", action="store_true", help="Use sync API instead of batch API")
    parser.add_argument("--verbose", "-v", action="store_true", help="Enable verbose output")
    args = parser.parse_args()

    setup_logging(args.verbose)
    log = logging.getLogger(__name__)

    args.output.mkdir(parents=True, exist_ok=True)
    job_marker = args.output / ".job_name"

    text = args.input.read_text()
    chunks = chunk(text, args.chunk_size)
    log.info("%d chunks · %s · %s", len(chunks), args.voice, args.model)

    config = TTSConfig(
        voice=args.voice,
        model=args.model,
        chunk_size=args.chunk_size,
        poll_interval=args.poll_interval,
        sample_rate=args.sample_rate,
        output_format=args.output_format,
        sync=args.sync,
        verbose=args.verbose,
    )

    if args.sync:
        log.info("Using sync API mode...")
        count = sync_tts(chunks, config.voice, config.model, args.output, config.sample_rate, config.output_format)
        log.info("Done — %d files in %s/", count, args.output)
        return

    client = genai.Client()

    if job_marker.exists():
        job_name = job_marker.read_text().strip()
        if job_name:
            count = handle_existing_job(client, job_name, job_marker, args.output, config.sample_rate, config.output_format)
            if count is not None:
                log.info("Done — %d files written to %s/", count, args.output)
                return

    try:
        log.debug("Submitting batch job...")
        count = submit_and_process_batch(
            client, chunks, config.voice, config.model,
            args.input.stem, args.output, job_marker,
            config.poll_interval, config.sample_rate, config.output_format,
        )
        log.info("Done — %d files written to %s/", count, args.output)
    except Exception as e:
        log.warning("Batch API failed: %s", e)
        log.info("Falling back to sync API...")
        count = sync_tts(chunks, config.voice, config.model, args.output, config.sample_rate, config.output_format)
        log.info("Done — %d files in %s/", count, args.output)

    job_marker.write_text("")


if __name__ == "__main__":
    main()

tts_book.py

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.12"
# dependencies = ["google-genai>=1.70"]
# ///
"""GEMINI_API_KEY=xxx uv run tts_book.py book.txt output/ [--voice Kore] [--model ...]"""
import argparse
import re
import struct
import time
from pathlib import Path

from google import genai
from google.genai import types


def chunk(text: str, n: int = 4000) -> list[str]:
    """Paragraph-aware text splitter."""
    parts, cur = [], ""
    for p in (p.strip() for p in re.split(r"\n\s*\n", text) if p.strip()):
        if cur and len(cur) + len(p) > n:
            parts.append(cur)
            cur = p
        else:
            cur = f"{cur}\n\n{p}" if cur else p
    return [*parts, cur] if cur else parts


def tts(client: genai.Client, text: str, model: str, cfg: types.GenerateContentConfig, retries: int = 3) -> bytes:
    for i in range(retries):
        try:
            r = client.models.generate_content(model=model, contents=text, config=cfg)
            return r.candidates[0].content.parts[0].inline_data.data
        except Exception as e:
            if i + 1 == retries:
                raise
            time.sleep(5 << i)  # 5s, 10s, 20s
            print(f"    retry {i + 1}: {e}")
    raise RuntimeError("unreachable")


def wav(pcm: bytes, sr: int = 24000) -> bytes:
    """Minimal WAV wrapper — mono 16-bit PCM."""
    n = len(pcm)
    return struct.pack("<4sI4s4sIHHIIHH4sI", b"RIFF", 36 + n, b"WAVE", b"fmt ", 16, 1, 1, sr, sr * 2, 2, 16, b"data", n) + pcm


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("input", type=Path)
    ap.add_argument("output", type=Path)
    ap.add_argument("--voice", default="Kore")
    ap.add_argument("--model", default="gemini-3.1-flash-tts-preview")
    ap.add_argument("--chunk-size", type=int, default=4000)
    args = ap.parse_args()
    args.output.mkdir(parents=True, exist_ok=True)

    chunks = chunk(args.input.read_text(), args.chunk_size)
    client = genai.Client()
    cfg = types.GenerateContentConfig(
        response_modalities=["AUDIO"],
        speech_config=types.SpeechConfig(
            voice_config=types.VoiceConfig(
                prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name=args.voice),
            ),
        ),
    )

    print(f"{len(chunks)} chunks · {args.voice} · {args.model}")
    for i, text in enumerate(chunks):
        out = args.output / f"{i:04d}.wav"
        if out.exists():
            continue
        print(f"  [{i + 1}/{len(chunks)}] {len(text)} chars")
        out.write_bytes(wav(tts(client, text, args.model, cfg)))

    print(f"Done — {len(list(args.output.glob('*.wav')))} files in {args.output}/")


if __name__ == "__main__":
    main()