import asyncio
import time
import json
import uuid
import os
import websockets
import gzip
import random
from urllib import parse
from tabulate import tabulate
from config.settings import load_config
import tempfile
import wave
import hmac
import base64
import hashlib
from datetime import datetime
from wsgiref.handlers import format_date_time
from time import mktime
description = "流式ASR首词延迟测试"
try:
    import dashscope
except ImportError:
    dashscope = None

class BaseASRTester:
    def __init__(self, config_key: str, config):
        self.config = config
        self.config_key = config_key
        self.asr_config = self.config.get("ASR", {}).get(config_key, {})
        self.test_audio_files = self._load_test_audio_files()
        self.results = []

    def _load_test_audio_files(self):
        audio_root = os.path.join(os.getcwd(), "config", "assets")
        test_files = []
        if os.path.exists(audio_root):
            for file_name in os.listdir(audio_root):
                if file_name.endswith(('.wav', '.pcm')):
                    file_path = os.path.join(audio_root, file_name)
                    with open(file_path, 'rb') as f:
                        test_files.append({
                            'data': f.read(),
                            'path': file_path,
                            'name': file_name
                        })
        return test_files

    async def test(self, test_count=5):
        raise NotImplementedError

    def _calculate_result(self, service_name, latencies, test_count):
        """计算测试结果（修复：正确处理None值，剔除失败测试）"""
        # 剔除None值（失败的测试）和无效延迟，只统计有效延迟
        valid_latencies = [l for l in latencies if l is not None and l > 0]
        if valid_latencies:
            avg_latency = sum(valid_latencies) / len(valid_latencies)
            status = f"成功（{len(valid_latencies)}/{test_count}次有效）"
        else:
            avg_latency = 0
            status = "失败: 所有测试均失败"
        return {"name": service_name, "latency": avg_latency, "status": status}


class DoubaoStreamASRTester(BaseASRTester):
    def __init__(self, config):
        super().__init__("DoubaoStreamASR", config)
        from core.providers.asr.doubao_stream import ASRProvider as DoubaoStreamProvider
        self.provider = DoubaoStreamProvider(self.asr_config, delete_audio_file=False)

    async def test(self, test_count=5):
        if not self.test_audio_files:
            return {"name": "豆包流式ASR", "latency": 0, "status": "失败: 未找到测试音频"}
        if not self.asr_config:
            return {"name": "豆包流式ASR", "latency": 0, "status": "失败: 未配置"}

        latencies = []
        for i in range(test_count):
            try:
                start_time = time.time()

                headers = self.provider.token_auth()

                async with websockets.connect(
                    self.provider.ws_url,
                    additional_headers=headers,
                    max_size=1000000000,
                    ping_interval=None,
                    ping_timeout=None,
                    close_timeout=10
                ) as ws:
                    request_params = self.provider.construct_request(str(uuid.uuid4()))

                    payload_bytes = str.encode(json.dumps(request_params))
                    payload_bytes = gzip.compress(payload_bytes)
                    full_client_request = bytearray(self.provider.generate_header())
                    full_client_request.extend((len(payload_bytes)).to_bytes(4, "big"))
                    full_client_request.extend(payload_bytes)
                    await ws.send(full_client_request)

                    init_res = await ws.recv()
                    result = self.provider.parse_response(init_res)
                    if "code" in result and result["code"] != 1000:
                        raise Exception(f"初始化失败: {result.get('payload_msg', {}).get('error', '未知错误')}")

                    audio_data = self.test_audio_files[0]['data']
                    if audio_data.startswith(b'RIFF'):
                        audio_data = audio_data[44:]

                    # 发送音频数据（使用最后一帧标记，告诉服务端音频已结束）
                    payload = gzip.compress(audio_data)
                    audio_request = bytearray(self.provider.generate_last_audio_default_header())
                    audio_request.extend(len(payload).to_bytes(4, "big"))
                    audio_request.extend(payload)
                    await ws.send(audio_request)

                    first_chunk = await ws.recv()
                    latency = time.time() - start_time
                    latencies.append(latency)
                    print(f"[豆包ASR] 第{i+1}次 首词延迟: {latency:.3f}s")
                    await ws.close()

            except Exception as e:
                print(f"[豆包ASR] 第{i+1}次测试失败: {str(e)}")
                latencies.append(None)

        return self._calculate_result("豆包流式ASR", latencies, test_count)


class QwenASRFlashTester(BaseASRTester):
    def __init__(self, config):
        super().__init__("Qwen3ASRFlash", config)

    async def _test_single(self, audio_file_info):
        temp_file_path = None

        try:
            audio_data = audio_file_info['data']

            # 优化：将临时文件准备工作移到计时前，减少磁盘IO对性能测试的影响
            with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as f:
                temp_file_path = f.name

            with wave.open(temp_file_path, 'wb') as wav_file:
                wav_file.setnchannels(1)
                wav_file.setsampwidth(2)
                wav_file.setframerate(16000)
                wav_file.writeframes(audio_data)

            messages = [
                {
                    "role": "user",
                    "content": [
                        {"audio": temp_file_path}
                    ]
                }
            ]

            api_key = self.asr_config.get("api_key") or os.getenv("DASHSCOPE_API_KEY")
            if not api_key:
                raise ValueError("未配置 api_key")

            if dashscope is None:
                raise RuntimeError("未安装 dashscope 库")

            dashscope.api_key = api_key

            # 统一计时起点：在API调用前开始计时（但文件准备已完成）
            start_time = time.time()

            response = dashscope.MultiModalConversation.call(
                model="qwen3-asr-flash",
                messages=messages,
                result_format="message",
                asr_options={"enable_lid": True, "enable_itn": False},
                stream=True
            )

            for chunk in response:
                latency = time.time() - start_time
                return latency

            raise Exception("流式结束，未收到任何响应")

        except Exception as e:
            raise Exception(f"通义ASR流式失败: {str(e)}")

        finally:
            if temp_file_path and os.path.exists(temp_file_path):
                try:
                    os.unlink(temp_file_path)
                except:
                    pass

    async def test(self, test_count=5):
        if not self.test_audio_files:
            return {"name": "通义千问ASR", "latency": 0, "status": "失败: 未找到测试音频"}
        if not self.asr_config and not os.getenv("DASHSCOPE_API_KEY"):
            return {"name": "通义千问ASR", "latency": 0, "status": "失败: 未配置 api_key"}

        latencies = []
        for i in range(test_count):
            try:
                # print(f"\n[通义ASR] 开始第 {i+1} 次测试...")
                latency = await self._test_single(self.test_audio_files[0])
                latencies.append(latency)
                print(f"[通义ASR] 第{i+1}次 首词延迟: {latency:.3f}s")
            except Exception as e:
                # print(f"[通义ASR] 第{i+1}次测试失败: {str(e)}")
                latencies.append(None)

        return self._calculate_result("通义千问ASR", latencies, test_count)


class XunfeiStreamASRTester(BaseASRTester):
    def __init__(self, config):
        super().__init__("XunfeiStreamASR", config)

    def _create_url(self):
        url = "wss://iat-api.xfyun.cn/v2/iat"
        now = datetime.now()
        date = format_date_time(mktime(now.timetuple()))

        signature_origin = f"host: iat-api.xfyun.cn\ndate: {date}\nGET /v2/iat HTTP/1.1"
        signature_sha = hmac.new(
            self.asr_config["api_secret"].encode('utf-8'),
            signature_origin.encode('utf-8'),
            hashlib.sha256
        ).digest()
        signature_sha = base64.b64encode(signature_sha).decode()

        authorization_origin = f'api_key="{self.asr_config["api_key"]}", algorithm="hmac-sha256", headers="host date request-line", signature="{signature_sha}"'
        authorization = base64.b64encode(authorization_origin.encode()).decode()

        v = {"authorization": authorization, "date": date, "host": "iat-api.xfyun.cn"}
        return url + "?" + parse.urlencode(v)

    async def test(self, test_count: int = 5):
        if not self.test_audio_files:
            return {"name": "讯飞流式ASR", "latency": 0, "status": "失败: 未找到测试音频"}
        if not self.asr_config:
            return {"name": "讯飞流式ASR", "latency": 0, "status": "失败: 未配置"}

        required = ["app_id", "api_key", "api_secret"]
        for k in required:
            if k not in self.asr_config:
                return {"name": "讯飞流式ASR", "latency": 0, "status": f"失败: 缺少配置 {k}"}

        latencies = []
        frame_size = 1280
        audio_raw = self.test_audio_files[0]['data']
        if audio_raw.startswith(b'RIFF'):
            audio_raw = audio_raw[44:]

        for i in range(test_count):
            try:
                start_time = time.time()
                ws_url = self._create_url()

                async with websockets.connect(
                    ws_url,
                    additional_headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"},
                    max_size=1 << 30,
                    ping_interval=None,
                    ping_timeout=None,
                    close_timeout=30,
                ) as ws:

                    # 第一帧：移除 punc 字段，避免未知参数错误
                    await ws.send(json.dumps({
                        "common": {"app_id": self.asr_config["app_id"]},
                        "business": {
                            "domain": "iat",
                            "language": "zh_cn",
                            "accent": "mandarin",
                            "dwa": "wpgs",
                            "vad_eos": 5000
                            # 已移除 "punc": True
                        },
                        "data": {
                            "status": 0,
                            "format": "audio/L16;rate=16000",
                            "encoding": "raw",
                            "audio": base64.b64encode(audio_raw[:frame_size]).decode()
                        }
                    }, ensure_ascii=False))

                    # 后续所有帧
                    pos = frame_size
                    while pos < len(audio_raw):
                        chunk = audio_raw[pos:pos + frame_size]
                        status = 2 if (pos + frame_size >= len(audio_raw)) else 1
                        await ws.send(json.dumps({
                            "data": {
                                "status": status,
                                "format": "audio/L16;rate=16000",
                                "encoding": "raw",
                                "audio": base64.b64encode(chunk).decode()
                            }
                        }, ensure_ascii=False))
                        if status == 2:
                            break
                        pos += frame_size

                    # 接收首词
                    first_token = True
                    async for message in ws:
                        data = json.loads(message)
                        if data.get("code") != 0:
                            raise Exception(f"讯飞错误: {data.get('message')}")

                        ws_result = data.get("data", {}).get("result", {}).get("ws")
                        if ws_result:
                            text = "".join(cw.get("w", "") for seg in ws_result for cw in seg.get("cw", []))
                            if text.strip() and first_token:
                                latency = time.time() - start_time
                                latencies.append(latency)
                                print(f"[讯飞ASR] 第{i+1}次 首词延迟: {latency:.3f}s")
                                first_token = False
                                break

            except Exception as e:
                print(f"[讯飞ASR] 第{i+1}次测试失败: {str(e)}")
                latencies.append(None)

        return self._calculate_result("讯飞流式ASR", latencies, test_count)
class ASRPerformanceSuite:
    def __init__(self, config):
        self.config = config
        self.testers = []
        self.results = []

    def register_tester(self, tester_class):
        try:
            tester = tester_class(self.config)
            self.testers.append(tester)
            print(f"已注册测试器: {tester.config_key}")
        except Exception as e:
            name_map = {
                "DoubaoStreamASRTester": "豆包流式ASR",
                "QwenASRFlashTester": "通义千问ASR",
                "XunfeiStreamASRTester": "讯飞流式ASR"
            }
            name = name_map.get(tester_class.__name__, tester_class.__name__)
            print(f"跳过 {name}: {str(e)}")

    def _print_results(self, test_count):
        if not self.results:
            print("没有有效的ASR测试结果")
            return

        print(f"\n{'='*60}")
        print("流式ASR首词响应时间测试结果")
        print(f"{'='*60}")
        print(f"测试次数: 每个ASR服务测试 {test_count} 次")

        success_results = sorted(
            [r for r in self.results if "成功" in r["status"]],
            key=lambda x: x["latency"]
        )
        failed_results = [r for r in self.results if "成功" not in r["status"]]

        table_data = [
            [r["name"], f"{r['latency']:.3f}s" if r['latency'] > 0 else "N/A", r["status"]]
            for r in success_results + failed_results
        ]

        print(tabulate(table_data, headers=["ASR服务", "首词延迟", "状态"], tablefmt="grid"))
        print("\n测试说明：")
        print("- 计时起点: 建立连接前（包含握手、发送音频、接收首个识别结果全流程）")
        print("- 通义千问优化: 临时文件准备在计时前完成，减少磁盘IO对测试的影响")
        print("- 错误处理: 失败的测试不计入平均值，只统计成功测试的延迟")
        print("- 排序规则: 成功的按延迟升序，失败的排在后面")

    async def run(self, test_count=5):
        print(f"开始流式ASR首词响应时间测试...")
        print(f"每个ASR服务测试次数: {test_count}次\n")

        self.results = []
        for tester in self.testers:
            print(f"\n--- 测试 {tester.config_key} ---")
            result = await tester.test(test_count)
            self.results.append(result)

        self._print_results(test_count)


async def main():
    import argparse
    parser = argparse.ArgumentParser(description="流式ASR首词响应时间测试工具")
    parser.add_argument("--count", type=int, default=5, help="测试次数")
    args = parser.parse_args()

    config = await load_config()
    suite = ASRPerformanceSuite(config)
    suite.register_tester(DoubaoStreamASRTester)
    suite.register_tester(QwenASRFlashTester)
    suite.register_tester(XunfeiStreamASRTester)

    await suite.run(args.count)


if __name__ == "__main__":
    asyncio.run(main())