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

"""SQLite-backed voice library index.

Audio files stay on disk. SQLite stores the computed voice-library payload so
the UI can load the last known library quickly while a scanner refreshes it.
"""
from __future__ import annotations

import logging
import threading
import time
from datetime import datetime, timezone
from pathlib import Path

from core.constants import _VOICES_DIR_DEFAULT
from core.database import (
    voice_index_delete,
    voice_index_get_all,
    voice_index_replace_all,
    voice_index_status,
    voice_index_upsert,
)
from core.voice import (
    _hidden_voices_dir,
    _active_voices_dir,
    _picture_path,
    _voice_audio_files,
    _voice_entry,
)

logger = logging.getLogger("uvicorn.error")

_REFRESH_MIN_INTERVAL_SEC = 45
_refresh_lock = threading.Lock()
_refresh_thread: threading.Thread | None = None
_last_refresh_started = 0.0


def _path_mtime(path: Path | None) -> float:
    if not path:
        return 0.0
    try:
        return path.stat().st_mtime
    except Exception:
        return 0.0


def _path_size(path: Path | None) -> int:
    if not path:
        return 0
    try:
        return path.stat().st_size
    except Exception:
        return 0


def _voice_sort_key(path: Path, active_dir: Path, hidden_dir: Path) -> tuple[str, int, str]:
    try:
        path.relative_to(active_dir)
        folder_rank = 0
    except ValueError:
        try:
            path.relative_to(hidden_dir)
            folder_rank = 1
        except ValueError:
            folder_rank = 2
    return (path.stem.lower(), folder_rank, str(path).lower())


def _sort_key_string(path: Path, active_dir: Path, hidden_dir: Path) -> str:
    stem, rank, full_path = _voice_sort_key(path, active_dir, hidden_dir)
    return f"{stem}\0{rank:02d}\0{full_path}"


def _apply_folder_enabled(entry: dict, path: Path, active_dir: Path, hidden_dir: Path) -> dict:
    try:
        path.relative_to(hidden_dir)
        entry["enabled"] = False
        return entry
    except ValueError:
        pass
    try:
        path.relative_to(active_dir)
        entry["enabled"] = True
    except ValueError:
        pass
    return entry


def _row_for_audio(settings: dict, audio: Path) -> dict:
    active_dir = _active_voices_dir(settings)
    hidden_dir = _hidden_voices_dir(settings)
    entry = _apply_folder_enabled(_voice_entry(audio), audio, active_dir, hidden_dir)
    meta = audio.with_suffix(".meta.json")
    ref = audio.with_suffix(".reference.txt")
    picture = _picture_path(audio)
    return {
        "voice_id": audio.stem,
        "path": str(audio),
        "sort_key": _sort_key_string(audio, active_dir, hidden_dir),
        "enabled": entry.get("enabled", True) is not False,
        "file_type": audio.suffix.lower().lstrip("."),
        "file_mtime": _path_mtime(audio),
        "file_size": _path_size(audio),
        "meta_mtime": _path_mtime(meta),
        "ref_mtime": _path_mtime(ref),
        "picture_mtime": _path_mtime(picture),
        "entry": entry,
    }


def rebuild_voice_index(settings: dict) -> list[dict]:
    scan_dir = Path(settings.get("voices_scan_dir", _VOICES_DIR_DEFAULT))
    active_dir = _active_voices_dir(settings)
    hidden_dir = _hidden_voices_dir(settings)
    rows: list[dict] = []
    if scan_dir.exists():
        seen: set[str] = set()
        all_audio = sorted(_voice_audio_files(scan_dir), key=lambda p: _voice_sort_key(p, active_dir, hidden_dir))
        for audio in all_audio:
            if audio.stem in seen:
                continue
            seen.add(audio.stem)
            try:
                rows.append(_row_for_audio(settings, audio))
            except Exception as exc:
                logger.warning("Could not index voice %s: %s", audio, exc)
    voice_index_replace_all(rows)
    return [row["entry"] for row in rows]


def upsert_voice_in_index(settings: dict, audio: Path) -> dict:
    row = _row_for_audio(settings, audio)
    voice_index_upsert(row)
    return row["entry"]


def remove_voice_from_index(voice_id: str) -> bool:
    return voice_index_delete(voice_id)


def _status_age(status: dict) -> float | None:
    raw = status.get("indexed_at")
    if not raw:
        return None
    try:
        stamped = datetime.fromisoformat(str(raw).replace("Z", "+00:00"))
        return max(0.0, (datetime.now(timezone.utc) - stamped).total_seconds())
    except Exception:
        return None


def _refresh_due(force: bool = False) -> bool:
    if force:
        return True
    status = voice_index_status()
    age = _status_age(status)
    return age is None or age >= _REFRESH_MIN_INTERVAL_SEC


def refresh_voice_index_background(settings: dict, *, force: bool = False) -> bool:
    global _refresh_thread, _last_refresh_started
    if not _refresh_due(force):
        return False
    with _refresh_lock:
        if _refresh_thread and _refresh_thread.is_alive():
            return False
        now = time.monotonic()
        if not force and now - _last_refresh_started < _REFRESH_MIN_INTERVAL_SEC:
            return False
        _last_refresh_started = now
        snapshot = dict(settings)

        def runner() -> None:
            try:
                rebuild_voice_index(snapshot)
            except Exception as exc:
                logger.warning("Voice index refresh failed: %s", exc)

        _refresh_thread = threading.Thread(target=runner, name="voice-index-refresh", daemon=True)
        _refresh_thread.start()
        return True


def indexed_voices(settings: dict, *, refresh: bool = False) -> list[dict]:
    status = voice_index_status()
    if refresh or not status.get("indexed_at"):
        return rebuild_voice_index(settings)
    voices = voice_index_get_all()
    refresh_voice_index_background(settings)
    return voices


def voice_index_info() -> dict:
    status = voice_index_status()
    age = _status_age(status)
    status["refreshing"] = bool(_refresh_thread and _refresh_thread.is_alive())
    status["age_sec"] = round(age, 2) if age is not None else None
    return status