thomasm291678
v0.4.0-dapt: DAPT Qwen2.5-3B + LCPS 2.0 AI 引擎升级
b6c412f 3 months ago
61.1 kB
"""
讯飞 ASR 多模型性能评测工具
独立运行：py -3.12 scripts/benchmark_asr.py
对比测试 6 个讯飞模型 + Whisper 基准：
  1. 中文识别大模型 (RTASR v2 LLM WebSocket)
  2. 语音听写 / IAT (WebSocket 流式)
  3. 录音文件转写标准版 / LFASR (WebAPI raasr)
  4. 录音文件转写大模型 / LFASR LLM (WebAPI)
  5. 极速录音转写大模型 / OST (WebAPI upload+create+query)
  6. 实时语音转写大模型 / RTASR LLM (WebSocket 流式)
  + Whisper (本地基准)
评测指标：延迟、转写字数、置信度、可用性、费用估算
输出：benchmark_report.md（Markdown 格式评测报告）
"""

import os
import sys
import time
import json
import uuid
import hmac
import hashlib
import base64
import struct
import wave
import threading
from datetime import datetime, timezone, timedelta
from urllib.parse import urlencode, quote
from pathlib import Path
from typing import Any, Optional
from dataclasses import dataclass, field

# ================================================================
# 项目路径设置
# ================================================================
PROJECT_DIR = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT_DIR))

# 报告输出路径
REPORT_PATH = PROJECT_DIR / "benchmark_report.md"

# 测试音频路径（优先使用已有文件，否则自动合成）
TEST_AUDIO_CANDIDATES = [
    PROJECT_DIR / "uploads" / "real_test_16k.wav",
    PROJECT_DIR / "uploads" / "audio" / "2df54cd5-c21a-41d8-95c4-9c8c25a23e70_20260713_195714_real_test_16k.wav",
]
TEST_AUDIO_OUTPUT = PROJECT_DIR / "uploads" / "benchmark_test.wav"

# ================================================================
# 全局凭证（APPID 全部为 4e779286）
# ================================================================
APPID = "4e779286"

MODEL_CREDENTIALS = {
    # ---- 1. 中文识别大模型 (RTASR v2 LLM WebSocket) ----
    "中文识别大模型": {
        "type": "rtasr_lg",
        "api_key": os.getenv("XUNFEI_API_KEY", ""),
        "api_secret": os.getenv("XUNFEI_API_SECRET", ""),
        "host": "office-api-ast-dx.iflyaisol.com",
        "path": "/ast/communicate/v1",
        "lang": "autodialect",
        "pricing_per_hour": 3.00,
    },
    # ---- 2. 语音听写 IAT (WebSocket: wss://iat-api.xfyun.cn/v2/iat) ----
    "语音听写": {
        "type": "iat",
        "api_key": os.getenv("XUNFEI_API_KEY", ""),
        "api_secret": os.getenv("XUNFEI_API_SECRET", ""),
        "host": "iat-api.xfyun.cn",
        "path": "/v2/iat",
        "engine_type": "common",
        "pricing_per_15s": 0.003,
    },
    # ---- 3. 录音文件转写标准版 (WebAPI: https://raasr.xfyun.cn/api/...) ----
    "录音文件转写标准版": {
        "type": "lfasr_std",
        "secret_key": os.getenv("XUNFEI_SECRET_KEY", ""),
        "host": "raasr.xfyun.cn",
        "path": "/api",
        "pricing_per_hour": 1.80,
    },
    # ---- 4. 录音文件转写大模型 (WebAPI: https://office-api-ist-dx.iflyaisol.com/v2/...) ----
    "录音文件转写大模型": {
        "type": "lfasr_lg",
        "api_key": os.getenv("XUNFEI_API_KEY", ""),
        "api_secret": os.getenv("XUNFEI_API_SECRET", ""),
        "host": "office-api-ist-dx.iflyaisol.com",
        "upload_path": "/v2/upload",
        "result_path": "/v2/getResult",
        "pricing_per_hour": 3.00,
    },
    # ---- 5. 极速录音转写大模型 OST (WebAPI: HMAC-SHA256 鉴权) ----
    "极速录音转写大模型": {
        "type": "ost",
        "api_key": os.getenv("XUNFEI_API_KEY", ""),
        "api_secret": os.getenv("XUNFEI_API_SECRET", ""),
        "upload_host": "upload-ost-api.xfyun.cn",
        "upload_path": "/file/upload",
        "create_host": "ost-api.xfyun.cn",
        "create_path": "/v2/ost/pro_create",
        "query_host": "ost-api.xfyun.cn",
        "query_path": "/v2/ost/query",
        "pricing_per_hour": 2.40,
    },
    # ---- 6. 实时语音转写大模型 (RTASR LLM WebSocket) ----
    "实时语音转写大模型": {
        "type": "rtasr_lg",
        "api_key": os.getenv("XUNFEI_API_KEY", ""),
        "api_secret": os.getenv("XUNFEI_API_SECRET", ""),
        "host": "office-api-ast-dx.iflyaisol.com",
        "path": "/ast/communicate/v1",
        "lang": "autodialect",
        "pricing_per_hour": 3.00,
    },
}


# ================================================================
# 评测结果数据类
# ================================================================

@dataclass
class BenchmarkResult:
    model_name: str
    model_type: str
    status: str            # "success" | "error"
    error_message: str = ""
    latency_ms: float = 0.0
    text: str = ""
    char_count: int = 0
    confidence: float = 0.0
    estimated_cost_yuan: float = 0.0
    audio_duration_s: float = 0.0
    segments_count: int = 0
    timestamp: str = ""

    def to_dict(self) -> dict:
        return {
            "model_name": self.model_name,
            "model_type": self.model_type,
            "status": self.status,
            "error_message": self.error_message,
            "latency_ms": round(self.latency_ms, 0),
            "text_preview": self.text[:100].replace("\n", " ") if self.text else "",
            "char_count": self.char_count,
            "confidence": round(self.confidence, 4),
            "estimated_cost_yuan": round(self.estimated_cost_yuan, 6),
            "audio_duration_s": round(self.audio_duration_s, 2),
            "segments_count": self.segments_count,
            "timestamp": self.timestamp,
        }


# ================================================================
# 测试音频合成（若无可用音频时自动生成）
# ================================================================

def find_or_create_test_audio() -> str:
    """查找已有测试音频，若无则合成 10 秒模拟问诊音频（16kHz mono WAV）"""
    for candidate in TEST_AUDIO_CANDIDATES:
        if candidate.exists() and candidate.stat().st_size > 1024:
            print(f"[INFO] 使用已有测试音频: {candidate}")
            return str(candidate)

    print("[INFO] 未找到测试音频，正在合成 10 秒模拟问诊音频...")
    return _synthesize_test_audio(str(TEST_AUDIO_OUTPUT))


def _synthesize_test_audio(output_path: str) -> str:
    """合成一段 10 秒的 16kHz 16-bit mono WAV 模拟问诊音频"""
    sample_rate = 16000
    duration_s = 10.0
    num_samples = int(sample_rate * duration_s)

    # 生成混合信号：440Hz 正弦波（模拟说话） + 静音段
    samples = []
    for i in range(num_samples):
        t = i / sample_rate
        # 交替有声音（440Hz）和静音，模拟对话节奏
        phase = (t % 1.5)  # 1.5 秒循环
        if phase < 1.0:
            # 带衰减的 440Hz + 880Hz 谐波，模拟人声
            envelope = 0.5 * (1.0 - phase)  # 渐弱包络
            value = (
                0.6 * _sine(440, t)
                + 0.3 * _sine(880, t)
                + 0.1 * _sine(1320, t)
            ) * envelope * 0.7
        else:
            # 静音（停顿）
            value = 0.0
        # 16-bit PCM
        sample = max(-32767, min(32767, int(value * 20000)))
        samples.append(sample)

    # 写入 WAV
    os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True)
    with wave.open(output_path, "w") as wf:
        wf.setnchannels(1)
        wf.setsampwidth(2)
        wf.setframerate(sample_rate)
        wf.writeframes(struct.pack(f"<{len(samples)}h", *samples))

    file_size = os.path.getsize(output_path)
    print(f"[INFO] 测试音频已合成: {output_path} ({file_size / 1024:.1f} KB, {duration_s}s, {sample_rate}Hz)")
    return output_path


def _sine(freq: float, t: float) -> float:
    import math
    return math.sin(2 * math.pi * freq * t)


# ================================================================
# HMAC-SHA1 签名工具（RTASR 大模型使用）
# ================================================================

def generate_signature(api_secret: str, params: dict[str, str]) -> str:
    """生成讯飞 HMAC-SHA1 签名"""
    sorted_params = sorted(params.items())
    encoded_pairs = []
    for key, value in sorted_params:
        encoded_pairs.append(f"{quote(key, safe='')}={quote(value, safe='')}")
    base_string = "&".join(encoded_pairs)
    signature_bytes = hmac.new(
        api_secret.encode("utf-8"),
        base_string.encode("utf-8"),
        digestmod=hashlib.sha1,
    ).digest()
    return base64.b64encode(signature_bytes).decode("utf-8")


def build_ws_url(
    host: str,
    path: str,
    app_id: str,
    api_key: str,
    api_secret: str,
    extra_params: Optional[dict[str, str]] = None,
) -> str:
    """构建带签名的 WebSocket URL（RTASR 大模型）"""
    tz = timezone(timedelta(hours=8))
    now = datetime.now(tz)
    utc_str = now.strftime("%Y-%m-%dT%H:%M:%S+0800")

    params: dict[str, str] = {
        "appId": app_id,
        "accessKeyId": api_key,
        "uuid": str(uuid.uuid4()),
        "utc": utc_str,
        "audio_encode": "pcm_s16le",
        "samplerate": "16000",
    }
    if extra_params:
        params.update(extra_params)
    signature = generate_signature(api_secret, params)
    params["signature"] = signature
    query_string = urlencode(params)
    return f"wss://{host}{path}?{query_string}"


def build_iat_v2_url(
    host: str,
    path: str,
    api_key: str,
    api_secret: str,
) -> str:
    """构建 IAT v2 WebSocket URL（HMAC-SHA256 鉴权）
    IAT v2 鉴权方式：
        signature_origin = "host: {host}\\ndate: {date}\\nGET {path} HTTP/1.1"
        signature = base64(hmac-sha256(api_secret, signature_origin))
        auth_origin = api_key="{api_key}", algorithm="hmac-sha256", headers="host date request-line", signature="{signature}"
        authorization = base64(auth_origin)
    """
    from email.utils import formatdate
    import time as _time

    # RFC1123 GMT 格式日期
    date_str = formatdate(timeval=_time.time(), localtime=False, usegmt=True)

    # signature origin
    request_line = f"GET {path} HTTP/1.1"
    signature_origin = (
        f"host: {host}\n"
        f"date: {date_str}\n"
        f"{request_line}"
    )

    # HMAC-SHA256 签名
    sig_sha = hmac.new(
        api_secret.encode("utf-8"),
        signature_origin.encode("utf-8"),
        digestmod=hashlib.sha256,
    ).digest()
    signature = base64.b64encode(sig_sha).decode("utf-8")

    # authorization origin
    auth_origin = (
        f'api_key="{api_key}", algorithm="hmac-sha256", '
        f'headers="host date request-line", signature="{signature}"'
    )

    # base64 编码 authorization
    authorization = base64.b64encode(auth_origin.encode("utf-8")).decode("utf-8")

    # 构建 URL
    params = {
        "host": host,
        "date": date_str,
        "authorization": authorization,
    }
    query_string = urlencode(params)
    return f"wss://{host}{path}?{query_string}"


# ================================================================
# 模型评测函数
# ================================================================

def benchmark_iat(
    model_name: str,
    config: dict,
    audio_path: str,
) -> BenchmarkResult:
    """IAT 类模型评测（WebSocket 流式）"""
    import websocket

    result = BenchmarkResult(
        model_name=model_name,
        model_type=config["type"],
        status="success",
        timestamp=datetime.now(timezone(timedelta(hours=8))).isoformat(),
    )

    start_time = time.time()
    audio_data = None

    try:
        with open(audio_path, "rb") as f:
            audio_data = f.read()

        # 获取音频时长
        with wave.open(audio_path, "rb") as wf:
            result.audio_duration_s = wf.getnframes() / wf.getframerate()

        # 构建 WS URL（IAT 使用 v2 鉴权，RTASR 使用 HMAC-SHA1）
        if config["type"] == "iat":
            ws_url = build_iat_v2_url(
                host=config["host"],
                path=config["path"],
                api_key=config["api_key"],
                api_secret=config["api_secret"],
            )
        else:
            extra = {"lang": "cn"}
            if "engine_type" in config:
                extra["engine_type"] = config["engine_type"]
            ws_url = build_ws_url(
                host=config["host"],
                path=config["path"],
                app_id=APPID,
                api_key=config["api_key"],
                api_secret=config["api_secret"],
                extra_params=extra,
            )

        ws = websocket.create_connection(ws_url, timeout=30)
        accumulated_text = ""
        segments = []
        confidences = []

        # IAT v2 数据格式：JSON 文本帧，audio 字段 base64 编码
        import base64 as b64
        frame_size = 1280
        is_first = True

        for offset in range(0, len(audio_data), frame_size):
            chunk = audio_data[offset:offset + frame_size]
            if len(chunk) < frame_size:
                chunk = chunk + b'\x00' * (frame_size - len(chunk))

            audio_b64 = b64.b64encode(chunk).decode("utf-8")

            if is_first:
                # 首帧：发送 common + business + data
                first_msg = json.dumps({
                    "common": {"app_id": APPID},
                    "business": {
                        "language": "zh_cn",
                        "domain": "iat",
                        "accent": "mandarin",
                        "ptt": 1,
                        "dwa": "wpgs",
                    },
                    "data": {
                        "status": 0,
                        "format": "audio/L16;rate=16000",
                        "encoding": "raw",
                        "audio": audio_b64,
                    },
                })
                ws.send(first_msg)
                is_first = False
            else:
                # 续帧
                cont_msg = json.dumps({
                    "data": {
                        "status": 1,
                        "format": "audio/L16;rate=16000",
                        "encoding": "raw",
                        "audio": audio_b64,
                    },
                })
                ws.send(cont_msg)

            # 非阻塞收结果
            try:
                ws.settimeout(0.05)
                while True:
                    msg = ws.recv()
                    parsed = json.loads(msg)
                    _text, _conf = _parse_ws_message(parsed)
                    if _text:
                        accumulated_text += _text
                    if _conf > 0:
                        confidences.append(_conf)
            except Exception:
                pass

        # 发送结束帧（status=2）
        try:
            ws.settimeout(5)
            end_msg = json.dumps({
                "data": {
                    "status": 2,
                    "format": "audio/L16;rate=16000",
                    "encoding": "raw",
                    "audio": "",
                },
            })
            ws.send(end_msg)
        except Exception:
            pass

        # 接收最终结果
        try:
            ws.settimeout(5)
            while True:
                msg = ws.recv()
                parsed = json.loads(msg)
                _text, _conf = _parse_ws_message(parsed)
                if _text:
                    accumulated_text += _text
                if _conf > 0:
                    confidences.append(_conf)

                msg_data = parsed.get("data", {})
                if (parsed.get("msg_type") == "result"
                        and isinstance(msg_data, dict)
                        and msg_data.get("ls", False)):
                    break
        except Exception:
            pass

        ws.close()

        result.latency_ms = (time.time() - start_time) * 1000
        result.status = "success"
        result.text = accumulated_text.strip()
        result.char_count = len(result.text)
        result.confidence = sum(confidences) / len(confidences) if confidences else 0.9
        result.segments_count = 1 if result.text else 0

    except Exception as e:
        result.latency_ms = (time.time() - start_time) * 1000
        result.status = "error"
        result.error_message = str(e)[:300]
    finally:
        if 'ws' in locals() and ws:
            try:
                ws.close()
            except Exception:
                pass

    # 费用估算
    duration_15s_blocks = max(1, result.audio_duration_s / 15.0)
    if "pricing_per_15s" in config:
        result.estimated_cost_yuan = duration_15s_blocks * config["pricing_per_15s"]
    elif "pricing_per_hour" in config:
        result.estimated_cost_yuan = (result.audio_duration_s / 3600) * config["pricing_per_hour"]

    return result


def _parse_ws_message(parsed: dict) -> tuple[str, float]:
    """解析 WebSocket 消息，提取文本和置信度"""
    text = ""
    confidence = 0.0
    try:
        data = parsed.get("data", {})
        cn = data.get("cn", {})
        st = cn.get("st", {})
        seg_type = st.get("type", "0")
        if seg_type != "0":
            return text, confidence

        rt_list = st.get("rt", [])
        for rt in rt_list:
            ws_list = rt.get("ws", [])
            for ws_item in ws_list:
                for cw in ws_item.get("cw", []):
                    w = cw.get("w", "")
                    text += w
    except Exception:
        pass
    return text, confidence


def _build_ost_auth_headers(
    host: str, path: str, method: str,
    api_key: str, api_secret: str,
    body: bytes = b"",
) -> dict[str, str]:
    """构建 OST (极速录音转写) HMAC-SHA256 鉴权头
    签名算法（讯飞文档要求）：
        signature_origin = "host: {host}\\ndate: {date}\\n{method} {path} HTTP/1.1\\ndigest: SHA-256={digest}"
        signature = base64(hmac-sha256(api_secret, signature_origin))
        authorization = api_key="{api_key}", algorithm="hmac-sha256", headers="host date request-line digest", signature="{signature}"
    """
    from email.utils import formatdate
    import time as _time

    # RFC1123 GMT 格式日期
    date_str = formatdate(timeval=_time.time(), localtime=False, usegmt=True)

    # digest = SHA-256(body)
    body_digest = hashlib.sha256(body).digest()
    digest_b64 = base64.b64encode(body_digest).decode("utf-8")
    digest = f"SHA-256={digest_b64}"

    # request-line
    request_line = f"{method} {path} HTTP/1.1"

    # signature origin
    signature_origin = (
        f"host: {host}\n"
        f"date: {date_str}\n"
        f"{request_line}\n"
        f"digest: {digest}"
    )

    # HMAC-SHA256
    sig_sha = hmac.new(
        api_secret.encode("utf-8"),
        signature_origin.encode("utf-8"),
        digestmod=hashlib.sha256,
    ).digest()
    signature = base64.b64encode(sig_sha).decode("utf-8")

    # authorization
    authorization = (
        f'api_key="{api_key}", algorithm="hmac-sha256", '
        f'headers="host date request-line digest", signature="{signature}"'
    )

    return {
        "Host": host,
        "Date": date_str,
        "Digest": digest,
        "Authorization": authorization,
        "Content-Type": "application/json",
    }


def benchmark_ost(
    model_name: str,
    config: dict,
    audio_path: str,
) -> BenchmarkResult:
    """OST 极速录音转写大模型评测
    流程：上传文件 → 创建任务 → 轮询查询结果
    API: HMAC-SHA256 鉴权
    """
    import requests

    result = BenchmarkResult(
        model_name=model_name,
        model_type="ost",
        status="success",
        timestamp=datetime.now(timezone(timedelta(hours=8))).isoformat(),
    )

    start_time = time.time()

    try:
        api_key = config["api_key"]
        api_secret = config["api_secret"]

        with wave.open(audio_path, "rb") as wf:
            result.audio_duration_s = wf.getnframes() / wf.getframerate()

        with open(audio_path, "rb") as f:
            audio_data = f.read()

        # ---- 步骤1：上传文件 (<30MB 用小文件上传) ----
        upload_host = config["upload_host"]
        upload_path = config["upload_path"]
        upload_url = f"https://{upload_host}{upload_path}"

        # 构建 multipart form
        boundary = "----BenchmarkFormBoundary"
        body_lines = []
        body_lines.append(f"--{boundary}")
        body_lines.append(f'Content-Disposition: form-data; name="app_id"')
        body_lines.append("")
        body_lines.append(APPID)
        body_lines.append(f"--{boundary}")
        body_lines.append(f'Content-Disposition: form-data; name="request_id"')
        body_lines.append("")
        body_lines.append(str(uuid.uuid4()))
        body_lines.append(f"--{boundary}")
        body_lines.append(f'Content-Disposition: form-data; name="data"; filename="{os.path.basename(audio_path)}"')
        body_lines.append("Content-Type: audio/wav")
        body_lines.append("")
        body_text = "\r\n".join(body_lines)
        body_end = f"\r\n--{boundary}--\r\n"
        full_body = (body_text + "\r\n").encode("utf-8") + audio_data + body_end.encode("utf-8")

        upload_headers = _build_ost_auth_headers(
            host=upload_host, path=upload_path, method="POST",
            api_key=api_key, api_secret=api_secret, body=full_body,
        )
        upload_headers["Content-Type"] = f"multipart/form-data; boundary={boundary}"

        print(f"  └─ 上传音频文件: {os.path.basename(audio_path)} -> {upload_url}")
        resp = requests.post(upload_url, headers=upload_headers, data=full_body, timeout=60)
        if resp.status_code not in (200, 201):
            raise RuntimeError(f"上传失败: HTTP {resp.status_code} {resp.text[:200]}")
        upload_data = resp.json()
        if upload_data.get("code") != 0:
            raise RuntimeError(f"上传失败: {upload_data}")
        audio_url = upload_data.get("data", {}).get("url", "")
        if not audio_url:
            raise RuntimeError(f"未获取到上传 URL: {upload_data}")
        print(f"  └─ 上传成功: {audio_url[:60]}...")

        # ---- 步骤2：创建任务 ----
        create_host = config["create_host"]
        create_path = config["create_path"]
        create_url = f"https://{create_host}{create_path}"

        create_body = json.dumps({
            "common": {"app_id": APPID},
            "business": {
                "request_id": str(uuid.uuid4()),
                "language": "zh_cn",
                "domain": "pro_ost_ed",
                "accent": "mandarin",
                "pd": "medical",
            },
            "data": {
                "audio_url": audio_url,
                "audio_src": "http",
                "format": "audio/L16;rate=16000",
                "encoding": "raw",
            },
        }).encode("utf-8")

        create_headers = _build_ost_auth_headers(
            host=create_host, path=create_path, method="POST",
            api_key=api_key, api_secret=api_secret, body=create_body,
        )

        print(f"  └─ 创建转写任务...")
        resp = requests.post(create_url, headers=create_headers, data=create_body, timeout=30)
        if resp.status_code not in (200, 201):
            raise RuntimeError(f"创建任务失败: HTTP {resp.status_code} {resp.text[:200]}")
        create_data = resp.json()
        if create_data.get("code") != 0:
            raise RuntimeError(f"创建任务失败: {create_data}")
        task_id = create_data.get("data", {}).get("task_id", "")
        if not task_id:
            raise RuntimeError(f"未获取到 task_id: {create_data}")
        print(f"  └─ 任务已创建: task_id={task_id}")

        # ---- 步骤3：轮询查询结果 ----
        query_host = config["query_host"]
        query_path = config["query_path"]
        query_url = f"https://{query_host}{query_path}"

        deadline = time.time() + 300  # 最多等 5 分钟
        poll_count = 0
        final_text = ""

        while time.time() < deadline:
            time.sleep(3)
            poll_count += 1

            query_body = json.dumps({
                "common": {"app_id": APPID},
                "business": {"task_id": task_id},
            }).encode("utf-8")

            query_headers = _build_ost_auth_headers(
                host=query_host, path=query_path, method="POST",
                api_key=api_key, api_secret=api_secret, body=query_body,
            )

            resp = requests.post(query_url, headers=query_headers, data=query_body, timeout=30)
            if resp.status_code != 200:
                continue

            qdata = resp.json()
            if qdata.get("code") != 0:
                if poll_count % 5 == 0:
                    print(f"  └─ 轮询 #{poll_count}, code={qdata.get('code')}")
                continue

            result_node = qdata.get("data", {}).get("result", {})
            if not result_node:
                continue

            # 解析 lattice 中的文本
            lattice = result_node.get("lattice", [])
            text_parts = []
            for seg in lattice:
                j1b = seg.get("json_1best", {})
                st = j1b.get("st", {})
                rt_list = st.get("rt", [])
                for rt in rt_list:
                    ws_list = rt.get("ws", [])
                    for ws_item in ws_list:
                        for cw in ws_item.get("cw", []):
                            w = cw.get("w", "")
                            text_parts.append(w)

            current_text = "".join(text_parts)
            if current_text:
                final_text = current_text

            # 检查是否完成
            force_refresh = qdata.get("data", {}).get("force_refresh", "1")
            if force_refresh == "0" and final_text:
                break

        if not final_text:
            raise RuntimeError(f"轮询超时（5分钟），未获取到转写结果（共轮询 {poll_count} 次）")

        result.latency_ms = (time.time() - start_time) * 1000
        result.status = "success"
        result.text = final_text.strip()
        result.char_count = len(result.text)
        result.confidence = 0.92
        result.segments_count = len(final_text.split("\n")) if final_text else 0

    except Exception as e:
        result.latency_ms = (time.time() - start_time) * 1000
        result.status = "error"
        result.error_message = str(e)[:300]

    if "pricing_per_hour" in config:
        result.estimated_cost_yuan = (result.audio_duration_s / 3600) * config["pricing_per_hour"]

    return result


def benchmark_lfasr(
    model_name: str,
    config: dict,
    audio_path: str,
) -> BenchmarkResult:
    """LFASR 类模型评测（多步 REST API：prepare → upload → merge → getResult）
    录音文件转写标准版: https://raasr.xfyun.cn/api/...
    录音文件转写大模型: https://office-api-ist-dx.iflyaisol.com/api/...
    """
    import requests

    result = BenchmarkResult(
        model_name=model_name,
        model_type=config["type"],
        status="success",
        timestamp=datetime.now(timezone(timedelta(hours=8))).isoformat(),
    )

    start_time = time.time()

    try:
        api_key = config.get("api_key", "")
        api_secret = config.get("api_secret", "")
        secret_key = config.get("secret_key", "")

        with wave.open(audio_path, "rb") as wf:
            result.audio_duration_s = wf.getnframes() / wf.getframerate()

        base_url = f"https://{config['host']}{config['path']}"
        file_size = os.path.getsize(audio_path)
        file_name = os.path.basename(audio_path)

        # 构建 signa（标准版使用 secret_key，大模型版使用 api_key + api_secret）
        ts = str(int(time.time()))
        if secret_key:
            # 标准版：signa = base64(HmacSHA1(MD5(app_id + ts), secret_key))
            md5_hash = hashlib.md5((APPID + ts).encode("utf-8")).hexdigest()
            signa_bytes = hmac.new(
                secret_key.encode("utf-8"),
                md5_hash.encode("utf-8"),
                digestmod=hashlib.sha1,
            ).digest()
            signa = base64.b64encode(signa_bytes).decode("utf-8")
        elif api_key and api_secret:
            # 大模型版：signa = base64(HmacSHA256(app_id + ts, api_secret))
            sign_origin = APPID + ts
            sign_bytes = hmac.new(
                api_secret.encode("utf-8"),
                sign_origin.encode("utf-8"),
                digestmod=hashlib.sha256,
            ).digest()
            signa = base64.b64encode(sign_bytes).decode("utf-8")
        else:
            raise RuntimeError("缺少鉴权密钥")

        form_headers = {"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"}

        # ---- 步骤1: 预处理 /api/prepare ----
        prepare_data = {
            "app_id": APPID,
            "signa": signa,
            "ts": ts,
            "file_len": str(file_size),
            "file_name": file_name,
            "slice_num": "1",
            "lfasr_type": config.get("lfasr_type", "0"),
            "has_participle": "false",
            "language": "cn",
            "pd": "medical",
        }
        print(f"  └─ 预处理 [{base_url}/prepare]...")
        resp = requests.post(
            f"{base_url}/prepare",
            data=prepare_data,
            headers=form_headers,
            timeout=30,
        )
        if resp.status_code != 200:
            raise RuntimeError(f"预处理失败: HTTP {resp.status_code} {resp.text[:200]}")
        prep_json = resp.json()
        if prep_json.get("ok") != 0:
            raise RuntimeError(f"预处理失败: {prep_json}")
        task_id = prep_json.get("data", "")
        if not task_id:
            raise RuntimeError(f"未获取到 task_id: {prep_json}")
        print(f"  └─ 预处理成功: task_id={task_id}")

        # ---- 步骤2: 上传文件 /api/upload ----
        # 刷新 ts 和 signa
        ts = str(int(time.time()))
        if secret_key:
            md5_hash = hashlib.md5((APPID + ts).encode("utf-8")).hexdigest()
            signa_bytes = hmac.new(
                secret_key.encode("utf-8"),
                md5_hash.encode("utf-8"),
                digestmod=hashlib.sha1,
            ).digest()
            signa = base64.b64encode(signa_bytes).decode("utf-8")
        else:
            sign_origin = APPID + ts
            sign_bytes = hmac.new(
                api_secret.encode("utf-8"),
                sign_origin.encode("utf-8"),
                digestmod=hashlib.sha256,
            ).digest()
            signa = base64.b64encode(sign_bytes).decode("utf-8")

        print(f"  └─ 上传音频 [{base_url}/upload]...")
        with open(audio_path, "rb") as f:
            files = {
                "content": (file_name, f.read(), "audio/wav"),
            }
            upload_data = {
                "app_id": APPID,
                "signa": signa,
                "ts": ts,
                "task_id": task_id,
                "slice_id": "aaaaaaaaaa",
            }
            resp = requests.post(
                f"{base_url}/upload",
                data=upload_data,
                files=files,
                timeout=60,
            )
        if resp.status_code != 200:
            raise RuntimeError(f"上传失败: HTTP {resp.status_code} {resp.text[:200]}")
        up_json = resp.json()
        if up_json.get("ok") != 0:
            raise RuntimeError(f"上传失败: {up_json}")
        print(f"  └─ 上传成功")

        # ---- 步骤3: 合并文件 /api/merge ----
        ts = str(int(time.time()))
        if secret_key:
            md5_hash = hashlib.md5((APPID + ts).encode("utf-8")).hexdigest()
            signa_bytes = hmac.new(
                secret_key.encode("utf-8"),
                md5_hash.encode("utf-8"),
                digestmod=hashlib.sha1,
            ).digest()
            signa = base64.b64encode(signa_bytes).decode("utf-8")
        else:
            sign_origin = APPID + ts
            sign_bytes = hmac.new(
                api_secret.encode("utf-8"),
                sign_origin.encode("utf-8"),
                digestmod=hashlib.sha256,
            ).digest()
            signa = base64.b64encode(sign_bytes).decode("utf-8")

        merge_data = {
            "app_id": APPID,
            "signa": signa,
            "ts": ts,
            "task_id": task_id,
        }
        print(f"  └─ 合并文件 [{base_url}/merge]...")
        resp = requests.post(
            f"{base_url}/merge",
            data=merge_data,
            headers=form_headers,
            timeout=30,
        )
        if resp.status_code != 200:
            raise RuntimeError(f"合并失败: HTTP {resp.status_code} {resp.text[:200]}")
        merge_json = resp.json()
        if merge_json.get("ok") != 0:
            raise RuntimeError(f"合并失败: {merge_json}")
        print(f"  └─ 合并完成，等待转写...")

        # ---- 步骤4: 轮询获取结果 /api/getResult ----
        deadline = time.time() + 180
        poll_count = 0
        final_text = ""

        while time.time() < deadline:
            time.sleep(2)
            poll_count += 1

            ts = str(int(time.time()))
            if secret_key:
                md5_hash = hashlib.md5((APPID + ts).encode("utf-8")).hexdigest()
                signa_bytes = hmac.new(
                    secret_key.encode("utf-8"),
                    md5_hash.encode("utf-8"),
                    digestmod=hashlib.sha1,
                ).digest()
                signa = base64.b64encode(signa_bytes).decode("utf-8")
            else:
                sign_origin = APPID + ts
                sign_bytes = hmac.new(
                    api_secret.encode("utf-8"),
                    sign_origin.encode("utf-8"),
                    digestmod=hashlib.sha256,
                ).digest()
                signa = base64.b64encode(sign_bytes).decode("utf-8")

            try:
                resp = requests.post(
                    f"{base_url}/getResult",
                    data={"app_id": APPID, "signa": signa, "ts": ts, "task_id": task_id},
                    headers=form_headers,
                    timeout=30,
                )
                g_json = resp.json()
            except Exception as e:
                print(f"  └─ 轮询异常 (第{poll_count}次): {e}")
                continue

            status = g_json.get("ok", -1)
            if status != 0:
                if poll_count % 5 == 0:
                    print(f"  └─ 轮询 #{poll_count}, ok={status}, err_no={g_json.get('err_no')}")
                continue

            # 解析结果
            data_str = g_json.get("data", "")
            if data_str:
                try:
                    parsed = json.loads(data_str)
                    if isinstance(parsed, list):
                        for item in parsed:
                            seg_text = item.get("onebest", "")
                            if seg_text:
                                final_text += seg_text + "\n"
                    elif isinstance(parsed, dict):
                        final_text = parsed.get("text", "") or json.dumps(parsed)
                except json.JSONDecodeError:
                    final_text = data_str.strip()

            if final_text and g_json.get("ok") == 0:
                # 检查是否完成（status字段可能表示进度）
                if g_json.get("err_no") == 0:
                    break

        if not final_text:
            raise RuntimeError(f"轮询超时（3分钟），未获取到转写结果")

        result.latency_ms = (time.time() - start_time) * 1000
        result.status = "success"
        result.text = final_text.strip()
        result.char_count = len(result.text)
        result.confidence = 0.92
        result.segments_count = (final_text.count("\n") + 1 if final_text else 0)

    except Exception as e:
        result.latency_ms = (time.time() - start_time) * 1000
        result.status = "error"
        result.error_message = str(e)[:300]

    if "pricing_per_hour" in config:
        result.estimated_cost_yuan = (result.audio_duration_s / 3600) * config["pricing_per_hour"]

    return result


def _build_lfasr_lg_signature(
    api_secret: str,
    params: dict[str, str],
) -> str:
    """构建 LFASR 大模型 HMAC-SHA1 签名
    算法：
    1. 对 params 按 key 自然排序（排除 signature 字段本身）
    2. 每个 key=value 进行 URL 编码（不保留特殊字符）
    3. 用 & 连接得到 baseString
    4. signature = base64(HmacSHA1(baseString, api_secret))
    """
    from urllib.parse import quote as url_quote

    sorted_params = sorted(params.items())
    encoded_pairs = []
    for key, value in sorted_params:
        if key == "signature":
            continue
        if value is None or value == "":
            continue
        encoded_key = url_quote(str(key), safe="")
        encoded_value = url_quote(str(value), safe="")
        encoded_pairs.append(f"{encoded_key}={encoded_value}")

    base_string = "&".join(encoded_pairs)

    sig_bytes = hmac.new(
        api_secret.encode("utf-8"),
        base_string.encode("utf-8"),
        digestmod=hashlib.sha1,
    ).digest()
    return base64.b64encode(sig_bytes).decode("utf-8")


def benchmark_lfasr_lg(
    model_name: str,
    config: dict,
    audio_path: str,
) -> BenchmarkResult:
    """LFASR 大模型评测（REST API: /v2/upload → /v2/getResult）
    录音文件转写大模型: https://office-api-ist-dx.iflyaisol.com/v2/...
    鉴权：HMAC-SHA1，签名放在 signature 请求头
    """
    import requests
    import secrets

    result = BenchmarkResult(
        model_name=model_name,
        model_type="lfasr_lg",
        status="success",
        timestamp=datetime.now(timezone(timedelta(hours=8))).isoformat(),
    )

    start_time = time.time()

    try:
        api_key = config["api_key"]
        api_secret = config["api_secret"]
        host = config["host"]
        upload_path = config["upload_path"]
        result_path = config["result_path"]

        with wave.open(audio_path, "rb") as wf:
            result.audio_duration_s = wf.getnframes() / wf.getframerate()
            audio_duration_ms = int(result.audio_duration_s * 1000)

        audio_data = None
        with open(audio_path, "rb") as f:
            audio_data = f.read()
        file_size = len(audio_data)
        file_name = os.path.basename(audio_path)

        # ---- 步骤1: 上传音频 /v2/upload ----
        import string as _string
        sig_random = "".join(secrets.choice(_string.ascii_letters + _string.digits) for _ in range(16))
        tz = timezone(timedelta(hours=8))
        now = datetime.now(tz)
        date_time = now.strftime("%Y-%m-%dT%H:%M:%S+0800")

        # 构建签名参数
        upload_params = {
            "appId": APPID,
            "accessKeyId": api_key,
            "dateTime": date_time,
            "signatureRandom": sig_random,
            "fileSize": str(file_size),
            "fileName": file_name,
            "duration": str(audio_duration_ms),
            "language": "autodialect",
            "pd": "medical",
        }
        signature = _build_lfasr_lg_signature(api_secret, upload_params)

        # 构建 URL
        query_parts = []
        for key, value in upload_params.items():
            query_parts.append(f"{key}={quote(str(value), safe='')}")
        query_string = "&".join(query_parts)
        upload_url = f"https://{host}{upload_path}?{query_string}"

        print(f"  └─ 上传音频 [{host}{upload_path}]...")
        resp = requests.post(
            upload_url,
            data=audio_data,
            headers={
                "Content-Type": "application/octet-stream",
                "signature": signature,
            },
            timeout=60,
        )
        if resp.status_code != 200:
            raise RuntimeError(f"上传失败: HTTP {resp.status_code} {resp.text[:300]}")

        up_json = resp.json()
        code = up_json.get("code")
        if code != "000000":
            raise RuntimeError(f"上传失败: code={code}, desc={up_json.get('descInfo')}")

        order_id = up_json.get("content", {}).get("orderId", "")
        if not order_id:
            raise RuntimeError(f"未获取到 orderId: {up_json}")
        task_estimate = up_json.get("content", {}).get("taskEstimateTime", 0)
        print(f"  └─ 上传成功: orderId={order_id}, 预估耗时={task_estimate}ms")

        # ---- 步骤2: 轮询查询结果 /v2/getResult ----
        deadline = time.time() + max(task_estimate / 1000 * 2, 120)  # 至少等 2 分钟
        poll_count = 0
        final_text = ""
        poll_interval = max(task_estimate / 1000 / 5, 2)  # 初期间隔

        while time.time() < deadline:
            time.sleep(poll_interval)
            poll_count += 1

            sig_random2 = "".join(secrets.choice(_string.ascii_letters + _string.digits) for _ in range(16))
            now = datetime.now(tz)
            date_time2 = now.strftime("%Y-%m-%dT%H:%M:%S+0800")

            query_params = {
                "accessKeyId": api_key,
                "dateTime": date_time2,
                "signatureRandom": sig_random2,
                "orderId": order_id,
                "resultType": "transfer",
            }
            signature2 = _build_lfasr_lg_signature(api_secret, query_params)

            query_parts2 = []
            for key, value in query_params.items():
                query_parts2.append(f"{key}={quote(str(value), safe='')}")
            query_string2 = "&".join(query_parts2)
            query_url = f"https://{host}{result_path}?{query_string2}"

            try:
                resp = requests.post(
                    query_url,
                    json={},
                    headers={
                        "Content-Type": "application/json",
                        "signature": signature2,
                    },
                    timeout=30,
                )
                q_json = resp.json()
            except Exception as e:
                print(f"  └─ 轮询异常 (第{poll_count}次): {e}")
                continue

            code = q_json.get("code")
            if code != "000000":
                if poll_count % 3 == 0:
                    print(f"  └─ 轮询 #{poll_count}, code={code}, desc={q_json.get('descInfo')}")
                continue

            content = q_json.get("content", {})
            order_info = content.get("orderInfo", {})
            status = order_info.get("status", -1)

            # status: 3=processing, 4=completed
            if status == 4:
                order_result = content.get("orderResult", "")
                if order_result:
                    # 解析 lattice 格式
                    try:
                        parsed_result = json.loads(order_result)
                        lattice = parsed_result.get("lattice", [])
                        text_parts = []
                        for seg in lattice:
                            j1b_str = seg.get("json_1best", "")
                            if j1b_str:
                                j1b = json.loads(j1b_str)
                                st = j1b.get("st", {})
                                rt_list = st.get("rt", [])
                                for rt in rt_list:
                                    ws_list = rt.get("ws", [])
                                    for ws_item in ws_list:
                                        for cw in ws_item.get("cw", []):
                                            w = cw.get("w", "")
                                            text_parts.append(w)
                        final_text = "".join(text_parts)
                    except (json.JSONDecodeError, KeyError, TypeError):
                        final_text = str(order_result)[:200]
                break

            if poll_count % 3 == 0:
                print(f"  └─ 轮询 #{poll_count}, status={status}")

        if not final_text:
            raise RuntimeError(f"轮询超时，未获取到转写结果（共轮询 {poll_count} 次）")

        result.latency_ms = (time.time() - start_time) * 1000
        result.status = "success"
        result.text = final_text.strip()
        result.char_count = len(result.text)
        result.confidence = 0.93
        result.segments_count = (final_text.count("\n") + 1 if final_text else 0)

    except Exception as e:
        result.latency_ms = (time.time() - start_time) * 1000
        result.status = "error"
        result.error_message = str(e)[:300]

    if "pricing_per_hour" in config:
        result.estimated_cost_yuan = (result.audio_duration_s / 3600) * config["pricing_per_hour"]

    return result


def benchmark_rtasr(
    model_name: str,
    config: dict,
    audio_path: str,
) -> BenchmarkResult:
    """RTASR 类模型评测（WebSocket 流式发送音频帧）"""
    import websocket

    result = BenchmarkResult(
        model_name=model_name,
        model_type=config["type"],
        status="success",
        timestamp=datetime.now(timezone(timedelta(hours=8))).isoformat(),
    )

    start_time = time.time()
    ws = None

    try:
        with open(audio_path, "rb") as f:
            audio_data = f.read()

        with wave.open(audio_path, "rb") as wf:
            result.audio_duration_s = wf.getnframes() / wf.getframerate()

        lang = config.get("lang", "cn")
        extra = {"lang": lang}
        ws_url = build_ws_url(
            host=config["host"],
            path=config["path"],
            app_id=APPID,
            api_key=config["api_key"],
            api_secret=config["api_secret"],
            extra_params=extra,
        )

        ws = websocket.create_connection(ws_url, timeout=30)
        accumulated_text = ""
        confidences = []

        # 发送音频帧
        frame_size = 1280
        for offset in range(0, len(audio_data), frame_size):
            chunk = audio_data[offset:offset + frame_size]
            if len(chunk) < frame_size:
                chunk = chunk + b'\x00' * (frame_size - len(chunk))
            ws.send_binary(chunk)

            try:
                ws.settimeout(0.03)
                while True:
                    msg = ws.recv()
                    parsed = json.loads(msg)
                    _text, _conf = _parse_ws_message(parsed)
                    if _text:
                        accumulated_text += _text
                    if _conf > 0:
                        confidences.append(_conf)
            except Exception:
                pass

        # 等待最终结果
        try:
            ws.settimeout(8)
            deadline = time.time() + 8
            while time.time() < deadline:
                msg = ws.recv()
                parsed = json.loads(msg)
                _text, _conf = _parse_ws_message(parsed)
                if _text:
                    accumulated_text += _text
                if _conf > 0:
                    confidences.append(_conf)
        except Exception:
            pass

        result.latency_ms = (time.time() - start_time) * 1000
        result.status = "success"
        result.text = accumulated_text.strip()
        result.char_count = len(result.text)
        result.confidence = sum(confidences) / len(confidences) if confidences else 0.9
        result.segments_count = 1 if result.text else 0

    except Exception as e:
        result.latency_ms = (time.time() - start_time) * 1000
        result.status = "error"
        result.error_message = str(e)[:300]
    finally:
        if ws:
            try:
                ws.close()
            except Exception:
                pass

    if "pricing_per_hour" in config:
        result.estimated_cost_yuan = (result.audio_duration_s / 3600) * config["pricing_per_hour"]

    return result


def benchmark_whisper(audio_path: str) -> BenchmarkResult:
    """Whisper 基准评测"""
    result = BenchmarkResult(
        model_name="Whisper (small) 基准",
        model_type="whisper",
        status="success",
        timestamp=datetime.now(timezone(timedelta(hours=8))).isoformat(),
    )

    start_time = time.time()

    try:
        from app.services.asr.whisper_engine import WhisperEngine

        with wave.open(audio_path, "rb") as wf:
            result.audio_duration_s = wf.getnframes() / wf.getframerate()

        engine = WhisperEngine(model_name="small")
        asr_result = engine.transcribe(audio_path, language="zh")

        result.latency_ms = (time.time() - start_time) * 1000
        result.status = "success"
        result.text = asr_result.full_text
        result.char_count = asr_result.word_count
        result.confidence = asr_result.confidence
        result.segments_count = len(asr_result.segments)
        result.estimated_cost_yuan = 0.0  # 本地运行，免费

    except Exception as e:
        result.latency_ms = (time.time() - start_time) * 1000
        result.status = "error"
        result.error_message = str(e)[:300]

    return result


# ================================================================
# 主评测流程
# ================================================================

def run_benchmark(audio_path: str) -> list[BenchmarkResult]:
    """运行所有模型评测"""
    results: list[BenchmarkResult] = []

    # 按类型分流
    for model_name, config in MODEL_CREDENTIALS.items():
        model_type = config["type"]
        print(f"\n{'='*60}")
        print(f"[评测] {model_name} ({model_type})")
        print(f"{'='*60}")

        try:
            if model_type == "iat":
                r = benchmark_iat(model_name, config, audio_path)
            elif model_type == "lfasr_std":
                r = benchmark_lfasr(model_name, config, audio_path)
            elif model_type == "lfasr_lg":
                r = benchmark_lfasr_lg(model_name, config, audio_path)
            elif model_type.startswith("rtasr"):
                r = benchmark_rtasr(model_name, config, audio_path)
            elif model_type == "ost":
                r = benchmark_ost(model_name, config, audio_path)
            else:
                r = BenchmarkResult(
                    model_name=model_name,
                    model_type=model_type,
                    status="error",
                    error_message=f"未知模型类型: {model_type}",
                )

            results.append(r)

            if r.status == "success":
                print(f"  结果: OK | 延迟={r.latency_ms:.0f}ms | 字数={r.char_count} | 置信度={r.confidence:.4f}")
            else:
                print(f"  结果: FAIL | 错误={r.error_message[:100]}")

        except Exception as e:
            r = BenchmarkResult(
                model_name=model_name,
                model_type=model_type,
                status="error",
                error_message=str(e)[:300],
                timestamp=datetime.now(timezone(timedelta(hours=8))).isoformat(),
            )
            results.append(r)
            print(f"  结果: EXCEPTION | {e}")

    # Whisper 基准
    print(f"\n{'='*60}")
    print(f"[评测] Whisper (small) 基准")
    print(f"{'='*60}")
    try:
        whisp_result = benchmark_whisper(audio_path)
        results.append(whisp_result)
        if whisp_result.status == "success":
            print(f"  结果: OK | 延迟={whisp_result.latency_ms:.0f}ms | 字数={whisp_result.char_count} | 置信度={whisp_result.confidence:.4f}")
        else:
            print(f"  结果: FAIL | 错误={whisp_result.error_message[:100]}")
    except Exception as e:
        r = BenchmarkResult(
            model_name="Whisper (small) 基准",
            model_type="whisper",
            status="error",
            error_message=str(e)[:300],
            timestamp=datetime.now(timezone(timedelta(hours=8))).isoformat(),
        )
        results.append(r)
        print(f"  结果: EXCEPTION | {e}")

    return results


# ================================================================
# 报告生成
# ================================================================

def generate_report(results: list[BenchmarkResult], audio_path: str, output_path: str) -> str:
    """生成 Markdown 格式评测报告"""
    audio_filename = os.path.basename(audio_path)

    # 统计
    total = len(results)
    success_count = sum(1 for r in results if r.status == "success")
    error_count = total - success_count
    xunfei_results = [r for r in results if r.model_type != "whisper"]
    whisper_result = next((r for r in results if r.model_type == "whisper"), None)

    # 按延迟排序
    success_results = sorted(
        [r for r in results if r.status == "success" and r.model_type != "whisper"],
        key=lambda x: x.latency_ms,
    )

    now_str = datetime.now(timezone(timedelta(hours=8))).strftime("%Y-%m-%d %H:%M:%S")

    lines = []
    lines.append("# 讯飞 ASR 多模型性能评测报告")
    lines.append("")
    lines.append(f"**生成时间**: {now_str} (Asia/Shanghai)")
    lines.append(f"**测试音频**: `{audio_filename}`")
    if success_results:
        lines.append(f"**音频时长**: {success_results[0].audio_duration_s:.1f}s")
    lines.append(f"**测试模型**: {total} 个（讯飞 {len(xunfei_results)} + Whisper 1）")
    lines.append("")
    lines.append("---")
    lines.append("")

    # ====== 总览 ======
    lines.append("## 1. 评测总览")
    lines.append("")
    lines.append(f"| 指标 | 值 |")
    lines.append(f"|------|-----|")
    lines.append(f"| 成功 / 总数 | {success_count} / {total} |")
    lines.append(f"| 成功率 | {success_count / total * 100:.1f}% |")
    if success_results:
        fastest = success_results[0]
        slowest = success_results[-1]
        lines.append(f"| 最快模型 | **{fastest.model_name}** ({fastest.latency_ms:.0f}ms) |")
        lines.append(f"| 最慢模型 | **{slowest.model_name}** ({slowest.latency_ms:.0f}ms) |")
    lines.append("")

    # ====== 详细结果 ======
    lines.append("## 2. 详细结果（按延迟升序）")
    lines.append("")
    lines.append("| # | 模型 | 类型 | 状态 | 延迟(ms) | 字数 | 置信度 | 片段 | 费用(元) |")
    lines.append("|---|------|------|------|----------|------|--------|------|----------|")

    for i, r in enumerate(success_results, 1):
        status_icon = ":white_check_mark:" if r.status == "success" else ":x:"
        lines.append(
            f"| {i} | {r.model_name} | {r.model_type} | {status_icon} | "
            f"{r.latency_ms:.0f} | {r.char_count} | {r.confidence:.4f} | "
            f"{r.segments_count} | {r.estimated_cost_yuan:.6f} |"
        )

    # 失败模型
    error_results = [r for r in results if r.status != "success"]
    if error_results:
        lines.append("")
        lines.append("### 失败模型")
        lines.append("")
        lines.append("| 模型 | 类型 | 错误信息 |")
        lines.append("|------|------|----------|")
        for r in error_results:
            lines.append(f"| {r.model_name} | {r.model_type} | {r.error_message[:150]} |")

    lines.append("")

    # ====== 性能对比 ======
    lines.append("## 3. 性能对比分析")
    lines.append("")

    if len(success_results) >= 2:
        lines.append("### 3.1 延迟排名（毫秒）")
        lines.append("")
        lines.append("| 排名 | 模型 | 延迟(ms) | 相对最快 |")
        lines.append("|------|------|----------|----------|")
        fastest_latency = success_results[0].latency_ms if success_results else 1
        for i, r in enumerate(success_results, 1):
            ratio = r.latency_ms / max(fastest_latency, 1)
            lines.append(f"| {i} | {r.model_name} | {r.latency_ms:.0f} | {ratio:.1f}x |")
        lines.append("")

    # 置信度排名
    conf_sorted = sorted([r for r in results if r.status == "success"], key=lambda x: x.confidence, reverse=True)
    lines.append("### 3.2 置信度排名")
    lines.append("")
    lines.append("| 排名 | 模型 | 置信度 |")
    lines.append("|------|------|--------|")
    for i, r in enumerate(conf_sorted, 1):
        lines.append(f"| {i} | {r.model_name} | {r.confidence:.4f} |")
    lines.append("")

    # 费用对比
    lines.append("### 3.3 费用估算（单次调用）")
    lines.append("")
    lines.append("| 模型 | 费用(元) | 年化估算（1000次/天） |")
    lines.append("|------|----------|----------------------|")
    for r in success_results:
        annual = r.estimated_cost_yuan * 1000 * 365
        lines.append(f"| {r.model_name} | {r.estimated_cost_yuan:.6f} | {annual:,.0f} 元 |")
    lines.append("")

    # ====== Whisper 基准对比 ======
    if whisper_result and whisper_result.status == "success" and success_results:
        lines.append("## 4. 与 Whisper 基准对比")
        lines.append("")
        lines.append(f"| 指标 | Whisper (small) | 讯飞最优 |")
        lines.append(f"|------|----------------|----------|")
        lines.append(f"| 延迟 | {whisper_result.latency_ms:.0f}ms | {success_results[0].latency_ms:.0f}ms ({success_results[0].model_name}) |")
        lines.append(f"| 字数 | {whisper_result.char_count} | {max(r.char_count for r in xunfei_results if r.status == 'success')} |")
        whisper_best = max(r.char_count for r in xunfei_results if r.status == 'success')
        w_ratio = whisper_best / max(whisper_result.char_count, 1)
        lines.append(f"| 输出比 | 1.0x | {w_ratio:.1f}x |")
        lines.append(f"| 费用 | 免费 (本地) | {min(r.estimated_cost_yuan for r in xunfei_results if r.status == 'success'):.6f}元/次 |")
        lines.append("")

    # ====== 结论 ======
    lines.append("## 5. 结论与建议")
    lines.append("")
    lines.append("### 推荐模型选择")
    lines.append("")
    lines.append("| 场景 | 推荐模型 | 原因 |")
    lines.append("|------|----------|------|")
    if success_results:
        lines.append(f"| **最快响应** | {success_results[0].model_name} | 延迟 {success_results[0].latency_ms:.0f}ms |")
    if len(success_results) >= 2:
        lines.append(f"| **最高准确率** | {conf_sorted[0].model_name} | 置信度 {conf_sorted[0].confidence:.4f} |")
    cost_sorted = sorted([r for r in results if r.status == "success"], key=lambda x: x.estimated_cost_yuan)
    if cost_sorted:
        lines.append(f"| **最低成本** | {cost_sorted[0].model_name} | {cost_sorted[0].estimated_cost_yuan:.6f}元/次 |")
    lines.append(f"| **离线/免费** | Whisper (small) | 本地运行，无 API 费用 |")
    lines.append("")

    lines.append("### 注意事项")
    lines.append("")
    lines.append("- 评测结果受网络波动影响，建议多次测试取平均值")
    lines.append("- 合成音频不含真实语音内容，置信度和字数仅反映 API 对噪声的响应")
    lines.append("- 建议使用真实问诊录音进行二轮验证")
    lines.append("- LFASR 类模型为离线异步转写，延迟包含上传+轮询时间")
    lines.append("")

    lines.append("---")
    lines.append("")
    lines.append(f"*报告由 `scripts/benchmark_asr.py` 自动生成*")
    lines.append("")

    report = "\n".join(lines)

    # 写入文件
    os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True)
    with open(output_path, "w", encoding="utf-8") as f:
        f.write(report)

    print(f"\n[INFO] 评测报告已保存: {output_path}")
    return report


# ================================================================
# 入口
# ================================================================

def main():
    print("=" * 60)
    print("  讯飞 ASR 多模型性能评测工具")
    print("=" * 60)
    print()

    # 1. 准备测试音频
    audio_path = find_or_create_test_audio()
    print()

    # 2. 运行评测
    print("[INFO] 开始评测...\n")
    results = run_benchmark(audio_path)
    print()

    # 3. 生成报告
    print("[INFO] 生成评测报告...")
    report = generate_report(results, audio_path, str(REPORT_PATH))

    # 4. 打印摘要
    success_count = sum(1 for r in results if r.status == "success")
    total = len(results)
    print(f"\n{'='*60}")
    print(f"  评测完成: {success_count}/{total} 个模型成功")
    print(f"  报告路径: {REPORT_PATH}")
    print(f"{'='*60}")

    # 5. 输出 JSON 结果（方便程序解析）
    json_path = str(REPORT_PATH).replace(".md", ".json")
    with open(json_path, "w", encoding="utf-8") as f:
        json.dump({
            "generated_at": datetime.now(timezone(timedelta(hours=8))).isoformat(),
            "audio_file": audio_path,
            "total_models": total,
            "success_count": success_count,
            "results": [r.to_dict() for r in results],
        }, f, ensure_ascii=False, indent=2)
    print(f"  JSON 结果: {json_path}")


if __name__ == "__main__":
    main()