life-echo/api/app/features/evaluation/asr_service.py

"""内部评测 ASR 转写服务。"""

from __future__ import annotations

from fastapi import UploadFile

from app.core.agent_logging import log_asr_transcript_result
from app.core.errors import ProviderError
from app.core.logging import get_logger
from app.features.evaluation.errors import EvaluationBadRequestError
from app.ports.asr import ASRProvider, ASRTranscriptionError

logger = get_logger(__name__)

MAX_AUDIO_BYTES = 100 * 1024 * 1024
_READ_CHUNK_BYTES = 1024 * 1024
_ALLOWED_FORMATS = frozenset(
    {"m4a", "mp3", "wav", "aac", "amr", "ogg-opus", "speex", "silk", "pcm"}
)

_EXT_TO_FORMAT: dict[str, str] = {
    "m4a": "m4a",
    "mp4": "m4a",
    "mp3": "mp3",
    "wav": "wav",
    "aac": "aac",
    "amr": "amr",
    "ogg": "ogg-opus",
    "opus": "ogg-opus",
    "webm": "ogg-opus",
    "pcm": "pcm",
    "speex": "speex",
    "silk": "silk",
}

_CONTENT_TYPE_TO_FORMAT: dict[str, str] = {
    "audio/mp4": "m4a",
    "audio/x-m4a": "m4a",
    "audio/m4a": "m4a",
    "audio/mpeg": "mp3",
    "audio/mp3": "mp3",
    "audio/wav": "wav",
    "audio/x-wav": "wav",
    "audio/wave": "wav",
    "audio/aac": "aac",
    "audio/amr": "amr",
    "audio/ogg": "ogg-opus",
    "audio/webm": "ogg-opus",
}


def resolve_voice_format(
    *,
    explicit: str | None,
    filename: str | None,
    content_type: str | None,
) -> str:
    if explicit:
        fmt = explicit.strip().lower()
        if fmt in _ALLOWED_FORMATS:
            return fmt
        raise EvaluationBadRequestError(f"不支持的音频格式：{fmt}")

    if filename:
        ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else ""
        if ext in _EXT_TO_FORMAT:
            return _EXT_TO_FORMAT[ext]

    if content_type:
        base = content_type.split(";", 1)[0].strip().lower()
        if base in _CONTENT_TYPE_TO_FORMAT:
            return _CONTENT_TYPE_TO_FORMAT[base]

    raise EvaluationBadRequestError(
        "无法识别音频格式，请上传 m4a/mp3/wav 或在请求中指定 format 参数"
    )


async def read_limited_upload(
    file: UploadFile,
    *,
    max_bytes: int = MAX_AUDIO_BYTES,
) -> bytes:
    chunks: list[bytes] = []
    total = 0
    while True:
        chunk = await file.read(_READ_CHUNK_BYTES)
        if not chunk:
            break
        total += len(chunk)
        if total > max_bytes:
            raise EvaluationBadRequestError("音频过大（上限 100MB）")
        chunks.append(chunk)
    return b"".join(chunks)


class EvalAsrService:
    def __init__(self, asr: ASRProvider) -> None:
        self._asr = asr

    async def transcribe(
        self,
        audio: bytes,
        *,
        voice_format: str,
        source_label: str = "eval_asr",
    ) -> str:
        if not audio:
            raise EvaluationBadRequestError("音频为空")
        if len(audio) > MAX_AUDIO_BYTES:
            raise EvaluationBadRequestError("音频过大（上限 100MB）")

        try:
            text = await self._asr.transcribe(audio, voice_format)
        except ASRTranscriptionError as e:
            raise ProviderError(str(e), provider="asr") from e

        log_asr_transcript_result(
            logger,
            text=text or "",
            conversation_id=None,
            source=source_label,
        )
        return text or ""