UmbraAudio/umbraaudio/library.py

"""Cross-project voice/name library.

The Library stores speakers (with their voice reference audio), confirmed
proper-name pronunciations, and (future) reusable utterances so they can be
used on other audiobook projects.  Data lives in the shared
``~/.umbra/voice_registry.db`` (same database as the Voice Registry) and the
copied voice samples live in ``~/.umbra/library/voices/``.
"""
from __future__ import annotations

import json
import shutil
import sqlite3
from pathlib import Path
from typing import Any

DEFAULT_LIBRARY_DB = Path.home() / ".umbra" / "voice_registry.db"
DEFAULT_LIBRARY_VOICES = Path.home() / ".umbra" / "library" / "voices"

LIBRARY_SCHEMA = """
CREATE TABLE IF NOT EXISTS library_speakers (
    id INTEGER PRIMARY KEY,
    primary_name TEXT NOT NULL,
    aliases TEXT NOT NULL DEFAULT '[]',
    gender TEXT,
    accent TEXT,
    vocal_traits TEXT NOT NULL DEFAULT '[]',
    distortion_effects TEXT NOT NULL DEFAULT '[]',
    is_narrator INTEGER NOT NULL DEFAULT 0,
    is_group INTEGER NOT NULL DEFAULT 0,
    description TEXT,
    source_project TEXT,
    voice_ref_path TEXT,
    created_at TEXT NOT NULL DEFAULT (datetime('now')),
    UNIQUE(primary_name)
);
CREATE TABLE IF NOT EXISTS library_names (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    phonetic TEXT,
    source TEXT,
    common INTEGER NOT NULL DEFAULT 0,
    confirmed INTEGER NOT NULL DEFAULT 0,
    source_project TEXT,
    created_at TEXT NOT NULL DEFAULT (datetime('now')),
    UNIQUE(name)
);
CREATE TABLE IF NOT EXISTS library_utterances (
    id INTEGER PRIMARY KEY,
    text TEXT NOT NULL,
    speaker_name TEXT,
    voice_ref_path TEXT,
    note TEXT,
    source_project TEXT,
    created_at TEXT NOT NULL DEFAULT (datetime('now'))
);
"""


def get_library_db_path() -> Path:
    return DEFAULT_LIBRARY_DB


def get_library_voices_dir() -> Path:
    return DEFAULT_LIBRARY_VOICES


def ensure_library_schema(db_path: Path | None = None) -> sqlite3.Connection:
    path = db_path or get_library_db_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(str(path))
    conn.row_factory = sqlite3.Row
    conn.executescript(LIBRARY_SCHEMA)
    conn.commit()
    return conn


def _json_str(value: Any) -> str:
    if isinstance(value, str):
        try:
            json.loads(value)
            return value
        except Exception:
            return json.dumps([value])
    if value is None:
        return "[]"
    return json.dumps(value)


def add_speakers(
    speakers: list[dict],
    source_project: str | None,
    voice_refs: dict[str, Path],
    db_path: Path | None = None,
    voices_dir: Path | None = None,
) -> dict:
    """Add speakers to the library, copying their voice reference into the
    library voices directory.

    ``speakers`` are the speaker dicts (from speakers.json).  ``voice_refs``
    maps speaker primary_name -> path to that speaker's ``*_ref.wav``.
    Returns ``{"added": [...], "already": [...], "skipped": [...]}``.
    """
    conn = ensure_library_schema(db_path)
    vdir = voices_dir or get_library_voices_dir()
    vdir.mkdir(parents=True, exist_ok=True)

    added, already, skipped = [], [], []
    inserted = 0
    for sp in speakers:
        name = (sp.get("primary_name") or "").strip()
        if not name:
            continue
        exists = conn.execute(
            "SELECT id FROM library_speakers WHERE primary_name = ?", (name,)
        ).fetchone()
        if exists:
            already.append(name)
            continue

        ref_src = voice_refs.get(name)
        copied_ref = None
        if ref_src and Path(ref_src).exists():
            dest = vdir / f"{name.lower().replace(' ', '_')}_ref.wav"
            try:
                shutil.copyfile(str(ref_src), str(dest))
                copied_ref = str(dest)
            except Exception:
                copied_ref = None

        conn.execute(
            """INSERT INTO library_speakers
                 (primary_name, aliases, gender, accent, vocal_traits,
                  distortion_effects, is_narrator, is_group, description,
                  source_project, voice_ref_path)
               VALUES (?,?,?,?,?,?,?,?,?,?,?)""",
            (
                name,
                _json_str(sp.get("aliases", [])),
                sp.get("gender") or sp.get("demographics", {}).get("gender"),
                sp.get("accent") or sp.get("demographics", {}).get("accent"),
                _json_str(sp.get("vocal_traits", [])),
                _json_str(sp.get("distortion_effects", [])),
                1 if sp.get("is_narrator") else 0,
                1 if sp.get("is_group") else 0,
                sp.get("description"),
                source_project,
                copied_ref,
            ),
        )
        inserted += 1
        added.append(name)

    conn.commit()
    conn.close()
    return {"added": added, "already": already, "skipped": skipped, "inserted": inserted}


def add_names(
    names: list[dict],
    source_project: str | None,
    db_path: Path | None = None,
) -> dict:
    """Add proper-name pronunciation records to the library.

    ``names`` are records with at least ``name`` and optionally ``phonetic``,
    ``source``, ``common``, ``confirmed``.
    """
    conn = ensure_library_schema(db_path)
    added, already = [], []
    for rec in names:
        name = (rec.get("name") or "").strip()
        if not name:
            continue
        exists = conn.execute(
            "SELECT id FROM library_names WHERE name = ?", (name,)
        ).fetchone()
        if exists:
            already.append(name)
            continue
        conn.execute(
            """INSERT INTO library_names (name, phonetic, source, common, confirmed, source_project)
               VALUES (?,?,?,?,?,?)""",
            (
                name,
                rec.get("phonetic"),
                rec.get("source"),
                1 if rec.get("common") else 0,
                1 if rec.get("confirmed") else 0,
                source_project,
            ),
        )
        added.append(name)
    conn.commit()
    conn.close()
    return {"added": added, "already": already}


def list_speakers(db_path: Path | None = None) -> list[dict]:
    conn = ensure_library_schema(db_path)
    rows = conn.execute(
        "SELECT * FROM library_speakers ORDER BY is_narrator DESC, primary_name"
    ).fetchall()
    data = [dict(r) for r in rows]
    conn.close()
    return data


def list_names(db_path: Path | None = None) -> list[dict]:
    conn = ensure_library_schema(db_path)
    rows = conn.execute(
        "SELECT * FROM library_names ORDER BY name"
    ).fetchall()
    data = [dict(r) for r in rows]
    conn.close()
    return data


def list_utterances(db_path: Path | None = None) -> list[dict]:
    conn = ensure_library_schema(db_path)
    rows = conn.execute(
        "SELECT * FROM library_utterances ORDER BY id"
    ).fetchall()
    data = [dict(r) for r in rows]
    conn.close()
    return data


def delete_speaker(speaker_id: int, db_path: Path | None = None) -> bool:
    conn = ensure_library_schema(db_path)
    cur = conn.execute("DELETE FROM library_speakers WHERE id = ?", (speaker_id,))
    conn.commit()
    conn.close()
    return cur.rowcount > 0


def delete_name(name_id: int, db_path: Path | None = None) -> bool:
    conn = ensure_library_schema(db_path)
    cur = conn.execute("DELETE FROM library_names WHERE id = ?", (name_id,))
    conn.commit()
    conn.close()
    return cur.rowcount > 0


def speaker_voice_ref_path(speaker_id: int, db_path: Path | None = None) -> Path | None:
    conn = ensure_library_schema(db_path)
    row = conn.execute(
        "SELECT voice_ref_path FROM library_speakers WHERE id = ?", (speaker_id,)
    ).fetchone()
    conn.close()
    if row and row["voice_ref_path"]:
        return Path(row["voice_ref_path"])
    return None