"""Voice profiles for clone-based TTS.

Voices live in ``$VOICEMODE_VOICES_DIR`` (default ``~/.voicemode/voices``).
Each subdirectory is a voice profile. For ``<name>/`` we look for a
``default.wav`` (or the first ``*.wav``) plus a sidecar transcript
``default.txt`` (or matching basename). SuperDirt-style: drop a folder in,
you get a voice. Symlink ``default.wav`` to swap which sample is active
without renaming files.

Each profile maps a voice name to a reference audio file and transcript,
plus model and endpoint routing info.

Voice expression syntax (``voice="<expr>"`` at converse time):

* ``samantha``           — the voice's ``default.wav``
* ``samantha[0]``        — the first ``*.wav`` in the dir (sorted)
* ``samantha[2]``        — the third ``*.wav`` (SuperDirt-style indexing)
* ``samantha/angry.wav`` — an explicit file inside the voice dir
* ``/abs/path.wav``      — absolute path passed straight to the TTS server
* ``./clip.wav``         — path relative to the CWD (also ``../`` and ``~/``);
  expanded to an absolute path before it reaches the server

Remote TTS servers (e.g. mlx-audio on ms2) need the ref_audio path that
exists on *their* filesystem, not ours. Set ``VOICEMODE_REMOTE_VOICES_DIR``
to the path where the voices directory is mirrored on the TTS host; we
rewrite the prefix when sending the request. If unset, the local
absolute path is sent (only useful when the TTS server runs locally).
"""

import logging
import os
import re
from dataclasses import dataclass, replace
from pathlib import Path
from typing import Dict, List, Optional, Tuple

import yaml

logger = logging.getLogger("voicemode")

VOICES_DIR = Path(os.path.expanduser(
    os.environ.get("VOICEMODE_VOICES_DIR", "~/.voicemode/voices")
))

# Path on the remote TTS server where VOICES_DIR is mirrored. When set,
# ref_audio paths sent to the server are rewritten with this prefix so
# the server can find the file on its own filesystem.
REMOTE_VOICES_DIR = os.environ.get("VOICEMODE_REMOTE_VOICES_DIR", "")

# Default mlx-audio endpoint for impressions (Qwen3-TTS).
# Defaults to a local mlx-audio server. Override via env vars when you
# want to point at a different host (e.g. a remote ms2 box on the LAN).
#
# VOICEMODE_CLONE_BASE_URL / VOICEMODE_CLONE_MODEL are honoured for one
# release with a deprecation warning (VM-1174). Removal in 8.8.0.
from voice_mode._env_deprecation import get_env_with_deprecation

DEFAULT_CLONE_BASE_URL = get_env_with_deprecation(
    "VOICEMODE_MLX_AUDIO_BASE_URL",
    "VOICEMODE_CLONE_BASE_URL",
    "http://127.0.0.1:8890/v1",
)
# 1.7B-Base-4bit: ~2× realtime on M-series, clean audio, ~2.2GB on disk.
# Picked as the default from the Apr 2026 quant matrix bench. The
# auto-generated voicemode.env lists alternatives (5-bit, 6-bit, bf16,
# 0.6B-5bit).
DEFAULT_CLONE_MODEL = get_env_with_deprecation(
    "VOICEMODE_IMPRESSIONS_MODEL",
    "VOICEMODE_CLONE_MODEL",
    "mlx-community/Qwen3-TTS-12Hz-1.7B-Base-4bit",
)

# Matches ``name[0]``, ``name[12]``. Captures (name, index).
_INDEX_RE = re.compile(r"^([^/\[\]]+)\[(\d+)\]$")


@dataclass
class VoiceProfile:
    """A voice cloning profile."""
    name: str
    ref_audio: str       # Absolute path to reference audio (server-side)
    ref_text: str        # Transcript of reference audio
    model: str           # TTS model to use
    base_url: str        # TTS endpoint URL
    description: str = ""
    voice_dir: str = ""  # Absolute path to the voice's own directory


_profiles: Dict[str, VoiceProfile] = {}
_loaded = False


def _resolve_default_wav(voice_dir: Path) -> Optional[Path]:
    """Pick the reference WAV inside a voice directory.

    Order of preference:
    1. ``default.wav`` (file or symlink) — the explicit default
    2. The single ``*.wav`` if there's only one — unambiguous

    Directories with multiple WAVs and no ``default.wav`` are treated as
    sample bins, not voices, and skipped. Add a ``default.wav`` symlink
    if you want such a directory to register as a voice.
    """
    default = voice_dir / "default.wav"
    if default.exists():
        return default

    wavs = sorted(voice_dir.glob("*.wav"))
    if not wavs:
        return None
    if len(wavs) == 1:
        return wavs[0]

    logger.debug(
        f"Skipping {voice_dir.name!r}: {len(wavs)} WAVs and no default.wav "
        f"(treat as a sample bin, not a voice; add default.wav symlink to "
        f"register)."
    )
    return None


def _resolve_transcript(wav_path: Path) -> str:
    """Read the matching transcript for a reference WAV.

    Resolution order:

    1. ``<basename>.txt`` sidecar next to the WAV.
    2. ``default.txt`` sidecar in the same directory.
    3. the ``transcript`` field of a ``voice.md`` frontmatter in the same
       directory — the layout ``voicemode clone add`` writes (VM-1439). This
       lazily repairs voice.md-only profiles (those created before clone add
       also wrote a ``default.txt``) so they resolve a non-empty ref_text with
       no migration step.

    Returns empty string if no transcript is found (caller will warn).
    """
    same_name = wav_path.with_suffix(".txt")
    if same_name.exists():
        return same_name.read_text().strip()

    fallback = wav_path.parent / "default.txt"
    if fallback.exists():
        return fallback.read_text().strip()

    return _transcript_from_voice_md(wav_path.parent / "voice.md")


def _extract_frontmatter(text: str) -> Optional[str]:
    """Return the YAML frontmatter delimited by the first two ``---`` fences.

    Returns ``None`` if ``text`` doesn't open with a ``---`` fence or the
    closing fence is missing.
    """
    lines = text.splitlines()
    if not lines or lines[0].strip() != "---":
        return None
    for i in range(1, len(lines)):
        if lines[i].strip() == "---":
            return "\n".join(lines[1:i])
    return None


def _transcript_from_voice_md(voice_md: Path) -> str:
    """Read the ``transcript`` field from a ``voice.md`` frontmatter.

    ``voicemode clone add`` records the reference transcript in the voice.md
    YAML frontmatter (``transcript:``). Reading it here lets the loader
    resolve a non-empty ref_text for cloned voices that have no ``.txt``
    sidecar. Returns empty string when the file is missing, has no
    frontmatter, or has no usable ``transcript`` field.
    """
    if not voice_md.exists():
        return ""
    try:
        front = _extract_frontmatter(voice_md.read_text())
    except OSError:
        return ""
    if front is None:
        return ""
    try:
        data = yaml.safe_load(front)
    except yaml.YAMLError:
        return ""
    if not isinstance(data, dict):
        return ""
    transcript = data.get("transcript")
    if not isinstance(transcript, str):
        return ""
    return transcript.strip()


def _read_description(voice_dir: Path) -> str:
    """Read the optional ``description.txt`` sidecar."""
    desc_path = voice_dir / "description.txt"
    if desc_path.exists():
        return desc_path.read_text().strip()
    return ""


def _derive_group(voice_dir: Path) -> str:
    """Joined relative path from ``VOICES_DIR`` to ``voice_dir.parent``.

    Returns the empty string for a top-level voice dir. For a multi-level
    nested voice (``star-trek/tng/picard``) the full lineage is preserved
    (``star-trek/tng``), not just the immediate parent.
    """
    rel_parent = voice_dir.parent.relative_to(VOICES_DIR)
    s = rel_parent.as_posix()
    return "" if s == "." else s


def _format_description(base_desc: str, group: str) -> str:
    """Append ``(from <group>)`` as a footer on its own line.

    Top-level voices (empty ``group``) keep their description unchanged.
    Voices with no ``description.txt`` get just the suffix.
    """
    if not group:
        return base_desc
    suffix = f"(from {group})"
    if not base_desc:
        return suffix
    return f"{base_desc}\n{suffix}"


def _build_profile(voice_dir: Path, wav: Path) -> VoiceProfile:
    """Construct a VoiceProfile for a directory that qualifies as a voice."""
    transcript = _resolve_transcript(wav)
    if not transcript:
        logger.warning(
            f"Voice {voice_dir.name!r}: no transcript found "
            f"(expected {wav.with_suffix('.txt').name} or default.txt). "
            f"ref_text will be empty."
        )

    base_desc = _read_description(voice_dir)
    group = _derive_group(voice_dir)
    return VoiceProfile(
        name=voice_dir.name,
        ref_audio=_translate_path(wav),
        ref_text=transcript,
        model=DEFAULT_CLONE_MODEL,
        base_url=DEFAULT_CLONE_BASE_URL,
        description=_format_description(base_desc, group),
        voice_dir=str(voice_dir),
    )


def _load_dir_profiles() -> Dict[str, VoiceProfile]:
    """Walk VOICES_DIR recursively and build profiles for each voice dir.

    A directory containing a resolvable WAV (see :func:`_resolve_default_wav`)
    is a voice; otherwise it's a group and we descend into it. Subdirectories
    of a voice dir are NOT walked further — voices and groups are disjoint.

    Voices are keyed by leaf directory name. If two directories anywhere in
    the tree share the same leaf name we treat it as a load-time
    configuration error: log a single ERROR listing every conflicting path
    and drop ALL conflicting candidates from the registry so neither
    resolves silently.
    """
    profiles: Dict[str, VoiceProfile] = {}
    seen: Dict[str, Path] = {}
    conflicts: Dict[str, List[Path]] = {}

    if not VOICES_DIR.exists() or not VOICES_DIR.is_dir():
        logger.debug(f"Voices directory not found at {VOICES_DIR}")
        return profiles

    def walk(dir_path: Path) -> None:
        wav = _resolve_default_wav(dir_path)
        if wav is not None:
            leaf = dir_path.name
            if leaf in seen:
                conflicts.setdefault(leaf, [seen[leaf]]).append(dir_path)
                return
            seen[leaf] = dir_path
            profiles[leaf] = _build_profile(dir_path, wav)
            return  # do NOT descend into a voice dir

        for child in sorted(p for p in dir_path.iterdir() if p.is_dir()):
            walk(child)

    for top in sorted(p for p in VOICES_DIR.iterdir() if p.is_dir()):
        walk(top)

    for leaf, paths in conflicts.items():
        rel_paths = [str(p.relative_to(VOICES_DIR)) for p in paths]
        logger.error(
            f"Voice name collision: leaf {leaf!r} appears at {rel_paths}. "
            f"Dropping ALL candidates — rename to disambiguate."
        )
        profiles.pop(leaf, None)

    if profiles:
        logger.info(
            f"Loaded {len(profiles)} voice profiles from {VOICES_DIR}: "
            f"{list(profiles.keys())}"
        )
    return profiles


def load_profiles() -> Dict[str, VoiceProfile]:
    """Load voice profiles by scanning VOICES_DIR."""
    global _profiles, _loaded
    _profiles = _load_dir_profiles()
    _loaded = True
    return _profiles


def parse_voice_expr(expr: str) -> Tuple[Optional[str], Optional[str]]:
    """Parse a voice expression into ``(voice_name, selector)``.

    Selector is either ``None`` (use default), a string like ``"[N]"``
    (indexed sample), or a relative file path inside the voice dir
    (e.g. ``"angry.wav"``). For a filesystem path the voice_name is
    ``None`` and the selector is the path itself.

    Filesystem paths are recognised when the expression starts with a
    path marker — ``/`` (absolute), ``./`` or ``../`` (relative to the
    CWD), or ``~`` (home). Relative and home forms are expanded and made
    absolute against the CWD so the rest of the pipeline — and the TTS
    server — always sees a concrete path. We use ``os.path.abspath``
    (lexical) rather than ``Path.resolve()`` so a symlinked
    ``default.wav`` is preserved: resolving the symlink would change
    which sidecar transcript (``<basename>.txt``) we pick up.

    A bare ``name/file.wav`` (no leading ``./``) stays the
    profile-selector form, so the path markers don't collide with it.

    Examples::

        parse_voice_expr("samantha")           == ("samantha", None)
        parse_voice_expr("samantha[0]")        == ("samantha", "[0]")
        parse_voice_expr("samantha/angry.wav") == ("samantha", "angry.wav")
        parse_voice_expr("/abs/path.wav")      == (None, "/abs/path.wav")
        parse_voice_expr("./clip.wav")         == (None, "/cwd/clip.wav")
        parse_voice_expr("~/clip.wav")         == (None, "/home/clip.wav")
    """
    if not expr:
        return None, None
    if expr.startswith("/"):
        return None, expr
    # Explicit-relative (./, ../) and home (~) paths → expand to an
    # absolute path and hand off to the absolute-path escape hatch.
    if expr.startswith(("./", "../", "~")):
        return None, os.path.abspath(os.path.expanduser(expr))

    m = _INDEX_RE.match(expr)
    if m:
        return m.group(1), f"[{m.group(2)}]"

    if "/" in expr:
        head, _, tail = expr.partition("/")
        return head, tail

    return expr, None


def _list_samples(voice_dir: Path) -> List[Path]:
    """Sorted list of ``*.wav`` files inside a voice directory."""
    return sorted(voice_dir.glob("*.wav"))


def _translate_path(local_path: Path) -> str:
    """Translate a local voices-dir path into the path the TTS server sees.

    If ``VOICEMODE_REMOTE_VOICES_DIR`` is set and ``local_path`` lives
    under ``VOICES_DIR``, replace the prefix. Otherwise return the local
    absolute path (correct when the TTS server is the local machine).
    """
    abs_local = local_path.resolve() if local_path.exists() else local_path
    if not REMOTE_VOICES_DIR:
        return str(abs_local)

    try:
        rel = abs_local.relative_to(VOICES_DIR.resolve())
    except ValueError:
        # Path isn't under VOICES_DIR — pass through untranslated
        return str(abs_local)

    return str(Path(REMOTE_VOICES_DIR) / rel)


def resolve_voice_expr(expr: str) -> Optional[VoiceProfile]:
    """Resolve a voice expression to a fully-populated ``VoiceProfile``.

    Returns a profile whose ``ref_audio`` is the path the TTS server
    should look up, and whose ``ref_text`` matches the chosen sample.
    Returns ``None`` if the expression doesn't refer to a clone voice.

    For the absolute-path escape hatch (``"/abs/path.wav"``), returns a
    minimal profile with the path passed through and an empty
    ``ref_text`` (caller may not have a transcript for arbitrary files).
    """
    if not _loaded:
        load_profiles()

    name, selector = parse_voice_expr(expr)

    # Absolute-path escape hatch: no profile lookup. If a sidecar
    # transcript sits next to the clip (``<basename>.txt`` or a
    # ``default.txt`` in the same dir) we pick it up so cloning has a
    # ref_text without registering a profile. If there's no sidecar,
    # ref_text stays empty and either a caller-supplied override
    # (VM-1278: converse(ref_text=...)) or the TTS server default applies.
    if name is None and selector and selector.startswith("/"):
        clip_path = Path(selector)
        sidecar_text = _resolve_transcript(clip_path) if clip_path.exists() else ""
        return VoiceProfile(
            name=expr,
            ref_audio=selector,
            ref_text=sidecar_text,
            model=DEFAULT_CLONE_MODEL,
            base_url=DEFAULT_CLONE_BASE_URL,
            description="(absolute path)",
        )

    if not name:
        return None

    profile = _profiles.get(name)
    if profile is None:
        return None

    # Bare name → use the profile as-is (already pointing at default.wav).
    if selector is None:
        return profile

    voice_dir = Path(profile.voice_dir) if profile.voice_dir else VOICES_DIR / name

    # Indexed sample: samantha[0]
    if selector.startswith("[") and selector.endswith("]"):
        try:
            idx = int(selector[1:-1])
        except ValueError:
            logger.error(f"Bad sample index in voice expr {expr!r}")
            return profile
        samples = _list_samples(voice_dir)
        if not samples:
            logger.warning(f"Voice {name!r}: no .wav samples for indexing")
            return profile
        if idx < 0 or idx >= len(samples):
            logger.error(
                f"Sample index {idx} out of range for {name!r} "
                f"({len(samples)} samples available)"
            )
            return profile
        wav = samples[idx]
        return replace(
            profile,
            ref_audio=_translate_path(wav),
            ref_text=_resolve_transcript(wav),
        )

    # Explicit relative file: samantha/angry.wav
    wav = voice_dir / selector
    if not wav.exists():
        logger.warning(
            f"Voice expr {expr!r}: {wav} does not exist locally — "
            f"sending path to server anyway in case it's mirrored."
        )
    return replace(
        profile,
        ref_audio=_translate_path(wav),
        ref_text=_resolve_transcript(wav) if wav.exists() else "",
    )


def get_profile(voice_expr: str) -> Optional[VoiceProfile]:
    """Get a voice profile resolved from a voice expression.

    Equivalent to :func:`resolve_voice_expr` — kept under the original
    name for back-compat with existing callers.
    """
    return resolve_voice_expr(voice_expr)


def is_clone_voice(voice_expr: str) -> bool:
    """Check if a voice expression refers to a clone profile.

    Recognises the selector syntax: ``samantha[0]`` and
    ``samantha/angry.wav`` are both clone voices if ``samantha`` is.
    Absolute paths always count as clone voices.
    """
    if not _loaded:
        load_profiles()
    if not voice_expr:
        return False
    name, selector = parse_voice_expr(voice_expr)
    if name is None and selector and selector.startswith("/"):
        return True
    return name in _profiles


def list_profiles() -> Dict[str, VoiceProfile]:
    """List all available voice profiles."""
    if not _loaded:
        load_profiles()
    return _profiles


def reload_profiles() -> Dict[str, VoiceProfile]:
    """Force a reload of voice profiles (clears the cache)."""
    global _loaded
    _loaded = False
    return load_profiles()
