xiaoxia-saas/apps/api/app/services/script_asr_service.py

"""文案提取 ASR 服务封装 — Issue #1893.

将已有的 ASR 服务工厂封装为面向文案提取场景的简单接口：
- transcribe_to_text(video_path) -> str：将视频/音频转写为纯文本
- 未配置 ASR 时抛 ASRNotConfiguredError（路由层映射为 503）
- ASR 调用失败时抛 ASRTranscriptionError（路由层映射为 502）
"""

from __future__ import annotations

import logging
from pathlib import Path

from packages.ports.asr_service import ASRServiceError

logger = logging.getLogger(__name__)


class ASRNotConfiguredError(Exception):
    """ASR 服务未配置."""


class ASRTranscriptionError(Exception):
    """ASR 转写失败."""


def transcribe_to_text(media_path: str | Path) -> str:
    """将视频/音频文件转写为纯文本.

    Args:
        media_path: 媒体文件路径

    Returns:
        转写出的文本

    Raises:
        ASRNotConfiguredError: ASR 服务未配置
        ASRTranscriptionError: ASR 调用失败
    """
    # 延迟导入，避免循环依赖和启动时副作用
    try:
        from apps.worker.services.asr_service_factory import get_asr_service
    except ImportError as exc:
        # API 镜像未打包 worker 代码（本地 ASR 依赖 worker 的 asr_service_factory）
        logger.warning("本地 ASR 不可用（apps.worker 未安装）: %s", exc)
        raise ASRNotConfiguredError("本地 ASR 服务不可用（worker 模块未安装）") from exc

    asr = get_asr_service()
    if asr is None:
        raise ASRNotConfiguredError("ASR 服务未配置，请联系管理员配置火山 MediaKit 或阿里云 ASR 密钥")

    try:
        timeline = asr.transcribe(Path(media_path))
        # 拼接所有分段的文本
        text = "".join(seg.text for seg in timeline.segments)
        return text.strip()
    except ASRNotConfiguredError:
        raise
    except ASRServiceError as exc:
        logger.error("ASR 转写失败: %s", exc)
        raise ASRTranscriptionError(f"语音识别失败: {exc}") from exc
    except Exception as exc:
        logger.error("ASR 转写异常: %s", exc)
        raise ASRTranscriptionError(f"语音识别失败: {exc}") from exc