178 lines
5.3 KiB
Python
178 lines
5.3 KiB
Python
"""
|
|
Suno API 클라이언트 모듈
|
|
|
|
API 문서: https://docs.sunoapi.org
|
|
|
|
## 사용법
|
|
```python
|
|
from app.utils.suno import SunoService
|
|
|
|
# config에서 자동으로 API 키를 가져옴
|
|
suno = SunoService()
|
|
|
|
# 또는 명시적으로 API 키 전달
|
|
suno = SunoService(api_key="your_api_key")
|
|
|
|
# 음악 생성 요청
|
|
task_id = await suno.generate(
|
|
prompt="[Verse]\\n오늘도 좋은 하루...",
|
|
style="K-Pop, Happy, 110 BPM",
|
|
title="좋은 하루"
|
|
)
|
|
|
|
# 상태 확인 (폴링 방식)
|
|
result = await suno.get_task_status(task_id)
|
|
```
|
|
|
|
## 콜백 URL 사용법
|
|
generate() 호출 시 callback_url 파라미터를 전달하면 생성 완료 시 해당 URL로 POST 요청이 전송됩니다.
|
|
콜백 요청 형식:
|
|
```json
|
|
{
|
|
"code": 200,
|
|
"msg": "All generated successfully.",
|
|
"data": {
|
|
"callbackType": "complete",
|
|
"task_id": "작업ID",
|
|
"data": [
|
|
{
|
|
"id": "clip_id",
|
|
"audio_url": "https://...",
|
|
"image_url": "https://...",
|
|
"title": "곡 제목",
|
|
"status": "complete"
|
|
}
|
|
]
|
|
}
|
|
}
|
|
```
|
|
콜백 주의사항:
|
|
- HTTPS 프로토콜 권장
|
|
- 15초 내 응답 필수
|
|
- 동일 task_id에 대해 여러 콜백 수신 가능 (멱등성 처리 필요)
|
|
"""
|
|
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from config import apikey_settings
|
|
|
|
|
|
class SunoService:
|
|
"""Suno API를 통한 AI 음악 생성 서비스"""
|
|
|
|
BASE_URL = "https://api.sunoapi.org/api/v1"
|
|
|
|
def __init__(self, api_key: str | None = None):
|
|
"""
|
|
Args:
|
|
api_key: Suno API 키 (Bearer token으로 사용)
|
|
None일 경우 config에서 자동으로 가져옴
|
|
"""
|
|
self.api_key = api_key or apikey_settings.SUNO_API_KEY
|
|
self.headers = {
|
|
"Authorization": f"Bearer {self.api_key}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
async def generate(
|
|
self,
|
|
prompt: str,
|
|
genre: str | None = None,
|
|
callback_url: str | None = None,
|
|
) -> str:
|
|
"""
|
|
음악 생성 요청
|
|
|
|
Args:
|
|
prompt: 가사 (customMode=true일 때 가사로 사용)
|
|
1분 이내 길이의 노래에 적합한 가사여야 함
|
|
genre: 음악 장르 (예: "K-Pop", "Pop", "R&B", "Hip-Hop", "Ballad", "EDM")
|
|
None일 경우 style 파라미터를 전송하지 않음
|
|
callback_url: 생성 완료 시 알림 받을 URL (None일 경우 config에서 기본값 사용)
|
|
|
|
Returns:
|
|
task_id: 작업 추적용 ID
|
|
|
|
Note:
|
|
- 스트림 URL: 30-40초 내 생성
|
|
- 다운로드 URL: 2-3분 내 생성
|
|
- 생성되는 노래는 약 1분 이내의 길이
|
|
"""
|
|
# 정확히 1분 길이의 노래 생성을 위한 프롬프트 조건 추가
|
|
formatted_prompt = f"[Song Duration: Exactly 1 minute - Must be precisely 60 seconds]\n{prompt}"
|
|
|
|
# callback_url이 없으면 config에서 기본값 사용 (Suno API 필수 파라미터)
|
|
actual_callback_url = callback_url or apikey_settings.SUNO_CALLBACK_URL
|
|
|
|
payload: dict[str, Any] = {
|
|
"model": "V5",
|
|
"customMode": True,
|
|
"instrumental": False,
|
|
"prompt": formatted_prompt,
|
|
"callBackUrl": actual_callback_url,
|
|
}
|
|
|
|
# genre가 있을 때만 style 추가
|
|
if genre:
|
|
payload["style"] = genre
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.post(
|
|
f"{self.BASE_URL}/generate",
|
|
headers=self.headers,
|
|
json=payload,
|
|
timeout=30.0,
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
# 응답: {"code": 200, "msg": "success", "data": {"taskId": "..."}}
|
|
# API 응답 검증
|
|
if data is None:
|
|
raise ValueError("Suno API returned empty response")
|
|
|
|
if data.get("code") != 200:
|
|
error_msg = data.get("msg", "Unknown error")
|
|
raise ValueError(f"Suno API error: {error_msg}")
|
|
|
|
response_data = data.get("data")
|
|
if response_data is None:
|
|
raise ValueError(f"Suno API response missing 'data' field: {data}")
|
|
|
|
task_id = response_data.get("taskId")
|
|
if task_id is None:
|
|
raise ValueError(f"Suno API response missing 'taskId': {response_data}")
|
|
|
|
return task_id
|
|
|
|
async def get_task_status(self, task_id: str) -> dict[str, Any]:
|
|
"""
|
|
음악 생성 작업 상태 확인
|
|
|
|
Args:
|
|
task_id: generate()에서 반환된 작업 ID
|
|
|
|
Returns:
|
|
작업 상태 정보 (status, audio_url, image_url 등 포함)
|
|
|
|
Note:
|
|
폴링 방식으로 상태 확인 시 사용.
|
|
콜백 URL을 사용하면 폴링 없이 결과를 받을 수 있음.
|
|
"""
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.get(
|
|
f"{self.BASE_URL}/generate/record-info",
|
|
headers=self.headers,
|
|
params={"taskId": task_id},
|
|
timeout=30.0,
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
if data is None:
|
|
raise ValueError("Suno API returned empty response for task status")
|
|
|
|
return data
|