diff --git a/music-lab/app/suno_provider.py b/music-lab/app/suno_provider.py index 073557a..7cc2be7 100644 --- a/music-lab/app/suno_provider.py +++ b/music-lab/app/suno_provider.py @@ -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", ""), }