from __future__ import annotations

import asyncio
import math
import logging
import os
import time
import audioop
from dataclasses import dataclass
from typing import Any, Callable, Mapping, Optional
from fastapi import WebSocket, WebSocketDisconnect

from app.utils.audio_backlog import (
    frames_from_ms,
    rms_threshold_from_dbfs,
    shed_queue_backlog,
)
from app.utils.logging import log_flow_event

logger = logging.getLogger("background")


def _out_env_int(name: str, default: int) -> int:
    try:
        return int(os.getenv(name, str(default)))
    except (TypeError, ValueError):
        return default


def _out_env_float(name: str, default: float) -> float:
    try:
        return float(os.getenv(name, str(default)))
    except (TypeError, ValueError):
        return default


def _out_env_bool(name: str, default: bool) -> bool:
    raw = os.getenv(name)
    if raw is None:
        return default
    return raw.strip().lower() in {"1", "true", "yes", "on"}


@dataclass
class AudioActivity:
    last_agent_out: float  # "última fala com energia" do agente (idle/mock-stop)


@dataclass
class BridgeOutputStats:
    total_ws_frames: int = 0
    agent_audio_frames: int = 0
    silence_frames: int = 0
    agent_audio_bursts: int = 0
    first_agent_audio_sent: bool = False
    last_agent_audio_sent_at: float = 0.0


@dataclass(frozen=True)
class AudioOutputBacklogConfig:
    """Instrumentacao e shed do caminho de saida (agente -> cliente).

    O backlog da agent_q e, em boa parte, fala do agente ja gerada esperando ser
    tocada em tempo real -- por isso o shed de saida e SEMPRE guiado por energia
    (so silencio sai; nunca corta voz). O descarte cego da entrada nao se aplica.
    """

    metrics_enabled: bool = True
    latency_alert_ms: int = 1000
    latency_log_interval_s: float = 15.0
    pacing_debt_threshold_ms: int = 200
    shed_enabled: bool = True
    shed_threshold_ms: int = 400
    shed_keep_ms: int = 200
    shed_check_interval_frames: int = 5
    silence_dbfs: float = -60.0

    def shed_threshold_frames(self, frame_ms: int) -> int:
        return frames_from_ms(self.shed_threshold_ms, frame_ms)

    def shed_keep_frames(self, frame_ms: int) -> int:
        return min(
            self.shed_threshold_frames(frame_ms),
            frames_from_ms(self.shed_keep_ms, frame_ms),
        )

    @property
    def silence_rms_threshold(self) -> int:
        return rms_threshold_from_dbfs(self.silence_dbfs)


def audio_output_backlog_config_from_env() -> AudioOutputBacklogConfig:
    return AudioOutputBacklogConfig(
        metrics_enabled=_out_env_bool("AUDIO_OUT_LATENCY_METRICS_ENABLED", True),
        latency_alert_ms=max(0, _out_env_int("AUDIO_OUT_LATENCY_ALERT_MS", 1000)),
        latency_log_interval_s=max(
            0.1, _out_env_float("AUDIO_OUT_LATENCY_LOG_INTERVAL_S", 15.0)
        ),
        pacing_debt_threshold_ms=max(
            0, _out_env_int("AUDIO_OUT_PACING_DEBT_THRESHOLD_MS", 200)
        ),
        shed_enabled=_out_env_bool("AUDIO_OUT_BACKLOG_SHED_ENABLED", True),
        shed_threshold_ms=max(0, _out_env_int("AUDIO_OUT_BACKLOG_SHED_THRESHOLD_MS", 400)),
        shed_keep_ms=max(0, _out_env_int("AUDIO_OUT_BACKLOG_SHED_KEEP_MS", 200)),
        shed_check_interval_frames=max(
            1, _out_env_int("AUDIO_OUT_BACKLOG_SHED_CHECK_INTERVAL_FRAMES", 5)
        ),
        silence_dbfs=_out_env_float("AUDIO_OUT_BACKLOG_SILENCE_DBFS", -60.0),
    )


class AudioOutputLatencyTracker:
    """Relogio + metrica + shed do caminho de saida, espelhando a entrada.

    - queue_ms/peak_queue_ms: quanto audio do agente espera na agent_q.
    - pacing_debt_dropped_ms: divida de pacing jogada fora quando o loop atrasa
      (antes era um resync mudo ``tick = now``, sem contador).
    - total_dropped_ms: silencio descartado pelo shed de saida.
    """

    def __init__(
        self,
        config: AudioOutputBacklogConfig,
        *,
        frame_ms: int,
        flow_logger: Any,
        timeline: Any = None,
        call_context: Optional[Mapping[str, Any]] = None,
        debug_event_publisher: Optional[Callable[[str, Mapping[str, Any]], None]] = None,
    ) -> None:
        self.config = config
        self.frame_ms = max(1, int(frame_ms))
        self.flow_logger = flow_logger
        self.timeline = timeline
        self.call_context = dict(call_context or {})
        self.debug_event_publisher = debug_event_publisher
        self.peak_queue_ms = 0
        self.total_dropped_frames = 0
        self.total_dropped_silent = 0
        self.pacing_debt_events = 0
        self.pacing_debt_dropped_ms = 0
        self.last_log_at: Optional[float] = None
        self.alert_active = False

    def _emit(self, step: str, payload: Mapping[str, Any]) -> None:
        log_flow_event(self.flow_logger, step, **payload)
        if self.timeline is not None:
            try:
                self.timeline.emit(step, **payload)
            except Exception:
                pass

    def note_pacing_debt(self, debt_ms: int, *, now: Optional[float] = None) -> None:
        debt_ms = max(0, int(debt_ms))
        if debt_ms <= 0:
            return
        self.pacing_debt_events += 1
        self.pacing_debt_dropped_ms += debt_ms
        if self.config.metrics_enabled:
            self._emit(
                "audio_out_pacing_debt",
                {
                    **self.call_context,
                    "debt_ms": debt_ms,
                    "pacing_debt_events": self.pacing_debt_events,
                    "pacing_debt_dropped_ms": self.pacing_debt_dropped_ms,
                },
            )

    def record_shed(
        self,
        *,
        dropped_frames: int,
        dropped_silent: int,
        mode: str,
        queue_before_frames: int,
        queue_after_frames: int,
    ) -> None:
        if dropped_frames <= 0:
            return
        self.total_dropped_frames += int(dropped_frames)
        self.total_dropped_silent += int(dropped_silent)
        if not self.config.metrics_enabled:
            return
        payload = {
            **self.call_context,
            "mode": mode,
            "content_policy": "silence_only" if mode == "energy" else "oldest_frames",
            "sync_action": "compress_silence" if mode == "energy" else "drop_audio_to_catch_up",
            "dropped_frames": int(dropped_frames),
            "dropped_ms": int(dropped_frames) * self.frame_ms,
            "dropped_silent_frames": int(dropped_silent),
            "queue_before_ms": int(queue_before_frames) * self.frame_ms,
            "queue_after_ms": int(queue_after_frames) * self.frame_ms,
            "total_dropped_ms": self.total_dropped_frames * self.frame_ms,
            "shed_threshold_ms": self.config.shed_threshold_ms,
            "shed_keep_ms": self.config.shed_keep_ms,
        }
        self._emit("audio_out_latency_shed", payload)
        if self.debug_event_publisher is not None:
            self.debug_event_publisher("bridge.audio_out.shed", payload)

    def maybe_log(
        self,
        *,
        queue_frames: int,
        now: float,
        reason: str,
        force: bool = False,
    ) -> None:
        queue_ms = max(0, int(queue_frames)) * self.frame_ms
        self.peak_queue_ms = max(self.peak_queue_ms, queue_ms)
        if not self.config.metrics_enabled:
            return
        alert_active = (
            self.config.latency_alert_ms > 0
            and queue_ms >= self.config.latency_alert_ms
        )
        alert_started = alert_active and not self.alert_active
        self.alert_active = alert_active
        if self.last_log_at is None:
            self.last_log_at = now
            if not force and not alert_started:
                return
        should_log = (
            force
            or alert_started
            or (now - self.last_log_at) >= self.config.latency_log_interval_s
        )
        if not should_log:
            return
        self.last_log_at = now
        self._emit(
            "audio_out_latency",
            {
                **self.call_context,
                "queue_frames": max(0, int(queue_frames)),
                "queue_ms": queue_ms,
                "peak_queue_ms": self.peak_queue_ms,
                "pacing_debt_events": self.pacing_debt_events,
                "pacing_debt_dropped_ms": self.pacing_debt_dropped_ms,
                "total_dropped_ms": self.total_dropped_frames * self.frame_ms,
                "reason": reason,
                "alert_ms": self.config.latency_alert_ms,
            },
        )
def _config_value(value: Any) -> str:
    if value in (None, ""):
        return ""
    return str(value).strip()


def _gain_from_linear(value: Any) -> Optional[float]:
    text = _config_value(value)
    if not text:
        return None
    try:
        gain = float(text)
        return max(0.0, min(10.0, gain)) if math.isfinite(gain) else None
    except Exception:
        return None


def _get_ws_output_gain(
    *,
    output_gain: Any = None,
) -> float:
    """
    Prioriza callConfig.ws.outputGain. Fallback: WS_OUTPUT_GAIN; default 1.0.
    """
    for gain in (
        _gain_from_linear(output_gain),
        _gain_from_linear(os.getenv("WS_OUTPUT_GAIN", "")),
    ):
        if gain is not None:
            return gain
    return 1.0


def _apply_pcm_gain(frame: bytes, sample_width: int, gain: float) -> bytes:
    if gain == 1.0:
        return frame

    try:
        return audioop.mul(frame, sample_width, gain)
    except Exception:
        return frame



def _agent_speech_rms_threshold() -> int:
    """Energia mínima para eventos de início de fala; não filtra o áudio."""
    try:
        dbfs = float(os.getenv("AGENT_OUT_SPEECH_DBFS", "-52"))
    except (TypeError, ValueError):
        dbfs = -52.0
    if not math.isfinite(dbfs):
        dbfs = -52.0
    return int(32768 * (10 ** (max(-90.0, min(0.0, dbfs)) / 20.0)))



async def ws_out_loop(
    ws: WebSocket,
    agent_q: asyncio.Queue[bytes],
    *,
    frame_ms: int,
    bytes_per_frame: int,
    first_agent_audio_sent: Optional[asyncio.Event] = None,
    flow_logger: Any = None,
    timeline: Any = None,
    call_context: Optional[Mapping[str, Any]] = None,
    output_stats: Optional[BridgeOutputStats] = None,
    output_gain: Any = None,
    recording_sink: Any = None,
    debug_event_publisher: Optional[Callable[[str, Mapping[str, Any]], None]] = None,
):
    frame_interval = frame_ms / 1000.0
    silence = b"\x00" * bytes_per_frame
    flow_logger = flow_logger or logger
    flow_context = dict(call_context or {})
    output_stats = output_stats or BridgeOutputStats()

    audio_burst_gap_s = float(os.getenv("FLOW_AUDIO_BURST_GAP_S", "0.8"))
    ws_output_gain = _get_ws_output_gain(
        output_gain=output_gain,
    )
    agent_speech_rms_threshold = _agent_speech_rms_threshold()
    if ws_output_gain != 1.0:
        logger.info("[ws-out] aplicando ganho global nos frames enviados ao WS gain=%.3f", ws_output_gain)

    # relogio + metrica + shed da saida (espelha a instrumentacao da entrada)
    out_config = audio_output_backlog_config_from_env()
    out_tracker = AudioOutputLatencyTracker(
        out_config,
        frame_ms=frame_ms,
        flow_logger=flow_logger,
        timeline=timeline,
        call_context=flow_context,
        debug_event_publisher=debug_event_publisher,
    )
    out_shed_threshold_frames = out_config.shed_threshold_frames(frame_ms)
    out_shed_keep_frames = out_config.shed_keep_frames(frame_ms)
    out_silence_rms = out_config.silence_rms_threshold
    pacing_debt_threshold_s = out_config.pacing_debt_threshold_ms / 1000.0
    if out_config.metrics_enabled:
        out_tracker._emit(
            "audio_out_backlog_config",
            {
                **flow_context,
                "shed_enabled": out_config.shed_enabled,
                "shed_threshold_ms": out_config.shed_threshold_ms,
                "shed_keep_ms": out_config.shed_keep_ms,
                "silence_dbfs": out_config.silence_dbfs,
                "latency_alert_ms": out_config.latency_alert_ms,
                "pacing_debt_threshold_ms": out_config.pacing_debt_threshold_ms,
                "shed_policy": "silence_only",
            },
        )

    def _emit_flow(step: str, **fields: Any) -> None:
        log_flow_event(flow_logger, step, **flow_context, **fields)

    tick = time.monotonic()
    n = 0

    while True:
        now = time.monotonic()

        if now < tick:
            await asyncio.sleep(tick - now)
            now = time.monotonic()

        if now - tick > pacing_debt_threshold_s:
            # loop atrasou: em vez de resync mudo, contabiliza a divida de pacing.
            out_tracker.note_pacing_debt(round((now - tick) * 1000), now=now)
            tick = now

        # 0) shed de backlog da saida: so silencio sai, nunca corta a fala do
        # agente. Rate-limited para nao redrenar a fila a cada tick.
        if (
            out_config.shed_enabled
            and (n % out_config.shed_check_interval_frames) == 0
        ):
            out_queue_before = agent_q.qsize()
            if out_queue_before > out_shed_threshold_frames:
                out_shed = shed_queue_backlog(
                    agent_q,
                    keep_frames=out_shed_keep_frames,
                    rms_threshold=out_silence_rms,
                    blind=False,
                )
                if out_shed.dropped:
                    out_tracker.record_shed(
                        dropped_frames=out_shed.dropped,
                        dropped_silent=out_shed.dropped_silent,
                        mode=out_shed.mode,
                        queue_before_frames=out_queue_before,
                        queue_after_frames=agent_q.qsize(),
                    )

        # 1) Consome exatamente um frame por tick. O frame e encaminhado como
        # veio; silencio temporal do agente so e removido pelo shed acima.
        agent_frame: Optional[bytes] = None
        agent_frame_rms = 0
        agent_frame_has_speech = False
        frame_source = "silence"
        try:
            cand = agent_q.get_nowait()
        except asyncio.QueueEmpty:
            cand = None

        if cand is not None:
            if len(cand) != bytes_per_frame:
                if len(cand) > bytes_per_frame:
                    cand = cand[:bytes_per_frame]
                else:
                    cand = cand + (b"\x00" * (bytes_per_frame - len(cand)))
            agent_frame = cand
            try:
                agent_frame_rms = audioop.rms(cand, 2)
            except Exception:
                agent_frame_rms = 0
            agent_frame_has_speech = agent_frame_rms > agent_speech_rms_threshold
        if agent_frame is not None:
            frame = agent_frame
            frame_source = "agent"
        else:
            # fila vazia: preenche o pacing de 20ms com silêncio
            frame = silence

        frame = _apply_pcm_gain(frame, 2, ws_output_gain)

        try:
            await ws.send_bytes(frame)
        except WebSocketDisconnect:
            return
        except RuntimeError as exc:
            if "close message has been sent" in str(exc):
                return
            raise

        if recording_sink is not None:
            try:
                recording_sink.record_output_frame(frame)
            except Exception:
                pass

        output_stats.total_ws_frames += 1
        if frame_source == "agent":
            output_stats.agent_audio_frames += 1
            new_burst = (
                output_stats.last_agent_audio_sent_at <= 0.0
                or (now - output_stats.last_agent_audio_sent_at) >= audio_burst_gap_s
            )
            if agent_frame_has_speech:
                output_stats.last_agent_audio_sent_at = now

            if agent_frame_has_speech and not output_stats.first_agent_audio_sent:
                output_stats.first_agent_audio_sent = True
                _emit_flow(
                    "audio_to_tia_first_frame",
                    destination="ws_client",
                    ws_frame=output_stats.total_ws_frames,
                    agent_audio_frames=output_stats.agent_audio_frames,
                    queued_agent_frames=agent_q.qsize(),
                    rms=agent_frame_rms,
                    bytes=len(frame),
                )
                if timeline is not None:
                    timeline.emit(
                        "agent_audio_first_frame_sent_to_tia",
                        **flow_context,
                        destination="ws_client",
                        ws_frame=output_stats.total_ws_frames,
                        agent_audio_frames=output_stats.agent_audio_frames,
                        queued_agent_frames=agent_q.qsize(),
                        rms=agent_frame_rms,
                        bytes=len(frame),
                    )

            if agent_frame_has_speech and new_burst:
                output_stats.agent_audio_bursts += 1
                _emit_flow(
                    "audio_to_tia",
                    destination="ws_client",
                    seq=output_stats.agent_audio_bursts,
                    ws_frame=output_stats.total_ws_frames,
                    agent_audio_frames=output_stats.agent_audio_frames,
                    queued_agent_frames=agent_q.qsize(),
                    rms=agent_frame_rms,
                    bytes=len(frame),
                )
                if timeline is not None:
                    timeline.emit(
                        "agent_audio_sent_to_tia",
                        **flow_context,
                        destination="ws_client",
                        seq=output_stats.agent_audio_bursts,
                        ws_frame=output_stats.total_ws_frames,
                        agent_audio_frames=output_stats.agent_audio_frames,
                        queued_agent_frames=agent_q.qsize(),
                        rms=agent_frame_rms,
                        bytes=len(frame),
                    )
        else:
            output_stats.silence_frames += 1

        if (
            agent_frame is not None
            and agent_frame_has_speech
            and first_agent_audio_sent is not None
            and not first_agent_audio_sent.is_set()
        ):
            first_agent_audio_sent.set()

        out_tracker.maybe_log(
            queue_frames=agent_q.qsize(),
            now=now,
            reason="tick",
        )

        n += 1
        tick += frame_interval
