From e1722e39630dd68b96a665be618f15c3e8bddedc Mon Sep 17 00:00:00 2001 From: gahusb Date: Tue, 19 May 2026 05:22:05 +0900 Subject: [PATCH] =?UTF-8?q?refactor(music-lab):=20suno=5Fprovider/local=5F?= =?UTF-8?q?provider=20=E2=86=92=20stub=20(SP-6)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 기존 13+1 외부 API 호출 함수는 web-ai/services/music-render/providers로 이식. NAS는 SUNO_MODELS (정적 데이터)만 잔존. SUNO_API_KEY = "" sentinel. Plan-B-Music Phase 3 (cutover 4/4). Co-Authored-By: Claude Opus 4.7 (1M context) --- music-lab/app/local_provider.py | 125 +--- music-lab/app/suno_provider.py | 1052 +------------------------------ 2 files changed, 15 insertions(+), 1162 deletions(-) diff --git a/music-lab/app/local_provider.py b/music-lab/app/local_provider.py index 6abf432..544a3fd 100644 --- a/music-lab/app/local_provider.py +++ b/music-lab/app/local_provider.py @@ -1,122 +1,5 @@ +"""DEPRECATED 2026-05-19 — MusicGen 호출은 Windows music-render로 이전. + +기존 run_local_generation은 web-ai/services/music-render/providers/local.py로 이식. +NAS는 Redis push (queue:music-render, job_type=local_generation)만 담당. """ -Local MusicGen Provider — Windows AI 서버(MusicGen)를 통한 음악 생성 -기존 _run_generation 로직을 그대로 분리. -""" - -import os -import time -import logging -import requests - -from .db import update_task, add_track - -logger = logging.getLogger(__name__) - -MUSIC_AI_SERVER_URL = os.getenv("MUSIC_AI_SERVER_URL", "") -MUSIC_DATA_DIR = "/app/data" -MUSIC_MEDIA_BASE = os.getenv("MUSIC_MEDIA_BASE", "/media/music") - - -def run_local_generation(task_id: str, params: dict) -> None: - """BackgroundTask: Windows AI 서버(MusicGen)에 생성 요청 → 파일 저장 → 라이브러리 등록""" - try: - update_task(task_id, "processing", 10, "AI 서버에 연결 중...") - - if not MUSIC_AI_SERVER_URL: - update_task(task_id, "failed", 0, "", - error="MUSIC_AI_SERVER_URL이 설정되지 않았습니다") - return - - update_task(task_id, "processing", 30, "음악 생성 중... (수 분 소요될 수 있습니다)") - - # 1단계: 생성 요청 → ai_task_id 반환 - resp = requests.post( - f"{MUSIC_AI_SERVER_URL}/generate", - json=params, - timeout=30, - ) - - if resp.status_code != 200: - update_task(task_id, "failed", 0, "", - error=f"AI 서버 오류: {resp.status_code} {resp.text[:200]}") - return - - ai_task_id = resp.json().get("task_id") - if not ai_task_id: - update_task(task_id, "failed", 0, "", - error="AI 서버 응답에 task_id가 없습니다") - return - - # 2단계: 상태 폴링 (최대 10분, 5초 간격) - remote_url = None - for _ in range(120): - time.sleep(5) - status_resp = requests.get( - f"{MUSIC_AI_SERVER_URL}/status/{ai_task_id}", timeout=10, - ) - status_data = status_resp.json() - ai_status = status_data.get("status") - - ai_progress = status_data.get("progress", 0) - ai_message = status_data.get("message", "음악 생성 중...") - scaled = 30 + int(ai_progress * 0.49) # 30% ~ 79% - update_task(task_id, "processing", scaled, ai_message) - - if ai_status == "succeeded": - remote_url = status_data.get("audio_url") - break - elif ai_status == "failed": - update_task(task_id, "failed", 0, "", - error=status_data.get("error", "AI 서버 생성 실패")) - return - - if not remote_url: - update_task(task_id, "failed", 0, "", - error="AI 서버 타임아웃 (10분 초과)") - return - - update_task(task_id, "processing", 80, "파일 저장 중...") - - filename = f"{task_id}.mp3" - file_path = os.path.join(MUSIC_DATA_DIR, filename) - - # 3단계: 오디오 파일 다운로드 - dl = requests.get(remote_url, timeout=120, stream=True) - with open(file_path, "wb") as f: - for chunk in dl.iter_content(chunk_size=8192): - f.write(chunk) - - audio_url = f"{MUSIC_MEDIA_BASE}/{filename}" - - # 라이브러리 자동 등록 - genre = params.get("genre", "") - moods = params.get("moods", []) - mood_str = moods[0] if moods else "Original" - title = params.get("title") or ( - f"{genre} — {mood_str} Mix" if genre else f"{mood_str} Mix" - ) - - add_track({ - "title": title, - "genre": genre, - "moods": params.get("moods", []), - "instruments": params.get("instruments", []), - "duration_sec": params.get("duration_sec"), - "bpm": params.get("bpm"), - "key": params.get("key", ""), - "scale": params.get("scale", ""), - "prompt": params.get("prompt", ""), - "audio_url": audio_url, - "file_path": file_path, - "task_id": task_id, - "provider": "local", - }) - - update_task(task_id, "succeeded", 100, "생성 완료", audio_url=audio_url) - - except requests.Timeout: - update_task(task_id, "failed", 0, "", - error="AI 서버 타임아웃 (10분 초과)") - except Exception as e: - logger.exception("Local generation error for task %s", task_id) - update_task(task_id, "failed", 0, "", error=str(e)) diff --git a/music-lab/app/suno_provider.py b/music-lab/app/suno_provider.py index 005924e..af31ee2 100644 --- a/music-lab/app/suno_provider.py +++ b/music-lab/app/suno_provider.py @@ -1,1050 +1,20 @@ +"""DEPRECATED 2026-05-19 — Suno API 호출 코드는 모두 Windows music-render로 이전. + +기존 13 함수 (run_suno_generation, run_suno_extend, ...)는 +web-ai/services/music-render/providers/suno.py로 이식됨. +NAS는 Redis push (queue:music-render)만 담당. + +SUNO_MODELS는 frontend 응답용 정적 데이터만 잔존. """ -Suno API Provider — sunoapi.org 래퍼를 통한 음악 생성 -https://docs.sunoapi.org/suno-api/quickstart -""" +from __future__ import annotations -import json -import os -import time -import logging -import requests -from typing import Optional +SUNO_API_KEY = "" # NAS에서 더 이상 보유 X — sentinel 유지 (다른 import 호환) -from .db import ( - update_task, add_track, - update_track_cover_images, update_track_wav_url, - update_track_video_url, update_track_stem_urls, -) - -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": "커스텀 모델, 최신 음악성"}, + {"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)) - if params.get("track_id") and image_urls: - update_track_cover_images(params["track_id"], 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) - if params.get("track_id") and wav_url: - update_track_wav_url(params["track_id"], 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) - if params.get("track_id") and wav_url: - update_track_wav_url(params["track_id"], 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)) - if params.get("track_id") and stems: - update_track_stem_urls(params["track_id"], 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 - - -# ── 오디오 업로드 + 커버 ───────────────────────────────────────────────────── - -def run_upload_cover(task_id: str, params: dict) -> None: - """외부 오디오를 Suno 스타일로 리메이크.""" - try: - if not SUNO_API_KEY: - update_task(task_id, "failed", 0, "", error="SUNO_API_KEY가 설정되지 않았습니다.") - return - - update_task(task_id, "processing", 5, "AI Cover 요청 중...") - - payload = { - "uploadUrl": params["upload_url"], - "customMode": params.get("custom_mode", True), - "instrumental": params.get("instrumental", False), - "model": params.get("model", "V4"), - "callBackUrl": "https://example.com/noop", - } - for key, api_key in [("prompt", "prompt"), ("style", "style"), ("title", "title"), - ("vocal_gender", "vocalGender"), ("negative_tags", "negativeTags"), - ("style_weight", "styleWeight"), ("audio_weight", "audioWeight")]: - if params.get(key): - payload[api_key] = params[key] - - resp = requests.post(f"{SUNO_BASE_URL}/generate/upload-cover", headers=_headers(), json=payload, timeout=30) - if resp.status_code != 200: - update_task(task_id, "failed", 0, "", error=f"Upload Cover API 오류: {resp.text[:300]}") - return - body = resp.json() - if body.get("code") != 200: - update_task(task_id, "failed", 0, "", error=f"Upload Cover 거부: {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="Upload Cover 응답에 taskId 없음") - return - update_task(task_id, "processing", 15, "AI Cover 생성 중...") - - 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="AI Cover 생성 완료했으나 트랙 없음") - return - - track = _download_and_register(task_id=task_id, song=completed_tracks[0], params=params, filename_suffix="") - if track: - update_task(task_id, "succeeded", 100, "AI Cover 완료", audio_url=track["audio_url"]) - - except Exception as e: - logger.exception("Upload cover error for task %s", task_id) - update_task(task_id, "failed", 0, "", error=str(e)) - - -# ── 오디오 업로드 + 확장 ───────────────────────────────────────────────────── - -def run_upload_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, "Upload Extend 요청 중...") - - payload = { - "uploadUrl": params["upload_url"], - "defaultParamFlag": params.get("default_param_flag", True), - "model": params.get("model", "V4"), - "callBackUrl": "https://example.com/noop", - } - for key, api_key in [("prompt", "prompt"), ("style", "style"), ("title", "title"), - ("continue_at", "continueAt"), ("instrumental", "instrumental"), - ("vocal_gender", "vocalGender"), ("negative_tags", "negativeTags")]: - if params.get(key) is not None: - payload[api_key] = params[key] - - resp = requests.post(f"{SUNO_BASE_URL}/generate/upload-extend", headers=_headers(), json=payload, timeout=30) - if resp.status_code != 200: - update_task(task_id, "failed", 0, "", error=f"Upload Extend API 오류: {resp.text[:300]}") - return - body = resp.json() - if body.get("code") != 200: - update_task(task_id, "failed", 0, "", error=f"Upload 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="Upload Extend 응답에 taskId 없음") - return - update_task(task_id, "processing", 15, "Upload Extend 생성 중...") - - 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="Upload Extend 완료했으나 트랙 없음") - return - - track = _download_and_register(task_id=task_id, song=completed_tracks[0], params=params, filename_suffix="") - if track: - update_task(task_id, "succeeded", 100, "Upload Extend 완료", audio_url=track["audio_url"]) - - except Exception as e: - logger.exception("Upload extend error for task %s", task_id) - update_task(task_id, "failed", 0, "", error=str(e)) - - -# ── 보컬 추가 ──────────────────────────────────────────────────────────────── - -def run_add_vocals(task_id: str, params: dict) -> None: - """인스트루멘탈에 AI 보컬을 추가.""" - try: - if not SUNO_API_KEY: - update_task(task_id, "failed", 0, "", error="SUNO_API_KEY가 설정되지 않았습니다.") - return - - update_task(task_id, "processing", 5, "보컬 추가 요청 중...") - - payload = { - "uploadUrl": params["upload_url"], - "prompt": params.get("prompt", ""), - "title": params.get("title", ""), - "style": params.get("style", ""), - "negativeTags": params.get("negative_tags", ""), - "callBackUrl": "https://example.com/noop", - } - for key, api_key in [("vocal_gender", "vocalGender"), ("model", "model"), - ("style_weight", "styleWeight"), ("audio_weight", "audioWeight")]: - if params.get(key) is not None: - payload[api_key] = params[key] - - resp = requests.post(f"{SUNO_BASE_URL}/generate/add-vocals", headers=_headers(), json=payload, timeout=30) - if resp.status_code != 200: - update_task(task_id, "failed", 0, "", error=f"Add Vocals API 오류: {resp.text[:300]}") - return - body = resp.json() - if body.get("code") != 200: - update_task(task_id, "failed", 0, "", error=f"Add Vocals 거부: {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="Add Vocals 응답에 taskId 없음") - return - update_task(task_id, "processing", 15, "AI 보컬 생성 중...") - - 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="보컬 추가 완료했으나 트랙 없음") - return - - track = _download_and_register(task_id=task_id, song=completed_tracks[0], params=params, filename_suffix="") - if track: - update_task(task_id, "succeeded", 100, "보컬 추가 완료", audio_url=track["audio_url"]) - - except Exception as e: - logger.exception("Add vocals error for task %s", task_id) - update_task(task_id, "failed", 0, "", error=str(e)) - - -# ── 인스트루멘탈 추가 ──────────────────────────────────────────────────────── - -def run_add_instrumental(task_id: str, params: dict) -> None: - """보컬에 AI 반주를 추가.""" - try: - if not SUNO_API_KEY: - update_task(task_id, "failed", 0, "", error="SUNO_API_KEY가 설정되지 않았습니다.") - return - - update_task(task_id, "processing", 5, "인스트루멘탈 추가 요청 중...") - - payload = { - "uploadUrl": params["upload_url"], - "title": params.get("title", ""), - "tags": params.get("tags", ""), - "negativeTags": params.get("negative_tags", ""), - "callBackUrl": "https://example.com/noop", - } - for key, api_key in [("vocal_gender", "vocalGender"), ("model", "model"), - ("style_weight", "styleWeight"), ("audio_weight", "audioWeight")]: - if params.get(key) is not None: - payload[api_key] = params[key] - - resp = requests.post(f"{SUNO_BASE_URL}/generate/add-instrumental", headers=_headers(), json=payload, timeout=30) - if resp.status_code != 200: - update_task(task_id, "failed", 0, "", error=f"Add Instrumental API 오류: {resp.text[:300]}") - return - body = resp.json() - if body.get("code") != 200: - update_task(task_id, "failed", 0, "", error=f"Add Instrumental 거부: {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="Add Instrumental 응답에 taskId 없음") - return - update_task(task_id, "processing", 15, "AI 반주 생성 중...") - - 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="인스트루멘탈 추가 완료했으나 트랙 없음") - return - - track = _download_and_register(task_id=task_id, song=completed_tracks[0], params=params, filename_suffix="") - if track: - update_task(task_id, "succeeded", 100, "인스트루멘탈 추가 완료", audio_url=track["audio_url"]) - - except Exception as e: - logger.exception("Add instrumental error for task %s", task_id) - update_task(task_id, "failed", 0, "", error=str(e)) - - -# ── 뮤직비디오 생성 ────────────────────────────────────────────────────────── - -def run_video_generate(task_id: str, params: dict) -> None: - """곡의 뮤직비디오(MP4) 생성.""" - try: - if not SUNO_API_KEY: - update_task(task_id, "failed", 0, "", error="SUNO_API_KEY가 설정되지 않았습니다.") - return - - update_task(task_id, "processing", 5, "뮤직비디오 생성 요청 중...") - - payload = { - "taskId": params["suno_task_id"], - "audioId": params["suno_id"], - "callBackUrl": "https://example.com/noop", - } - if params.get("author"): - payload["author"] = params["author"][:50] - if params.get("domain_name"): - payload["domainName"] = params["domain_name"][:50] - - resp = requests.post(f"{SUNO_BASE_URL}/mp4/generate", headers=_headers(), json=payload, timeout=30) - if resp.status_code != 200: - update_task(task_id, "failed", 0, "", error=f"Video API 오류: {resp.text[:300]}") - return - body = resp.json() - if body.get("code") != 200: - update_task(task_id, "failed", 0, "", error=f"Video 생성 거부: {body.get('msg', 'unknown')}") - return - - video_task_id = body.get("data", {}).get("taskId", params.get("suno_task_id", "")) - update_task(task_id, "processing", 15, "뮤직비디오 렌더링 중...") - - response = _poll_suno_record( - "/mp4/record-info", video_task_id, task_id, - max_attempts=60, interval=10, - progress_msg_map={"PENDING": "비디오 렌더링 대기 중...", "GENERATING": "비디오 렌더링 중..."}, - ) - if not response: - return - - video_url = "" - suno_data = response.get("sunoData") or [] - if suno_data and isinstance(suno_data, list) and isinstance(suno_data[0], dict): - video_url = suno_data[0].get("videoUrl") or suno_data[0].get("video_url", "") - if not video_url: - video_url = response.get("video_url") or response.get("videoUrl", "") - - update_task(task_id, "succeeded", 100, "뮤직비디오 생성 완료", audio_url=video_url) - if params.get("track_id") and video_url: - update_track_video_url(params["track_id"], video_url) - - except Exception as e: - logger.exception("Video generate error for task %s", task_id) - update_task(task_id, "failed", 0, "", error=str(e))