""" Suno API Provider — Suno REST API를 통한 음악 생성 https://apicast.suno.ai/v1 """ 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://apicast.suno.ai/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) 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}, timeout=30, ) if resp.status_code == 200: return resp.json() except Exception as e: logger.warning("Suno lyrics API error: %s", e) 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) resp = requests.post( f"{SUNO_BASE_URL}/songs", headers=_headers(), json=payload, timeout=30, ) if resp.status_code not in (200, 201): 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 응답이 비어있습니다") return primary_song = songs[0] suno_song_id = primary_song.get("id", "") update_task(task_id, "processing", 15, "곡 생성 대기열에 등록됨...") # ── 2단계: 상태 폴링 ── completed_song = _poll_until_complete(task_id, suno_song_id) if not completed_song: return # 에러는 _poll_until_complete 내부에서 처리 update_task(task_id, "processing", 80, "오디오 파일 다운로드 중...") # ── 3단계: 메인 트랙 다운로드 + 등록 ── track = _download_and_register( task_id=task_id, song=completed_song, 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(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 # 보조 변형 실패는 무시 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 → Suno API 요청 형식으로 변환.""" payload = {} # 프롬프트 조합: 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" # 스타일 태그 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) # 제목 if params.get("title"): payload["title"] = params["title"] # 가사 / 인스트루멘탈 if params.get("instrumental", False): payload["instrumental"] = True elif params.get("lyrics"): payload["lyrics"] = params["lyrics"] 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_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: 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 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", "") if not audio_url_remote: update_task(task_id, "failed", 0, "", error="Suno 응답에 audio_url이 없습니다") 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}" # 메타데이터 조합 genre = params.get("genre", song.get("style", "")) 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": 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", ""), "suno_id": song.get("id", ""), } return add_track(track_data)