shay7sev
a875fe9f63
Emit correlated JSON logs from the API and FunASR services, propagate request IDs to the client, and configure bounded opt-in Docker logging. Document the host-wide Grafana/Loki query workflow and keep observability deployment ownership outside the application repository.
272 lines
8.3 KiB
Python
272 lines
8.3 KiB
Python
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()
|