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

"""Boot-time defaults, path constants, in-memory log buffer, and routing log."""
from __future__ import annotations

import logging
import os
import shutil
from datetime import datetime, timezone
from pathlib import Path

# ── Version ───────────────────────────────────────────────────────────────────

def _read_version() -> str:
    v_file = Path(__file__).parent.parent / "VERSION"
    try:
        return v_file.read_text().strip()
    except Exception:
        return "0.0.0"

__version__: str = _read_version()

# ── Boot-time defaults ────────────────────────────────────────────────────────

_VOICES_DIR_DEFAULT  = os.environ.get("VOICES_DIR",  "/voices")
_OUTPUT_DIR_DEFAULT  = os.environ.get("OUTPUT_DIR",  "/voices/active_voices")
_WHISPER_DEFAULT     = os.environ.get("WHISPER_URL", "http://host.docker.internal:8010")
_TTS_DEFAULT         = os.environ.get("TTS_URL",     "http://host.docker.internal:8020")
_TTS_STREAM_DEFAULT  = os.environ.get("TTS_STREAM_URL", "http://host.docker.internal:8023")
_CUSTOMVOICE_DEFAULT = os.environ.get("CUSTOMVOICE_URL", "http://host.docker.internal:8022")
_VOICE_DESIGN_DEFAULT = os.environ.get("VOICE_DESIGN_URL", "http://host.docker.internal:8021")
_NVIDIA_ROUTER_DEFAULT = os.environ.get("NVIDIA_SPEECH_ROUTER_URL", "http://host.docker.internal:8090")
_NVIDIA_TTS_DEFAULT = os.environ.get("NVIDIA_MAGPIE_TTS_URL", "http://host.docker.internal:8091")
_NVIDIA_ASR_DEFAULT = os.environ.get("NVIDIA_PARAKEET_ASR_URL", "http://host.docker.internal:8092")
_NVIDIA_CLONE_DEFAULT = os.environ.get("NVIDIA_TTS_CLONE_URL", "http://host.docker.internal:8093")
_NVIDIA_ZEROSHOT_DEFAULT = os.environ.get("NVIDIA_ZEROSHOT_TTS_URL", _NVIDIA_CLONE_DEFAULT)
_NVIDIA_FLOW_DEFAULT = os.environ.get("NVIDIA_FLOW_TTS_URL", "http://host.docker.internal:8094")
_FASTER_WHISPER_DEFAULT = os.environ.get("FASTER_WHISPER_URL", "http://host.docker.internal:8000")
_WHISPER_CPP_DEFAULT    = os.environ.get("WHISPER_CPP_URL",    "http://host.docker.internal:8085")
_GROQ_STT_ENDPOINT      = "https://api.groq.com/openai/v1"
_KOKORO_DEFAULT         = os.environ.get("KOKORO_URL",         "http://host.docker.internal:8880/v1")
_VIBEVOICE_DEFAULT      = os.environ.get("VIBEVOICE_URL",      "http://192.168.178.8:8027")
_XTTS_DEFAULT           = os.environ.get("XTTS_URL",           "http://host.docker.internal:8024")
_FISHSPEECH_DEFAULT     = os.environ.get("FISHSPEECH_URL",     "http://host.docker.internal:38080")
_TTS_CONTAINER       = os.environ.get("TTS_CONTAINER_NAME", "faster-qwen3-tts")
_TTS_CONTAINERS_RAW  = os.environ.get("TTS_CONTAINER_NAMES", "")  # comma-separated override
_VOICE_DESIGN_MODEL  = os.environ.get("VOICE_DESIGN_MODEL", "Qwen3-TTS-12Hz-1.7B-VoiceDesign")
_VOICE_TARGET_DBFS   = float(os.environ.get("VOICE_TARGET_DBFS", "-20.0"))
_VOICE_PEAK_DBFS     = float(os.environ.get("VOICE_PEAK_DBFS", "-1.0"))
_MAX_UPLOAD_BYTES    = int(os.environ.get("MAX_UPLOAD_MB", "1024")) * 1024 * 1024
_MAX_PICTURE_BYTES   = int(os.environ.get("MAX_PICTURE_MB", "10")) * 1024 * 1024
_MAX_SOUND_BYTES     = int(os.environ.get("MAX_SOUND_MB", "25")) * 1024 * 1024
_MAX_TTS_OUTPUT_SECONDS = float(os.environ.get("MAX_TTS_OUTPUT_SECONDS", "30"))
_STT_REQUEST_TIMEOUT = float(os.environ.get("STT_REQUEST_TIMEOUT", os.environ.get("REQUEST_TIMEOUT", "900")))
_BENCHMARK_TEXT = os.environ.get("VOICE_BENCHMARK_TEXT", "This is a short realtime voice benchmark.")
_BENCHMARK_SENTENCES = [
    ("short", _BENCHMARK_TEXT),
    ("medium", "The quick brown fox jumps over the lazy dog near the river bank."),
    (
        "long",
        "Artificial intelligence is transforming the way we interact with technology. "
        "From voice assistants to autonomous vehicles, machine learning models are becoming "
        "an integral part of everyday life.",
    ),
]
_ALLOW_PRIVATE_DOWNLOADS = os.environ.get("ALLOW_PRIVATE_DOWNLOADS", "").lower() in {"1", "true", "yes"}

# ── Config dir calculation ────────────────────────────────────────────────────

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


def _default_config_dir() -> Path:
    env_dir = os.environ.get("CONFIG_DIR")
    if env_dir:
        return Path(env_dir)
    current = Path("/home/app/.config/tts-voice-creator")
    legacy = Path("/home/app/.config/voice-clone-factory")
    if legacy.exists():
        legacy.mkdir(parents=True, exist_ok=True)
        if current.exists():
            for name in ("settings.json", "voice_design_presets.json", "tts_routes.json"):
                src, dst = current / name, legacy / name
                if src.exists() and not dst.exists():
                    try:
                        shutil.copy2(src, dst)
                    except Exception as exc:
                        logger.warning("Could not migrate config %s to mounted config dir: %s", name, exc)
        return legacy
    return current


CONFIG_DIR  = _default_config_dir()
CONFIG_FILE = CONFIG_DIR / "settings.json"
DESIGN_PRESETS_FILE = CONFIG_DIR / "voice_design_presets.json"
TTS_ROUTES_FILE = CONFIG_DIR / "tts_routes.json"

# ── Static dir ────────────────────────────────────────────────────────────────

STATIC_DIR = Path(__file__).parent.parent / "static"
STATIC_DIR.mkdir(exist_ok=True)

# ── In-memory log buffer ───────────────────────────────────────────────────────

_LOG_BUFFER_MAX = 400
_log_buffer: list[dict] = []


class _BufferHandler(logging.Handler):
    def emit(self, record: logging.LogRecord) -> None:
        try:
            _log_buffer.insert(0, {
                "ts": datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat(),
                "level": record.levelname,
                "name": record.name,
                "msg": record.getMessage(),
            })
            del _log_buffer[_LOG_BUFFER_MAX:]
        except Exception:
            pass


# ── Routing log ───────────────────────────────────────────────────────────────

_ROUTING_LOG_MAX = int(os.environ.get("TTS_ROUTING_LOG_MAX", "120"))
_routing_log: list[dict] = []


def _routing_log_add(**entry) -> None:
    item = {
        "ts": datetime.now(timezone.utc).isoformat(),
        **entry,
    }
    _routing_log.insert(0, item)
    del _routing_log[_ROUTING_LOG_MAX:]