[project.optional-dependencies]
voice = [
    "PyNaCl>=1.5",
    "discord-ext-voice-recv>=0.5",
    "silero-vad>=5.0",
    "audioop-lts>=0.2; python_version>='3.13'",
]
elevenlabs = ["elevenlabs>=1.0"]
deepgram = ["deepgram-sdk>=3.0"]
openai-voice = ["openai>=1.0"]
local-voice = ["piper-tts>=1.2", "faster-whisper>=1.0"]
all-voice = [
    "discord-cli-agent[voice,elevenlabs,deepgram]",
]
dev = [
    "pytest>=7.0",
    "pytest-asyncio>=0.21",
]

import pytest
from discli.tts import get_tts_provider, TTSProvider, TTSError


def test_get_tts_provider_unknown_raises():
    with pytest.raises(TTSError, match="Unknown TTS provider"):
        get_tts_provider("nonexistent")


def test_tts_provider_protocol_has_required_methods():
    """TTSProvider protocol defines synthesize and close."""
    assert hasattr(TTSProvider, "synthesize")
    assert hasattr(TTSProvider, "close")


def test_get_tts_provider_elevenlabs_missing_key():
    """ElevenLabs provider raises if no API key."""
    import os
    old = os.environ.pop("ELEVENLABS_API_KEY", None)
    try:
        with pytest.raises(TTSError, match="ELEVENLABS_API_KEY"):
            get_tts_provider("elevenlabs")
    finally:
        if old:
            os.environ["ELEVENLABS_API_KEY"] = old

"""TTS provider protocol and implementations."""

from __future__ import annotations

import os
from typing import AsyncIterator, Protocol, runtime_checkable


class TTSError(Exception):
    """Raised when TTS synthesis fails."""


@runtime_checkable
class TTSProvider(Protocol):
    """Protocol for text-to-speech providers."""

    async def synthesize(
        self, text: str, *, voice: str = "default", speed: float = 1.0
    ) -> AsyncIterator[bytes]:
        """Yield PCM audio chunks (16-bit, 48kHz, stereo) from text."""
        ...

    async def close(self) -> None:
        """Release resources."""
        ...


class ElevenLabsTTS:
    """ElevenLabs TTS provider with streaming support."""

    def __init__(self, api_key: str, default_voice: str = "default"):
        try:
            from elevenlabs.client import AsyncElevenLabs
        except ImportError:
            raise TTSError(
                "elevenlabs package not installed. Run: pip install discord-cli-agent[elevenlabs]"
            )
        self._client = AsyncElevenLabs(api_key=api_key)
        self._default_voice = default_voice

    async def synthesize(
        self, text: str, *, voice: str = "default", speed: float = 1.0
    ) -> AsyncIterator[bytes]:
        voice_id = voice if voice != "default" else self._default_voice
        try:
            response = await self._client.text_to_speech.convert_as_stream(
                text=text,
                voice_id=voice_id,
                output_format="pcm_24000",
            )
            async for chunk in response:
                yield chunk
        except Exception as e:
            raise TTSError(f"ElevenLabs synthesis failed: {e}") from e

    async def close(self) -> None:
        await self._client.close()


class OpenAITTS:
    """OpenAI TTS provider."""

    def __init__(self, api_key: str, default_voice: str = "alloy"):
        try:
            from openai import AsyncOpenAI
        except ImportError:
            raise TTSError(
                "openai package not installed. Run: pip install discord-cli-agent[openai-voice]"
            )
        self._client = AsyncOpenAI(api_key=api_key)
        self._default_voice = default_voice

    async def synthesize(
        self, text: str, *, voice: str = "default", speed: float = 1.0
    ) -> AsyncIterator[bytes]:
        voice_name = voice if voice != "default" else self._default_voice
        try:
            response = await self._client.audio.speech.create(
                model="tts-1",
                voice=voice_name,
                input=text,
                speed=speed,
                response_format="pcm",
            )
            async for chunk in response.iter_bytes(chunk_size=4096):
                yield chunk
        except Exception as e:
            raise TTSError(f"OpenAI TTS failed: {e}") from e

    async def close(self) -> None:
        await self._client.close()


_PROVIDERS = {
    "elevenlabs": lambda: ElevenLabsTTS(
        api_key=_require_env("ELEVENLABS_API_KEY"),
    ),
    "openai": lambda: OpenAITTS(
        api_key=_require_env("OPENAI_API_KEY"),
    ),
}


def _require_env(name: str) -> str:
    val = os.environ.get(name)
    if not val:
        raise TTSError(f"{name} environment variable is required")
    return val


def get_tts_provider(name: str) -> TTSProvider:
    """Get a TTS provider by name."""
    factory = _PROVIDERS.get(name)
    if not factory:
        raise TTSError(f"Unknown TTS provider: {name}. Available: {list(_PROVIDERS.keys())}")
    return factory()

import pytest
from discli.stt import get_stt_provider, STTProvider, STTError, TranscriptionResult


def test_get_stt_provider_unknown_raises():
    with pytest.raises(STTError, match="Unknown STT provider"):
        get_stt_provider("nonexistent")


def test_stt_provider_protocol_has_required_methods():
    assert hasattr(STTProvider, "transcribe")
    assert hasattr(STTProvider, "close")


def test_transcription_result_fields():
    r = TranscriptionResult(text="hello", confidence=0.95, is_final=True, user_id="123")
    assert r.text == "hello"
    assert r.confidence == 0.95
    assert r.is_final is True
    assert r.user_id == "123"


def test_get_stt_provider_deepgram_missing_key():
    import os
    old = os.environ.pop("DEEPGRAM_API_KEY", None)
    try:
        with pytest.raises(STTError, match="DEEPGRAM_API_KEY"):
            get_stt_provider("deepgram")
    finally:
        if old:
            os.environ["DEEPGRAM_API_KEY"] = old

"""STT provider protocol and implementations."""

from __future__ import annotations

import os
from dataclasses import dataclass
from typing import AsyncIterator, Protocol, runtime_checkable


class STTError(Exception):
    """Raised when speech-to-text fails."""


@dataclass
class TranscriptionResult:
    """A transcription result from STT."""

    text: str
    confidence: float
    is_final: bool
    user_id: str | None = None


@runtime_checkable
class STTProvider(Protocol):
    """Protocol for speech-to-text providers."""

    async def transcribe(
        self, audio_stream: AsyncIterator[bytes], *, sample_rate: int = 48000
    ) -> AsyncIterator[TranscriptionResult]:
        """Yield transcription results from an audio stream."""
        ...

    async def close(self) -> None:
        """Release resources."""
        ...


class DeepgramSTT:
    """Deepgram STT provider with real-time WebSocket streaming."""

    def __init__(self, api_key: str):
        try:
            from deepgram import DeepgramClient
        except ImportError:
            raise STTError(
                "deepgram-sdk not installed. Run: pip install discord-cli-agent[deepgram]"
            )
        self._client = DeepgramClient(api_key)

    async def transcribe(
        self, audio_stream: AsyncIterator[bytes], *, sample_rate: int = 48000
    ) -> AsyncIterator[TranscriptionResult]:
        from deepgram import LiveTranscriptionEvents, LiveOptions

        options = LiveOptions(
            model="nova-2",
            language="en",
            smart_format=True,
            encoding="linear16",
            sample_rate=sample_rate,
            channels=2,
            interim_results=True,
        )

        connection = self._client.listen.asyncwebsocket.v("1")
        results: list[TranscriptionResult] = []

        async def on_transcript(_, result, **kwargs):
            alt = result.channel.alternatives[0] if result.channel.alternatives else None
            if alt and alt.transcript.strip():
                results.append(
                    TranscriptionResult(
                        text=alt.transcript.strip(),
                        confidence=alt.confidence,
                        is_final=result.is_final,
                    )
                )

        connection.on(LiveTranscriptionEvents.Transcript, on_transcript)
        await connection.start(options)

        try:
            async for chunk in audio_stream:
                await connection.send(chunk)
                while results:
                    yield results.pop(0)
        finally:
            await connection.finish()

    async def close(self) -> None:
        pass


class OpenAISTT:
    """OpenAI Whisper STT provider (batch, not streaming)."""

    def __init__(self, api_key: str):
        try:
            from openai import AsyncOpenAI
        except ImportError:
            raise STTError(
                "openai not installed. Run: pip install discord-cli-agent[openai-voice]"
            )
        self._client = AsyncOpenAI(api_key=api_key)

    async def transcribe(
        self, audio_stream: AsyncIterator[bytes], *, sample_rate: int = 48000
    ) -> AsyncIterator[TranscriptionResult]:
        import io
        import wave

        # Collect all audio (Whisper is batch-only)
        audio_data = b""
        async for chunk in audio_stream:
            audio_data += chunk

        # Wrap in WAV container
        buf = io.BytesIO()
        with wave.open(buf, "wb") as wf:
            wf.setnchannels(2)
            wf.setsampwidth(2)
            wf.setframerate(sample_rate)
            wf.writeframes(audio_data)
        buf.seek(0)
        buf.name = "audio.wav"

        result = await self._client.audio.transcriptions.create(
            model="whisper-1",
            file=buf,
        )
        yield TranscriptionResult(
            text=result.text,
            confidence=1.0,
            is_final=True,
        )

    async def close(self) -> None:
        await self._client.close()


_PROVIDERS = {
    "deepgram": lambda: DeepgramSTT(api_key=_require_env("DEEPGRAM_API_KEY")),
    "openai": lambda: OpenAISTT(api_key=_require_env("OPENAI_API_KEY")),
}


def _require_env(name: str) -> str:
    val = os.environ.get(name)
    if not val:
        raise STTError(f"{name} environment variable is required")
    return val


def get_stt_provider(name: str) -> STTProvider:
    """Get an STT provider by name."""
    factory = _PROVIDERS.get(name)
    if not factory:
        raise STTError(f"Unknown STT provider: {name}. Available: {list(_PROVIDERS.keys())}")
    return factory()

import pytest
from discli.voice_engine import VoiceEngine, VoiceError


def test_voice_engine_instantiates():
    engine = VoiceEngine()
    assert engine.connections == {}


def test_voice_engine_get_connection_not_connected():
    engine = VoiceEngine()
    with pytest.raises(VoiceError, match="Not connected"):
        engine.get_connection("123")


def test_voice_engine_config_defaults():
    engine = VoiceEngine()
    assert engine.config["tts_provider"] == "elevenlabs"
    assert engine.config["stt_provider"] == "deepgram"
    assert engine.config["vad_threshold"] == 0.5
    assert engine.config["silence_duration_ms"] == 800
    assert engine.config["playback_volume"] == 1.0


def test_voice_engine_update_config():
    engine = VoiceEngine()
    engine.update_config({"tts_provider": "openai", "vad_threshold": 0.7})
    assert engine.config["tts_provider"] == "openai"
    assert engine.config["vad_threshold"] == 0.7
    # Unchanged values stay
    assert engine.config["stt_provider"] == "deepgram"

"""Voice engine — manages voice connections, audio playback, and listening."""

from __future__ import annotations

import asyncio
from typing import Any, Callable

import discord

from discli.tts import TTSProvider, get_tts_provider, TTSError
from discli.stt import STTProvider, get_stt_provider, STTError, TranscriptionResult


class VoiceError(Exception):
    """Raised when a voice operation fails."""


DEFAULT_CONFIG = {
    "tts_provider": "elevenlabs",
    "tts_voice": "default",
    "stt_provider": "deepgram",
    "vad_threshold": 0.5,
    "silence_duration_ms": 800,
    "playback_volume": 1.0,
}


class AudioPlayer:
    """Queue-based audio player for a voice connection."""

    def __init__(self, voice_client: discord.VoiceClient, volume: float = 1.0):
        self._vc = voice_client
        self._volume = volume
        self._queue: asyncio.Queue[discord.AudioSource | None] = asyncio.Queue()
        self._current: discord.AudioSource | None = None
        self._task: asyncio.Task | None = None
        self._playing = asyncio.Event()

    def start(self) -> None:
        self._task = asyncio.create_task(self._player_loop())

    async def _player_loop(self) -> None:
        while True:
            source = await self._queue.get()
            if source is None:
                break
            self._current = source
            self._playing.set()
            volume_source = discord.PCMVolumeTransformer(source, volume=self._volume)
            done = asyncio.Event()
            self._vc.play(volume_source, after=lambda e: done.set())
            await done.wait()
            self._current = None
            self._playing.clear()

    async def enqueue(self, source: discord.AudioSource) -> None:
        await self._queue.put(source)

    async def enqueue_file(self, path: str) -> None:
        source = discord.FFmpegPCMAudio(path)
        await self.enqueue(source)

    async def enqueue_url(self, url: str) -> None:
        source = discord.FFmpegPCMAudio(url, before_options="-reconnect 1")
        await self.enqueue(source)

    async def enqueue_pcm(self, pcm_data: bytes, sample_rate: int = 48000) -> None:
        import io
        source = discord.FFmpegPCMAudio(
            io.BytesIO(pcm_data), pipe=True,
        )
        await self.enqueue(source)

    def stop(self) -> None:
        if self._vc.is_playing():
            self._vc.stop()

    def pause(self) -> None:
        if self._vc.is_playing():
            self._vc.pause()

    def resume(self) -> None:
        if self._vc.is_paused():
            self._vc.resume()

    def set_volume(self, volume: float) -> None:
        self._volume = max(0.0, min(2.0, volume))

    @property
    def is_playing(self) -> bool:
        return self._vc.is_playing()

    async def close(self) -> None:
        await self._queue.put(None)
        if self._task:
            await self._task


class AudioListener:
    """Listens to voice channel audio, segments by VAD, feeds to STT."""

    def __init__(
        self,
        voice_client: discord.VoiceClient,
        stt: STTProvider,
        vad_threshold: float = 0.5,
        silence_duration_ms: int = 800,
        on_transcription: Callable[[TranscriptionResult], Any] | None = None,
    ):
        self._vc = voice_client
        self._stt = stt
        self._vad_threshold = vad_threshold
        self._silence_ms = silence_duration_ms
        self._on_transcription = on_transcription
        self._task: asyncio.Task | None = None
        self._running = False
        self._user_buffers: dict[int, bytearray] = {}

    async def start(self) -> None:
        """Start listening to the voice channel."""
        self._running = True

        try:
            from voice_recv import VoiceData
        except ImportError:
            raise VoiceError(
                "discord-ext-voice-recv not installed. "
                "Run: pip install discord-cli-agent[voice]"
            )

        def callback(user, data: VoiceData):
            if user is None:
                return
            uid = user.id
            if uid not in self._user_buffers:
                self._user_buffers[uid] = bytearray()
            self._user_buffers[uid].extend(data.pcm)

        self._vc.listen(callback)
        self._task = asyncio.create_task(self._vad_loop())

    async def _vad_loop(self) -> None:
        """Periodically check buffers, run VAD, and send to STT."""
        try:
            import torch
            model, utils = torch.hub.load(
                repo_or_dir="snakers4/silero-vad", model="silero_vad"
            )
            (get_speech_timestamps, _, _, _, _) = utils
        except ImportError:
            raise VoiceError("silero-vad not installed. Run: pip install discord-cli-agent[voice]")

        while self._running:
            await asyncio.sleep(self._silence_ms / 1000.0)

            for uid, buf in list(self._user_buffers.items()):
                if len(buf) < 9600:  # Less than 100ms of audio at 48kHz stereo 16-bit
                    continue

                audio_bytes = bytes(buf)
                self._user_buffers[uid] = bytearray()

                # Run VAD in thread to avoid blocking
                import numpy as np
                audio_np = np.frombuffer(audio_bytes, dtype=np.int16).astype(np.float32) / 32768.0
                # Convert stereo to mono for VAD
                if len(audio_np) % 2 == 0:
                    audio_np = audio_np.reshape(-1, 2).mean(axis=1)

                import torch
                tensor = torch.from_numpy(audio_np)
                speech_timestamps = get_speech_timestamps(
                    tensor, model, threshold=self._vad_threshold, sampling_rate=16000
                )

                if speech_timestamps:
                    # Speech detected — send to STT
                    async def audio_gen():
                        yield audio_bytes

                    async for result in self._stt.transcribe(audio_gen()):
                        result.user_id = str(uid)
                        if self._on_transcription:
                            self._on_transcription(result)

    async def stop(self) -> None:
        self._running = False
        self._vc.stop_listening()
        if self._task:
            self._task.cancel()
            try:
                await self._task
            except asyncio.CancelledError:
                pass


class VoiceEngine:
    """Manages voice connections across guilds."""

    def __init__(self, config: dict | None = None):
        self.config = {**DEFAULT_CONFIG, **(config or {})}
        self.connections: dict[str, discord.VoiceClient] = {}  # guild_id -> VoiceClient
        self._players: dict[str, AudioPlayer] = {}  # guild_id -> AudioPlayer
        self._listeners: dict[str, AudioListener] = {}  # guild_id -> AudioListener
        self._tts: TTSProvider | None = None
        self._stt: STTProvider | None = None
        self._on_event: Callable[[dict], Any] | None = None

    def set_event_handler(self, handler: Callable[[dict], Any]) -> None:
        self._on_event = handler

    def _emit(self, event: dict) -> None:
        if self._on_event:
            self._on_event(event)

    def update_config(self, updates: dict) -> None:
        self.config.update(updates)

    def get_connection(self, guild_id: str) -> discord.VoiceClient:
        vc = self.connections.get(guild_id)
        if not vc:
            raise VoiceError(f"Not connected to voice in guild {guild_id}")
        return vc

    async def connect(self, channel: discord.VoiceChannel) -> discord.VoiceClient:
        guild_id = str(channel.guild.id)
        if guild_id in self.connections:
            vc = self.connections[guild_id]
            if vc.is_connected():
                await vc.move_to(channel)
                self._emit({"event": "voice_connected", "guild_id": guild_id, "channel_id": str(channel.id)})
                return vc

        vc = await channel.connect()
        self.connections[guild_id] = vc

        player = AudioPlayer(vc, volume=self.config["playback_volume"])
        player.start()
        self._players[guild_id] = player

        self._emit({"event": "voice_connected", "guild_id": guild_id, "channel_id": str(channel.id)})
        return vc

    async def disconnect(self, guild_id: str) -> None:
        vc = self.connections.pop(guild_id, None)
        if not vc:
            raise VoiceError(f"Not connected to voice in guild {guild_id}")

        if guild_id in self._listeners:
            await self._listeners[guild_id].stop()
            del self._listeners[guild_id]

        if guild_id in self._players:
            await self._players[guild_id].close()
            del self._players[guild_id]

        await vc.disconnect()
        self._emit({"event": "voice_disconnected", "guild_id": guild_id})

    async def move(self, channel: discord.VoiceChannel) -> None:
        guild_id = str(channel.guild.id)
        vc = self.get_connection(guild_id)
        await vc.move_to(channel)
        self._emit({"event": "voice_connected", "guild_id": guild_id, "channel_id": str(channel.id)})

    async def speak(self, guild_id: str, text: str, *, voice: str = "default", speed: float = 1.0) -> None:
        self.get_connection(guild_id)  # Verify connected
        player = self._players[guild_id]

        if not self._tts:
            self._tts = get_tts_provider(self.config["tts_provider"])

        pcm_chunks = []
        async for chunk in self._tts.synthesize(text, voice=voice, speed=speed):
            pcm_chunks.append(chunk)

        pcm_data = b"".join(pcm_chunks)
        self._emit({"event": "voice_playback_started", "guild_id": guild_id, "source": "tts"})
        await player.enqueue_pcm(pcm_data)

    async def play(self, guild_id: str, source: str) -> None:
        self.get_connection(guild_id)
        player = self._players[guild_id]
        self._emit({"event": "voice_playback_started", "guild_id": guild_id, "source": source})

        if source.startswith(("http://", "https://")):
            await player.enqueue_url(source)
        else:
            await player.enqueue_file(source)

    def stop(self, guild_id: str) -> None:
        player = self._players.get(guild_id)
        if player:
            player.stop()

    def pause(self, guild_id: str) -> None:
        player = self._players.get(guild_id)
        if player:
            player.pause()

    def resume(self, guild_id: str) -> None:
        player = self._players.get(guild_id)
        if player:
            player.resume()

    async def listen_start(self, guild_id: str) -> None:
        vc = self.get_connection(guild_id)

        if not self._stt:
            self._stt = get_stt_provider(self.config["stt_provider"])

        def on_transcription(result: TranscriptionResult):
            self._emit({
                "event": "voice_transcription",
                "guild_id": guild_id,
                "user_id": result.user_id,
                "text": result.text,
                "confidence": result.confidence,
                "is_final": result.is_final,
            })

        listener = AudioListener(
            vc, self._stt,
            vad_threshold=self.config["vad_threshold"],
            silence_duration_ms=self.config["silence_duration_ms"],
            on_transcription=on_transcription,
        )
        await listener.start()
        self._listeners[guild_id] = listener

    async def listen_stop(self, guild_id: str) -> None:
        listener = self._listeners.pop(guild_id, None)
        if listener:
            await listener.stop()

    def status(self) -> list[dict]:
        result = []
        for gid, vc in self.connections.items():
            result.append({
                "guild_id": gid,
                "channel_id": str(vc.channel.id) if vc.channel else None,
                "connected": vc.is_connected(),
                "playing": self._players.get(gid, None) is not None and self._players[gid].is_playing,
                "listening": gid in self._listeners,
            })
        return result

    async def close(self) -> None:
        for gid in list(self.connections.keys()):
            try:
                await self.disconnect(gid)
            except Exception:
                pass
        if self._tts:
            await self._tts.close()
        if self._stt:
            await self._stt.close()

import pytest
from discli.interact_engine import (
    InteractEngine,
    InteractError,
    WorkflowDefinition,
    WorkflowStep,
    DashboardDefinition,
    DashboardPage,
)


def test_interact_engine_instantiates():
    engine = InteractEngine()
    assert engine.workflows == {}
    assert engine.dashboards == {}


def test_workflow_definition():
    steps = [
        WorkflowStep(
            step_id="intro",
            step_type="message",
            content="Welcome!",
            components=[{"type": "button", "label": "Next", "custom_id": "next"}],
        ),
        WorkflowStep(
            step_id="name",
            step_type="modal",
            content="Enter your name",
            fields=[{"label": "Name", "style": "short", "required": True}],
        ),
    ]
    wf = WorkflowDefinition(workflow_id="onboard", steps=steps)
    assert wf.workflow_id == "onboard"
    assert len(wf.steps) == 2
    assert wf.steps[0].step_type == "message"


def test_dashboard_definition():
    pages = [
        DashboardPage(
            embed={"title": "Page 1", "description": "First page"},
        ),
        DashboardPage(
            embed={"title": "Page 2", "description": "Second page"},
        ),
    ]
    dash = DashboardDefinition(dashboard_id="stats", pages=pages, refresh_interval=60)
    assert dash.dashboard_id == "stats"
    assert len(dash.pages) == 2
    assert dash.refresh_interval == 60


def test_interaction_router_prefix_routing():
    engine = InteractEngine()
    assert engine.route_custom_id("modal:signup") == "modal"
    assert engine.route_custom_id("wf:step1:next") == "workflow"
    assert engine.route_custom_id("dash:stats:page") == "dashboard"
    assert engine.route_custom_id("unknown:thing") is None

"""Interactive engine — modals, workflows, and dashboards."""

from __future__ import annotations

import asyncio
import json
import uuid
from dataclasses import dataclass, field
from typing import Any, Callable

import discord


class InteractError(Exception):
    """Raised when an interactive operation fails."""


# ── Data Models ──────────────────────────────────────────────────────


@dataclass
class WorkflowStep:
    step_id: str
    step_type: str  # "message", "select", "modal", "confirm"
    content: str = ""
    components: list[dict] = field(default_factory=list)
    fields: list[dict] = field(default_factory=list)  # For modal steps
    next: dict[str, str] | str | None = None  # Conditional or direct next step
    timeout: int = 300  # seconds


@dataclass
class WorkflowDefinition:
    workflow_id: str
    steps: list[WorkflowStep]

    def get_step(self, step_id: str) -> WorkflowStep | None:
        for s in self.steps:
            if s.step_id == step_id:
                return s
        return None


@dataclass
class WorkflowState:
    workflow_id: str
    user_id: str
    channel_id: str
    current_step: str
    collected_data: dict = field(default_factory=dict)
    message_id: str | None = None


@dataclass
class DashboardPage:
    embed: dict = field(default_factory=dict)
    components: list[dict] = field(default_factory=list)


@dataclass
class DashboardDefinition:
    dashboard_id: str
    pages: list[DashboardPage]
    refresh_interval: int = 0  # 0 = no auto-refresh
    components: list[dict] = field(default_factory=list)  # Global components (e.g., pagination)


@dataclass
class DashboardState:
    dashboard_id: str
    channel_id: str
    message_id: str | None = None
    current_page: int = 0
    data: dict = field(default_factory=dict)  # Dynamic data for updates


# ── Interaction Router ───────────────────────────────────────────────

_PREFIX_MAP = {
    "modal:": "modal",
    "wf:": "workflow",
    "dash:": "dashboard",
    "voice:": "voice",
}


# ── Engine ───────────────────────────────────────────────────────────


class InteractEngine:
    """Manages interactive Discord components: modals, workflows, dashboards."""

    def __init__(self):
        self.workflows: dict[str, WorkflowState] = {}  # (user_id:workflow_id) -> state
        self.dashboards: dict[str, DashboardState] = {}  # dashboard_id -> state
        self._workflow_defs: dict[str, WorkflowDefinition] = {}
        self._dashboard_defs: dict[str, DashboardDefinition] = {}
        self._timeout_tasks: dict[str, asyncio.Task] = {}
        self._refresh_tasks: dict[str, asyncio.Task] = {}
        self._on_event: Callable[[dict], Any] | None = None

    def set_event_handler(self, handler: Callable[[dict], Any]) -> None:
        self._on_event = handler

    def _emit(self, event: dict) -> None:
        if self._on_event:
            self._on_event(event)

    @staticmethod
    def route_custom_id(custom_id: str) -> str | None:
        """Route a custom_id to its handler type based on prefix."""
        for prefix, handler_type in _PREFIX_MAP.items():
            if custom_id.startswith(prefix):
                return handler_type
        return None

    # ── Modals ───────────────────────────────────────────────────────

    @staticmethod
    def build_modal(
        title: str,
        custom_id: str,
        fields: list[dict],
    ) -> discord.ui.Modal:
        """Build a Discord modal from a field spec.

        Each field dict: {label, style ("short"|"paragraph"), required, placeholder, default, max_length}
        """
        modal = discord.ui.Modal(title=title, custom_id=f"modal:{custom_id}")

        for f in fields:
            style = discord.TextStyle.short if f.get("style") == "short" else discord.TextStyle.paragraph
            item = discord.ui.TextInput(
                label=f["label"],
                style=style,
                required=f.get("required", False),
                placeholder=f.get("placeholder", ""),
                default=f.get("default", ""),
                max_length=f.get("max_length", 4000),
                custom_id=f.get("custom_id", f["label"].lower().replace(" ", "_")),
            )
            modal.add_item(item)

        return modal

    # ── Workflows ────────────────────────────────────────────────────

    async def workflow_start(
        self,
        channel: discord.TextChannel,
        user_id: str,
        definition: WorkflowDefinition,
    ) -> str:
        """Start a workflow for a user. Returns workflow key."""
        key = f"{user_id}:{definition.workflow_id}"
        self._workflow_defs[definition.workflow_id] = definition

        state = WorkflowState(
            workflow_id=definition.workflow_id,
            user_id=user_id,
            channel_id=str(channel.id),
            current_step=definition.steps[0].step_id,
        )
        self.workflows[key] = state

        await self._send_workflow_step(channel, state, definition.steps[0])
        self._start_timeout(key, definition.steps[0].timeout)

        return key

    async def _send_workflow_step(
        self,
        channel: discord.TextChannel,
        state: WorkflowState,
        step: WorkflowStep,
    ) -> None:
        """Send the current workflow step to the channel."""
        if step.step_type in ("message", "confirm", "select"):
            view = discord.ui.View(timeout=step.timeout)

            if step.step_type == "confirm":
                # Add Yes/No buttons
                yes_btn = discord.ui.Button(
                    label="Yes", style=discord.ButtonStyle.success,
                    custom_id=f"wf:{state.workflow_id}:{step.step_id}:yes",
                )
                no_btn = discord.ui.Button(
                    label="No", style=discord.ButtonStyle.danger,
                    custom_id=f"wf:{state.workflow_id}:{step.step_id}:no",
                )
                view.add_item(yes_btn)
                view.add_item(no_btn)
            elif step.step_type == "select":
                select = discord.ui.Select(
                    custom_id=f"wf:{state.workflow_id}:{step.step_id}:select",
                    options=[
                        discord.SelectOption(label=c["label"], value=c.get("value", c["label"]))
                        for c in step.components
                    ],
                )
                view.add_item(select)
            else:
                # Generic buttons from components
                for comp in step.components:
                    btn = discord.ui.Button(
                        label=comp["label"],
                        style=discord.ButtonStyle.primary,
                        custom_id=f"wf:{state.workflow_id}:{step.step_id}:{comp['custom_id']}",
                    )
                    view.add_item(btn)

            msg = await channel.send(content=step.content, view=view)
            state.message_id = str(msg.id)

        # Modal steps are triggered by a button in a prior step — handled in handle_workflow_interaction

    def _start_timeout(self, key: str, timeout: int) -> None:
        if key in self._timeout_tasks:
            self._timeout_tasks[key].cancel()

        async def _timeout():
            await asyncio.sleep(timeout)
            state = self.workflows.pop(key, None)
            if state:
                self._emit({
                    "event": "workflow_timeout",
                    "workflow_id": state.workflow_id,
                    "user_id": state.user_id,
                    "step_id": state.current_step,
                })

        self._timeout_tasks[key] = asyncio.create_task(_timeout())

    async def handle_workflow_interaction(
        self,
        interaction: discord.Interaction,
        custom_id: str,
    ) -> None:
        """Handle a workflow-related interaction."""
        # Parse: wf:{workflow_id}:{step_id}:{action}
        parts = custom_id.split(":")
        if len(parts) < 4:
            return

        _, wf_id, step_id, action = parts[0], parts[1], parts[2], parts[3]
        user_id = str(interaction.user.id)
        key = f"{user_id}:{wf_id}"

        state = self.workflows.get(key)
        if not state or state.current_step != step_id:
            await interaction.response.send_message("This workflow has expired.", ephemeral=True)
            return

        wf_def = self._workflow_defs.get(wf_id)
        if not wf_def:
            return

        step = wf_def.get_step(step_id)
        if not step:
            return

        # Record input
        state.collected_data[step_id] = action

        self._emit({
            "event": "workflow_step_completed",
            "workflow_id": wf_id,
            "step_id": step_id,
            "user_id": user_id,
            "user_input": action,
        })

        # Determine next step
        next_step_id = None
        if isinstance(step.next, dict):
            next_step_id = step.next.get(action)
        elif isinstance(step.next, str):
            next_step_id = step.next

        if next_step_id:
            next_step = wf_def.get_step(next_step_id)
            if next_step:
                state.current_step = next_step_id
                channel = interaction.channel
                if next_step.step_type == "modal":
                    modal = self.build_modal(
                        title=next_step.content,
                        custom_id=f"wf_modal:{wf_id}:{next_step_id}",
                        fields=next_step.fields,
                    )
                    await interaction.response.send_modal(modal)
                else:
                    await interaction.response.defer()
                    await self._send_workflow_step(channel, state, next_step)
                self._start_timeout(key, next_step.timeout)
                return

        # No next step — workflow complete
        self.workflows.pop(key, None)
        if key in self._timeout_tasks:
            self._timeout_tasks[key].cancel()
            del self._timeout_tasks[key]

        await interaction.response.send_message("Workflow complete!", ephemeral=True)
        self._emit({
            "event": "workflow_finished",
            "workflow_id": wf_id,
            "user_id": user_id,
            "collected_data": state.collected_data,
        })

    def workflow_cancel(self, user_id: str, workflow_id: str) -> bool:
        key = f"{user_id}:{workflow_id}"
        state = self.workflows.pop(key, None)
        if key in self._timeout_tasks:
            self._timeout_tasks[key].cancel()
            del self._timeout_tasks[key]
        return state is not None

    # ── Dashboards ───────────────────────────────────────────────────

    async def dashboard_create(
        self,
        channel: discord.TextChannel,
        definition: DashboardDefinition,
    ) -> str:
        """Create a persistent dashboard message. Returns dashboard_id."""
        self._dashboard_defs[definition.dashboard_id] = definition

        state = DashboardState(
            dashboard_id=definition.dashboard_id,
            channel_id=str(channel.id),
        )

        msg = await self._render_dashboard(channel, definition, state)
        state.message_id = str(msg.id)
        self.dashboards[definition.dashboard_id] = state

        if definition.refresh_interval > 0:
            self._start_refresh(definition.dashboard_id, definition.refresh_interval, channel)

        return definition.dashboard_id

    async def _render_dashboard(
        self,
        channel: discord.TextChannel,
        definition: DashboardDefinition,
        state: DashboardState,
    ) -> discord.Message:
        """Render the current dashboard page."""
        page = definition.pages[state.current_page]

        embed = discord.Embed(
            title=page.embed.get("title", ""),
            description=page.embed.get("description", ""),
        )
        if "color" in page.embed:
            embed.color = discord.Color(int(str(page.embed["color"]).lstrip("#"), 16))
        for f in page.embed.get("fields", []):
            embed.add_field(name=f["name"], value=f["value"], inline=f.get("inline", True))

        view = discord.ui.View(timeout=None)

        # Pagination buttons
        if len(definition.pages) > 1:
            prev_btn = discord.ui.Button(
                label="Previous", style=discord.ButtonStyle.secondary,
                custom_id=f"dash:{definition.dashboard_id}:prev",
                disabled=state.current_page == 0,
            )
            next_btn = discord.ui.Button(
                label="Next", style=discord.ButtonStyle.secondary,
                custom_id=f"dash:{definition.dashboard_id}:next",
                disabled=state.current_page >= len(definition.pages) - 1,
            )
            page_label = discord.ui.Button(
                label=f"{state.current_page + 1}/{len(definition.pages)}",
                style=discord.ButtonStyle.secondary,
                disabled=True,
                custom_id=f"dash:{definition.dashboard_id}:page_label",
            )
            view.add_item(prev_btn)
            view.add_item(page_label)
            view.add_item(next_btn)

        # Custom components from page
        for comp in page.components:
            if comp.get("type") == "button":
                btn = discord.ui.Button(
                    label=comp["label"],
                    style=getattr(discord.ButtonStyle, comp.get("style", "primary")),
                    custom_id=f"dash:{definition.dashboard_id}:{comp['custom_id']}",
                )
                view.add_item(btn)

        if state.message_id:
            try:
                msg = await channel.fetch_message(int(state.message_id))
                await msg.edit(embed=embed, view=view)
                return msg
            except discord.NotFound:
                pass

        return await channel.send(embed=embed, view=view)

    def _start_refresh(self, dashboard_id: str, interval: int, channel: discord.TextChannel) -> None:
        async def _refresh_loop():
            while dashboard_id in self.dashboards:
                await asyncio.sleep(interval)
                state = self.dashboards.get(dashboard_id)
                definition = self._dashboard_defs.get(dashboard_id)
                if state and definition:
                    await self._render_dashboard(channel, definition, state)

        self._refresh_tasks[dashboard_id] = asyncio.create_task(_refresh_loop())

    async def handle_dashboard_interaction(
        self,
        interaction: discord.Interaction,
        custom_id: str,
    ) -> None:
        """Handle a dashboard-related interaction."""
        parts = custom_id.split(":")
        if len(parts) < 3:
            return

        _, dash_id, action = parts[0], parts[1], parts[2]

        state = self.dashboards.get(dash_id)
        definition = self._dashboard_defs.get(dash_id)
        if not state or not definition:
            await interaction.response.send_message("Dashboard not found.", ephemeral=True)
            return

        if action == "prev" and state.current_page > 0:
            state.current_page -= 1
        elif action == "next" and state.current_page < len(definition.pages) - 1:
            state.current_page += 1
        else:
            # Custom component interaction
            self._emit({
                "event": "dashboard_interaction",
                "dashboard_id": dash_id,
                "user_id": str(interaction.user.id),
                "component_id": action,
            })
            await interaction.response.defer()
            return

        await interaction.response.defer()
        await self._render_dashboard(interaction.channel, definition, state)

    async def dashboard_update(self, dashboard_id: str, updates: dict, channel: discord.TextChannel) -> None:
        """Update a dashboard's data and re-render."""
        state = self.dashboards.get(dashboard_id)
        definition = self._dashboard_defs.get(dashboard_id)
        if not state or not definition:
            raise InteractError(f"Dashboard {dashboard_id} not found")

        state.data.update(updates)

        # Update page content from updates if provided
        if "pages" in updates:
            for i, page_update in enumerate(updates["pages"]):
                if i < len(definition.pages):
                    if "embed" in page_update:
                        definition.pages[i].embed.update(page_update["embed"])

        await self._render_dashboard(channel, definition, state)

    async def dashboard_delete(self, dashboard_id: str, channel: discord.TextChannel) -> None:
        """Delete a dashboard."""
        state = self.dashboards.pop(dashboard_id, None)
        self._dashboard_defs.pop(dashboard_id, None)

        if dashboard_id in self._refresh_tasks:
            self._refresh_tasks[dashboard_id].cancel()
            del self._refresh_tasks[dashboard_id]

        if state and state.message_id:
            try:
                msg = await channel.fetch_message(int(state.message_id))
                await msg.delete()
            except discord.NotFound:
                pass

    async def close(self) -> None:
        for task in self._timeout_tasks.values():
            task.cancel()
        for task in self._refresh_tasks.values():
            task.cancel()
        self._timeout_tasks.clear()
        self._refresh_tasks.clear()
        self.workflows.clear()
        self.dashboards.clear()