58 lines
1.7 KiB
Python
58 lines
1.7 KiB
Python
"""Demo ASR adapter for local voice co-creation smoke tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from app.services.adapters.asr.models import TranscriptionOutput
|
|
from app.services.adapters.base import BaseAdapter
|
|
from app.services.adapters.registry import AdapterRegistry
|
|
|
|
|
|
@AdapterRegistry.register("asr", "demo")
|
|
class DemoASRAdapter(BaseAdapter[TranscriptionOutput]):
|
|
"""Return transcript hints or text uploads without external ASR services."""
|
|
|
|
adapter_type = "asr"
|
|
adapter_name = "demo"
|
|
|
|
async def execute(
|
|
self,
|
|
audio_bytes: bytes,
|
|
file_name: str | None = None,
|
|
mime_type: str | None = None,
|
|
transcript_hint: str | None = None,
|
|
**kwargs,
|
|
) -> TranscriptionOutput:
|
|
hint = (transcript_hint or "").strip()
|
|
if hint:
|
|
return TranscriptionOutput(
|
|
transcript_text=hint,
|
|
confidence=1.0,
|
|
provider=self.adapter_name,
|
|
)
|
|
|
|
if mime_type and mime_type.startswith("text/"):
|
|
text = audio_bytes.decode("utf-8", errors="ignore").strip()
|
|
if text:
|
|
return TranscriptionOutput(
|
|
transcript_text=text,
|
|
confidence=1.0,
|
|
provider=self.adapter_name,
|
|
)
|
|
|
|
raise HTTPException(
|
|
status_code=503,
|
|
detail=(
|
|
"当前环境未配置真实语音转写，请先使用文本共创模式，"
|
|
"或在开发模式下提供 transcript_hint。"
|
|
),
|
|
)
|
|
|
|
async def health_check(self) -> bool:
|
|
return True
|
|
|
|
@property
|
|
def estimated_cost(self) -> float:
|
|
return 0.0
|