hyungi_document_server/services/stt/server.py

"""STT 마이크로서비스 — faster-whisper (GPU) 기반 음성 전사.

filePath → {text, segments:[{start,end,text}]}. 모델은 첫 요청 시 lazy loading.
기본 모델 large-v3 (VRAM ~3GB, float16). 환경변수로 교체 가능.
"""

import os
import unicodedata
from pathlib import Path

from fastapi import FastAPI

app = FastAPI()

_model = None
_MODEL_NAME = os.getenv("WHISPER_MODEL", "large-v3")
_DEVICE = os.getenv("WHISPER_DEVICE", "cuda")
_COMPUTE_TYPE = os.getenv("WHISPER_COMPUTE_TYPE", "float16")


def _resolve_path(file_path: str) -> Path | None:
    """NFC(DB) vs NFD(NFS) 한글 경로 정규화 차이 흡수. OCR 서비스와 동일 패턴."""
    candidates = [
        file_path,
        unicodedata.normalize("NFD", file_path),
        unicodedata.normalize("NFC", file_path),
    ]
    for c in candidates:
        p = Path(c)
        if p.exists():
            return p
    # 마지막 fallback: parent 디렉토리에서 이름을 NFC 로 매칭
    parent = Path(file_path).parent
    if parent.exists():
        target = unicodedata.normalize("NFC", Path(file_path).name)
        for child in parent.iterdir():
            if unicodedata.normalize("NFC", child.name) == target:
                return child
    return None


def _load_model():
    """faster-whisper lazy loading — 첫 호출 시만 VRAM 점유."""
    global _model
    if _model is not None:
        return _model
    from faster_whisper import WhisperModel

    _model = WhisperModel(_MODEL_NAME, device=_DEVICE, compute_type=_COMPUTE_TYPE)
    return _model


def _cuda_device_count() -> int:
    try:
        import ctranslate2
        return ctranslate2.get_cuda_device_count()
    except Exception:
        return 0


@app.get("/health")
def health():
    """Liveness — Docker healthcheck 용, 프로세스 생존 확인."""
    return {"status": "ok", "service": "stt-faster-whisper"}


@app.get("/ready")
def ready():
    """Readiness — CUDA + 모델 상태. 배포 검증용."""
    count = _cuda_device_count()
    cuda_ok = count > 0
    models_loaded = _model is not None
    return {
        "ready": cuda_ok and models_loaded,
        "cuda": cuda_ok,
        "cuda_device_count": count,
        "models_loaded": models_loaded,
        "model": _MODEL_NAME,
        "compute_type": _COMPUTE_TYPE,
    }


@app.post("/transcribe")
async def transcribe(body: dict):
    """오디오 파일 전사.

    입력:
      {
        "filePath": "/documents/PKM/Recordings/2026-04-23_회의.mp3",
        "langs":    ["ko"]?,     # 단일 언어 지정 or 생략(자동감지)
        "beamSize": 5?           # 기본 5
      }

    출력:
      {
        "text": "전체 전사 텍스트",
        "segments": [{"start": 0.0, "end": 2.4, "text": "..."}, ...],
        "language": "ko",
        "language_probability": 0.99,
        "duration": 1832.5
      }
    """
    raw_path = body["filePath"]
    langs = body.get("langs")
    beam_size = int(body.get("beamSize", 5))

    resolved = _resolve_path(raw_path)
    if resolved is None:
        return {"error": f"파일 없음: {raw_path}", "text": "", "segments": []}

    model = _load_model()

    language = None
    if isinstance(langs, list) and len(langs) == 1:
        language = langs[0]

    segments_iter, info = model.transcribe(
        str(resolved),
        beam_size=beam_size,
        language=language,
        vad_filter=True,
    )

    segments = []
    parts = []
    for seg in segments_iter:
        segments.append({
            "start": round(float(seg.start), 2),
            "end": round(float(seg.end), 2),
            "text": seg.text.strip(),
        })
        parts.append(seg.text)

    return {
        "text": " ".join(p.strip() for p in parts).strip(),
        "segments": segments,
        "language": getattr(info, "language", None),
        "language_probability": float(getattr(info, "language_probability", 0.0) or 0.0),
        "duration": float(getattr(info, "duration", 0.0) or 0.0),
    }