jwkim
277378477a
- 경기 맥락(전날 결과·순위·연전 차수·선발투수) 조립 → LLM 작사 (콜앤리스폰스·섹션 태그 형식) → Suno(sunoapi.org) 생성 → songs 테이블 - 워커: 킥오프 150분 전 윈도우 진입 시 생성 시작 + 2분 주기 폴링 - API: GET /api/songs/today · POST /api/songs/callback(싱크대) · POST /api/songs/generate(관리자 강제 생성) - 프론트: MusicBar 가 경기 상세·메인에서 오늘의 응원가를 동적 로드 (정적 플레이리스트는 폴백 유지) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
105 lines
3.3 KiB
Python
105 lines
3.3 KiB
Python
"""Suno 음악 생성 클라이언트 — sunoapi.org 서드파티 게이트웨이.
|
|
|
|
공식 Suno API 가 없어 게이트웨이를 쓴다. 흐름:
|
|
POST /api/v1/generate (customMode: 가사·스타일·제목) → taskId
|
|
GET /api/v1/generate/record-info?taskId= → status 폴링 → sunoData[] (보통 2곡)
|
|
|
|
오디오/커버는 게이트웨이 CDN URL 을 그대로 쓴다 (당일 소비 콘텐츠).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from ..config import settings
|
|
|
|
log = logging.getLogger("triplepick.suno")
|
|
|
|
# record-info status 값
|
|
SUNO_DONE = "SUCCESS"
|
|
SUNO_FAILED = {
|
|
"CREATE_TASK_FAILED",
|
|
"GENERATE_AUDIO_FAILED",
|
|
"CALLBACK_EXCEPTION",
|
|
"SENSITIVE_WORD_ERROR",
|
|
}
|
|
|
|
|
|
class SunoUnavailable(RuntimeError):
|
|
"""SUNO_API_KEY 미설정 등으로 호출 불가."""
|
|
|
|
|
|
class SunoError(RuntimeError):
|
|
"""게이트웨이가 에러 코드를 반환."""
|
|
|
|
|
|
def _headers() -> dict:
|
|
if not settings.suno_api_key:
|
|
raise SunoUnavailable("SUNO_API_KEY 미설정")
|
|
return {
|
|
"Authorization": f"Bearer {settings.suno_api_key}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
|
|
async def start_generation(title: str, style: str, lyrics: str) -> str:
|
|
"""생성 작업 시작 → taskId. customMode: 가사(prompt)·style·title 을 그대로 사용."""
|
|
import httpx
|
|
|
|
payload = {
|
|
"prompt": lyrics[:4900],
|
|
"style": style[:950],
|
|
"title": title[:80],
|
|
"customMode": True,
|
|
"instrumental": False,
|
|
"model": settings.suno_model,
|
|
# 게이트웨이 필수 파라미터 — 실제 완료 감지는 폴링으로 한다(콜백은 싱크대).
|
|
"callBackUrl": f"{settings.public_origin}/api/songs/callback",
|
|
}
|
|
async with httpx.AsyncClient(timeout=30) as c:
|
|
r = await c.post(
|
|
f"{settings.suno_api_base}/api/v1/generate",
|
|
json=payload,
|
|
headers=_headers(),
|
|
)
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
if data.get("code") != 200 or not (data.get("data") or {}).get("taskId"):
|
|
raise SunoError(f"generate 실패: {data.get('code')} {data.get('msg')}")
|
|
return data["data"]["taskId"]
|
|
|
|
|
|
async def get_task(task_id: str) -> dict:
|
|
"""작업 상태 조회 — {status, response: {sunoData: [...]}} 형태의 data 반환."""
|
|
import httpx
|
|
|
|
async with httpx.AsyncClient(timeout=30) as c:
|
|
r = await c.get(
|
|
f"{settings.suno_api_base}/api/v1/generate/record-info",
|
|
params={"taskId": task_id},
|
|
headers=_headers(),
|
|
)
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
if data.get("code") != 200:
|
|
raise SunoError(f"record-info 실패: {data.get('code')} {data.get('msg')}")
|
|
return data.get("data") or {}
|
|
|
|
|
|
def extract_tracks(task_data: dict) -> list[dict]:
|
|
"""record-info data → 표준 트랙 목록 [{title, audioUrl, imageUrl, duration}]."""
|
|
items = ((task_data.get("response") or {}).get("sunoData")) or []
|
|
out = []
|
|
for it in items:
|
|
url = it.get("audioUrl") or it.get("sourceAudioUrl") or ""
|
|
if not url:
|
|
continue
|
|
out.append(
|
|
{
|
|
"title": it.get("title") or "",
|
|
"audioUrl": url,
|
|
"imageUrl": it.get("imageUrl") or it.get("sourceImageUrl") or "",
|
|
"duration": it.get("duration"),
|
|
}
|
|
)
|
|
return out
|