from __future__ import annotations import asyncio import logging import os import tempfile import time import uuid from contextlib import asynccontextmanager from pathlib import Path from typing import Any from fastapi import FastAPI, File, Form, HTTPException, Request, UploadFile from funasr import AutoModel from pydantic import BaseModel from logging_config import configure_logging, log_event, redirect_standard_streams configure_logging() MAX_AUDIO_BYTES = 15 * 1024 * 1024 CHUNK_BYTES = 1024 * 1024 SUPPORTED_FORMATS = {"mp3", "aac", "wav", "m4a", "amr"} class Settings: def __init__(self) -> None: self.model = os.getenv("ASR_MODEL", "paraformer-zh") self.vad_model = os.getenv("ASR_VAD_MODEL", "fsmn-vad") self.punc_model = os.getenv("ASR_PUNC_MODEL", "ct-punc") self.device = os.getenv("ASR_DEVICE", "cpu") self.max_concurrency = positive_int("ASR_MAX_CONCURRENCY", 1) self.batch_size_seconds = positive_int("ASR_BATCH_SIZE_SECONDS", 300) class HealthResponse(BaseModel): status: str model: str device: str class TranscriptionResponse(BaseModel): text: str duration_ms: int request_id: str def positive_int(name: str, default: int) -> int: raw = os.getenv(name, str(default)) try: value = int(raw) except ValueError as error: raise RuntimeError(f"{name} must be an integer") from error if value <= 0: raise RuntimeError(f"{name} must be greater than 0") return value def load_model(settings: Settings) -> Any: log_event( logging.INFO, "asr_model_loading", "loading FunASR model", model=settings.model, device=settings.device, ) model = AutoModel( model=settings.model, vad_model=settings.vad_model, punc_model=settings.punc_model, device=settings.device, disable_update=True, ) log_event(logging.INFO, "asr_model_ready", "FunASR model is ready") return model @asynccontextmanager async def lifespan(app: FastAPI): redirect_standard_streams() settings = Settings() app.state.settings = settings app.state.model = await asyncio.to_thread(load_model, settings) app.state.inference_slots = asyncio.Semaphore(settings.max_concurrency) yield app = FastAPI( title="Teaching Feedback FunASR Service", version=os.getenv("APP_VERSION", "0.2.0"), lifespan=lifespan, ) @app.middleware("http") async def request_id_middleware(request: Request, call_next: Any): request_id = valid_request_id(request.headers.get("x-request-id")) request.state.request_id = request_id try: response = await call_next(request) except Exception: log_event( logging.ERROR, "asr_http_request_failed", "ASR HTTP request failed", request_id=request_id, status=500, error_class="unhandled_error", ) raise response.headers["X-Request-ID"] = request_id return response @app.get("/health", response_model=HealthResponse) async def health() -> HealthResponse: settings: Settings = app.state.settings return HealthResponse(status="ok", model=settings.model, device=settings.device) @app.post("/v1/transcriptions", response_model=TranscriptionResponse) async def create_transcription( request: Request, file: UploadFile = File(...), language: str = Form("zh"), duration_ms: int = Form(0), hotword: str = Form(""), ) -> TranscriptionResponse: request_id = request.state.request_id started_at = time.perf_counter() audio_format = "unknown" audio_size_bytes = 0 temporary_path: Path | None = None normalized_duration_ms = max(0, duration_ms) log_event( logging.INFO, "asr_request_started", "ASR request started", request_id=request_id, audio_duration_ms=normalized_duration_ms, audio_format=audio_format, ) try: if language not in {"zh", "zh-cn", "cmn"}: raise HTTPException(status_code=422, detail="this service is configured for Chinese audio") audio_format = file_extension(file.filename) temporary_path, audio_size_bytes = await save_upload(file, audio_format) async with app.state.inference_slots: text = await asyncio.to_thread( run_inference, app.state.model, temporary_path, app.state.settings.batch_size_seconds, hotword.strip(), ) if not text: raise HTTPException(status_code=422, detail="no clear speech was recognized") except HTTPException as error: log_event( logging.WARNING, "asr_request_failed", "ASR request failed", request_id=request_id, status=error.status_code, latency_ms=elapsed_ms(started_at), audio_size_bytes=audio_size_bytes, audio_duration_ms=normalized_duration_ms, audio_format=audio_format, error_class=http_error_class(error.status_code), ) raise except Exception: log_event( logging.ERROR, "asr_request_failed", "ASR request failed", request_id=request_id, status=500, latency_ms=elapsed_ms(started_at), audio_size_bytes=audio_size_bytes, audio_duration_ms=normalized_duration_ms, audio_format=audio_format, error_class="inference_failed", ) raise HTTPException(status_code=500, detail="speech recognition failed") from None finally: if temporary_path is not None: temporary_path.unlink(missing_ok=True) log_event( logging.INFO, "asr_request_finished", "ASR request finished", request_id=request_id, status=200, latency_ms=elapsed_ms(started_at), audio_size_bytes=audio_size_bytes, audio_duration_ms=normalized_duration_ms, audio_format=audio_format, ) return TranscriptionResponse( text=text, duration_ms=normalized_duration_ms, request_id=request_id, ) def valid_request_id(value: str | None) -> str: candidate = (value or "").strip() if candidate and len(candidate) <= 128 and candidate.isascii() and candidate.isprintable(): return candidate return str(uuid.uuid4()) def elapsed_ms(started_at: float) -> int: return int((time.perf_counter() - started_at) * 1000) def http_error_class(status: int) -> str: if status == 413: return "audio_too_large" if status == 415: return "unsupported_audio_format" if status == 422: return "invalid_audio" return "request_error" def file_extension(filename: str | None) -> str: suffix = Path(filename or "segment.mp3").suffix.lower().lstrip(".") if suffix not in SUPPORTED_FORMATS: raise HTTPException(status_code=415, detail="unsupported audio format") return suffix async def save_upload(file: UploadFile, audio_format: str) -> tuple[Path, int]: handle = tempfile.NamedTemporaryFile(suffix=f".{audio_format}", delete=False) path = Path(handle.name) total_bytes = 0 try: with handle: while chunk := await file.read(CHUNK_BYTES): total_bytes += len(chunk) if total_bytes > MAX_AUDIO_BYTES: raise HTTPException(status_code=413, detail="audio file exceeds 15 MB") handle.write(chunk) except Exception: path.unlink(missing_ok=True) raise finally: await file.close() if total_bytes == 0: path.unlink(missing_ok=True) raise HTTPException(status_code=422, detail="audio file is empty") return path, total_bytes def run_inference(model: Any, path: Path, batch_size_seconds: int, hotword: str) -> str: options: dict[str, Any] = { "input": str(path), "batch_size_s": batch_size_seconds, "use_itn": True, } if hotword: options["hotword"] = hotword result = model.generate(**options) if not isinstance(result, list): return "" texts = [str(item.get("text", "")).strip() for item in result if isinstance(item, dict)] return "\n".join(text for text in texts if text).strip()