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

"""효과음 API 라우터

Freesound API 연동
"""

import uuid
from datetime import datetime
from typing import Optional, List

from fastapi import APIRouter, HTTPException, Depends, Query
from fastapi.responses import Response
from pydantic import BaseModel

from app.database import Database, get_db
from app.services.freesound_client import freesound_client

router = APIRouter(prefix="/api/v1/sound-effects", tags=["sound-effects"])


# ========================================
# Pydantic 모델
# ========================================

class SoundEffectResponse(BaseModel):
    """효과음 응답"""
    id: str
    freesound_id: Optional[int] = None
    name: str
    description: str
    duration: float
    tags: List[str] = []
    preview_url: Optional[str] = None
    license: str = ""
    username: Optional[str] = None
    source: str = "freesound"  # freesound | local


class SoundEffectSearchResponse(BaseModel):
    """효과음 검색 응답"""
    count: int
    page: int
    page_size: int
    results: List[SoundEffectResponse]


class SoundEffectImportRequest(BaseModel):
    """효과음 가져오기 요청"""
    freesound_id: int


# ========================================
# API 엔드포인트
# ========================================

@router.get("/search", response_model=SoundEffectSearchResponse)
async def search_sound_effects(
    query: str = Query(..., min_length=1, description="검색어"),
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    min_duration: Optional[float] = Query(None, ge=0, description="최소 길이 (초)"),
    max_duration: Optional[float] = Query(None, ge=0, description="최대 길이 (초)"),
    sort: str = Query("score", description="정렬 (score, duration_asc, duration_desc)"),
):
    """Freesound에서 효과음 검색"""
    try:
        result = await freesound_client.search(
            query=query,
            page=page,
            page_size=page_size,
            min_duration=min_duration,
            max_duration=max_duration,
            sort=sort,
        )

        # 응답 형식 변환
        sounds = []
        for item in result["results"]:
            sounds.append(SoundEffectResponse(
                id=f"fs_{item['freesound_id']}",
                freesound_id=item["freesound_id"],
                name=item["name"],
                description=item["description"],
                duration=item["duration"],
                tags=item["tags"],
                preview_url=item["preview_url"],
                license=item["license"],
                username=item.get("username"),
                source="freesound",
            ))

        return SoundEffectSearchResponse(
            count=result["count"],
            page=page,
            page_size=page_size,
            results=sounds,
        )

    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Search failed: {str(e)}")


@router.get("/library", response_model=SoundEffectSearchResponse)
async def list_local_sound_effects(
    page: int = Query(1, ge=1),
    page_size: int = Query(20, ge=1, le=100),
    category: Optional[str] = Query(None, description="카테고리 필터"),
    db: Database = Depends(get_db),
):
    """로컬 효과음 라이브러리 조회"""
    query = {}
    if category:
        query["categories"] = category

    total = await db.sound_effects.count_documents(query)
    skip = (page - 1) * page_size

    cursor = db.sound_effects.find(query).sort("created_at", -1).skip(skip).limit(page_size)

    sounds = []
    async for doc in cursor:
        sounds.append(SoundEffectResponse(
            id=str(doc["_id"]),
            freesound_id=doc.get("source_id"),
            name=doc["name"],
            description=doc.get("description", ""),
            duration=doc.get("duration_seconds", 0),
            tags=doc.get("tags", []),
            preview_url=None,  # 로컬 파일은 별도 엔드포인트로 제공
            license=doc.get("license", ""),
            source="local",
        ))

    return SoundEffectSearchResponse(
        count=total,
        page=page,
        page_size=page_size,
        results=sounds,
    )


@router.post("/import", response_model=SoundEffectResponse)
async def import_sound_effect(
    request: SoundEffectImportRequest,
    db: Database = Depends(get_db),
):
    """Freesound에서 효과음 가져오기 (로컬 캐시)"""
    try:
        # Freesound에서 상세 정보 조회
        sound_info = await freesound_client.get_sound(request.freesound_id)

        # 프리뷰 다운로드
        preview_url = sound_info.get("previews", {}).get("preview-hq-mp3", "")
        if not preview_url:
            raise HTTPException(status_code=400, detail="Preview not available")

        audio_bytes = await freesound_client.download_preview(preview_url)

        # GridFS에 저장
        file_id = await db.save_audio(
            audio_bytes,
            f"sfx_{request.freesound_id}.mp3",
            content_type="audio/mpeg",
            metadata={"freesound_id": request.freesound_id},
        )

        # DB에 메타데이터 저장
        now = datetime.utcnow()
        doc = {
            "name": sound_info.get("name", ""),
            "description": sound_info.get("description", ""),
            "source": "freesound",
            "source_id": request.freesound_id,
            "source_url": f"https://freesound.org/s/{request.freesound_id}/",
            "audio_file_id": file_id,
            "duration_seconds": sound_info.get("duration", 0),
            "format": "mp3",
            "categories": [],
            "tags": sound_info.get("tags", [])[:20],  # 최대 20개
            "license": sound_info.get("license", ""),
            "attribution": sound_info.get("username", ""),
            "created_at": now,
            "updated_at": now,
        }

        result = await db.sound_effects.insert_one(doc)

        return SoundEffectResponse(
            id=str(result.inserted_id),
            freesound_id=request.freesound_id,
            name=doc["name"],
            description=doc["description"],
            duration=doc["duration_seconds"],
            tags=doc["tags"],
            license=doc["license"],
            source="local",
        )

    except HTTPException:
        raise
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Import failed: {str(e)}")


@router.get("/{sound_id}")
async def get_sound_effect_info(
    sound_id: str,
    db: Database = Depends(get_db),
):
    """효과음 상세 정보 조회"""
    # Freesound ID인 경우
    if sound_id.startswith("fs_"):
        freesound_id = int(sound_id[3:])
        try:
            sound_info = await freesound_client.get_sound(freesound_id)
            return SoundEffectResponse(
                id=sound_id,
                freesound_id=freesound_id,
                name=sound_info.get("name", ""),
                description=sound_info.get("description", ""),
                duration=sound_info.get("duration", 0),
                tags=sound_info.get("tags", []),
                preview_url=sound_info.get("previews", {}).get("preview-hq-mp3", ""),
                license=sound_info.get("license", ""),
                source="freesound",
            )
        except Exception as e:
            raise HTTPException(status_code=404, detail="Sound not found")

    # 로컬 ID인 경우
    from bson import ObjectId
    try:
        doc = await db.sound_effects.find_one({"_id": ObjectId(sound_id)})
    except:
        raise HTTPException(status_code=400, detail="Invalid sound ID")

    if not doc:
        raise HTTPException(status_code=404, detail="Sound not found")

    return SoundEffectResponse(
        id=str(doc["_id"]),
        freesound_id=doc.get("source_id"),
        name=doc["name"],
        description=doc.get("description", ""),
        duration=doc.get("duration_seconds", 0),
        tags=doc.get("tags", []),
        license=doc.get("license", ""),
        source="local",
    )


@router.get("/{sound_id}/audio")
async def get_sound_effect_audio(
    sound_id: str,
    db: Database = Depends(get_db),
):
    """효과음 오디오 스트리밍"""
    # Freesound ID인 경우 프리뷰 리다이렉트
    if sound_id.startswith("fs_"):
        freesound_id = int(sound_id[3:])
        try:
            sound_info = await freesound_client.get_sound(freesound_id)
            preview_url = sound_info.get("previews", {}).get("preview-hq-mp3", "")
            if preview_url:
                audio_bytes = await freesound_client.download_preview(preview_url)
                return Response(
                    content=audio_bytes,
                    media_type="audio/mpeg",
                    headers={"Content-Disposition": f'inline; filename="{freesound_id}.mp3"'},
                )
        except Exception as e:
            raise HTTPException(status_code=404, detail="Audio not found")

    # 로컬 ID인 경우
    from bson import ObjectId
    try:
        doc = await db.sound_effects.find_one({"_id": ObjectId(sound_id)})
    except:
        raise HTTPException(status_code=400, detail="Invalid sound ID")

    if not doc or not doc.get("audio_file_id"):
        raise HTTPException(status_code=404, detail="Audio not found")

    audio_bytes = await db.get_audio(doc["audio_file_id"])
    content_type = "audio/mpeg" if doc.get("format") == "mp3" else "audio/wav"

    return Response(
        content=audio_bytes,
        media_type=content_type,
        headers={"Content-Disposition": f'inline; filename="{sound_id}.{doc.get("format", "wav")}"'},
    )


@router.get("/categories")
async def list_categories(
    db: Database = Depends(get_db),
):
    """효과음 카테고리 목록"""
    # 로컬 라이브러리의 카테고리 집계
    pipeline = [
        {"$unwind": "$categories"},
        {"$group": {"_id": "$categories", "count": {"$sum": 1}}},
        {"$sort": {"count": -1}},
    ]

    categories = []
    async for doc in db.sound_effects.aggregate(pipeline):
        categories.append({
            "name": doc["_id"],
            "count": doc["count"],
        })

    return {"categories": categories}


@router.delete("/{sound_id}")
async def delete_sound_effect(
    sound_id: str,
    db: Database = Depends(get_db),
):
    """로컬 효과음 삭제"""
    if sound_id.startswith("fs_"):
        raise HTTPException(status_code=400, detail="Cannot delete Freesound reference")

    from bson import ObjectId
    try:
        doc = await db.sound_effects.find_one({"_id": ObjectId(sound_id)})
    except:
        raise HTTPException(status_code=400, detail="Invalid sound ID")

    if not doc:
        raise HTTPException(status_code=404, detail="Sound not found")

    # 오디오 파일 삭제
    if doc.get("audio_file_id"):
        await db.delete_audio(doc["audio_file_id"])

    # 문서 삭제
    await db.sound_effects.delete_one({"_id": ObjectId(sound_id)})

    return {"status": "deleted", "sound_id": sound_id}