404 lines
15 KiB
Python
404 lines
15 KiB
Python
"""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 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")
|
|
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
|
|
|
|
|
|
# ── 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
|