hyungi_document_server/services/stt/server.py

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

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

D-1 (plan crawl-24x7-1, 2026-06-10) — idle-unload 운영 전환:
  STT_PRELOAD=0          : startup eager preload 끔 (첫 요청 시 lazy load)
  STT_IDLE_UNLOAD_MINUTES: N분 유휴 시 모델 해제 (0=비활성, 기존 동작).
                           faster-whisper=CTranslate2 라 torch 미설치 — 해제는
                           참조 제거 + gc (CTranslate2 가 소멸 시 VRAM 반환).
콜드로드 수초~수십 초는 호출측(stt_worker read=1800s)이 흡수. healthcheck 는
cuda 가용성 기준 (compose) — 모델 적재는 더 이상 상시 상태가 아니다.
"""

import asyncio
import gc
import logging
import os
import threading
import time
import unicodedata
from contextlib import asynccontextmanager
from pathlib import Path

from fastapi import FastAPI

logger = logging.getLogger("stt")

_IDLE_UNLOAD_MINUTES = int(os.getenv("STT_IDLE_UNLOAD_MINUTES", "0"))


@asynccontextmanager
async def lifespan(_app: FastAPI):
    # startup: 모델 eager preload 시도. 실패해도 프로세스는 살아 있고
    # /ready 의 models_loaded 가 false 로 남는다.
    if os.getenv("STT_PRELOAD", "1") != "0":
        try:
            _load_model()
            logger.info("stt model preloaded: %s (%s, %s)", _MODEL_NAME, _DEVICE, _COMPUTE_TYPE)
        except Exception as e:
            logger.exception("stt model preload failed: %s", e)
    reaper = None
    if _IDLE_UNLOAD_MINUTES > 0:
        reaper = asyncio.create_task(_idle_reaper())
        logger.info("stt idle-unload 활성: %d분", _IDLE_UNLOAD_MINUTES)
    yield
    if reaper:
        reaper.cancel()


app = FastAPI(lifespan=lifespan)

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

# load/unload/inflight 상태 전이는 전부 이 lock 아래 (cold 동시 요청 이중 로드 방지 포함)
_model_lock = threading.Lock()
_inflight = 0
_last_used = time.monotonic()


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 점유. lock 으로 이중 로드 방지."""
    global _model
    if _model is not None:
        return _model
    with _model_lock:
        if _model is None:
            from faster_whisper import WhisperModel
            logger.info("stt model loading: %s (%s, %s)", _MODEL_NAME, _DEVICE, _COMPUTE_TYPE)
            _model = WhisperModel(_MODEL_NAME, device=_DEVICE, compute_type=_COMPUTE_TYPE)
        return _model


def _maybe_unload() -> None:
    """유휴 시 모델 해제. 처리 중(inflight>0)이면 절대 해제하지 않는다."""
    global _model
    with _model_lock:
        if _model is None or _inflight > 0:
            return
        if time.monotonic() - _last_used < _IDLE_UNLOAD_MINUTES * 60:
            return
        _model = None
    gc.collect()
    logger.info("stt idle-unload: whisper 모델 해제 (유휴 %d분 초과)", _IDLE_UNLOAD_MINUTES)


async def _idle_reaper():
    while True:
        await asyncio.sleep(60)
        try:
            _maybe_unload()
        except Exception:
            logger.exception("stt idle reaper 오류")


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 + 모델 상태. healthcheck 는 cuda 만 본다 (D-1 idle-unload)."""
    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,
        "idle_unload_minutes": _IDLE_UNLOAD_MINUTES,
        "inflight": _inflight,
    }


@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
      }
    """
    global _inflight, _last_used
    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": []}

    with _model_lock:
        _inflight += 1
    try:
        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)
    finally:
        with _model_lock:
            _inflight -= 1
            _last_used = time.monotonic()

    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),
    }