app.py

# app.py
from __future__ import annotations
import os
import tarfile
import io
import shutil
from typing import Optional

from app.gradio_app import build_demo
from models.tts_router import cleanup_old_audio, ensure_runtime_audio_dir
import huggingface_hub


# -------------------------
# Helpers: models & Piper
# -------------------------
def _env(name: str, default: Optional[str] = None) -> Optional[str]:
    # Environment takes precedence over .env defaults (pydantic loads .env)
    return os.environ.get(name, default)

def ensure_model() -> str:
    """
    Ensure a local llama.cpp GGUF exists at LLAMACPP_MODEL_PATH.
    If missing, download from HF_MODEL_REPO (and optional HF_MODEL_FILE).
    """
    model_path = _env("LLAMACPP_MODEL_PATH")
    repo_id    = _env("HF_MODEL_REPO")
    file_name  = _env("HF_MODEL_FILE") or (os.path.basename(model_path) if model_path else None)

    if not model_path or not repo_id or not file_name:
        raise RuntimeError(
            "Missing config: set LLAMACPP_MODEL_PATH and HF_MODEL_REPO in .env (optionally HF_MODEL_FILE)."
        )

    if os.path.exists(model_path):
        print(f"[MODEL] Found: {model_path}")
        return model_path

    os.makedirs(os.path.dirname(model_path), exist_ok=True)
    print(f"[MODEL] Downloading {file_name} from {repo_id} …")
    local_path = huggingface_hub.hf_hub_download(
        repo_id=repo_id,
        filename=file_name,
        local_dir=os.path.dirname(model_path),
        local_dir_use_symlinks=False,
    )
    # If hf_hub_download stored under a hashed subdir, move to exact target path
    if os.path.abspath(local_path) != os.path.abspath(model_path):
        shutil.copy2(local_path, model_path)
    print(f"[MODEL] Ready at {model_path}")
    return model_path

def ensure_piper() -> tuple[str, str]:
    """
    Ensure Piper binary + voice model exist, and set env vars PIPER_BIN, PIPER_MODEL.
    Returns (piper_bin, piper_model).
    """
    # --- Ensure binary ---
    desired_bin = _env("PIPER_BIN", os.path.abspath("./bin/piper"))
    if not os.path.exists(desired_bin):
        os.makedirs(os.path.dirname(desired_bin), exist_ok=True)
        # Download a small Linux x86_64 Piper binary tarball from the official repo
        # (Using a known release; if HF changes base image/arch you may need to adjust)
        print("[PIPER] Downloading Piper binary …")
        # v1.2.0 is a common stable tag; adjust if needed
        piper_repo = "rhasspy/piper"
        piper_asset = "piper_linux_x86_64.tar.gz"
        # Pull via HF Hub to avoid GitHub rate limiting on Spaces runners
        # We mirror by leveraging HF's cached files capability (requires file to exist in repo).
        # If you have your own mirror, point HF_PIPER_REPO/HF_PIPER_FILE env vars to it.
        piper_hf_repo = _env("HF_PIPER_REPO")
        piper_hf_file = _env("HF_PIPER_FILE")
        if piper_hf_repo and piper_hf_file:
            tar_path = huggingface_hub.hf_hub_download(
                repo_id=piper_hf_repo,
                filename=piper_hf_file,
                local_dir="./bin",
                local_dir_use_symlinks=False,
            )
            with tarfile.open(tar_path, "r:gz") as tf:
                tf.extractall("./bin")
        else:
            # Direct HTTP fallback (works on Spaces runners)
            import requests
            url = f"https://github.com/{piper_repo}/releases/download/v1.2.0/{piper_asset}"
            r = requests.get(url, timeout=60)
            r.raise_for_status()
            with tarfile.open(fileobj=io.BytesIO(r.content), mode="r:gz") as tf:
                tf.extractall("./bin")

        # Find the extracted "piper" binary
        candidate = os.path.join("./bin", "piper")
        if not os.path.exists(candidate):
            # sometimes archives unpack into a subdir
            for root, _, files in os.walk("./bin"):
                if "piper" in files:
                    candidate = os.path.join(root, "piper")
                    break
        if not os.path.exists(candidate):
            raise RuntimeError("[PIPER] Could not locate extracted 'piper' binary.")
        os.chmod(candidate, 0o755)
        desired_bin = os.path.abspath(candidate)

    os.environ["PIPER_BIN"] = desired_bin
    print(f"[PIPER] BIN = {desired_bin}")

    # --- Ensure voice model ---
    desired_model = _env("PIPER_MODEL", os.path.abspath("models/piper/en_US-amy-medium.onnx"))
    if not os.path.exists(desired_model):
        print("[PIPER] Downloading voice model (en_US-amy-medium.onnx) …")
        local_dir = os.path.dirname(desired_model)
        os.makedirs(local_dir, exist_ok=True)
        # Use the canonical voice repo on HF
        voice_path = huggingface_hub.hf_hub_download(
            repo_id="rhasspy/piper-voices",
            filename="en/en_US-amy-medium.onnx",
            local_dir=local_dir,
            local_dir_use_symlinks=False,
        )
        if os.path.abspath(voice_path) != os.path.abspath(desired_model):
            shutil.copy2(voice_path, desired_model)

    os.environ["PIPER_MODEL"] = desired_model
    print(f"[PIPER] MODEL = {desired_model}")
    return desired_bin, desired_model


# -------------------------
# App entry
# -------------------------
def main():
    # 1) Clean runtime/audio on boot
    audio_dir = ensure_runtime_audio_dir()
    print(f"[BOOT] Cleaning audio dir: {audio_dir}")
    cleanup_old_audio(keep_latest=None)

    # 2) Ensure model + Piper assets
    ensure_model()
    ensure_piper()

    # 3) Launch Gradio
    demo = build_demo()
    demo.launch(share=True)  # Spaces-friendly


if __name__ == "__main__":
    main()