feat: Refactor pronunciation storage from SQLite to JSON for improved simplicity and performance

abogen/pronunciation_store.py

@@ -1,9 +1,10 @@
 from __future__ import annotations
 
-import sqlite3
+import json
 import shutil
 import threading
 import time
+import uuid
 from pathlib import Path
 from typing import Any, Dict, Iterable, List, Mapping, Optional
 
@@ -19,125 +20,64 @@ def _store_path() -> Path:
         base_dir = Path(get_user_settings_dir())
     except ModuleNotFoundError:
         base_dir = Path(get_internal_cache_path("pronunciations"))
-    target = base_dir / "pronunciations.db"
+    target = base_dir / "overrides.json"
     target.parent.mkdir(parents=True, exist_ok=True)
-
-    if not target.exists():
-        try:
-            legacy_dir = Path(get_internal_cache_path("pronunciations"))
-            legacy_path = legacy_dir / "pronunciations.db"
-            if legacy_path.exists() and legacy_path != target:
-                shutil.move(str(legacy_path), target)
-        except Exception:
-            pass
-
     return target
 
 
-def _connect() -> sqlite3.Connection:
+def _load_db() -> Dict[str, Any]:
-    connection = sqlite3.connect(_store_path())
+    path = _store_path()
-    connection.execute("PRAGMA journal_mode=WAL")
+    if not path.exists():
-    connection.execute("PRAGMA foreign_keys=ON")
+        return {"version": _SCHEMA_VERSION, "overrides": {}}
-    return connection
+    try:
+        with open(path, "r", encoding="utf-8") as f:
+            return json.load(f)
+    except (json.JSONDecodeError, OSError):
+        return {"version": _SCHEMA_VERSION, "overrides": {}}
 
 
-def _ensure_schema(conn: sqlite3.Connection) -> None:
+def _save_db(data: Dict[str, Any]) -> None:
-    conn.execute(
+    path = _store_path()
-        """
+    # Atomic write
-        CREATE TABLE IF NOT EXISTS overrides (
+    temp_path = path.with_suffix(".tmp")
-            id INTEGER PRIMARY KEY AUTOINCREMENT,
+    with open(temp_path, "w", encoding="utf-8") as f:
-            normalized TEXT NOT NULL,
+        json.dump(data, f, indent=2, ensure_ascii=False)
-            token TEXT NOT NULL,
+    shutil.move(str(temp_path), str(path))
-            language TEXT NOT NULL,
-            pronunciation TEXT,
-            voice TEXT,
-            notes TEXT,
-            context TEXT,
-            usage_count INTEGER NOT NULL DEFAULT 0,
-            created_at REAL NOT NULL,
-            updated_at REAL NOT NULL,
-            UNIQUE(normalized, language)
-        )
-        """
-    )
-    conn.execute(
-        """
-        CREATE TABLE IF NOT EXISTS metadata (
-            key TEXT PRIMARY KEY,
-            value INTEGER NOT NULL
-        )
-        """
-    )
-    row = conn.execute("SELECT value FROM metadata WHERE key='schema_version'").fetchone()
-    if row is None:
-        conn.execute(
-            "INSERT OR REPLACE INTO metadata(key, value) VALUES('schema_version', ?)",
-            (_SCHEMA_VERSION,),
-        )
-    conn.commit()
-
-
-def _row_to_dict(row: sqlite3.Row) -> Dict[str, Any]:
-    return {
-        "id": row["id"],
-        "normalized": row["normalized"],
-        "token": row["token"],
-        "language": row["language"],
-        "pronunciation": row["pronunciation"],
-        "voice": row["voice"],
-        "notes": row["notes"],
-        "context": row["context"],
-        "usage_count": row["usage_count"],
-        "created_at": row["created_at"],
-        "updated_at": row["updated_at"],
-    }
 
 
 def load_overrides(language: str, tokens: Iterable[str]) -> Dict[str, Dict[str, Any]]:
     normalized_tokens = {normalize_token(token) for token in tokens if token}
     if not normalized_tokens:
         return {}
-    # Use parameterized queries to prevent SQL injection
+    
-    placeholders = ",".join("?" for _ in normalized_tokens)
     with _DB_LOCK:
-        conn = _connect()
+        db = _load_db()
-        try:
+        lang_overrides = db.get("overrides", {}).get(language, {})
-            _ensure_schema(conn)
+        
-            conn.row_factory = sqlite3.Row
+        results: Dict[str, Dict[str, Any]] = {}
-            cursor = conn.execute(
+        for normalized in normalized_tokens:
-                f"SELECT * FROM overrides WHERE language=? AND normalized IN ({placeholders})",
+            if normalized in lang_overrides:
-                (language, *normalized_tokens),
+                results[normalized] = lang_overrides[normalized]
-            )
+        return results
-            results: Dict[str, Dict[str, Any]] = {}
-            for row in cursor.fetchall():
-                payload = _row_to_dict(row)
-                results[payload["normalized"]] = payload
-            return results
-        finally:
-            conn.close()
 
 
 def search_overrides(language: str, query: str, *, limit: int = 15) -> List[Dict[str, Any]]:
     if not query:
         return []
-    pattern = f"%{query.lower()}%"
+    
+    query = query.lower()
     with _DB_LOCK:
-        conn = _connect()
+        db = _load_db()
-        try:
+        lang_overrides = db.get("overrides", {}).get(language, {})
-            _ensure_schema(conn)
+        
-            conn.row_factory = sqlite3.Row
+        matches = []
-            cursor = conn.execute(
+        for entry in lang_overrides.values():
-                """
+            if query in entry["normalized"] or query in entry["token"].lower():
-                SELECT * FROM overrides
+                matches.append(entry)
-                WHERE language = ? AND (normalized LIKE ? OR LOWER(token) LIKE ?)
+        
-                ORDER BY usage_count DESC, updated_at DESC
+        # Sort by usage count desc, then updated_at desc
-                LIMIT ?
+        matches.sort(key=lambda x: (x.get("usage_count", 0), x.get("updated_at", 0)), reverse=True)
-                """,
+        return matches[:limit]
-                (language, pattern, pattern, limit),
-            )
-            return [_row_to_dict(row) for row in cursor.fetchall()]
-        finally:
-            conn.close()
 
 
 def save_override(
@@ -152,116 +92,95 @@ def save_override(
     normalized = normalize_token(token)
     if not normalized:
         raise ValueError("Provide a token to override")
+    
     timestamp = time.time()
     with _DB_LOCK:
-        conn = _connect()
+        db = _load_db()
-        try:
+        overrides = db.setdefault("overrides", {})
-            _ensure_schema(conn)
+        lang_overrides = overrides.setdefault(language, {})
-            conn.execute(
+        
-                """
+        existing = lang_overrides.get(normalized)
-                INSERT INTO overrides (normalized, token, language, pronunciation, voice, notes, context, usage_count, created_at, updated_at)
+        
-                VALUES (?, ?, ?, ?, ?, ?, ?, 0, ?, ?)
+        if existing:
-                ON CONFLICT(normalized, language) DO UPDATE SET
+            entry = existing
-                    token=excluded.token,
+            entry["token"] = token
-                    pronunciation=excluded.pronunciation,
+            entry["pronunciation"] = pronunciation
-                    voice=excluded.voice,
+            entry["voice"] = voice
-                    notes=excluded.notes,
+            entry["notes"] = notes
-                    context=excluded.context,
+            entry["context"] = context
-                    updated_at=excluded.updated_at
+            entry["updated_at"] = timestamp
-                """,
+        else:
-                (
+            entry = {
-                    normalized,
+                "id": str(uuid.uuid4()),
-                    token,
+                "normalized": normalized,
-                    language,
+                "token": token,
-                    pronunciation,
+                "language": language,
-                    voice,
+                "pronunciation": pronunciation,
-                    notes,
+                "voice": voice,
-                    context,
+                "notes": notes,
-                    timestamp,
+                "context": context,
-                    timestamp,
+                "usage_count": 0,
-                ),
+                "created_at": timestamp,
-            )
+                "updated_at": timestamp,
-            conn.commit()
+            }
-            conn.row_factory = sqlite3.Row
+            lang_overrides[normalized] = entry
-            row = conn.execute(
+            
-                "SELECT * FROM overrides WHERE normalized=? AND language=?",
+        _save_db(db)
-                (normalized, language),
+        return entry
-            ).fetchone()
-            if row is None:  # pragma: no cover - defensive guard
-                raise RuntimeError("Override save failed")
-            return _row_to_dict(row)
-        finally:
-            conn.close()
 
 
 def delete_override(*, language: str, token: str) -> None:
     normalized = normalize_token(token)
     if not normalized:
         return
+    
     with _DB_LOCK:
-        conn = _connect()
+        db = _load_db()
-        try:
+        lang_overrides = db.get("overrides", {}).get(language, {})
-            _ensure_schema(conn)
+        
-            conn.execute("DELETE FROM overrides WHERE normalized=? AND language=?", (normalized, language))
+        if normalized in lang_overrides:
-            conn.commit()
+            del lang_overrides[normalized]
-        finally:
+            _save_db(db)
-            conn.close()
 
 
 def all_overrides(language: str) -> List[Dict[str, Any]]:
     with _DB_LOCK:
-        conn = _connect()
+        db = _load_db()
-        try:
+        lang_overrides = db.get("overrides", {}).get(language, {})
-            _ensure_schema(conn)
+        
-            conn.row_factory = sqlite3.Row
+        results = list(lang_overrides.values())
-            cursor = conn.execute(
+        results.sort(key=lambda x: x.get("updated_at", 0), reverse=True)
-                "SELECT * FROM overrides WHERE language=? ORDER BY updated_at DESC",
+        return results
-                (language,),
-            )
-            return [_row_to_dict(row) for row in cursor.fetchall()]
-        finally:
-            conn.close()
 
 
 def increment_usage(*, language: str, token: str, amount: int = 1) -> None:
     normalized = normalize_token(token)
     if not normalized:
         return
+    
     with _DB_LOCK:
-        conn = _connect()
+        db = _load_db()
-        try:
+        lang_overrides = db.get("overrides", {}).get(language, {})
-            _ensure_schema(conn)
+        
-            conn.execute(
+        if normalized in lang_overrides:
-                "UPDATE overrides SET usage_count = usage_count + ?, updated_at = ? WHERE normalized=? AND language=?",
+            entry = lang_overrides[normalized]
-                (amount, time.time(), normalized, language),
+            entry["usage_count"] = entry.get("usage_count", 0) + amount
-            )
+            entry["updated_at"] = time.time()
-            conn.commit()
+            _save_db(db)
-        finally:
-            conn.close()
 
 
 def get_override_stats(language: str) -> Dict[str, int]:
     with _DB_LOCK:
-        conn = _connect()
+        db = _load_db()
-        try:
+        lang_overrides = db.get("overrides", {}).get(language, {})
-            _ensure_schema(conn)
+        
-            cursor = conn.execute(
+        total = len(lang_overrides)
-                """
+        with_pronunciation = sum(1 for x in lang_overrides.values() if x.get("pronunciation"))
-                SELECT 
+        with_voice = sum(1 for x in lang_overrides.values() if x.get("voice"))
-                    COUNT(*) as total,
+        
-                    COUNT(CASE WHEN pronunciation IS NOT NULL AND pronunciation != '' THEN 1 END) as with_pronunciation,
+        return {
-                    COUNT(CASE WHEN voice IS NOT NULL AND voice != '' THEN 1 END) as with_voice
+            "total": total,
-                FROM overrides 
+            "filtered": total,
-                WHERE language=?
+            "with_pronunciation": with_pronunciation,
-                """,
+            "with_voice": with_voice,
-                (language,),
+        }
-            )
-            row = cursor.fetchone()
-            return {
-                "total": row[0],
-                "filtered": row[0],
-                "with_pronunciation": row[1],
-                "with_voice": row[2],
-            }
-        finally:
-            conn.close()