""" 讯飞 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()