tts-voice-creator-clone-and-design-2/core/database.py

"""SQLite persistence layer for user-owned app data.

Single-file database at CONFIG_DIR/tts_creator.db.
Designed for easy migration to PostgreSQL / Supabase later:
  - JSON columns map 1-to-1 to JSONB
  - TEXT PKs (characters) and INTEGER PKs (rehearsals) both survive
  - No SQLite-specific pragmas that would break compat
"""
from __future__ import annotations

import json
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from core.constants import CONFIG_DIR

_DB_PATH = CONFIG_DIR / "tts_creator.db"

_DDL = """
CREATE TABLE IF NOT EXISTS characters (
    id       TEXT PRIMARY KEY,
    book     TEXT NOT NULL DEFAULT '',
    name     TEXT NOT NULL DEFAULT '',
    tags     TEXT NOT NULL DEFAULT '',
    voice    TEXT,
    image    TEXT,
    sheet    TEXT NOT NULL DEFAULT '{}',
    analysis TEXT,
    created  TEXT,
    updated  TEXT
);
CREATE INDEX IF NOT EXISTS idx_characters_book ON characters(book);

CREATE TABLE IF NOT EXISTS rehearsals (
    id             INTEGER PRIMARY KEY AUTOINCREMENT,
    title          TEXT NOT NULL DEFAULT '',
    script         TEXT NOT NULL DEFAULT '',
    cast           TEXT NOT NULL DEFAULT '{}',
    emotions       TEXT NOT NULL DEFAULT '{}',
    notes          TEXT NOT NULL DEFAULT '{}',
    ignored        TEXT NOT NULL DEFAULT '{}',
    hidden_lines   TEXT NOT NULL DEFAULT '{}',
    backend        TEXT NOT NULL DEFAULT '',
    narrator_voice TEXT NOT NULL DEFAULT '',
    line_index     INTEGER NOT NULL DEFAULT 0,
    created        TEXT,
    updated        TEXT
);
CREATE INDEX IF NOT EXISTS idx_rehearsals_title ON rehearsals(title);

CREATE TABLE IF NOT EXISTS reader_scripts (
    doc_id   TEXT NOT NULL,
    name     TEXT NOT NULL DEFAULT 'cast',
    title    TEXT NOT NULL DEFAULT '',
    payload  TEXT NOT NULL DEFAULT '{}',
    saved_at TEXT,
    created  TEXT,
    updated  TEXT,
    PRIMARY KEY (doc_id, name)
);
CREATE INDEX IF NOT EXISTS idx_reader_scripts_doc ON reader_scripts(doc_id);

CREATE TABLE IF NOT EXISTS app_state (
    key     TEXT PRIMARY KEY,
    value   TEXT NOT NULL DEFAULT 'null',
    updated TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS voice_index (
    voice_id      TEXT PRIMARY KEY,
    path          TEXT NOT NULL,
    sort_key      TEXT NOT NULL DEFAULT '',
    enabled       INTEGER NOT NULL DEFAULT 1,
    file_type     TEXT NOT NULL DEFAULT '',
    file_mtime    REAL NOT NULL DEFAULT 0,
    file_size     INTEGER NOT NULL DEFAULT 0,
    meta_mtime    REAL NOT NULL DEFAULT 0,
    ref_mtime     REAL NOT NULL DEFAULT 0,
    picture_mtime REAL NOT NULL DEFAULT 0,
    entry_json    TEXT NOT NULL DEFAULT '{}',
    indexed_at    TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_voice_index_enabled ON voice_index(enabled);
CREATE INDEX IF NOT EXISTS idx_voice_index_sort_key ON voice_index(sort_key);
"""


def _open() -> sqlite3.Connection:
    _DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(str(_DB_PATH), check_same_thread=False)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    return conn


def _init() -> None:
    with _open() as conn:
        conn.executescript(_DDL)


_init()


# ── Helpers ───────────────────────────────────────────────────────────────────

def _j(v: Any) -> str:
    return json.dumps(v, ensure_ascii=False, default=str)


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _row_to_char(row: sqlite3.Row) -> dict:
    d = dict(row)
    d["sheet"]    = json.loads(d.get("sheet")    or "{}")
    d["analysis"] = json.loads(d.get("analysis") or "null")
    d["color"]    = d["sheet"].get("color")
    return d


def _row_to_reh(row: sqlite3.Row) -> dict:
    d = dict(row)
    d["cast"]          = json.loads(d.get("cast")         or "{}")
    d["emotions"]      = json.loads(d.get("emotions")     or "{}")
    d["notes"]         = json.loads(d.get("notes")        or "{}")
    d["ignored"]       = json.loads(d.get("ignored")      or "{}")
    d["hidden"]        = json.loads(d.get("hidden_lines") or "{}")
    d["narratorVoice"] = d.pop("narrator_voice", "")
    d["lineIndex"]     = d.pop("line_index", 0)
    d.pop("hidden_lines", None)
    return d


# ── Characters ────────────────────────────────────────────────────────────────

def char_get_all() -> list[dict]:
    with _open() as conn:
        rows = conn.execute("SELECT * FROM characters ORDER BY book, name").fetchall()
    return [_row_to_char(r) for r in rows]


def char_get(char_id: str) -> dict | None:
    with _open() as conn:
        row = conn.execute("SELECT * FROM characters WHERE id=?", (char_id,)).fetchone()
    return _row_to_char(row) if row else None


def char_put(rec: dict) -> dict:
    with _open() as conn:
        conn.execute("""
            INSERT INTO characters (id, book, name, tags, voice, image, sheet, analysis, created, updated)
            VALUES (:id,:book,:name,:tags,:voice,:image,:sheet,:analysis,:created,:updated)
            ON CONFLICT(id) DO UPDATE SET
                book=excluded.book, name=excluded.name, tags=excluded.tags,
                voice=excluded.voice, image=excluded.image, sheet=excluded.sheet,
                analysis=excluded.analysis, updated=excluded.updated
        """, {
            "id":       rec.get("id", ""),
            "book":     rec.get("book", ""),
            "name":     rec.get("name", ""),
            "tags":     rec.get("tags", ""),
            "voice":    rec.get("voice"),
            "image":    rec.get("image"),
            "sheet":    _j(rec.get("sheet", {})),
            "analysis": _j(rec.get("analysis")) if rec.get("analysis") is not None else None,
            "created":  str(rec.get("created", "")),
            "updated":  str(rec.get("updated", "")),
        })
        conn.commit()
    return rec


def char_delete(char_id: str) -> bool:
    with _open() as conn:
        conn.execute("DELETE FROM characters WHERE id=?", (char_id,))
        conn.commit()
    return True


# ── Rehearsals ────────────────────────────────────────────────────────────────

def reh_get_all() -> list[dict]:
    with _open() as conn:
        rows = conn.execute(
            "SELECT * FROM rehearsals ORDER BY updated DESC"
        ).fetchall()
    return [_row_to_reh(r) for r in rows]


def reh_get(reh_id: int) -> dict | None:
    with _open() as conn:
        row = conn.execute("SELECT * FROM rehearsals WHERE id=?", (reh_id,)).fetchone()
    return _row_to_reh(row) if row else None


def reh_put(rec: dict) -> dict:
    """Insert (no id) or full replace (id present). Returns record with id."""
    reh_id = rec.get("id")
    row = {
        "title":          rec.get("title", ""),
        "script":         rec.get("script", ""),
        "cast":           _j(rec.get("cast", {})),
        "emotions":       _j(rec.get("emotions", {})),
        "notes":          _j(rec.get("notes", {})),
        "ignored":        _j(rec.get("ignored", {})),
        "hidden_lines":   _j(rec.get("hidden", {})),
        "backend":        rec.get("backend", ""),
        "narrator_voice": rec.get("narratorVoice", ""),
        "line_index":     int(rec.get("lineIndex", 0)),
        "created":        str(rec.get("created", "")),
        "updated":        str(rec.get("updated", "")),
    }
    with _open() as conn:
        if reh_id:
            row["id"] = int(reh_id)
            conn.execute("""
                INSERT INTO rehearsals
                    (id,title,script,cast,emotions,notes,ignored,hidden_lines,
                     backend,narrator_voice,line_index,created,updated)
                VALUES
                    (:id,:title,:script,:cast,:emotions,:notes,:ignored,:hidden_lines,
                     :backend,:narrator_voice,:line_index,:created,:updated)
                ON CONFLICT(id) DO UPDATE SET
                    title=excluded.title, script=excluded.script, cast=excluded.cast,
                    emotions=excluded.emotions, notes=excluded.notes, ignored=excluded.ignored,
                    hidden_lines=excluded.hidden_lines, backend=excluded.backend,
                    narrator_voice=excluded.narrator_voice, line_index=excluded.line_index,
                    updated=excluded.updated
            """, row)
            conn.commit()
        else:
            cur = conn.execute("""
                INSERT INTO rehearsals
                    (title,script,cast,emotions,notes,ignored,hidden_lines,
                     backend,narrator_voice,line_index,created,updated)
                VALUES
                    (:title,:script,:cast,:emotions,:notes,:ignored,:hidden_lines,
                     :backend,:narrator_voice,:line_index,:created,:updated)
            """, row)
            conn.commit()
            reh_id = cur.lastrowid
    rec = dict(rec)
    rec["id"] = reh_id
    return rec


def reh_delete(reh_id: int) -> bool:
    with _open() as conn:
        conn.execute("DELETE FROM rehearsals WHERE id=?", (reh_id,))
        conn.commit()
    return True


# ── Reader scripts / audiobook casts ─────────────────────────────────────────

def reader_script_list(doc_id: str) -> list[dict]:
    with _open() as conn:
        rows = conn.execute(
            "SELECT name, title, payload, saved_at, updated FROM reader_scripts WHERE doc_id=? ORDER BY updated DESC",
            (doc_id,),
        ).fetchall()
    out: list[dict] = []
    for row in rows:
        payload: dict = {}
        try:
            payload = json.loads(row["payload"] or "{}")
        except Exception:
            payload = {}
        out.append({
            "name": row["name"],
            "title": row["title"],
            "savedAt": row["saved_at"] or payload.get("savedAt"),
            "updated": row["updated"],
            "segments": len(payload.get("segments", [])) if isinstance(payload.get("segments"), list) else 0,
            "roster": payload.get("roster", []),
        })
    return out


def reader_script_get(doc_id: str, name: str = "cast") -> dict | None:
    with _open() as conn:
        row = conn.execute(
            "SELECT payload FROM reader_scripts WHERE doc_id=? AND name=?",
            (doc_id, name),
        ).fetchone()
    if not row:
        return None
    try:
        data = json.loads(row["payload"] or "{}")
        return data if isinstance(data, dict) else None
    except Exception:
        return None


def reader_script_put(doc_id: str, name: str, data: dict) -> dict:
    now = _now()
    with _open() as conn:
        existing = conn.execute(
            "SELECT created FROM reader_scripts WHERE doc_id=? AND name=?",
            (doc_id, name),
        ).fetchone()
        created = existing["created"] if existing else now
        conn.execute("""
            INSERT INTO reader_scripts (doc_id, name, title, payload, saved_at, created, updated)
            VALUES (?, ?, ?, ?, ?, ?, ?)
            ON CONFLICT(doc_id, name) DO UPDATE SET
                title=excluded.title,
                payload=excluded.payload,
                saved_at=excluded.saved_at,
                updated=excluded.updated
        """, (
            doc_id,
            name,
            str(data.get("title", "")),
            _j(data),
            str(data.get("savedAt", "")),
            created,
            now,
        ))
        conn.commit()
    return data


def reader_script_delete(doc_id: str, name: str = "cast") -> bool:
    with _open() as conn:
        conn.execute("DELETE FROM reader_scripts WHERE doc_id=? AND name=?", (doc_id, name))
        conn.commit()
    return True


# ── Generic app state ─────────────────────────────────────────────────────────

def state_get(key: str, default: Any = None) -> Any:
    with _open() as conn:
        row = conn.execute("SELECT value FROM app_state WHERE key=?", (key,)).fetchone()
    if not row:
        return default
    try:
        return json.loads(row["value"])
    except Exception:
        return default


def state_updated(key: str) -> str:
    with _open() as conn:
        row = conn.execute("SELECT updated FROM app_state WHERE key=?", (key,)).fetchone()
    return str(row["updated"]) if row else ""


def state_put(key: str, value: Any) -> Any:
    updated = _now()
    with _open() as conn:
        conn.execute("""
            INSERT INTO app_state (key, value, updated)
            VALUES (?, ?, ?)
            ON CONFLICT(key) DO UPDATE SET
                value=excluded.value,
                updated=excluded.updated
        """, (key, _j(value), updated))
        conn.commit()
    return value


def state_delete(key: str) -> bool:
    with _open() as conn:
        conn.execute("DELETE FROM app_state WHERE key=?", (key,))
        conn.commit()
    return True


# ── Voice library index ───────────────────────────────────────────────────────

def voice_index_get_all() -> list[dict]:
    with _open() as conn:
        rows = conn.execute(
            "SELECT entry_json FROM voice_index ORDER BY sort_key, voice_id"
        ).fetchall()
    voices: list[dict] = []
    for row in rows:
        try:
            entry = json.loads(row["entry_json"] or "{}")
            if isinstance(entry, dict):
                voices.append(entry)
        except Exception:
            continue
    return voices


def voice_index_count() -> int:
    with _open() as conn:
        row = conn.execute("SELECT COUNT(*) AS n FROM voice_index").fetchone()
    return int(row["n"] if row else 0)


def voice_index_replace_all(rows: list[dict]) -> None:
    indexed_at = _now()
    with _open() as conn:
        conn.execute("DELETE FROM voice_index")
        conn.executemany("""
            INSERT INTO voice_index (
                voice_id, path, sort_key, enabled, file_type, file_mtime,
                file_size, meta_mtime, ref_mtime, picture_mtime, entry_json, indexed_at
            )
            VALUES (
                :voice_id, :path, :sort_key, :enabled, :file_type, :file_mtime,
                :file_size, :meta_mtime, :ref_mtime, :picture_mtime, :entry_json, :indexed_at
            )
        """, [
            {
                "voice_id": str(row.get("voice_id", "")),
                "path": str(row.get("path", "")),
                "sort_key": str(row.get("sort_key", "")),
                "enabled": 1 if row.get("enabled", True) else 0,
                "file_type": str(row.get("file_type", "")),
                "file_mtime": float(row.get("file_mtime") or 0),
                "file_size": int(row.get("file_size") or 0),
                "meta_mtime": float(row.get("meta_mtime") or 0),
                "ref_mtime": float(row.get("ref_mtime") or 0),
                "picture_mtime": float(row.get("picture_mtime") or 0),
                "entry_json": _j(row.get("entry", {})),
                "indexed_at": indexed_at,
            }
            for row in rows
            if row.get("voice_id") and row.get("path")
        ])
        conn.execute("""
            INSERT INTO app_state (key, value, updated)
            VALUES ('voice_index_status', ?, ?)
            ON CONFLICT(key) DO UPDATE SET
                value=excluded.value,
                updated=excluded.updated
        """, (_j({"count": len(rows), "indexed_at": indexed_at}), indexed_at))
        conn.commit()


def voice_index_upsert(row: dict) -> None:
    indexed_at = _now()
    with _open() as conn:
        conn.execute("""
            INSERT INTO voice_index (
                voice_id, path, sort_key, enabled, file_type, file_mtime,
                file_size, meta_mtime, ref_mtime, picture_mtime, entry_json, indexed_at
            )
            VALUES (
                :voice_id, :path, :sort_key, :enabled, :file_type, :file_mtime,
                :file_size, :meta_mtime, :ref_mtime, :picture_mtime, :entry_json, :indexed_at
            )
            ON CONFLICT(voice_id) DO UPDATE SET
                path=excluded.path,
                sort_key=excluded.sort_key,
                enabled=excluded.enabled,
                file_type=excluded.file_type,
                file_mtime=excluded.file_mtime,
                file_size=excluded.file_size,
                meta_mtime=excluded.meta_mtime,
                ref_mtime=excluded.ref_mtime,
                picture_mtime=excluded.picture_mtime,
                entry_json=excluded.entry_json,
                indexed_at=excluded.indexed_at
        """, {
            "voice_id": str(row.get("voice_id", "")),
            "path": str(row.get("path", "")),
            "sort_key": str(row.get("sort_key", "")),
            "enabled": 1 if row.get("enabled", True) else 0,
            "file_type": str(row.get("file_type", "")),
            "file_mtime": float(row.get("file_mtime") or 0),
            "file_size": int(row.get("file_size") or 0),
            "meta_mtime": float(row.get("meta_mtime") or 0),
            "ref_mtime": float(row.get("ref_mtime") or 0),
            "picture_mtime": float(row.get("picture_mtime") or 0),
            "entry_json": _j(row.get("entry", {})),
            "indexed_at": indexed_at,
        })
        conn.commit()


def voice_index_delete(voice_id: str) -> bool:
    with _open() as conn:
        conn.execute("DELETE FROM voice_index WHERE voice_id=?", (voice_id,))
        conn.commit()
    return True


def voice_index_status() -> dict:
    status = state_get("voice_index_status", {}) or {}
    if not isinstance(status, dict):
        status = {}
    status.setdefault("count", voice_index_count())
    status.setdefault("indexed_at", state_updated("voice_index_status"))
    return status