"""CosyVoice 한국어 합성 워커 (전용 venv에서 실행).

모델을 1회만 로드해 상주시키고, 메인 앱이 HTTP로 합성을 요청한다.
- SFT 프리셋 화자('韩语女' = 한국어)로 참조음성 없이 합성
- 속도(speed)는 CosyVoice 네이티브, 피치(pitch)는 librosa 후처리
"""
from __future__ import annotations

import io
import os
import sys

# 벤더링된 CosyVoice 코드 경로
sys.path.insert(0, "/opt/CosyVoice")
sys.path.insert(0, "/opt/CosyVoice/third_party/Matcha-TTS")

import numpy as np
import soundfile as sf
import torch

try:
    import librosa
except Exception:  # pragma: no cover
    librosa = None

from fastapi import FastAPI
from fastapi.responses import Response
from pydantic import BaseModel

from cosyvoice.cli.cosyvoice import CosyVoice

MODEL_DIR = os.environ.get(
    "COSYVOICE_MODEL_DIR", "/models/cosyvoice/CosyVoice-300M-SFT"
)
DEFAULT_SPK = "韩语女"  # 한국어 여성 프리셋

_model = None


def get_model() -> CosyVoice:
    global _model
    if _model is None:
        _model = CosyVoice(
            MODEL_DIR, load_jit=False, load_trt=False, fp16=False
        )
    return _model


app = FastAPI(title="CosyVoice Worker")


class SynthReq(BaseModel):
    text: str
    speaker: str = DEFAULT_SPK
    speed: float = 1.0
    pitch: float = 0.0


@app.on_event("startup")
def _startup() -> None:
    get_model()


@app.get("/health")
def health() -> dict:
    m = get_model()
    return {
        "status": "ok",
        "speakers": m.list_available_spks(),
        "sr": m.sample_rate,
        "cuda": torch.cuda.is_available(),
    }


@app.post("/synth")
def synth(req: SynthReq) -> Response:
    text = (req.text or "").strip()
    if not text:
        return Response(content="empty text", status_code=400)

    m = get_model()
    spks = m.list_available_spks()
    spk = req.speaker if req.speaker in spks else (
        DEFAULT_SPK if DEFAULT_SPK in spks else spks[0]
    )
    speed = float(max(0.5, min(2.0, req.speed)))
    pitch = float(max(-12.0, min(12.0, req.pitch)))

    outs = list(m.inference_sft(text, spk, stream=False, speed=speed))
    audio = (
        torch.cat([o["tts_speech"] for o in outs], dim=1)
        .squeeze(0)
        .cpu()
        .numpy()
        .astype(np.float32)
    )
    sr = m.sample_rate

    if abs(pitch) > 1e-3 and librosa is not None:
        audio = librosa.effects.pitch_shift(audio, sr=sr, n_steps=pitch)

    buf = io.BytesIO()
    sf.write(buf, audio, sr, format="WAV", subtype="PCM_16")
    buf.seek(0)
    return Response(content=buf.read(), media_type="audio/wav")