语音识别与降级链路#

1
def transcribe_audio_to_text(audio_path: str) -> str:
2
    """
3
    将本地音频文件转为文字。
4
    失败时自动降级，保证程序不崩溃。
5
    """

1
# 为什么要 Base64 编码？
2
# 音频是二进制数据，HTTP JSON 只能传文本
3
# Base64 把二进制转为 ASCII 字符串（体积增大约 33%）
4

5
ext = Path(audio_path).suffix.lower().lstrip('.') or 'wav'
6

7
# 部分后缀到 MIME 子类型的映射
8
fmt_map = {'m4a': 'mp4', 'ogg': 'ogg', 'flac': 'flac', 'mp3': 'mp3'}
9
audio_fmt = fmt_map.get(ext, 'wav')
10
mime_type = f"audio/{audio_fmt}"   # 例如 "audio/wav"
11

12
with open(audio_path, "rb") as f:
13
    audio_b64 = base64.b64encode(f.read()).decode('utf-8')
14

15
# Data URI 格式：data:<MIME>;base64,<数据>
16
# 这样就能把文件「内嵌」进 JSON，不依赖文件服务器
17
data_uri = f"data:{mime_type};base64,{audio_b64}"

1
resp = httpx.post(
2
    DASHSCOPE_MULTIMODAL_URL,  # 原生多模态接口，不是 compatible-mode！
3
    headers={
4
        "Authorization": f"Bearer {api_key}",
5
        "Content-Type": "application/json",
6
    },
7
    json={
8
        "model": "qwen3-asr-flash",
9
        "input": {
10
            "messages": [
11
                # system 消息必须存在，但 ASR 模型不支持自定义 prompt，留空
12
                {"role": "system", "content": [{"text": ""}]},
13
                # audio 字段直接放 Data URI 字符串（不是字典！）
14
                {"role": "user", "content": [{"audio": data_uri}]},
15
            ]
16
        },
17
        "parameters": {
18
            "asr_options": {
19
                # enable_itn=False：保留口语数字，不转阿拉伯数字
20
                # "三点一四" 不会变成 "3.14"
21
                "enable_itn": False
22
            }
23
        },
24
    },
25
    timeout=30,
26
)
27

28
# 原生接口的返回结构（与 OpenAI 兼容接口不同！）
29
result = resp.json()
30
text = result["output"]["choices"][0]["message"]["content"][0]["text"].strip()

1
# ── 第一步：提交任务 ──────────────────────────────────
2
submit_resp = httpx.post(
3
    DASHSCOPE_ASR_SUBMIT_URL,
4
    headers={
5
        "Authorization": f"Bearer {api_key}",
6
        "Content-Type": "application/json",
7
        "X-DashScope-Async": "enable",   # ← 必须！否则走同步接口会 404
8
    },
9
    json={
10
        "model": "sensevoice-v1",
11
        "input": {"file_url": data_uri},  # sensevoice 也支持 Data URI
12
        "parameters": {},
13
    },
14
)
15
task_id = submit_resp.json()["output"]["task_id"]
16

17
# ── 第二步：轮询结果 ──────────────────────────────────
18
for attempt in range(POLL_MAX_RETRIES):  # 最多等 40 秒
19
    time.sleep(POLL_INTERVAL_SECONDS)
20
    poll_resp = httpx.get(
21
        DASHSCOPE_TASK_QUERY_URL.format(task_id=task_id),
22
        headers={"Authorization": f"Bearer {api_key}"},
23
    )
24
    status = poll_resp.json()["output"]["task_status"]
25

26
    if status == "SUCCEEDED":
27
        text = poll_resp.json()["output"]["results"][0]["transcription"]
28
        return text
29
    elif status in ("FAILED", "CANCELED"):
30
        break  # 失败，退出轮询
31
    # PENDING / RUNNING：继续等待