drama-studio/audio-studio-api/app/routers/recordings.py

"""녹음 관리 API 라우터"""

import uuid
import io
from typing import List

from fastapi import APIRouter, HTTPException, Depends, UploadFile, File, Form
from fastapi.responses import Response
import soundfile as sf
import numpy as np

from app.database import Database, get_db
from app.models.voice import RecordingValidateResponse, RecordingUploadResponse

router = APIRouter(prefix="/api/v1/recordings", tags=["recordings"])


def analyze_audio(audio_bytes: bytes) -> dict:
    """오디오 파일 분석

    Returns:
        duration, sample_rate, quality_score, issues
    """
    try:
        # 오디오 로드
        audio_data, sample_rate = sf.read(io.BytesIO(audio_bytes))

        # 모노로 변환
        if len(audio_data.shape) > 1:
            audio_data = audio_data.mean(axis=1)

        duration = len(audio_data) / sample_rate

        # 품질 분석
        issues = []
        quality_score = 1.0

        # 길이 체크
        if duration < 1.0:
            issues.append("오디오가 너무 짧습니다 (최소 1초 이상)")
            quality_score -= 0.3
        elif duration < 3.0:
            issues.append("Voice Clone에는 3초 이상의 오디오가 권장됩니다")
            quality_score -= 0.1

        # RMS 레벨 체크 (볼륨)
        rms = np.sqrt(np.mean(audio_data ** 2))
        if rms < 0.01:
            issues.append("볼륨이 너무 낮습니다")
            quality_score -= 0.2
        elif rms > 0.5:
            issues.append("볼륨이 너무 높습니다 (클리핑 가능성)")
            quality_score -= 0.1

        # 피크 체크
        peak = np.max(np.abs(audio_data))
        if peak > 0.99:
            issues.append("오디오가 클리핑되었습니다")
            quality_score -= 0.2

        # 노이즈 체크 (간단한 휴리스틱)
        # 실제로는 더 정교한 노이즈 감지 필요
        silence_threshold = 0.01
        silent_samples = np.sum(np.abs(audio_data) < silence_threshold)
        silence_ratio = silent_samples / len(audio_data)

        if silence_ratio > 0.7:
            issues.append("대부분이 무음입니다")
            quality_score -= 0.3
        elif silence_ratio > 0.5:
            issues.append("무음 구간이 많습니다")
            quality_score -= 0.1

        quality_score = max(0.0, min(1.0, quality_score))

        return {
            "duration": duration,
            "sample_rate": sample_rate,
            "quality_score": quality_score,
            "issues": issues,
            "rms": float(rms),
            "peak": float(peak),
        }

    except Exception as e:
        return {
            "duration": 0,
            "sample_rate": 0,
            "quality_score": 0,
            "issues": [f"오디오 분석 실패: {str(e)}"],
        }


@router.post("/validate", response_model=RecordingValidateResponse)
async def validate_recording(
    audio: UploadFile = File(..., description="검증할 오디오 파일"),
):
    """녹음 품질 검증

    Voice Clone에 사용할 녹음의 품질을 검증합니다.
    """
    audio_bytes = await audio.read()

    if len(audio_bytes) < 1000:
        raise HTTPException(status_code=400, detail="파일이 너무 작습니다")

    analysis = analyze_audio(audio_bytes)

    return RecordingValidateResponse(
        valid=analysis["quality_score"] > 0.5 and analysis["duration"] > 1.0,
        duration=analysis["duration"],
        sample_rate=analysis["sample_rate"],
        quality_score=analysis["quality_score"],
        issues=analysis["issues"],
    )


@router.post("/upload", response_model=RecordingUploadResponse)
async def upload_recording(
    audio: UploadFile = File(..., description="업로드할 오디오 파일"),
    transcript: str = Form(None, description="오디오의 텍스트 내용"),
    db: Database = Depends(get_db),
):
    """녹음 파일 업로드

    Voice Clone에 사용할 녹음을 업로드합니다.
    """
    audio_bytes = await audio.read()

    # 품질 분석
    analysis = analyze_audio(audio_bytes)

    if analysis["duration"] < 0.5:
        raise HTTPException(status_code=400, detail="오디오가 너무 짧습니다")

    # GridFS에 저장
    file_id = await db.save_audio(
        audio_bytes,
        audio.filename or f"recording_{uuid.uuid4()}.wav",
        metadata={
            "type": "recording",
            "transcript": transcript,
            "duration": analysis["duration"],
            "sample_rate": analysis["sample_rate"],
            "quality_score": analysis["quality_score"],
        },
    )

    return RecordingUploadResponse(
        file_id=file_id,
        filename=audio.filename or "recording.wav",
        duration=analysis["duration"],
        sample_rate=analysis["sample_rate"],
    )


@router.get("/{file_id}")
async def get_recording(
    file_id: str,
    db: Database = Depends(get_db),
):
    """녹음 파일 다운로드"""
    try:
        audio_bytes = await db.get_audio(file_id)
        return Response(
            content=audio_bytes,
            media_type="audio/wav",
            headers={"Content-Disposition": f'attachment; filename="{file_id}.wav"'},
        )
    except Exception as e:
        raise HTTPException(status_code=404, detail="Recording not found")


@router.delete("/{file_id}")
async def delete_recording(
    file_id: str,
    db: Database = Depends(get_db),
):
    """녹음 파일 삭제"""
    try:
        await db.delete_audio(file_id)
        return {"status": "deleted", "file_id": file_id}
    except Exception as e:
        raise HTTPException(status_code=404, detail="Recording not found")