"""Piper voice library — browse + install voices from the HuggingFace catalog.

Piper voices live in https://huggingface.co/rhasspy/piper-voices, with a
machine-readable catalog at:

    https://huggingface.co/rhasspy/piper-voices/raw/main/voices.json

That catalog enumerates every available voice (~250+ across many languages)
with file sizes and download paths. We cache it in memory with a TTL so
the Settings UI's "browse voices" panel doesn't hit HuggingFace every time
the page opens, but a manual "refresh" button can force a re-fetch.

Downloads land in /data/voices so they persist across container restarts.
Bundled voices in /opt/piper-voices are read-only and cannot be deleted
via the admin UI.
"""
from __future__ import annotations

import asyncio
import json
import logging
import re
import shutil
import time
import urllib.request
from pathlib import Path

logger = logging.getLogger(__name__)

CATALOG_URL = (
    "https://huggingface.co/rhasspy/piper-voices/raw/main/voices.json"
)
VOICE_BASE_URL = (
    "https://huggingface.co/rhasspy/piper-voices/resolve/main"
)
_USER_VOICES_DIR = Path("/data/voices")
_BUNDLED_VOICES_DIR = Path("/opt/piper-voices")

# Voice IDs are like `en_US-amy-medium`, `fr_FR-siwis-low` — strict
# allowlist to prevent path traversal in download/delete handlers.
_VOICE_ID_RE = re.compile(r"^[a-zA-Z]{2,3}(?:_[a-zA-Z]{2,4})?-[a-zA-Z0-9_+]+-(?:x_low|low|medium|high)$")

# In-memory cache. Catalog rarely changes (kokoro-style "model update"
# checks would be overkill — admin can force-refresh from the UI).
_CATALOG_CACHE: dict | None = None
_CATALOG_FETCHED_AT: float = 0.0
_CATALOG_TTL_SECS = 24 * 3600  # 24h — long, but the manual refresh button overrides
_CATALOG_LOCK = asyncio.Lock()


def _is_valid_voice_id(voice_id: str) -> bool:
    return bool(_VOICE_ID_RE.match(voice_id))


async def fetch_catalog(force_refresh: bool = False) -> dict:
    """Return the piper-voices catalog dict, fetching from HF if stale.

    Cached in memory across requests. force_refresh=True bypasses the TTL.
    Raises on network error so the UI can surface the failure.
    """
    global _CATALOG_CACHE, _CATALOG_FETCHED_AT
    async with _CATALOG_LOCK:
        now = time.monotonic()
        fresh_enough = (
            _CATALOG_CACHE is not None
            and (now - _CATALOG_FETCHED_AT) < _CATALOG_TTL_SECS
        )
        if fresh_enough and not force_refresh:
            return _CATALOG_CACHE  # type: ignore[return-value]

        loop = asyncio.get_running_loop()

        def _do_fetch() -> dict:
            with urllib.request.urlopen(CATALOG_URL, timeout=30) as resp:
                raw = resp.read().decode("utf-8")
            return json.loads(raw)

        try:
            catalog = await loop.run_in_executor(None, _do_fetch)
        except Exception:
            logger.exception("Failed to fetch piper voices catalog from %s", CATALOG_URL)
            raise

        _CATALOG_CACHE = catalog
        _CATALOG_FETCHED_AT = now
        logger.info("Piper voices catalog refreshed: %d entries", len(catalog))
        return catalog


def _is_installed(voice_id: str) -> tuple[bool, str | None]:
    """Return (installed, source) where source is 'bundled' / 'user' / None."""
    if (_BUNDLED_VOICES_DIR / f"{voice_id}.onnx").is_file():
        return True, "bundled"
    if (_USER_VOICES_DIR / f"{voice_id}.onnx").is_file():
        return True, "user"
    return False, None


def shape_catalog_for_ui(catalog: dict) -> list[dict]:
    """Project the raw HF catalog into a list of UI-friendly voice cards.

    The HF catalog uses voice IDs as keys with nested file-size metadata.
    The UI just needs id, language, dataset, quality, size, install state.
    Sorted by language then dataset for stable display.
    """
    out: list[dict] = []
    for voice_id, info in catalog.items():
        if not isinstance(info, dict):
            continue
        installed, source = _is_installed(voice_id)
        lang_info = info.get("language") or {}
        files = info.get("files") or {}
        # Sum the file sizes for the .onnx + .onnx.json pair so the UI
        # can show "≈ 62 MB" before the user clicks Download.
        total_bytes = 0
        for f_info in files.values():
            if isinstance(f_info, dict):
                total_bytes += int(f_info.get("size_bytes", 0) or 0)
        out.append({
            "id": voice_id,
            "name": info.get("name") or voice_id.split("-")[1],
            "language_code": lang_info.get("code") or voice_id.split("-")[0],
            "language_name": lang_info.get("name_native")
                or lang_info.get("name_english")
                or lang_info.get("code", ""),
            "country": lang_info.get("country_english") or lang_info.get("region", ""),
            "quality": info.get("quality") or voice_id.split("-")[-1],
            "num_speakers": int(info.get("num_speakers") or 1),
            "size_bytes": total_bytes,
            "installed": installed,
            "installed_source": source,
        })
    out.sort(key=lambda v: (v["language_code"], v["name"], v["quality"]))
    return out


def _resolve_file_urls(catalog: dict, voice_id: str) -> dict[str, str]:
    """Return absolute URLs for the .onnx + .onnx.json files of `voice_id`.

    Walks the catalog entry's `files` dict; that's the authoritative path
    list (the layout in the repo can drift, e.g. some voices live under
    different sub-paths). Returns {extension: url}.
    """
    info = catalog.get(voice_id)
    if not isinstance(info, dict):
        raise KeyError(f"Voice {voice_id!r} not in catalog")
    files = info.get("files") or {}
    urls: dict[str, str] = {}
    for rel_path in files.keys():
        if rel_path.endswith(".onnx"):
            urls["onnx"] = f"{VOICE_BASE_URL}/{rel_path}"
        elif rel_path.endswith(".onnx.json"):
            urls["onnx.json"] = f"{VOICE_BASE_URL}/{rel_path}"
    if "onnx" not in urls or "onnx.json" not in urls:
        raise ValueError(
            f"Catalog entry for {voice_id!r} is missing .onnx or .onnx.json file"
        )
    return urls


async def install_voice(voice_id: str) -> dict:
    """Download a voice's .onnx + .onnx.json into /data/voices.

    Idempotent — if already installed under /data, returns success without
    re-downloading. Bundled voices in /opt are not touched; the same voice
    can also live under /data if the admin wants to override it.
    Returns {"id", "size_bytes", "skipped"}.
    """
    if not _is_valid_voice_id(voice_id):
        raise ValueError(f"Invalid voice id: {voice_id!r}")

    _USER_VOICES_DIR.mkdir(parents=True, exist_ok=True)
    onnx_path = _USER_VOICES_DIR / f"{voice_id}.onnx"
    sidecar_path = _USER_VOICES_DIR / f"{voice_id}.onnx.json"

    if onnx_path.is_file() and sidecar_path.is_file():
        return {
            "id": voice_id,
            "size_bytes": onnx_path.stat().st_size + sidecar_path.stat().st_size,
            "skipped": True,
        }

    catalog = await fetch_catalog()
    urls = _resolve_file_urls(catalog, voice_id)

    loop = asyncio.get_running_loop()

    def _download() -> int:
        total = 0
        # Use atomic write — download to .tmp then rename. Means a failed
        # partial download doesn't leave a corrupt model in place that
        # would later cause PiperVoice.load() to throw.
        for ext, url in urls.items():
            target = _USER_VOICES_DIR / f"{voice_id}.{ext}"
            tmp = target.with_suffix(target.suffix + ".tmp")
            logger.info("Downloading piper voice file %s → %s", url, target)
            with urllib.request.urlopen(url, timeout=120) as resp:
                with open(tmp, "wb") as fh:
                    shutil.copyfileobj(resp, fh, length=1024 * 1024)
            tmp.replace(target)
            total += target.stat().st_size
        return total

    try:
        size = await loop.run_in_executor(None, _download)
    except Exception:
        # Clean up partials.
        for ext in ("onnx", "onnx.json"):
            for p in (
                _USER_VOICES_DIR / f"{voice_id}.{ext}",
                _USER_VOICES_DIR / f"{voice_id}.{ext}.tmp",
            ):
                try:
                    if p.exists():
                        p.unlink()
                except Exception:
                    pass
        logger.exception("Voice install failed for %s — cleaned up partials", voice_id)
        raise

    logger.info("Installed piper voice %s (%d bytes)", voice_id, size)
    return {"id": voice_id, "size_bytes": size, "skipped": False}


async def uninstall_voice(voice_id: str) -> dict:
    """Remove a /data voice. Bundled voices in /opt cannot be removed via API.

    Returns {"id", "removed": bool}.
    """
    if not _is_valid_voice_id(voice_id):
        raise ValueError(f"Invalid voice id: {voice_id!r}")

    # Refuse to delete a bundled voice. If a /data override exists alongside
    # a bundled one with the same id, only the /data copy is removed.
    if not (_USER_VOICES_DIR / f"{voice_id}.onnx").is_file() and not (
        _USER_VOICES_DIR / f"{voice_id}.onnx.json"
    ).is_file():
        if (_BUNDLED_VOICES_DIR / f"{voice_id}.onnx").is_file():
            raise PermissionError(
                f"Voice {voice_id!r} is bundled with the image — cannot be removed"
            )
        return {"id": voice_id, "removed": False}

    removed_any = False
    for ext in ("onnx", "onnx.json"):
        p = _USER_VOICES_DIR / f"{voice_id}.{ext}"
        if p.is_file():
            p.unlink()
            removed_any = True
    logger.info("Uninstalled piper voice %s (files removed: %s)", voice_id, removed_any)
    return {"id": voice_id, "removed": removed_any}