"""
D3RO-VOICE STT Sidecar (FastAPI HTTP 서버)
faster-whisper를 사용한 로컬 음성 인식 서비스.

사용법:
    python main.py --port 18765

엔드포인트:
    GET  /health     - 헬스체크
    POST /load       - Whisper 모델 로딩
    POST /transcribe - 오디오 전사 (multipart)
    POST /shutdown   - 서버 종료
"""

from __future__ import annotations

import argparse
import logging
import os
import signal
import sys
import time
from contextlib import asynccontextmanager
from typing import AsyncGenerator

import numpy as np
import uvicorn
from fastapi import FastAPI, File, Form, UploadFile
from fastapi.responses import JSONResponse

# ── 로깅 설정 ──────────────────────────────────────────────

logging.basicConfig(
    level=logging.INFO,
    format="[%(asctime)s] [%(levelname)s] [%(name)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
    stream=sys.stdout,
)
logger = logging.getLogger("sidecar")

# ── 전역 상태 ──────────────────────────────────────────────

_model: "WhisperModel | None" = None
_model_id: str | None = None
_gpu_available: bool = False
_server: uvicorn.Server | None = None

# ── FastAPI 앱 ─────────────────────────────────────────────


@asynccontextmanager
async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
    """서버 시작/종료 생명주기."""
    logger.info("Sidecar 서버 시작")
    _detect_gpu()
    yield
    logger.info("Sidecar 서버 종료")


app = FastAPI(title="D3RO-VOICE STT Sidecar", lifespan=lifespan)


def _detect_gpu() -> None:
    """GPU(CUDA) 사용 가능 여부를 감지한다."""
    global _gpu_available
    try:
        import torch

        _gpu_available = torch.cuda.is_available()
        if _gpu_available:
            device_name = torch.cuda.get_device_name(0)
            logger.info("GPU 감지: %s", device_name)
        else:
            logger.info("GPU 미감지, CPU 모드로 동작")
    except ImportError:
        _gpu_available = False
        logger.info("PyTorch 미설치, CPU 모드로 동작")


# ── 엔드포인트 ─────────────────────────────────────────────


@app.get("/health")
async def health() -> JSONResponse:
    """헬스체크. sidecar가 준비되었는지 확인한다."""
    return JSONResponse(
        content={
            "status": "ready" if _model is not None else "no_model",
            "model": _model_id,
            "gpu": _gpu_available,
        }
    )


@app.post("/load")
async def load_model(body: dict) -> JSONResponse:  # noqa: ANN001
    """Whisper 모델을 로딩한다.

    Request body:
        { "model_id": "base" }  -- tiny, base, small, medium, large-v3

    Returns:
        { "status": "loaded", "model_id": "base", "load_time_ms": 1234 }
    """
    global _model, _model_id

    model_id: str = body.get("model_id", "base")
    logger.info("모델 로딩 시작: %s", model_id)

    start_time = time.monotonic()

    try:
        from faster_whisper import WhisperModel

        device = "cuda" if _gpu_available else "cpu"
        compute_type = "float16" if _gpu_available else "int8"

        # 모델 크기별 compute_type 조정
        if model_id in ("large-v3", "medium") and not _gpu_available:
            compute_type = "int8"

        _model = WhisperModel(
            model_id,
            device=device,
            compute_type=compute_type,
        )
        _model_id = model_id

        load_time_ms = int((time.monotonic() - start_time) * 1000)
        logger.info(
            "모델 로딩 완료: %s (device=%s, compute=%s, %dms)",
            model_id,
            device,
            compute_type,
            load_time_ms,
        )

        return JSONResponse(
            content={
                "status": "loaded",
                "model_id": model_id,
                "load_time_ms": load_time_ms,
            }
        )

    except Exception as exc:
        logger.error("모델 로딩 실패: %s", exc)
        return JSONResponse(
            status_code=500,
            content={"status": "error", "message": str(exc)},
        )


@app.post("/transcribe")
async def transcribe(
    audio: UploadFile = File(...),
    language: str = Form("auto"),
    vad_filter: str = Form("true"),
    initial_prompt: str = Form(""),
) -> JSONResponse:
    """오디오 파일을 전사한다.

    Multipart form:
        audio         - PCM16 16kHz mono 바이너리 파일
        language      - 언어 코드 ('auto', 'ko', 'en', ...)
        vad_filter    - VAD 필터 활성화 ('true' / 'false')
        initial_prompt - 초기 프롬프트 (컨텍스트 힌트)

    Returns:
        {
            "text": "전사된 텍스트",
            "segments": [...],
            "language": "ko",
            "duration": 3.5,
            "processing_time": 1234
        }
    """
    if _model is None:
        return JSONResponse(
            status_code=503,
            content={"status": "error", "message": "모델이 로딩되지 않았습니다"},
        )

    start_time = time.monotonic()

    try:
        # PCM16 바이너리 읽기
        pcm_bytes = await audio.read()

        if len(pcm_bytes) == 0:
            return JSONResponse(
                status_code=400,
                content={"status": "error", "message": "오디오 데이터가 비어있습니다"},
            )

        # PCM16 → float32 변환 (-1.0 ~ 1.0)
        audio_array = (
            np.frombuffer(pcm_bytes, dtype=np.int16).astype(np.float32) / 32768.0
        )

        # 오디오 길이 계산 (16kHz mono 기준)
        sample_rate = 16000
        audio_duration = len(audio_array) / sample_rate

        logger.info(
            "전사 시작: %.1f초 오디오, language=%s, vad=%s",
            audio_duration,
            language,
            vad_filter,
        )

        # 전사 옵션 구성
        transcribe_kwargs: dict = {
            "vad_filter": vad_filter.lower() == "true",
            "beam_size": 5,
        }

        if language != "auto":
            transcribe_kwargs["language"] = language

        if initial_prompt:
            transcribe_kwargs["initial_prompt"] = initial_prompt

        # faster-whisper 전사 실행
        # VAD가 전체 오디오를 제거하면 max() 에러 발생 → VAD 없이 재시도
        try:
            segments_iter, info = _model.transcribe(audio_array, **transcribe_kwargs)
        except ValueError as ve:
            if "empty sequence" in str(ve) and transcribe_kwargs.get("vad_filter"):
                logger.warning("VAD가 전체 오디오를 제거함 → VAD 없이 재시도")
                transcribe_kwargs["vad_filter"] = False
                segments_iter, info = _model.transcribe(audio_array, **transcribe_kwargs)
            else:
                raise

        # 세그먼트 수집
        segments_list: list[dict] = []
        full_text_parts: list[str] = []

        for segment in segments_iter:
            seg_dict = {
                "text": segment.text.strip(),
                "start": round(segment.start, 3),
                "end": round(segment.end, 3),
                "avg_logprob": round(segment.avg_logprob, 4),
            }
            segments_list.append(seg_dict)
            full_text_parts.append(segment.text.strip())

        full_text = " ".join(full_text_parts).strip()
        processing_time = int((time.monotonic() - start_time) * 1000)

        detected_language = info.language if info.language else "unknown"

        logger.info(
            "전사 완료: '%s' (lang=%s, %.1f초, %dms)",
            full_text[:80],
            detected_language,
            audio_duration,
            processing_time,
        )

        return JSONResponse(
            content={
                "text": full_text,
                "segments": segments_list,
                "language": detected_language,
                "duration": round(audio_duration, 3),
                "processing_time": processing_time,
            }
        )

    except Exception as exc:
        logger.error("전사 실패: %s", exc, exc_info=True)
        return JSONResponse(
            status_code=500,
            content={"status": "error", "message": str(exc)},
        )


# ── 화자 구분 (Phase 15.5) ────────────────────────────────

_diarization_pipeline = None


def _get_diarization_pipeline(hf_token: str):
    """pyannote speaker diarization 파이프라인을 로드한다 (캐시)."""
    global _diarization_pipeline
    if _diarization_pipeline is not None:
        return _diarization_pipeline

    try:
        from pyannote.audio import Pipeline

        logger.info("Diarization 파이프라인 로딩 시작")
        _diarization_pipeline = Pipeline.from_pretrained(
            "pyannote/speaker-diarization-3.1",
            use_auth_token=hf_token,
        )
        if _gpu_available:
            import torch
            _diarization_pipeline.to(torch.device("cuda"))
            logger.info("Diarization 파이프라인 로딩 완료 (GPU)")
        else:
            logger.info("Diarization 파이프라인 로딩 완료 (CPU)")
    except Exception as exc:
        logger.error("Diarization 파이프라인 로딩 실패: %s", exc)
        raise

    return _diarization_pipeline


@app.post("/diarize")
async def diarize(
    audio: UploadFile = File(...),
    hf_token: str = Form(default=""),
    num_speakers: int = Form(default=0),
) -> JSONResponse:
    """오디오 파일의 화자 구분을 수행한다."""
    if not hf_token:
        return JSONResponse(
            status_code=400,
            content={"status": "error", "message": "HuggingFace 토큰이 필요합니다"},
        )

    start_time = time.monotonic()

    try:
        import tempfile
        import os

        pipeline = _get_diarization_pipeline(hf_token)

        # 오디오 파일 임시 저장 (pyannote는 파일 경로 필요)
        pcm_bytes = await audio.read()
        if len(pcm_bytes) == 0:
            return JSONResponse(
                status_code=400,
                content={"status": "error", "message": "오디오 데이터가 비어있습니다"},
            )

        # PCM16 → WAV 변환
        import wave
        tmp_fd, tmp_path = tempfile.mkstemp(suffix=".wav")
        try:
            with wave.open(tmp_path, "wb") as wf:
                wf.setnchannels(1)
                wf.setsampwidth(2)  # 16-bit
                wf.setframerate(16000)
                wf.writeframes(pcm_bytes)

            # diarization 실행
            diarize_kwargs = {}
            if num_speakers > 0:
                diarize_kwargs["num_speakers"] = num_speakers

            logger.info("화자 구분 시작: %.1f초 오디오", len(pcm_bytes) / (16000 * 2))
            diarization = pipeline(tmp_path, **diarize_kwargs)

            # 결과 파싱
            segments = []
            speakers = set()
            for turn, _, speaker in diarization.itertracks(yield_label=True):
                segments.append({
                    "speaker": speaker,
                    "start": round(turn.start, 3),
                    "end": round(turn.end, 3),
                })
                speakers.add(speaker)

            processing_time = int((time.monotonic() - start_time) * 1000)
            logger.info(
                "화자 구분 완료: %d개 세그먼트, %d명 화자, %dms",
                len(segments),
                len(speakers),
                processing_time,
            )

            return JSONResponse(
                content={
                    "segments": segments,
                    "num_speakers": len(speakers),
                    "processing_time": processing_time,
                }
            )
        finally:
            os.close(tmp_fd)
            os.unlink(tmp_path)

    except Exception as exc:
        logger.error("화자 구분 실패: %s", exc, exc_info=True)
        return JSONResponse(
            status_code=500,
            content={"status": "error", "message": str(exc)},
        )


@app.post("/shutdown")
async def shutdown() -> JSONResponse:
    """서버를 graceful하게 종료한다."""
    logger.info("종료 요청 수신")

    if _server is not None:
        _server.should_exit = True

    return JSONResponse(content={"status": "shutting_down"})


# ── 메인 ───────────────────────────────────────────────────


def main() -> None:
    """CLI 진입점."""
    global _server

    parser = argparse.ArgumentParser(description="D3RO-VOICE STT Sidecar")
    parser.add_argument(
        "--port",
        type=int,
        default=18765,
        help="HTTP 서버 포트 (기본: 18765)",
    )
    parser.add_argument(
        "--host",
        type=str,
        default="127.0.0.1",
        help="HTTP 서버 호스트 (기본: 127.0.0.1)",
    )
    args = parser.parse_args()

    logger.info("D3RO-VOICE STT Sidecar 시작 (port=%d)", args.port)

    # SIGINT/SIGTERM 핸들러
    def signal_handler(signum: int, _frame: object) -> None:
        sig_name = signal.Signals(signum).name
        logger.info("시그널 수신: %s, 종료 시작", sig_name)
        if _server is not None:
            _server.should_exit = True

    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)

    # uvicorn 서버 구성 및 시작
    config = uvicorn.Config(
        app=app,
        host=args.host,
        port=args.port,
        log_level="warning",  # uvicorn 자체 로그는 최소화 (우리 로거 사용)
        access_log=False,
    )
    _server = uvicorn.Server(config)
    _server.run()

    logger.info("Sidecar 서버 종료 완료")


if __name__ == "__main__":
    main()