""" Suno API Provider — sunoapi.org 래퍼를 통한 음악 생성 https://docs.sunoapi.org/suno-api/quickstart """ import os import time import logging import requests from typing import Optional from .db import update_task, add_track logger = logging.getLogger(__name__) 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 = 8 # 초 (Suno 생성은 30초~3분 소요) POLL_MAX_ATTEMPTS = 40 # 최대 ~5분 20초 (8초 × 40) def _headers() -> dict: return { "Authorization": f"Bearer {SUNO_API_KEY}", "Content-Type": "application/json", } def generate_lyrics(prompt: str) -> Optional[dict]: """Suno 가사 생성 API 호출. 곡 생성 전 가사 미리보기용.""" if not SUNO_API_KEY: return None try: resp = requests.post( f"{SUNO_BASE_URL}/lyrics", headers=_headers(), json={"prompt": prompt[:200]}, # max 200 chars timeout=30, ) if resp.status_code == 200: 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 다운로드 → 라이브러리 등록. Suno는 1회 생성 시 2개 변형을 반환하므로, 첫 번째를 메인으로 저장하고 두 번째는 별도 트랙으로 추가 등록한다. """ try: # ── 사전 검증 ── if not SUNO_API_KEY: update_task(task_id, "failed", 0, "", error="SUNO_API_KEY가 설정되지 않았습니다. .env 파일을 확인하세요.") return update_task(task_id, "processing", 5, "Suno API에 연결 중...") # ── 1단계: 곡 생성 요청 ── payload = _build_suno_payload(params) print(f"[SUNO DEBUG] generate payload: {payload}") resp = requests.post( f"{SUNO_BASE_URL}/generate", headers=_headers(), json=payload, timeout=30, ) 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 body = resp.json() if body.get("code") != 200: update_task(task_id, "failed", 0, "", error=f"Suno API 거부: {body.get('msg', 'unknown error')}") return print(f"[SUNO DEBUG] generate response: {body}") 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_tracks = _poll_until_complete(task_id, suno_task_id) if not completed_tracks: return # 에러는 _poll_until_complete 내부에서 처리 update_task(task_id, "processing", 80, "오디오 파일 다운로드 중...") # ── 3단계: 메인 트랙 다운로드 + 등록 ── track = _download_and_register( task_id=task_id, song=completed_tracks[0], params=params, filename_suffix="", ) if not track: return audio_url = track["audio_url"] update_task(task_id, "succeeded", 100, "생성 완료", audio_url=audio_url) # ── 4단계: 두 번째 변형이 있으면 추가 등록 ── 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 타임아웃") except Exception as e: logger.exception("Suno generation error for task %s", task_id) update_task(task_id, "failed", 0, "", error=str(e)) def _build_suno_payload(params: dict) -> dict: """프론트엔드 params → sunoapi.org 요청 형식으로 변환.""" instrumental = params.get("instrumental", False) has_lyrics = bool(params.get("lyrics")) # customMode: 가사 또는 스타일 지정 시 활성화 custom_mode = has_lyrics or bool(params.get("genre")) or bool(params.get("moods")) payload = { "customMode": custom_mode, "instrumental": instrumental, "model": "V4", # 안정적인 기본 모델 "callBackUrl": "https://example.com/noop", # 필수 파라미터, 폴링 방식이므로 더미 URL } 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 "" # 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 _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", } for attempt in range(POLL_MAX_ATTEMPTS): time.sleep(POLL_INTERVAL) 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": # 완료 — 트랙 추출 (디버깅용 print) import json as _json print(f"[SUNO DEBUG] SUCCESS response body: {_json.dumps(body, ensure_ascii=False, default=str)[:2000]}") tracks = data.get("sunoData") or data.get("data") or data.get("tracks") or [] # data 자체가 리스트일 수도 있음 if not tracks and isinstance(data, list): tracks = data if tracks: return tracks update_task(task_id, "failed", 0, "", error=f"Suno 생성 완료했으나 트랙 데이터 없음 (keys: {list(data.keys())})") 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 update_task(task_id, "failed", 0, "", error="Suno 생성 타임아웃 (5분 초과)") return None def _download_and_register( task_id: str, song: dict, params: dict, filename_suffix: str = "", ) -> Optional[dict]: """Suno CDN에서 MP3 다운로드 → 로컬 저장 → 라이브러리 등록.""" # 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 응답에 audioUrl이 없습니다") return None filename = f"{task_id}{filename_suffix}.mp3" file_path = os.path.join(MUSIC_DATA_DIR, filename) try: dl = requests.get(audio_url_remote, timeout=120, stream=True) dl.raise_for_status() with open(file_path, "wb") as f: for chunk in dl.iter_content(chunk_size=8192): f.write(chunk) except Exception as e: update_task(task_id, "failed", 0, "", error=f"오디오 다운로드 실패: {e}") return None local_audio_url = f"{MUSIC_MEDIA_BASE}/{filename}" # 메타데이터 조합 (sunoapi.org 필드는 camelCase) genre = params.get("genre", song.get("tags", "")) moods = params.get("moods", []) mood_str = moods[0] if moods else "Original" title = ( song.get("title") or params.get("title") or (f"{genre} — {mood_str} Mix" if genre else f"{mood_str} Mix") ) track_data = { "title": title, "genre": genre, "moods": moods, "instruments": params.get("instruments", []), "duration_sec": int(song["duration"]) if song.get("duration") else params.get("duration_sec"), "bpm": params.get("bpm"), "key": params.get("key", ""), "scale": params.get("scale", ""), "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("prompt", params.get("lyrics", "")), "image_url": song.get("imageUrl") or song.get("image_url", ""), "suno_id": song.get("id", ""), } return add_track(track_data)