"""Tests for the shared LLM speech adapter and speech job handlers."""

from __future__ import annotations

import asyncio
import base64
import json
import tempfile
from pathlib import Path

import httpx
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

BACKEND_DIR = Path(__file__).resolve().parents[1]
if str(BACKEND_DIR) not in __import__("sys").path:
    __import__("sys").path.insert(0, str(BACKEND_DIR))

import job_handlers  # noqa: E402
import tts_asr  # noqa: E402
from audit_store import BaseAuditStore  # noqa: E402


def _wav_bytes(duration_ms: int = 100) -> bytes:
    sample_rate = 16000
    frames = max(1, int(sample_rate * duration_ms / 1000))
    data = b"".join((i % 32768).to_bytes(2, "little", signed=False) for i in range(frames))
    data_size = len(data)
    return (
        b"RIFF" + (36 + data_size).to_bytes(4, "little")
        + b"WAVE"
        + b"fmt " + (16).to_bytes(4, "little")
        + (1).to_bytes(2, "little")
        + (1).to_bytes(2, "little")
        + sample_rate.to_bytes(4, "little")
        + sample_rate.to_bytes(4, "little")
        + (2).to_bytes(2, "little")
        + (16).to_bytes(2, "little")
        + b"data" + data_size.to_bytes(4, "little")
        + data
    )


def _run_async(coro):
    return asyncio.run(coro)


class _CaptureAuditStore(BaseAuditStore):
    def __init__(self) -> None:
        self.llm_calls: list[dict] = []

    def record_llm_call(self, payload: dict) -> None:
        self.llm_calls.append(payload)


def test_tts_calls_shared_llm_speech_endpoint(monkeypatch):
    captured: dict[str, object] = {}
    monkeypatch.setattr(tts_asr, "LLM_API_KEY", "test-api-key")
    monkeypatch.setattr(tts_asr, "TTS_MODEL_ID", "Qwen3-TTS-12Hz-1.7B-VoiceDesign-8bit")

    def transport(request: httpx.Request):
        captured["url"] = str(request.url)
        captured["headers"] = dict(request.headers)
        captured["json"] = json.loads(request.read().decode("utf-8"))
        return httpx.Response(200, content=b"speech-ok", headers={"x-request-id": "tts-req-1"})

    async def run():
        client = httpx.AsyncClient(
            base_url="https://speech.example/v1",
            transport=httpx.MockTransport(transport),
        )
        try:
            tts_asr._httpx_client = client
            return await tts_asr.generate_tts_response(
                "你好世界",
                instruct="A warm Mandarin voice.",
                speaker="Vivian",
                output_format="wav",
            )
        finally:
            await client.aclose()
            tts_asr._httpx_client = None

    result = _run_async(run())

    assert result["format"] == "wav"
    assert result["speaker"] == "Vivian"
    assert result["model"] == "Qwen3-TTS-12Hz-1.7B-VoiceDesign-8bit"
    assert result["upstream_request_id"] == "tts-req-1"
    assert base64.b64decode(result["audio_base64"]) == b"speech-ok"
    assert captured["url"] == "https://speech.example/v1/audio/speech"
    assert captured["headers"]["authorization"] == "Bearer test-api-key"
    payload = captured["json"]
    assert payload["model"] == "Qwen3-TTS-12Hz-1.7B-VoiceDesign-8bit"
    assert payload["voice"] == "Vivian"
    assert payload["input"] == "你好世界"
    assert payload["instructions"] == "A warm Mandarin voice."
    assert "instruction" not in payload


def test_tts_uses_nonempty_default_instructions(monkeypatch):
    captured: dict[str, object] = {}

    def transport(request: httpx.Request):
        captured["json"] = json.loads(request.read().decode("utf-8"))
        return httpx.Response(200, content=b"speech-ok")

    async def run():
        client = httpx.AsyncClient(
            base_url="https://speech.example/v1",
            transport=httpx.MockTransport(transport),
        )
        try:
            tts_asr._httpx_client = client
            return await tts_asr.generate_tts_response("你好世界")
        finally:
            await client.aclose()
            tts_asr._httpx_client = None

    _run_async(run())

    payload = captured["json"]
    assert payload["instructions"] == tts_asr.DEFAULT_TTS_INSTRUCTIONS
    assert payload["instructions"].strip()


def test_asr_calls_shared_llm_transcriptions_endpoint(monkeypatch):
    captured: dict[str, object] = {}
    monkeypatch.setattr(tts_asr, "LLM_API_KEY", "test-api-key")
    monkeypatch.setattr(tts_asr, "ASR_MODEL_ID", "Qwen3-ASR-0.6B-8bit")

    def transport(request: httpx.Request):
        captured["url"] = str(request.url)
        captured["headers"] = dict(request.headers)
        captured["content"] = request.read()
        return httpx.Response(200, json={"text": "hello world", "language": "zh"}, headers={"x-request-id": "asr-req-1"})

    async def run():
        client = httpx.AsyncClient(
            base_url="https://speech.example/v1",
            transport=httpx.MockTransport(transport),
        )
        try:
            tts_asr._httpx_client = client
            return await tts_asr.generate_asr_response(_wav_bytes(), language="zh-CN")
        finally:
            await client.aclose()
            tts_asr._httpx_client = None

    result = _run_async(run())

    assert result["text"] == "hello world"
    assert result["language"] == "zh"
    assert result["model"] == "Qwen3-ASR-0.6B-8bit"
    assert result["upstream_request_id"] == "asr-req-1"
    assert captured["url"] == "https://speech.example/v1/audio/transcriptions"
    assert captured["headers"]["authorization"] == "Bearer test-api-key"
    content = captured["content"]
    assert b'name="model"' in content
    assert b"Qwen3-ASR-0.6B-8bit" in content
    assert b'name="language"' in content
    assert b"zh" in content


def test_invalid_tts_text_returns_http_exception():
    with pytest.raises(tts_asr.HTTPException) as exc:
        _run_async(tts_asr._call_tts_api("", speaker="Vivian"))
    assert exc.value.status_code == 400


def test_invalid_asr_audio_returns_http_exception():
    with pytest.raises(tts_asr.HTTPException) as exc:
        _run_async(tts_asr._call_asr_api(b"", language="zh-CN"))
    assert exc.value.status_code == 400


def test_status_config_routes(monkeypatch):
    app = FastAPI()
    app.include_router(tts_asr.meta_router)
    monkeypatch.setattr(tts_asr, "LLM_BASE_URL", "https://speech.example/v1")
    monkeypatch.setattr(tts_asr, "LLM_API_KEY", "")
    monkeypatch.setattr(tts_asr, "TTS_MODEL_ID", "Qwen3-TTS-12Hz-1.7B-VoiceDesign-8bit")
    monkeypatch.setattr(tts_asr, "ASR_MODEL_ID", "Qwen3-ASR-0.6B-8bit")

    with TestClient(app) as client:
        status = client.get("/status")
        config = client.get("/config")

    assert status.status_code == 200
    assert config.status_code == 200
    assert status.json()["llm_url"] == "https://speech.example/v1"
    assert status.json()["tts_model"] == "Qwen3-TTS-12Hz-1.7B-VoiceDesign-8bit"
    assert status.json()["asr_model"] == "Qwen3-ASR-0.6B-8bit"
    assert status.json()["status"]["api_key_configured"] is False
    assert status.json()["status"]["max_connections"] == tts_asr.SPEECH_MAX_CONNECTIONS


def test_tts_concurrent_requests_respect_connection_limit(monkeypatch):
    monkeypatch.setattr(tts_asr, "SPEECH_MAX_CONNECTIONS", 4)
    monkeypatch.setattr(tts_asr, "SPEECH_MAX_KEEPALIVE_CONNECTIONS", 1)

    class LimitedClient:
        def __init__(self):
            self.semaphore = asyncio.Semaphore(4)
            self.active = 0
            self.max_active = 0

        async def post(self, url: str, **kwargs):
            async with self.semaphore:
                self.active += 1
                self.max_active = max(self.max_active, self.active)
                await asyncio.sleep(0.01)
                self.active -= 1
                return httpx.Response(200, content=b"speech-ok", request=httpx.Request("POST", f"https://speech.example{url}"))

    async def run():
        client = LimitedClient()

        async def get_client():
            return client

        monkeypatch.setattr(tts_asr, "_get_speech_client", get_client)
        await asyncio.gather(*(tts_asr.generate_tts_response(f"文本 {index}") for index in range(20)))
        return client

    client = _run_async(run())
    assert client.max_active <= 4


def test_tts_asr_handlers_record_audit(monkeypatch):
    audit_store = _CaptureAuditStore()

    async def fake_tts(*args, **kwargs):
        return {
            "audio_base64": base64.b64encode(b"ok").decode("utf-8"),
            "format": "wav",
            "duration_ms": 1200,
            "audio_bytes": 2,
            "text_chars": 2,
            "speaker": "Vivian",
            "model": "Qwen3-TTS-12Hz-1.7B-VoiceDesign-8bit",
            "request_ms": 45,
            "upstream_request_id": "tts-upstream",
        }

    async def fake_asr(*args, **kwargs):
        return {
            "text": "hello world",
            "language": "zh",
            "audio_bytes": len(_wav_bytes()),
            "model": "Qwen3-ASR-0.6B-8bit",
            "request_ms": 80,
            "upstream_request_id": "asr-upstream",
        }

    monkeypatch.setattr(job_handlers, "generate_tts_response", fake_tts)
    monkeypatch.setattr(job_handlers, "generate_asr_response", fake_asr)
    monkeypatch.setattr(job_handlers, "get_audit_store", lambda *_args, **_kwargs: audit_store)

    base_payload = {
        "request_id": "req-1",
        "risk": {
            "request_id": "req-1",
            "session_hash": "session",
            "ip_hash": "ip",
            "estimated_input_tokens": 12,
            "estimated_cost": 0.0,
            "policy": {
                "job_type": "tts",
                "model": "Qwen3-TTS-12Hz-1.7B-VoiceDesign-8bit",
                "profile": "speech_tts",
                "max_output_tokens": 0,
            },
        },
        "job_context": {
            "created_at": 1000,
            "started_at": 1200,
            "queue_ms": 200,
        },
    }

    async def run():
        events = []

        async def emit(event: str, data: dict):
            events.append((event, data))

        tts_payload = {
            **base_payload,
            "text": "你好",
            "speaker": "Vivian",
            "format": "wav",
        }
        await job_handlers.tts_handler(tts_payload, emit, lambda: False)

        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as handle:
            handle.write(_wav_bytes())
            audio_path = handle.name

        try:
            asr_payload = {
                **base_payload,
                "risk": {
                    **base_payload["risk"],
                    "policy": {
                        "job_type": "asr",
                        "model": "Qwen3-ASR-0.6B-8bit",
                        "profile": "speech_asr",
                        "max_output_tokens": 0,
                    },
                },
                "input_path": audio_path,
                "language": "zh-CN",
            }
            await job_handlers.asr_handler(asr_payload, emit, lambda: False)
        finally:
            job_handlers._safe_unlink(audio_path)

        return events

    events = _run_async(run())

    assert any(event == "result" for event, _data in events)
    assert len(audit_store.llm_calls) == 2
    tts_audit = audit_store.llm_calls[0]
    asr_audit = audit_store.llm_calls[1]
    assert tts_audit["job_type"] == "tts"
    assert tts_audit["queue_ms"] == 200
    assert tts_audit["metadata"]["duration_ms"] == 1200
    assert tts_audit["metadata"]["upstream_request_id"] == "tts-upstream"
    assert asr_audit["job_type"] == "asr"
    assert asr_audit["metadata"]["language"] == "zh"
    assert asr_audit["metadata"]["upstream_request_id"] == "asr-upstream"