UmbraAudio/scripts/create_vocals_project.py

#!/usr/bin/env python3
# Copyright 2026 Umbra Tools (umbratools.com)
# SPDX-License-Identifier: Apache-2.0

"""Create the standalone "Vocals" example-vocals project (no upload needed).

The project has a single chapter that speaks the Rainbow Passage once per
emotion variation with the Narrator voice, then repeats the exact same set
with the Deal Hunter voice. Speakers, voice references, and voice profiles are
copied from an existing project's library so both voices match the audiobook.

Usage:
    python scripts/create_vocals_project.py [--force] [--generate]

- Without options: build (or leave in place) the Vocals project structure.
- ``--force``: wipe and rebuild the project from scratch.
- ``--generate``: synthesize every variation, stitch the chapter, and export an
  MP3 to the project export dir (requires the Qwen TTS server on port 8881).
"""
from __future__ import annotations

import argparse
import json
import shutil
import sqlite3
import subprocess
import sys
from pathlib import Path
from typing import Any

sys.path.insert(0, str(Path(__file__).parent.parent))

from umbraaudio.config import load_config  # noqa: E402
from umbraaudio.database import Database  # noqa: E402

DEFAULT_SOURCE_PROJECT = "DealHunterShip_4.0e_whole-to-audio"
PROJECT_NAME = "Vocals"
CHAPTER_TITLE = "Rainbow Passage Variations"

RAINBOW_SENTENCES = [
    "When the sunlight strikes raindrops in the air, they act as a prism and form a rainbow.",  # noqa: E501
    "The rainbow is a division of white light into many beautiful colors.",
    "These take the shape of a long round arch, with its path high above, "
    "and its two ends apparently beyond the horizon.",
    "There is, according to legend, a boiling pot of gold at one end.",
    "People look, but no one ever finds it.",
    "When a man looks for something beyond his reach, "
    "his friends say he is looking for the pot of gold at the end of the rainbow.",  # noqa: E501
]

# Mirrors the emotion options in the per-utterance Variation editor so every
# variation the narrator/he ro can perform is demonstrated once.
VARIATIONS = [
    "neutral", "angry", "sad", "happy", "fearful", "surprised", "whisper",
    "pleading", "sarcastic", "authoritative", "thoughtful", "excited",
    "anxious", "tired", "romantic", "disgusted", "amused", "nostalgic",
]

SPEAKERS = ["Narrator", "Deal Hunter"]
VOICE_REF_FILES = ["narrator_ref.wav", "deal_hunter_ref.wav"]


def plan_utterances() -> list[dict[str, str]]:
    """One utterance per (speaker, variation) pair, cycling rainbow sentences."""
    plan: list[dict[str, str]] = []
    for speaker in SPEAKERS:
        for i, variation in enumerate(VARIATIONS):
            plan.append(
                {
                    "speaker_name": speaker,
                    "text": RAINBOW_SENTENCES[i % len(RAINBOW_SENTENCES)],
                    "emotion": variation,
                    "pace": "moderate",
                }
            )
    return plan


def _copy_library_speakers(source_dir: Path) -> list[dict[str, Any]]:
    """Return the Narrator + Deal Hunter entries from the source speakers.json."""
    source_file = source_dir / "speakers.json"
    data = json.loads(source_file.read_text())
    wanted = {s.lower() for s in SPEAKERS}
    return [s for s in data.get("speakers", []) if s.get("primary_name", "").lower() in wanted]


def _copy_voice_references(source_dir: Path, dest_voices: Path) -> list[Path]:
    copied: list[Path] = []
    for name in VOICE_REF_FILES:
        src = source_dir / "voices" / name
        if src.exists():
            dst = dest_voices / name
            shutil.copy2(src, dst)
            copied.append(dst)
    return copied


def setup(force: bool) -> Path:
    source_cfg = load_config(DEFAULT_SOURCE_PROJECT)
    cfg = load_config(PROJECT_NAME)
    project_dir = cfg.project_dir

    if project_dir.exists() and not force:
        print(f"Project '{PROJECT_NAME}' already exists at {project_dir} (use --force to rebuild)")
        return project_dir
    if project_dir.exists():
        shutil.rmtree(project_dir)

    cfg.ensure_dirs()

    speakers = _copy_library_speakers(source_cfg.data_dir)
    if len(speakers) != len(SPEAKERS):
        found = [s["primary_name"] for s in speakers]
        raise SystemExit(f"Source library missing speakers: found {found}")

    db = Database(cfg.db_path)
    import asyncio

    async def _bootstrap() -> int:
        await db.connect()
        project_id = await db.create_project(
            name=PROJECT_NAME, source_path="", source_format="json"
        )
        await db.update_project_status(project_id, "parsed")
        full_passage = " ".join(RAINBOW_SENTENCES)
        chapter_id = await db.insert_chapter(
            project_id=project_id,
            index=0,
            title=CHAPTER_TITLE,
            source_text=full_passage,
            word_count=len(full_passage.split()),
        )
        await db.close()
        return chapter_id

    chapter_id = asyncio.run(_bootstrap())

    conn = sqlite3.connect(str(cfg.db_path))
    for i, utt in enumerate(plan_utterances()):
        conn.execute(
            "INSERT INTO utterances "
            "(chapter_id, utterance_index, text, type, emotion, pace, speaker_name, status) "
            "VALUES (?, ?, ?, 'narration', ?, ?, ?, 'pending')",
            (chapter_id, i, utt["text"], utt["emotion"], utt["pace"], utt["speaker_name"]),
        )
    conn.commit()
    conn.close()

    speakers_file = cfg.data_dir / "speakers.json"
    payload = {
        "speakers": speakers,
        "book_title": PROJECT_NAME,
        "total_chapters": 1,
        "total_words": sum(len(u["text"].split()) for u in plan_utterances()),
    }
    speakers_file.write_text(json.dumps(payload, indent=2))

    copied = _copy_voice_references(source_cfg.data_dir, cfg.voices_dir)
    print(f"Voice references copied: {[p.name for p in copied]}")

    profiles_src = source_cfg.data_dir / "voice_profiles.json"
    if profiles_src.exists():
        profiles = json.loads(profiles_src.read_text())
        wanted = {s.lower() for s in SPEAKERS}
        kept = {k: v for k, v in profiles.items() if k.lower() in wanted}
        if kept:
            (cfg.data_dir / "voice_profiles.json").write_text(json.dumps(kept, indent=2))

    print(f"Created '{PROJECT_NAME}' project at {project_dir}")
    plan = plan_utterances()
    print(f"Chapter {chapter_id}: {CHAPTER_TITLE} ({len(plan)} utterances, "
          f"{len(plan) // len(SPEAKERS)} variations per speaker)")
    return project_dir


def export_chapter_mp3() -> None:
    """Stitch (if needed) and converts the single chapter to a 192k MP3."""
    cfg = load_config(PROJECT_NAME)
    db = sqlite3.connect(str(cfg.db_path))
    chapter_id = db.execute("SELECT id FROM chapters ORDER BY id LIMIT 1").fetchone()[0]
    db.close()

    chapter_dir = cfg.data_dir / "chapter_audio"
    full_wav = chapter_dir / f"ch{chapter_id}_full.wav"
    if not full_wav.exists():
        hint = f"stitch ch{chapter_id} or run with --generate first"
        raise SystemExit(f"No stitched audio at {full_wav} - {hint}")
    cfg.export_dir.mkdir(parents=True, exist_ok=True)
    out = cfg.export_dir / f"{PROJECT_NAME}_rainbow_variations.mp3"
    result = subprocess.run(
        [
            "ffmpeg", "-y", "-i", str(full_wav),
            "-ac", "1", "-ar", "44100",
            "-c:a", "libmp3lame", "-b:a", "192k",
            str(out),
        ],
        capture_output=True, text=True,
    )
    if result.returncode != 0:
        raise SystemExit(f"MP3 export failed: {result.stderr[-300:]}")
    print(f"Exported {out} ({out.stat().st_size / 1048576:.1f} MB)")


def generate() -> None:
    from scripts.generate_all_audio import main as gen_main

    gen_main(project=PROJECT_NAME, emotion_fx=True)
    export_chapter_mp3()


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--force", action="store_true", help="rebuild the project from scratch")
    parser.add_argument("--generate", action="store_true", help="synthesize audio and export MP3")
    args = parser.parse_args()

    setup(force=args.force)
    if args.generate:
        generate()


if __name__ == "__main__":
    main()