gahusb/web-page-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

music-lab: Suno API를 sunoapi.org 래퍼로 전환 (URL·요청·응답 형식 수정)

Browse Source 
- Base URL: apicast.suno.ai → api.sunoapi.org/api/v1
- 생성: POST /generate (customMode, model, instrumental 필드)
- 폴링: GET /generate/record-info?taskId=xxx (PENDING→SUCCESS)
- 가사: /lyrics 비동기 폴링 방식으로 변경
- 응답 필드: camelCase (audioUrl, imageUrl, sunoData) 대응

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
gahusb
2026-04-03 08:38:20 +09:00
parent f5c58a5aa5
commit 2a1d8716c7
1 changed files with 171 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files

272
music-lab/app/suno_provider.py
View File

@@ -1,6 +1,6 @@
"""
Suno API Provider — Suno REST API를 통한 음악 생성
https://apicast.suno.ai/v1
Suno API Provider — sunoapi.org 래퍼를 통한 음악 생성
https://docs.sunoapi.org/suno-api/quickstart
"""
import os
@@ -13,14 +13,14 @@ from .db import update_task, add_track
logger = logging.getLogger(__name__)
SUNO_BASE_URL = "https://apicast.suno.ai/v1"
SUNO_BASE_URL = "https://api.sunoapi.org/api/v1"
SUNO_API_KEY = os.getenv("SUNO_API_KEY", "")
MUSIC_DATA_DIR = "/app/data/music"
MUSIC_MEDIA_BASE = os.getenv("MUSIC_MEDIA_BASE", "/media/music")
# 폴링 설정
POLL_INTERVAL = 6 # 초
POLL_MAX_ATTEMPTS = 50 # 최대 5분 (6초 × 50)
POLL_INTERVAL = 8 # 초 (Suno 생성은 30초~3분 소요)
POLL_MAX_ATTEMPTS = 40 # 최대 ~5분 20초 (8초 × 40)
def _headers() -> dict:
@@ -38,16 +38,54 @@ def generate_lyrics(prompt: str) -> Optional[dict]:
resp = requests.post(
f"{SUNO_BASE_URL}/lyrics",
headers=_headers(),
json={"prompt": prompt},
json={"prompt": prompt[:200]}, # max 200 chars
timeout=30,
)
if resp.status_code == 200:
return resp.json()
body = resp.json()
# sunoapi.org 래퍼 응답: {code, msg, data: {taskId}}
if body.get("code") == 200:
task_id = body.get("data", {}).get("taskId", "")
if task_id:
# 가사 생성도 비동기 — 폴링으로 결과 대기
return _poll_lyrics(task_id)
return body
except Exception as e:
logger.warning("Suno lyrics API error: %s", e)
return None
def _poll_lyrics(lyrics_task_id: str) -> Optional[dict]:
"""가사 생성 결과를 폴링하여 반환."""
for _ in range(15): # 최대 ~45초
time.sleep(3)
try:
resp = requests.get(
f"{SUNO_BASE_URL}/lyrics/record-info",
headers=_headers(),
params={"taskId": lyrics_task_id},
timeout=15,
)
if resp.status_code != 200:
continue
body = resp.json()
data = body.get("data", {})
status = data.get("status", "")
if status == "complete":
items = data.get("data") or data.get("sunoData") or []
if items and isinstance(items, list):
return {
"id": lyrics_task_id,
"status": "complete",
"text": items[0].get("text", ""),
"title": items[0].get("title", ""),
}
return {"id": lyrics_task_id, "status": "complete", "text": ""}
except Exception:
continue
return None
def run_suno_generation(task_id: str, params: dict) -> None:
"""
BackgroundTask: Suno API로 곡 생성 → MP3 다운로드 → 라이브러리 등록.
@@ -65,33 +103,37 @@ def run_suno_generation(task_id: str, params: dict) -> None:
# ── 1단계: 곡 생성 요청 ──
payload = _build_suno_payload(params)
logger.info("Suno generate request: %s", payload)
resp = requests.post(
f"{SUNO_BASE_URL}/songs",
f"{SUNO_BASE_URL}/generate",
headers=_headers(),
json=payload,
timeout=30,
)
if resp.status_code not in (200, 201):
if resp.status_code != 200:
error_detail = resp.text[:300] if resp.text else f"HTTP {resp.status_code}"
update_task(task_id, "failed", 0, "", error=f"Suno API 오류: {error_detail}")
return
song_data = resp.json()
# Suno 응답 형태: 단일 객체 또는 리스트
songs = song_data if isinstance(song_data, list) else [song_data]
if not songs:
update_task(task_id, "failed", 0, "", error="Suno API 응답이 비어있습니다")
body = resp.json()
if body.get("code") != 200:
update_task(task_id, "failed", 0, "",
error=f"Suno API 거부: {body.get('msg', 'unknown error')}")
return
primary_song = songs[0]
suno_song_id = primary_song.get("id", "")
suno_task_id = body.get("data", {}).get("taskId", "")
if not suno_task_id:
update_task(task_id, "failed", 0, "", error="Suno 응답에 taskId가 없습니다")
return
update_task(task_id, "processing", 15, "곡 생성 대기열에 등록됨...")
logger.info("Suno task created: %s (internal: %s)", suno_task_id, task_id)
# ── 2단계: 상태 폴링 ──
completed_song = _poll_until_complete(task_id, suno_song_id)
if not completed_song:
completed_tracks = _poll_until_complete(task_id, suno_task_id)
if not completed_tracks:
return # 에러는 _poll_until_complete 내부에서 처리
update_task(task_id, "processing", 80, "오디오 파일 다운로드 중...")
@@ -99,7 +141,7 @@ def run_suno_generation(task_id: str, params: dict) -> None:
# ── 3단계: 메인 트랙 다운로드 + 등록 ──
track = _download_and_register(
task_id=task_id,
song=completed_song,
song=completed_tracks[0],
params=params,
filename_suffix="",
)
@@ -110,20 +152,16 @@ def run_suno_generation(task_id: str, params: dict) -> None:
update_task(task_id, "succeeded", 100, "생성 완료", audio_url=audio_url)
# ── 4단계: 두 번째 변형이 있으면 추가 등록 ──
if len(songs) > 1:
second_id = songs[1].get("id", "")
if second_id:
try:
second_song = _fetch_song(second_id)
if second_song and second_song.get("status") == "complete":
_download_and_register(
task_id=f"{task_id}_v2",
song=second_song,
params=params,
filename_suffix="_v2",
)
except Exception:
pass # 보조 변형 실패는 무시
if len(completed_tracks) > 1:
try:
_download_and_register(
task_id=f"{task_id}_v2",
song=completed_tracks[1],
params=params,
filename_suffix="_v2",
)
except Exception:
pass # 보조 변형 실패는 무시
except requests.Timeout:
update_task(task_id, "failed", 0, "", error="Suno API 타임아웃")
@@ -133,81 +171,112 @@ def run_suno_generation(task_id: str, params: dict) -> None:
def _build_suno_payload(params: dict) -> dict:
"""프론트엔드 params → Suno API 요청 형식으로 변환."""
payload = {}
"""프론트엔드 params → sunoapi.org 요청 형식으로 변환."""
instrumental = params.get("instrumental", False)
has_lyrics = bool(params.get("lyrics"))
# 프롬프트 조합: prompt + genre + moods
parts = []
if params.get("prompt"):
parts.append(params["prompt"])
if params.get("genre"):
parts.append(params["genre"])
if params.get("moods"):
parts.append(", ".join(params["moods"]))
payload["prompt"] = " ".join(parts) if parts else "instrumental music"
# customMode: 가사 또는 스타일 지정 시 활성화
custom_mode = has_lyrics or bool(params.get("genre")) or bool(params.get("moods"))
# 스타일 태그
style_parts = []
if params.get("genre"):
style_parts.append(params["genre"])
if params.get("moods"):
style_parts.extend(params["moods"])
if params.get("instruments"):
style_parts.extend(params["instruments"][:3]) # 너무 많으면 잘림
if style_parts:
payload["style"] = " ".join(style_parts)
payload = {
"customMode": custom_mode,
"instrumental": instrumental,
"model": "V4", # 안정적인 기본 모델
}
# 제목
if params.get("title"):
payload["title"] = params["title"]
if custom_mode:
# ── Custom Mode ──
# prompt: 가사 (instrumental이면 빈 문자열)
if instrumental:
payload["prompt"] = ""
elif has_lyrics:
payload["prompt"] = params["lyrics"][:3000]
else:
# 가사 없이 커스텀 모드 — 프롬프트를 가사 위치에
prompt_text = params.get("prompt", "")
payload["prompt"] = prompt_text[:3000] if prompt_text else ""
# 가사 / 인스트루멘탈
if params.get("instrumental", False):
payload["instrumental"] = True
elif params.get("lyrics"):
payload["lyrics"] = params["lyrics"]
# style: 장르 + 분위기 + 악기 조합
style_parts = []
if params.get("genre"):
style_parts.append(params["genre"])
if params.get("moods"):
style_parts.extend(params["moods"])
if params.get("instruments"):
style_parts.extend(params["instruments"][:3])
if style_parts:
payload["style"] = ", ".join(style_parts)[:200]
# title
if params.get("title"):
payload["title"] = params["title"][:80]
else:
# ── Simple Mode ──
# prompt: 자연어 설명
parts = []
if params.get("prompt"):
parts.append(params["prompt"])
if params.get("genre"):
parts.append(params["genre"])
if params.get("moods"):
parts.append(", ".join(params["moods"]))
payload["prompt"] = " ".join(parts)[:500] if parts else "instrumental music"
return payload
def _fetch_song(song_id: str) -> Optional[dict]:
"""Suno에서 단일 곡 상태 조회."""
try:
resp = requests.get(
f"{SUNO_BASE_URL}/songs/{song_id}",
headers=_headers(),
timeout=15,
)
if resp.status_code == 200:
return resp.json()
except Exception as e:
logger.warning("Suno fetch song error: %s", e)
return None
def _poll_until_complete(task_id: str, suno_task_id: str) -> Optional[list]:
"""sunoapi.org record-info를 폴링하여 SUCCESS가 될 때까지 대기."""
error_statuses = {
"CREATE_TASK_FAILED", "GENERATE_AUDIO_FAILED",
"CALLBACK_EXCEPTION", "SENSITIVE_WORD_ERROR",
}
def _poll_until_complete(task_id: str, suno_song_id: str) -> Optional[dict]:
"""Suno 곡 상태를 폴링하여 complete가 될 때까지 대기."""
for attempt in range(POLL_MAX_ATTEMPTS):
time.sleep(POLL_INTERVAL)
song = _fetch_song(suno_song_id)
if not song:
try:
resp = requests.get(
f"{SUNO_BASE_URL}/generate/record-info",
headers=_headers(),
params={"taskId": suno_task_id},
timeout=15,
)
if resp.status_code != 200:
continue
body = resp.json()
if body.get("code") != 200:
continue
data = body.get("data", {})
status = data.get("status", "")
progress = min(15 + int((attempt / POLL_MAX_ATTEMPTS) * 65), 79)
if status == "PENDING":
update_task(task_id, "processing", progress, "대기열에서 대기 중...")
elif status == "TEXT_SUCCESS":
update_task(task_id, "processing", progress, "가사 생성 완료, 음악 생성 중...")
elif status == "FIRST_SUCCESS":
update_task(task_id, "processing", max(progress, 60), "첫 번째 트랙 완료, 두 번째 생성 중...")
elif status == "SUCCESS":
# 완료 — sunoData 배열에서 트랙 추출
tracks = data.get("sunoData") or data.get("data") or []
if tracks:
return tracks
update_task(task_id, "failed", 0, "", error="Suno 생성 완료했으나 트랙 데이터 없음")
return None
elif status in error_statuses:
error_msg = data.get("errorMessage") or data.get("msg") or f"Suno 생성 실패 ({status})"
update_task(task_id, "failed", 0, "", error=error_msg)
return None
else:
update_task(task_id, "processing", progress, f"처리 중... ({status})")
except Exception as e:
logger.warning("Suno poll error (attempt %d): %s", attempt, e)
continue
status = song.get("status", "")
progress = min(15 + int((attempt / POLL_MAX_ATTEMPTS) * 65), 79)
if status == "streaming":
update_task(task_id, "processing", progress, "AI가 음악을 작곡 중...")
elif status == "complete":
return song
elif status == "error":
error_msg = song.get("error_message", "Suno 생성 실패")
update_task(task_id, "failed", 0, "", error=error_msg)
return None
else:
update_task(task_id, "processing", progress, f"대기 중... ({status})")
update_task(task_id, "failed", 0, "", error="Suno 생성 타임아웃 (5분 초과)")
return None
@@ -216,9 +285,10 @@ def _download_and_register(
task_id: str, song: dict, params: dict, filename_suffix: str = "",
) -> Optional[dict]:
"""Suno CDN에서 MP3 다운로드 → 로컬 저장 → 라이브러리 등록."""
audio_url_remote = song.get("audio_url", "")
# sunoapi.org 응답 필드: audioUrl (camelCase)
audio_url_remote = song.get("audioUrl") or song.get("audio_url", "")
if not audio_url_remote:
update_task(task_id, "failed", 0, "", error="Suno 응답에 audio_url이 없습니다")
update_task(task_id, "failed", 0, "", error="Suno 응답에 audioUrl이 없습니다")
return None
filename = f"{task_id}{filename_suffix}.mp3"
@@ -236,8 +306,8 @@ def _download_and_register(
local_audio_url = f"{MUSIC_MEDIA_BASE}/{filename}"
# 메타데이터 조합
genre = params.get("genre", song.get("style", ""))
# 메타데이터 조합 (sunoapi.org 필드는 camelCase)
genre = params.get("genre", song.get("tags", ""))
moods = params.get("moods", [])
mood_str = moods[0] if moods else "Original"
title = (
@@ -255,13 +325,13 @@ def _download_and_register(
"bpm": params.get("bpm"),
"key": params.get("key", ""),
"scale": params.get("scale", ""),
"prompt": params.get("prompt", ""),
"prompt": song.get("prompt", params.get("prompt", "")),
"audio_url": local_audio_url,
"file_path": file_path,
"task_id": task_id,
"provider": "suno",
"lyrics": song.get("lyrics", params.get("lyrics", "")),
"image_url": song.get("image_url", ""),
"lyrics": song.get("prompt", params.get("lyrics", "")),
"image_url": song.get("imageUrl") or song.get("image_url", ""),
"suno_id": song.get("id", ""),
}