760 lines
29 KiB
Python
760 lines
29 KiB
Python
"""
|
||
Suno API Provider — sunoapi.org 래퍼를 통한 음악 생성
|
||
https://docs.sunoapi.org/suno-api/quickstart
|
||
"""
|
||
|
||
import json
|
||
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_MEDIA_BASE = os.getenv("MUSIC_MEDIA_BASE", "/media/music")
|
||
|
||
# 폴링 설정
|
||
POLL_INTERVAL = 8 # 초 (Suno 생성은 30초~3분 소요)
|
||
POLL_MAX_ATTEMPTS = 40 # 최대 ~5분 20초 (8초 × 40)
|
||
|
||
# 사용 가능한 모델 목록
|
||
SUNO_MODELS = [
|
||
{"id": "V4", "name": "V4", "max_duration": "4분", "description": "안정적 품질, 빠른 생성"},
|
||
{"id": "V4_5", "name": "V4.5", "max_duration": "8분", "description": "향상된 장르 블렌딩"},
|
||
{"id": "V4_5PLUS", "name": "V4.5+", "max_duration": "8분", "description": "강화된 음악성"},
|
||
{"id": "V4_5ALL", "name": "V4.5 All", "max_duration": "8분", "description": "더 나은 곡 구조"},
|
||
{"id": "V5", "name": "V5", "max_duration": "8분", "description": "최신, 빠른 생성 + 뛰어난 음악성"},
|
||
{"id": "V5_5", "name": "V5.5", "max_duration": "8분", "description": "커스텀 모델, 최신 음악성"},
|
||
]
|
||
|
||
|
||
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)
|
||
|
||
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
|
||
|
||
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단계: 상태 폴링 ──
|
||
response = _poll_suno_record("/generate/record-info", suno_task_id, task_id)
|
||
if not response:
|
||
return
|
||
completed_tracks = response.get("sunoData") or []
|
||
if not completed_tracks:
|
||
update_task(task_id, "failed", 0, "", error="Suno 생성 완료했으나 트랙 데이터 없음")
|
||
return
|
||
|
||
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="",
|
||
)
|
||
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": params.get("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"
|
||
|
||
if params.get("vocal_gender"):
|
||
payload["vocalGender"] = params["vocal_gender"]
|
||
if params.get("negative_tags"):
|
||
payload["negativeTags"] = params["negative_tags"]
|
||
if params.get("style_weight") is not None:
|
||
payload["styleWeight"] = params["style_weight"]
|
||
if params.get("audio_weight") is not None:
|
||
payload["audioWeight"] = params["audio_weight"]
|
||
|
||
return payload
|
||
|
||
|
||
def _poll_suno_record(
|
||
record_info_path: str,
|
||
suno_task_id: str,
|
||
task_id: str,
|
||
max_attempts: int = POLL_MAX_ATTEMPTS,
|
||
interval: int = POLL_INTERVAL,
|
||
progress_msg_map: dict = None,
|
||
) -> Optional[dict]:
|
||
"""범용 Suno 작업 폴링. SUCCESS 시 response 객체 반환."""
|
||
error_statuses = {
|
||
"CREATE_TASK_FAILED", "GENERATE_AUDIO_FAILED",
|
||
"CALLBACK_EXCEPTION", "SENSITIVE_WORD_ERROR",
|
||
}
|
||
default_msgs = {
|
||
"PENDING": "대기열에서 대기 중...",
|
||
"TEXT_SUCCESS": "가사 생성 완료, 음악 생성 중...",
|
||
"FIRST_SUCCESS": "첫 번째 트랙 완료, 두 번째 생성 중...",
|
||
"GENERATING": "생성 중...",
|
||
}
|
||
msgs = {**default_msgs, **(progress_msg_map or {})}
|
||
|
||
for attempt in range(max_attempts):
|
||
time.sleep(interval)
|
||
try:
|
||
resp = requests.get(
|
||
f"{SUNO_BASE_URL}{record_info_path}",
|
||
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 / max_attempts) * 65), 79)
|
||
|
||
if status == "SUCCESS":
|
||
return data.get("response", data)
|
||
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:
|
||
msg = msgs.get(status, f"처리 중... ({status})")
|
||
if status == "FIRST_SUCCESS":
|
||
progress = max(progress, 60)
|
||
update_task(task_id, "processing", progress, msg)
|
||
except Exception as e:
|
||
logger.warning("Suno poll error (attempt %d): %s", attempt, e)
|
||
continue
|
||
|
||
update_task(task_id, "failed", 0, "", error="Suno 작업 타임아웃")
|
||
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)
|
||
|
||
|
||
# ── 크레딧 조회 ──────────────────────────────────────────────────────────────
|
||
|
||
def get_credits() -> Optional[dict]:
|
||
"""Suno API 잔여 크레딧 조회. 두 엔드포인트 폴백."""
|
||
if not SUNO_API_KEY:
|
||
return None
|
||
for path in ["/generate/credit", "/get-credits"]:
|
||
try:
|
||
resp = requests.get(f"{SUNO_BASE_URL}{path}", headers=_headers(), timeout=15)
|
||
if resp.status_code == 200:
|
||
body = resp.json()
|
||
data = body.get("data", body)
|
||
if isinstance(data, (int, float)):
|
||
return {"credits_left": int(data)}
|
||
return data
|
||
except Exception as e:
|
||
logger.warning("Suno credits API error (%s): %s", path, e)
|
||
return None
|
||
|
||
|
||
# ── 곡 연장 (Extend) ────────────────────────────────────────────────────────
|
||
|
||
def run_suno_extend(task_id: str, params: dict) -> None:
|
||
"""기존 곡을 특정 지점부터 연장."""
|
||
try:
|
||
if not SUNO_API_KEY:
|
||
update_task(task_id, "failed", 0, "",
|
||
error="SUNO_API_KEY가 설정되지 않았습니다.")
|
||
return
|
||
|
||
update_task(task_id, "processing", 5, "곡 연장 요청 중...")
|
||
|
||
payload = {
|
||
"audioId": params["suno_id"],
|
||
"defaultParamFlag": not bool(params.get("prompt")),
|
||
"prompt": params.get("prompt", ""),
|
||
"continueAt": params.get("continue_at", 0),
|
||
"model": params.get("model", "V4"),
|
||
"callBackUrl": "https://example.com/noop",
|
||
}
|
||
if params.get("style"):
|
||
payload["style"] = params["style"]
|
||
if params.get("title"):
|
||
payload["title"] = params["title"]
|
||
|
||
resp = requests.post(
|
||
f"{SUNO_BASE_URL}/generate/extend",
|
||
headers=_headers(),
|
||
json=payload,
|
||
timeout=30,
|
||
)
|
||
|
||
if resp.status_code != 200:
|
||
update_task(task_id, "failed", 0, "",
|
||
error=f"Suno Extend API 오류: {resp.text[:300]}")
|
||
return
|
||
|
||
body = resp.json()
|
||
if body.get("code") != 200:
|
||
update_task(task_id, "failed", 0, "",
|
||
error=f"Suno Extend 거부: {body.get('msg', 'unknown')}")
|
||
return
|
||
|
||
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, "곡 연장 대기열에 등록됨...")
|
||
|
||
response = _poll_suno_record("/generate/record-info", suno_task_id, task_id)
|
||
if not response:
|
||
return
|
||
completed_tracks = response.get("sunoData") or []
|
||
if not completed_tracks:
|
||
update_task(task_id, "failed", 0, "", error="Suno 연장 완료했으나 트랙 데이터 없음")
|
||
return
|
||
|
||
update_task(task_id, "processing", 80, "연장된 오디오 다운로드 중...")
|
||
|
||
track = _download_and_register(
|
||
task_id=task_id, song=completed_tracks[0],
|
||
params=params, filename_suffix="",
|
||
)
|
||
if not track:
|
||
return
|
||
|
||
update_task(task_id, "succeeded", 100, "곡 연장 완료",
|
||
audio_url=track["audio_url"])
|
||
|
||
except Exception as e:
|
||
logger.exception("Suno extend error for task %s", task_id)
|
||
update_task(task_id, "failed", 0, "", error=str(e))
|
||
|
||
|
||
# ── 보컬 분리 ────────────────────────────────────────────────────────────────
|
||
|
||
def run_vocal_removal(task_id: str, params: dict) -> None:
|
||
"""트랙에서 보컬과 인스트루멘탈을 분리."""
|
||
try:
|
||
if not SUNO_API_KEY:
|
||
update_task(task_id, "failed", 0, "",
|
||
error="SUNO_API_KEY가 설정되지 않았습니다.")
|
||
return
|
||
|
||
update_task(task_id, "processing", 5, "보컬 분리 요청 중...")
|
||
|
||
# suno_id로 원본 Suno 곡 참조
|
||
payload = {
|
||
"audioId": params["suno_id"],
|
||
"callBackUrl": "https://example.com/noop",
|
||
}
|
||
|
||
resp = requests.post(
|
||
f"{SUNO_BASE_URL}/vocal-removal/generate",
|
||
headers=_headers(),
|
||
json=payload,
|
||
timeout=30,
|
||
)
|
||
|
||
if resp.status_code != 200:
|
||
update_task(task_id, "failed", 0, "",
|
||
error=f"Suno Vocal Removal 오류: {resp.text[:300]}")
|
||
return
|
||
|
||
body = resp.json()
|
||
if body.get("code") != 200:
|
||
update_task(task_id, "failed", 0, "",
|
||
error=f"Suno Vocal Removal 거부: {body.get('msg', 'unknown')}")
|
||
return
|
||
|
||
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, "보컬 분리 처리 중...")
|
||
|
||
# 보컬 분리 결과 폴링
|
||
response = _poll_suno_record("/vocal-removal/record-info", suno_task_id, task_id)
|
||
if not response:
|
||
return
|
||
completed_tracks = response.get("sunoData") or []
|
||
if not completed_tracks:
|
||
update_task(task_id, "failed", 0, "", error="보컬 분리 완료했으나 트랙 데이터 없음")
|
||
return
|
||
|
||
update_task(task_id, "processing", 80, "분리된 오디오 다운로드 중...")
|
||
|
||
# 첫 번째: 보컬 트랙, 두 번째: 인스트루멘탈 트랙
|
||
vocal_params = {**params, "title": f"{params.get('title', 'Track')} (Vocals)"}
|
||
track = _download_and_register(
|
||
task_id=task_id, song=completed_tracks[0],
|
||
params=vocal_params, filename_suffix="",
|
||
)
|
||
|
||
if len(completed_tracks) > 1:
|
||
inst_params = {**params, "title": f"{params.get('title', 'Track')} (Instrumental)"}
|
||
_download_and_register(
|
||
task_id=f"{task_id}_inst", song=completed_tracks[1],
|
||
params=inst_params, filename_suffix="",
|
||
)
|
||
|
||
if track:
|
||
update_task(task_id, "succeeded", 100, "보컬 분리 완료",
|
||
audio_url=track["audio_url"])
|
||
else:
|
||
update_task(task_id, "failed", 0, "", error="분리 결과 저장 실패")
|
||
|
||
except Exception as e:
|
||
logger.exception("Suno vocal removal error for task %s", task_id)
|
||
update_task(task_id, "failed", 0, "", error=str(e))
|
||
|
||
|
||
# ── 커버 이미지 생성 ────────────────────────────────────────────────────────
|
||
|
||
def run_cover_image(task_id: str, params: dict) -> None:
|
||
"""Suno 곡의 커버 이미지 2장을 생성."""
|
||
try:
|
||
if not SUNO_API_KEY:
|
||
update_task(task_id, "failed", 0, "", error="SUNO_API_KEY가 설정되지 않았습니다.")
|
||
return
|
||
update_task(task_id, "processing", 5, "커버 이미지 생성 요청 중...")
|
||
suno_task_id = params.get("suno_task_id", "")
|
||
if not suno_task_id:
|
||
update_task(task_id, "failed", 0, "", error="suno_task_id가 필요합니다")
|
||
return
|
||
payload = {
|
||
"taskId": suno_task_id,
|
||
"callBackUrl": "https://example.com/noop",
|
||
}
|
||
resp = requests.post(f"{SUNO_BASE_URL}/suno/cover/generate", headers=_headers(), json=payload, timeout=30)
|
||
if resp.status_code != 200:
|
||
update_task(task_id, "failed", 0, "", error=f"커버 이미지 API 오류: {resp.text[:300]}")
|
||
return
|
||
body = resp.json()
|
||
if body.get("code") != 200:
|
||
update_task(task_id, "failed", 0, "", error=f"커버 이미지 거부: {body.get('msg', 'unknown')}")
|
||
return
|
||
cover_task_id = body.get("data", {}).get("taskId", suno_task_id)
|
||
update_task(task_id, "processing", 15, "커버 이미지 생성 중...")
|
||
response = _poll_suno_record(
|
||
"/suno/cover/record-info", cover_task_id, task_id,
|
||
max_attempts=30, interval=5,
|
||
progress_msg_map={"PENDING": "이미지 생성 대기 중...", "GENERATING": "이미지 생성 중..."},
|
||
)
|
||
if not response:
|
||
return
|
||
images = response.get("images") or response.get("sunoData") or []
|
||
image_urls = []
|
||
if isinstance(images, list):
|
||
for img in images:
|
||
if isinstance(img, str):
|
||
image_urls.append(img)
|
||
elif isinstance(img, dict):
|
||
image_urls.append(img.get("imageUrl") or img.get("image_url", ""))
|
||
update_task(task_id, "succeeded", 100, "커버 이미지 생성 완료", audio_url=json.dumps(image_urls))
|
||
except Exception as e:
|
||
logger.exception("Cover image generation error for task %s", task_id)
|
||
update_task(task_id, "failed", 0, "", error=str(e))
|
||
|
||
|
||
# ── WAV 변환 ─────────────────────────────────────────────────────────────────
|
||
|
||
def run_wav_convert(task_id: str, params: dict) -> None:
|
||
"""곡을 WAV 포맷으로 변환."""
|
||
try:
|
||
if not SUNO_API_KEY:
|
||
update_task(task_id, "failed", 0, "", error="SUNO_API_KEY가 설정되지 않았습니다.")
|
||
return
|
||
|
||
update_task(task_id, "processing", 5, "WAV 변환 요청 중...")
|
||
|
||
payload = {
|
||
"taskId": params["suno_task_id"],
|
||
"audioId": params["suno_id"],
|
||
"callBackUrl": "https://example.com/noop",
|
||
}
|
||
|
||
resp = requests.post(
|
||
f"{SUNO_BASE_URL}/wav/generate",
|
||
headers=_headers(),
|
||
json=payload,
|
||
timeout=30,
|
||
)
|
||
|
||
if resp.status_code == 409:
|
||
body = resp.json()
|
||
wav_url = body.get("data", {}).get("audioWavUrl", "")
|
||
if wav_url:
|
||
update_task(task_id, "succeeded", 100, "WAV 변환 완료 (캐시)", audio_url=wav_url)
|
||
return
|
||
|
||
if resp.status_code != 200:
|
||
update_task(task_id, "failed", 0, "", error=f"WAV API 오류: {resp.text[:300]}")
|
||
return
|
||
|
||
body = resp.json()
|
||
if body.get("code") != 200:
|
||
update_task(task_id, "failed", 0, "", error=f"WAV 변환 거부: {body.get('msg', 'unknown')}")
|
||
return
|
||
|
||
wav_task_id = body.get("data", {}).get("taskId", params["suno_task_id"])
|
||
update_task(task_id, "processing", 15, "WAV 변환 처리 중...")
|
||
|
||
response = _poll_suno_record(
|
||
"/wav/record-info", wav_task_id, task_id,
|
||
max_attempts=30, interval=5,
|
||
progress_msg_map={"PENDING": "WAV 변환 대기 중...", "GENERATING": "WAV 변환 중..."},
|
||
)
|
||
if not response:
|
||
return
|
||
|
||
wav_url = ""
|
||
suno_data = response.get("sunoData") or []
|
||
if suno_data and isinstance(suno_data, list):
|
||
wav_url = suno_data[0].get("audioWavUrl", "") if isinstance(suno_data[0], dict) else ""
|
||
if not wav_url:
|
||
wav_url = response.get("audioWavUrl", "")
|
||
|
||
update_task(task_id, "succeeded", 100, "WAV 변환 완료", audio_url=wav_url)
|
||
|
||
except Exception as e:
|
||
logger.exception("WAV convert error for task %s", task_id)
|
||
update_task(task_id, "failed", 0, "", error=str(e))
|
||
|
||
|
||
# ── 12스템 분리 ──────────────────────────────────────────────────────────────
|
||
|
||
def run_stem_split(task_id: str, params: dict) -> None:
|
||
"""곡을 12개 스템으로 분리 (50 크레딧 소모)."""
|
||
try:
|
||
if not SUNO_API_KEY:
|
||
update_task(task_id, "failed", 0, "", error="SUNO_API_KEY가 설정되지 않았습니다.")
|
||
return
|
||
|
||
update_task(task_id, "processing", 5, "12스템 분리 요청 중...")
|
||
|
||
payload = {
|
||
"taskId": params["suno_task_id"],
|
||
"audioId": params["suno_id"],
|
||
"type": "split_stem",
|
||
"callBackUrl": "https://example.com/noop",
|
||
}
|
||
|
||
resp = requests.post(
|
||
f"{SUNO_BASE_URL}/vocal-removal/generate",
|
||
headers=_headers(),
|
||
json=payload,
|
||
timeout=30,
|
||
)
|
||
|
||
if resp.status_code != 200:
|
||
update_task(task_id, "failed", 0, "", error=f"스템 분리 API 오류: {resp.text[:300]}")
|
||
return
|
||
|
||
body = resp.json()
|
||
if body.get("code") != 200:
|
||
update_task(task_id, "failed", 0, "", error=f"스템 분리 거부: {body.get('msg', 'unknown')}")
|
||
return
|
||
|
||
stem_task_id = body.get("data", {}).get("taskId", "")
|
||
if not stem_task_id:
|
||
update_task(task_id, "failed", 0, "", error="스템 분리 응답에 taskId 없음")
|
||
return
|
||
|
||
update_task(task_id, "processing", 15, "12스템 분리 처리 중 (약 2~3분)...")
|
||
|
||
response = _poll_suno_record(
|
||
"/vocal-removal/record-info", stem_task_id, task_id,
|
||
max_attempts=40, interval=8,
|
||
progress_msg_map={"PENDING": "스템 분리 대기 중...", "GENERATING": "스템 분리 중..."},
|
||
)
|
||
if not response:
|
||
return
|
||
|
||
suno_data = response.get("sunoData") or []
|
||
stems = {}
|
||
stem_names = ["vocal", "backing_vocals", "drums", "bass", "guitar", "keyboard",
|
||
"strings", "brass", "woodwinds", "percussion", "synth", "fx"]
|
||
for i, item in enumerate(suno_data):
|
||
if isinstance(item, dict):
|
||
name = stem_names[i] if i < len(stem_names) else f"stem_{i}"
|
||
stems[name] = item.get("audioUrl") or item.get("audio_url", "")
|
||
|
||
update_task(task_id, "succeeded", 100, "12스템 분리 완료",
|
||
audio_url=json.dumps(stems))
|
||
|
||
except Exception as e:
|
||
logger.exception("Stem split error for task %s", task_id)
|
||
update_task(task_id, "failed", 0, "", error=str(e))
|
||
|
||
|
||
# ── 타임스탬프 가사 ──────────────────────────────────────────────────────────
|
||
|
||
def get_timestamped_lyrics(suno_task_id: str, suno_id: str) -> Optional[dict]:
|
||
"""타임스탬프가 포함된 가사 데이터 조회 (동기)."""
|
||
if not SUNO_API_KEY:
|
||
return None
|
||
try:
|
||
resp = requests.post(
|
||
f"{SUNO_BASE_URL}/generate/get-timestamped-lyrics",
|
||
headers=_headers(),
|
||
json={"taskId": suno_task_id, "audioId": suno_id},
|
||
timeout=30,
|
||
)
|
||
if resp.status_code == 200:
|
||
body = resp.json()
|
||
return body.get("data", body)
|
||
except Exception as e:
|
||
logger.warning("Timestamped lyrics error: %s", e)
|
||
return None
|
||
|
||
|
||
# ── 스타일 부스트 ────────────────────────────────────────────────────────────
|
||
|
||
def generate_style_boost(content: str) -> Optional[dict]:
|
||
"""AI로 최적 스타일 텍스트 생성 (동기)."""
|
||
if not SUNO_API_KEY:
|
||
return None
|
||
try:
|
||
resp = requests.post(
|
||
f"{SUNO_BASE_URL}/style/generate",
|
||
headers=_headers(),
|
||
json={"content": content},
|
||
timeout=30,
|
||
)
|
||
if resp.status_code == 200:
|
||
body = resp.json()
|
||
return body.get("data", body)
|
||
except Exception as e:
|
||
logger.warning("Style boost error: %s", e)
|
||
return None
|