wecom_it_smart_desk/backend/app/api/voice_asr.py

# =============================================================================
# 企微IT智能服务台 — 百度语音识别（ASR）API
# =============================================================================
# 说明：接收前端上传的 PCM 音频数据，调用百度 ASR API 转换为文字
#   1. POST /api/voice/asr — 语音转文字（上传PCM音频，返回识别文字）
#
# 音频格式要求：PCM, 16kHz, 16-bit, 单声道
# 百度ASR凭证通过环境变量配置（BAIDU_ASR_APP_ID/API_KEY/SECRET_KEY）
# access_token 缓存在 Redis（key=baidu:asr:token，TTL=30天）
# =============================================================================

import logging
from typing import Optional

import httpx
from fastapi import APIRouter, File, HTTPException, UploadFile

from app.config import settings
from app.utils.response import success_response, error_response

router = APIRouter(prefix="/voice", tags=["语音识别"])
logger = logging.getLogger(__name__)

# --------------------------------------------------------------------------
# 百度 ASR 配置常量
# --------------------------------------------------------------------------
# 百度 ASR 极速版 API 端点（raw binary POST 方式）
BAIDU_ASR_API_URL = "https://vop.baidu.com/pro_api"
# 百度 OAuth2 token 获取端点
BAIDU_TOKEN_URL = "https://aip.baidubce.com/oauth/2.0/token"
# dev_pid=80001 表示普通话极速版
BAIDU_ASR_DEV_PID = "80001"
# Redis 缓存 key
BAIDU_TOKEN_CACHE_KEY = "baidu:asr:token"
# token 缓存 TTL（30天，秒）— 与百度 token有效期一致
BAIDU_TOKEN_TTL = 2592000


async def get_baidu_token() -> str:
    """获取百度 ASR access_token，带 Redis 缓存。

    优先从 Redis 读取缓存的 token（TTL 30天）；
    缓存不存在时调用百度 OAuth2 接口获取新 token 并缓存。

    Returns:
        str: 百度 ASR access_token

    Raises:
        HTTPException: 获取 token 失败时抛出 500 错误
    """
    redis = settings.create_redis_client()
    try:
        # 1. 尝试从 Redis 缓存读取 token
        cached_token: Optional[str] = await redis.get(BAIDU_TOKEN_CACHE_KEY)
        if cached_token:
            logger.debug("[BaiduASR] 使用缓存的 access_token")
            return cached_token

        # 2. 缓存不存在，调用百度 OAuth2 获取新 token
        logger.info("[BaiduASR] 缓存未命中，请求新 access_token")
        async with httpx.AsyncClient(timeout=10) as client:
            resp = await client.post(
                BAIDU_TOKEN_URL,
                params={
                    "grant_type": "client_credentials",
                    "client_id": settings.baidu_asr_api_key,
                    "client_secret": settings.baidu_asr_secret_key,
                },
            )
            data = resp.json()
            token = data.get("access_token")
            if not token:
                error_msg = data.get("error_description", "未知错误")
                logger.error(f"[BaiduASR] 获取 access_token 失败: {error_msg}")
                raise HTTPException(
                    status_code=500,
                    detail=f"获取百度ASR token失败: {error_msg}",
                )

        # 3. 缓存 token 到 Redis（TTL 30天）
        await redis.set(BAIDU_TOKEN_CACHE_KEY, token, ex=BAIDU_TOKEN_TTL)
        logger.info("[BaiduASR] access_token 已缓存到 Redis（TTL=30天）")
        return token
    finally:
        # 关闭 Redis 连接（每次创建新客户端，用完即关）
        await redis.aclose()


@router.post("/asr")
async def transcribe_audio(
    audio: UploadFile = File(..., description="PCM 音频数据（16kHz, 16-bit, mono）"),
):
    """语音转文字 — 上传 PCM 音频，返回百度 ASR 识别结果。

    处理流程：
    1. 接收前端上传的 PCM 音频数据（multipart/form-data，字段名 audio）
    2. 获取百度 access_token（Redis 缓存，TTL 30天）
    3. 调用百度 ASR API（raw binary POST 方式）
    4. 解析返回结果，返回识别文字

    **请求格式**: multipart/form-data
    **字段**: audio — PCM 音频文件（16kHz, 16-bit, 单声道）

    **成功响应**: {code: 0, data: {text: "识别的文字"}, message: "success"}
    **错误响应**: {code: 3001, data: null, message: "百度ASR错误: xxx"}

    Args:
        audio: FastAPI UploadFile 对象，包含 PCM 音频数据

    Returns:
        Dict: 统一响应格式
    """
    # 1. 读取 PCM 音频数据
    pcm_data = await audio.read()
    if not pcm_data:
        raise HTTPException(status_code=400, detail="未收到音频数据")

    logger.info(
        f"[BaiduASR] 收到音频数据: {len(pcm_data)} bytes, "
        f"filename={audio.filename}"
    )

    # 2. 获取百度 access_token
    try:
        token = await get_baidu_token()
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"[BaiduASR] 获取 token 异常: {e}")
        return error_response(code=3001, message=f"获取百度ASR token失败: {str(e)}")

    # 3. 调用百度 ASR API（raw binary POST）
    try:
        async with httpx.AsyncClient(timeout=30) as client:
            resp = await client.post(
                BAIDU_ASR_API_URL,
                params={
                    "dev_pid": BAIDU_ASR_DEV_PID,
                    "cuid": settings.baidu_asr_app_id,
                    "token": token,
                },
                content=pcm_data,
                headers={"Content-Type": "audio/pcm;rate=16000"},
            )
            result = resp.json()

        # 4. 解析返回结果
        # 百度 ASR 返回格式: {"err_no": 0, "result": ["识别的文字"]}
        # err_no=0 表示成功，非0表示错误
        if result.get("err_no") != 0:
            err_msg = result.get("err_msg", "未知错误")
            err_no = result.get("err_no", -1)
            logger.error(f"[BaiduASR] 识别失败: err_no={err_no}, err_msg={err_msg}")
            return error_response(
                code=3001,
                message=f"百度ASR错误: {err_msg}",
            )

        # 成功：提取识别文字
        text_list = result.get("result", [])
        recognized_text = text_list[0] if text_list else ""

        logger.info(
            f"[BaiduASR] 识别成功: text='{recognized_text[:50]}...' "
            f"(len={len(recognized_text)})"
        )
        return success_response(data={"text": recognized_text})

    except httpx.TimeoutException:
        logger.error("[BaiduASR] 调用百度 ASR API 超时（30秒）")
        return error_response(code=3001, message="百度ASR识别超时，请重试")
    except Exception as e:
        logger.error(f"[BaiduASR] 调用百度 ASR API 异常: {e}")
        return error_response(
            code=3001,
            message=f"语音识别失败: {str(e)}",
        )