214 lines
7.9 KiB
Python
214 lines
7.9 KiB
Python
"""SQLite persistence layer for characters and rehearsals.
|
|
|
|
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 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);
|
|
"""
|
|
|
|
|
|
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 _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
|