from __future__ import annotations

import json
import logging
import unittest
import uuid
from unittest import mock

import httpx

from app.providers.stt_internal_livekit import (
    InternalHTTPSTT,
    InternalSTTConfig,
    _prepend_pcm16le_silence,
    stt_text_with_single_word_threshold,
)
from app.providers.stt_config import override_config
from app.utils import logging as logging_utils
from app.utils.turn_ids import (
    peek_started_turn_message_id,
    register_started_turn_message_id,
    reset_turn_message_sequence,
)


def _assert_uuid(testcase: unittest.TestCase, value: object) -> str:
    text = str(value or "")
    testcase.assertEqual(str(uuid.UUID(text)), text)
    return text


def _single_word_payload(text: str, probability: float) -> dict:
    return {
        "data": {
            "text": text,
            "words": [{"word": text, "probability": probability}],
        }
    }


class InternalHTTPSTTTests(unittest.IsolatedAsyncioTestCase):
    async def test_provider_can_be_instantiated(self) -> None:
        client = httpx.AsyncClient()
        try:
            stt = InternalHTTPSTT(
                InternalSTTConfig(
                    url="https://example.invalid/stt",
                    api_key="test-key",
                ),
                client=client,
            )
        finally:
            await client.aclose()

        self.assertEqual(stt.provider, "unknown")

    def test_default_sofya_vad_padding_keeps_more_prefix_audio(self) -> None:
        vad_params = override_config["processor"]["config"]["extra_params"]["vad_parameters"]

        self.assertEqual(vad_params["speech_pad_ms"], 1000)

    def test_stt_input_prefix_padding_prepends_silence(self) -> None:
        pcm = b"\x01\x00" * 16000

        padded, padding_ms = _prepend_pcm16le_silence(
            pcm,
            sample_rate=16000,
            channels=1,
            padding_ms=250,
        )

        self.assertEqual(padding_ms, 250)
        self.assertEqual(len(padded), len(pcm) + 8000)
        self.assertTrue(padded.startswith(b"\x00" * 8000))
        self.assertTrue(padded.endswith(pcm))

    async def test_config_override_is_used_as_json_object(self) -> None:
        client = httpx.AsyncClient()
        try:
            stt = InternalHTTPSTT(
                InternalSTTConfig(
                    url="https://example.invalid/stt",
                    api_key="test-key",
                    config_override='{"processor":{"strategy":"faster_default"}}',
                ),
                client=client,
            )
            override = json.loads(stt._build_override_config_json())
        finally:
            await client.aclose()

        self.assertEqual(override["processor"]["strategy"], "faster_default")

    async def test_http_success_with_empty_text_does_not_create_structured_turn(self) -> None:
        published_events = []
        empty_transcript_calls = []
        captured_headers: dict[str, str] = {}

        def _handler(request: httpx.Request) -> httpx.Response:
            captured_headers["connection"] = request.headers.get("connection", "")
            return httpx.Response(200, json={"data": {"text": ""}}, request=request)

        logger = logging.getLogger("test.internal_http_stt.structured")
        logger.handlers = [logging.NullHandler()]
        logger.setLevel(logging.INFO)
        structured_context = logging_utils.StructuredLogContext(
            callid="call-1",
            session_id="session-1",
            num_telefone="5511999999999",
            cod_ani="1234",
            nome_agente="conta",
        )
        reset_turn_message_sequence(structured_context, clear_pending=True)
        upload_message_id = "12345678-1234-4234-9234-123456789abc"
        register_started_turn_message_id(structured_context, upload_message_id)

        client = httpx.AsyncClient(transport=httpx.MockTransport(_handler))
        try:
            stt = InternalHTTPSTT(
                InternalSTTConfig(
                    url="https://example.invalid/stt",
                    api_key="test-key",
                    output_mode="api_text",
                ),
                client=client,
                structured_log_context=structured_context,
                structured_logger=logger,
                empty_transcript_handler=lambda: empty_transcript_calls.append("called"),
            )
            with (
                mock.patch("app.utils.logging.publish_structured_event", side_effect=published_events.append),
                mock.patch("app.utils.logging.publish_structured_span"),
                mock.patch("app.providers.stt_internal_livekit.log_flow_event") as flow_log,
                mock.patch("app.providers.stt_internal_livekit.logger.info") as logger_info,
            ):
                event = await stt._post_internal_http(
                    b"\0\0",
                    b"fake-wav",
                    req_id="req-1",
                    started_ns=1_700_000_000_000_000_000,
                    language="pt-BR",
                    message_id=upload_message_id,
                )
        finally:
            await client.aclose()

        self.assertEqual(event.alternatives[0].text, "")
        self.assertEqual(captured_headers["connection"], "close")
        self.assertEqual(published_events, [])
        self.assertEqual(peek_started_turn_message_id(structured_context), "")
        self.assertTrue(
            any(
                call.args[1] == "stt_done"
                and call.kwargs["request_id"] == "req-1"
                and call.kwargs["text_len"] == 0
                for call in flow_log.call_args_list
            )
        )
        self.assertEqual(empty_transcript_calls, ["called"])
        logger_info.assert_any_call(
            "[stt][response_json] req_id=%s status=%s took=%.0fms json=%s",
            "req-1",
            200,
            mock.ANY,
            '{"data": {"text": ""}}',
        )

    async def test_http_success_marks_structured_interruption(self) -> None:
        published_events = []
        provider_metrics = []

        def _handler(request: httpx.Request) -> httpx.Response:
            return httpx.Response(200, json={"data": {"text": "sim"}}, request=request)

        logger = logging.getLogger("test.internal_http_stt.interruption")
        logger.handlers = [logging.NullHandler()]
        logger.setLevel(logging.INFO)
        structured_context = logging_utils.StructuredLogContext(
            callid="call-1",
            session_id="session-1",
            num_telefone="5511999999999",
            cod_ani="1234",
            nome_agente="conta",
        )
        reset_turn_message_sequence(structured_context, clear_pending=True)

        client = httpx.AsyncClient(transport=httpx.MockTransport(_handler))
        try:
            stt = InternalHTTPSTT(
                InternalSTTConfig(
                    url="https://example.invalid/stt",
                    api_key="test-key",
                    output_mode="api_text",
                ),
                client=client,
                structured_log_context=structured_context,
                structured_logger=logger,
                structured_interruption_flag=lambda: True,
                metrics_handler=provider_metrics.append,
            )
            with (
                mock.patch("app.utils.logging.publish_structured_event", side_effect=published_events.append),
                mock.patch("app.utils.logging.publish_structured_span"),
            ):
                await stt._post_internal_http(
                    b"\0\0",
                    b"fake-wav",
                    req_id="req-1",
                    started_ns=1_700_000_000_000_000_000,
                    language="pt-BR",
                    audio_duration_ms=1250,
                    original_audio_duration_ms=1000,
                    input_padding_ms=250,
                    level_dbfs=-24.5,
                )
        finally:
            await client.aclose()

        self.assertEqual(len(published_events), 1)
        self.assertEqual(published_events[0]["interrupcao"], 1)
        self.assertEqual(len(provider_metrics), 1)
        self.assertEqual(provider_metrics[0]["event"], "completed")
        self.assertEqual(provider_metrics[0]["audio_duration_ms"], 1250)
        self.assertEqual(provider_metrics[0]["original_audio_duration_ms"], 1000)
        self.assertEqual(provider_metrics[0]["input_padding_ms"], 250)
        self.assertEqual(provider_metrics[0]["input_dbfs"], -24.5)
        self.assertEqual(provider_metrics[0]["retry_count"], 0)
        self.assertEqual(provider_metrics[0]["http_status"], 200)
        self.assertFalse(provider_metrics[0]["empty_transcript"])
        self.assertEqual(provider_metrics[0]["text_length"], 3)

    async def test_http_success_uses_supplied_message_id_for_structured_event(self) -> None:
        published_events = []

        def _handler(request: httpx.Request) -> httpx.Response:
            return httpx.Response(200, json={"data": {"text": "sim"}}, request=request)

        logger = logging.getLogger("test.internal_http_stt.message_id")
        logger.handlers = [logging.NullHandler()]
        logger.setLevel(logging.INFO)
        structured_context = logging_utils.StructuredLogContext(
            callid="call-1",
            session_id="session-1",
            num_telefone="5511999999999",
            cod_ani="1234",
            nome_agente="conta",
        )
        reset_turn_message_sequence(structured_context, clear_pending=True)

        client = httpx.AsyncClient(transport=httpx.MockTransport(_handler))
        try:
            stt = InternalHTTPSTT(
                InternalSTTConfig(
                    url="https://example.invalid/stt",
                    api_key="test-key",
                    output_mode="api_text",
                ),
                client=client,
                structured_log_context=structured_context,
                structured_logger=logger,
            )
            with (
                mock.patch("app.utils.logging.publish_structured_event", side_effect=published_events.append),
                mock.patch("app.utils.logging.publish_structured_span"),
            ):
                await stt._post_internal_http(
                    b"\0\0",
                    b"fake-wav",
                    req_id="req-1",
                    started_ns=1_700_000_000_000_000_000,
                    language="pt-BR",
                    message_id="message-from-upload-path",
                )
        finally:
            await client.aclose()

        self.assertEqual(len(published_events), 1)
        self.assertEqual(published_events[0]["message_id"], "message-from-upload-path")

    async def test_http_500_retries_once_then_uses_success(self) -> None:
        attempts = 0

        def _handler(request: httpx.Request) -> httpx.Response:
            nonlocal attempts
            attempts += 1
            if attempts == 1:
                return httpx.Response(500, text="temporary", request=request)
            return httpx.Response(200, json={"data": {"text": "sim"}}, request=request)

        client = httpx.AsyncClient(transport=httpx.MockTransport(_handler))
        try:
            stt = InternalHTTPSTT(
                InternalSTTConfig(
                    url="https://example.invalid/stt",
                    api_key="test-key",
                    output_mode="api_text",
                ),
                client=client,
            )
            with mock.patch("app.providers.stt_internal_livekit.log_flow_event") as flow_log:
                event = await stt._post_internal_http(
                    b"\0\0",
                    b"fake-wav",
                    req_id="req-retry",
                    started_ns=1_700_000_000_000_000_000,
                    language="pt-BR",
                )
        finally:
            await client.aclose()

        self.assertEqual(attempts, 2)
        self.assertEqual(event.alternatives[0].text, "sim")
        self.assertTrue(
            any(
                call.args[1] == "stt_http_retry"
                and call.kwargs["request_id"] == "req-retry"
                and call.kwargs["attempt"] == 1
                and call.kwargs["max_retries"] == 1
                for call in flow_log.call_args_list
            )
        )

    async def test_http_500_after_retry_returns_empty_transcript_without_raising(self) -> None:
        attempts = 0
        published_events = []

        def _handler(request: httpx.Request) -> httpx.Response:
            nonlocal attempts
            attempts += 1
            return httpx.Response(500, text="still failing", request=request)

        logger = logging.getLogger("test.internal_http_stt.http_500_nonfatal")
        logger.handlers = [logging.NullHandler()]
        logger.setLevel(logging.INFO)
        structured_context = logging_utils.StructuredLogContext(
            callid="call-1",
            session_id="session-1",
            num_telefone="5511999999999",
            cod_ani="1234",
            nome_agente="conta",
        )
        reset_turn_message_sequence(structured_context, clear_pending=True)

        client = httpx.AsyncClient(transport=httpx.MockTransport(_handler))
        try:
            stt = InternalHTTPSTT(
                InternalSTTConfig(
                    url="https://example.invalid/stt",
                    api_key="test-key",
                    output_mode="api_text",
                ),
                client=client,
                structured_log_context=structured_context,
                structured_logger=logger,
            )
            with (
                mock.patch("app.utils.logging.publish_structured_event", side_effect=published_events.append),
                mock.patch("app.utils.logging.publish_structured_span"),
                mock.patch("app.providers.stt_internal_livekit.log_flow_event") as flow_log,
            ):
                event = await stt._post_internal_http(
                    b"\0\0",
                    b"fake-wav",
                    req_id="req-500",
                    started_ns=1_700_000_000_000_000_000,
                    language="pt-BR",
                )
        finally:
            await client.aclose()

        self.assertEqual(attempts, 2)
        self.assertEqual(event.alternatives[0].text, "")
        self.assertEqual(len(published_events), 1)
        self.assertEqual(published_events[0]["erro_msg"], "Falha STT")
        self.assertEqual(published_events[0]["http_cod_status"], 500)
        self.assertTrue(
            any(
                call.args[1] == "stt_error_nonfatal"
                and call.kwargs["request_id"] == "req-500"
                and call.kwargs["action"] == "return_empty_transcript"
                for call in flow_log.call_args_list
            )
        )

    def test_single_word_allowlist_accepts_low_confidence_sim(self) -> None:
        self.assertEqual(
            stt_text_with_single_word_threshold(
                _single_word_payload("sim", 0.02),
                min_prob_single_word=0.03,
            ),
            "sim",
        )

    def test_single_word_allowlist_accepts_extremely_low_confidence_sim(self) -> None:
        self.assertEqual(
            stt_text_with_single_word_threshold(
                _single_word_payload("sim", 0.009),
                min_prob_single_word=0.03,
            ),
            "sim",
        )

    def test_single_word_filter_rejects_non_allowlisted_low_confidence_word(self) -> None:
        self.assertIsNone(
            stt_text_with_single_word_threshold(
                _single_word_payload("talvez", 0.02),
                min_prob_single_word=0.03,
            )
        )

    def test_single_word_filter_accepts_non_allowlisted_high_confidence_word(self) -> None:
        self.assertEqual(
            stt_text_with_single_word_threshold(
                _single_word_payload("talvez", 0.04),
                min_prob_single_word=0.03,
            ),
            "talvez",
        )

    def test_single_word_filter_keeps_api_text_empty_as_empty(self) -> None:
        self.assertIsNone(
            stt_text_with_single_word_threshold(
                {
                    "data": {
                        "text": "",
                        "words": [{"word": "sim", "probability": 0.5}],
                    }
                },
                min_prob_single_word=0.03,
            )
        )

    async def test_single_word_allowlist_logs_low_confidence_reason(self) -> None:
        client = httpx.AsyncClient()
        try:
            stt = InternalHTTPSTT(
                InternalSTTConfig(
                    url="https://example.invalid/stt",
                    api_key="test-key",
                    min_prob_single_word=0.03,
                ),
                client=client,
            )
            with mock.patch("app.providers.stt_internal_livekit.log_flow_event") as flow_log:
                text = stt._format_stt_output(
                    _single_word_payload("sim", 0.02),
                    request_id="req-allowlist",
                )
        finally:
            await client.aclose()

        self.assertEqual(text, "sim")
        self.assertTrue(
            any(
                call.args[1] == "stt_payload"
                and call.kwargs["request_id"] == "req-allowlist"
                and call.kwargs["filter_reason"] == "single_word_allowlist_low_confidence"
                and call.kwargs["allowlist_min_prob"] == 0.0
                for call in flow_log.call_args_list
            )
        )


if __name__ == "__main__":
    unittest.main()
