from __future__ import annotations

import json
import threading
import urllib.error
import urllib.request
from http.server import ThreadingHTTPServer
from typing import ClassVar

from cognitive_agent.auth import AuthConfig, AuthRuntime
from cognitive_agent.infrastructure.speech import SpeechRecognitionError
from cognitive_agent.server import SeatDemoHandler


class SpeechStub:
    preview_calls: ClassVar[list[dict[str, object]]] = []
    final_calls: ClassVar[list[dict[str, object]]] = []
    preview_error: ClassVar[SpeechRecognitionError | None] = None

    @staticmethod
    def status() -> dict[str, object]:
        return {
            "enabled": True,
            "model": "SenseVoiceSmall",
            "decoder": "offline-sensevoice-live-preview-and-final",
            "live_preview": {"enabled": True, "recommended_interval_ms": 850},
            "input_modes": {
                "supported": ["manual", "wake"],
                "header": "X-Speech-Input-Mode",
                "preview_default": "wake",
                "final_default": "manual",
            },
            "wake_word": {
                "enabled": True,
                "primary": "你好，双十一",
                "aliases": ["你好，双十一", "你好，双11"],
            },
        }

    @classmethod
    def transcribe_preview_wav(
        cls,
        content: bytes,
        *,
        utterance_id: str | None = None,
        sequence: int,
    ) -> dict[str, object]:
        if cls.preview_error is not None:
            raise cls.preview_error
        if sequence < 0:
            raise SpeechRecognitionError(
                "invalid_preview_sequence", "实时转写序号无效。", status=400
            )
        cls.preview_calls.append(
            {
                "content": content,
                "utterance_id": utterance_id,
                "sequence": sequence,
            }
        )
        return {
            "utterance_id": utterance_id,
            "sequence": sequence,
            "state": "partial",
            "raw_text": "你好双十一打开空",
            "text": "你好双十一打开空",
            "live_text": "打开空",
            "wake_word": {"detected": True, "remainder": "打开空"},
            "semantic_cleaned": False,
            "stale": False,
            "idempotent_replay": False,
        }

    @classmethod
    def transcribe_wav(
        cls,
        content: bytes,
        *,
        utterance_id: str | None = None,
        require_wake_word: bool = False,
    ) -> dict[str, object]:
        cls.final_calls.append(
            {
                "content": content,
                "utterance_id": utterance_id,
                "require_wake_word": require_wake_word,
            }
        )
        return {
            "utterance_id": utterance_id,
            "captured_text": "你好双十一车上有点热",
            "raw_text": "车上有点热",
            "assistant_text": "车上有点热。",
            "text": "车上有点热。",
            "clean_text": "",
            "control_text": "",
            "commands": [],
            "execution_eligible": False,
            "wake_word": {"detected": True, "remainder": "车上有点热"},
        }


class SpeechApiHandler(SeatDemoHandler):
    speech = SpeechStub()
    auth = AuthRuntime(AuthConfig(False, "", "", b"", 3600, 5, 300, 120, False))


class TestSpeechLiveWebApi:
    early_rejection_body = b"RIFF" + (b"\0" * (256 * 1024))

    @classmethod
    def setup_class(cls) -> None:
        cls.server = ThreadingHTTPServer(("127.0.0.1", 0), SpeechApiHandler)
        cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True)
        cls.thread.start()
        cls.base_url = f"http://127.0.0.1:{cls.server.server_port}"

    @classmethod
    def teardown_class(cls) -> None:
        cls.server.shutdown()
        cls.server.server_close()
        cls.thread.join(timeout=3)

    def setup_method(self) -> None:
        SpeechStub.preview_calls.clear()
        SpeechStub.final_calls.clear()
        SpeechStub.preview_error = None

    @classmethod
    def post_audio(
        cls,
        path: str,
        *,
        body: bytes = b"RIFF-live-speech",
        headers: dict[str, str] | None = None,
    ) -> tuple[int, dict[str, object]]:
        request = urllib.request.Request(
            f"{cls.base_url}{path}",
            method="POST",
            headers={"Content-Type": "audio/wav", **(headers or {})},
            data=body,
        )
        try:
            with urllib.request.urlopen(request, timeout=5) as response:
                cls.last_response_headers = dict(response.headers.items())
                return response.status, json.loads(response.read().decode("utf-8"))
        except urllib.error.HTTPError as exc:
            cls.last_response_headers = dict(exc.headers.items())
            return exc.code, json.loads(exc.read().decode("utf-8"))

    def test_status_advertises_live_preview_and_configured_wake_word(self) -> None:
        with urllib.request.urlopen(f"{self.base_url}/api/speech/status", timeout=5) as response:
            payload = json.loads(response.read().decode("utf-8"))

        assert payload["decoder"] == "offline-sensevoice-live-preview-and-final"
        assert payload["live_preview"]["enabled"] is True
        assert payload["input_modes"] == {
            "supported": ["manual", "wake"],
            "header": "X-Speech-Input-Mode",
            "preview_default": "wake",
            "final_default": "manual",
        }
        assert payload["wake_word"]["enabled"] is True
        assert payload["wake_word"]["primary"] == "你好，双十一"

    def test_preview_endpoint_forwards_accumulated_audio_identity_and_sequence(self) -> None:
        status, payload = self.post_audio(
            "/api/speech/preview",
            headers={
                "X-Speech-Input-Mode": "wake",
                "X-Speech-Utterance-Id": "web-live-preview-1",
                "X-Speech-Preview-Sequence": "7",
            },
        )

        assert status == 200
        assert SpeechStub.preview_calls == [
            {
                "content": b"RIFF-live-speech",
                "utterance_id": "web-live-preview-1",
                "sequence": 7,
            }
        ]
        assert payload["state"] == "partial"
        assert payload["sequence"] == 7
        assert payload["raw_text"] == "你好双十一打开空"
        assert payload["live_text"] == "打开空"
        assert payload["semantic_cleaned"] is False

    def test_final_endpoint_requires_wake_and_preserves_assistant_only_text(self) -> None:
        status, payload = self.post_audio(
            "/api/speech/transcribe",
            headers={
                "X-Speech-Input-Mode": "wake",
                "X-Speech-Utterance-Id": "web-live-final-1",
                "X-Speech-Wake-Word-Required": "1",
            },
        )

        assert status == 200
        assert SpeechStub.final_calls == [
            {
                "content": b"RIFF-live-speech",
                "utterance_id": "web-live-final-1",
                "require_wake_word": True,
            }
        ]
        assert payload["raw_text"] == "车上有点热"
        assert payload["assistant_text"] == "车上有点热。"
        assert payload["text"] == "车上有点热。"
        assert payload["control_text"] == ""
        assert payload["commands"] == []
        assert payload["execution_eligible"] is False

    def test_manual_final_endpoint_does_not_require_a_wake_word(self) -> None:
        status, payload = self.post_audio(
            "/api/speech/transcribe",
            headers={
                "X-Speech-Input-Mode": "manual",
                "X-Speech-Utterance-Id": "web-manual-final-1",
                "X-Speech-Wake-Word-Required": "0",
            },
        )

        assert status == 200
        assert SpeechStub.final_calls == [
            {
                "content": b"RIFF-live-speech",
                "utterance_id": "web-manual-final-1",
                "require_wake_word": False,
            }
        ]
        assert payload["assistant_text"] == "车上有点热。"

    def test_final_endpoint_defaults_to_manual_when_wake_header_is_absent(self) -> None:
        status, _payload = self.post_audio(
            "/api/speech/transcribe",
            headers={
                "X-Speech-Input-Mode": "manual",
                "X-Speech-Utterance-Id": "web-manual-default-1",
            },
        )

        assert status == 200
        assert SpeechStub.final_calls[0]["require_wake_word"] is False

    def test_preview_rejects_missing_sequence_without_invoking_transcriber(self) -> None:
        status, payload = self.post_audio(
            "/api/speech/preview",
            body=self.early_rejection_body,
            headers={"X-Speech-Utterance-Id": "web-live-preview-invalid"},
        )

        assert status == 400
        assert payload["error"] == "invalid_preview_sequence"
        assert SpeechStub.preview_calls == []

    def test_preview_http_gate_rejects_overload_before_reading_or_transcribing(self) -> None:
        original_slots = SpeechApiHandler.speech_preview_http_slots
        constrained_slots = threading.BoundedSemaphore(1)
        SpeechApiHandler.speech_preview_http_slots = constrained_slots
        assert constrained_slots.acquire(blocking=False)
        try:
            status, payload = self.post_audio(
                "/api/speech/preview",
                body=self.early_rejection_body,
                headers={
                    "X-Speech-Utterance-Id": "web-live-preview-overloaded",
                    "X-Speech-Preview-Sequence": "0",
                },
            )
        finally:
            constrained_slots.release()
            SpeechApiHandler.speech_preview_http_slots = original_slots

        assert status == 429
        assert payload["error"] == "preview_overloaded"
        assert payload["retry_after_seconds"] == 1
        assert self.last_response_headers["Retry-After"] == "1"
        assert SpeechStub.preview_calls == []

    def test_preview_transcriber_rate_error_preserves_retry_contract(self) -> None:
        SpeechStub.preview_error = SpeechRecognitionError(
            "preview_rate_limited",
            "实时语音更新过快，请稍后重试。",
            status=429,
            retry_after_seconds=2,
        )
        status, payload = self.post_audio(
            "/api/speech/preview",
            headers={
                "X-Speech-Utterance-Id": "web-live-preview-throttled",
                "X-Speech-Preview-Sequence": "1",
            },
        )

        assert status == 429
        assert payload["error"] == "preview_rate_limited"
        assert payload["retry_after_seconds"] == 2
        assert self.last_response_headers["Retry-After"] == "2"

    def test_live_endpoints_reject_non_wav_content_type(self) -> None:
        for path in ("/api/speech/preview", "/api/speech/transcribe"):
            status, payload = self.post_audio(
                path,
                headers={"Content-Type": "application/json"},
            )
            assert status == 415, path
            assert payload["error"] == "unsupported_media_type", path