diff --git a/CLAUDE.md b/CLAUDE.md index 6cf6a46..8d3758e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -259,12 +259,30 @@ docker compose up -d | 메서드 | 경로 | 설명 | |--------|------|------| | GET | `/api/music/providers` | 사용 가능한 프로바이더 목록 | -| POST | `/api/music/generate` | 음악 생성 시작 (provider, lyrics, instrumental 지원) | -| GET | `/api/music/status/{task_id}` | 생성 상태 폴링 (queued→processing→succeeded/failed) | -| POST | `/api/music/lyrics` | Suno AI 가사 생성 (곡 생성 전 미리보기용) | +| GET | `/api/music/models` | Suno 모델 목록 (V4~V5.5) | +| GET | `/api/music/credits` | Suno 크레딧 조회 | +| POST | `/api/music/generate` | 음악 생성 (provider, model, vocal_gender, negative_tags, style_weight, audio_weight) | +| GET | `/api/music/status/{task_id}` | 생성 상태 폴링 | +| POST | `/api/music/lyrics` | Suno AI 가사 생성 | | GET | `/api/music/library` | 라이브러리 전체 조회 | -| POST | `/api/music/library` | 트랙 수동 추가 (201) | -| DELETE | `/api/music/library/{id}` | 트랙 삭제 (로컬 파일 포함) | +| POST | `/api/music/library` | 트랙 수동 추가 | +| DELETE | `/api/music/library/{id}` | 트랙 삭제 | +| POST | `/api/music/extend` | 곡 연장 | +| POST | `/api/music/vocal-removal` | 보컬/인스트 분리 (2트랙) | +| POST | `/api/music/cover-image` | 커버 이미지 2장 생성 | +| POST | `/api/music/wav` | WAV 고음질 변환 | +| POST | `/api/music/stem-split` | 12스템 분리 (50cr) | +| GET | `/api/music/timestamped-lyrics` | 타임스탬프 가사 (가라오케) | +| POST | `/api/music/style-boost` | AI 스타일 프롬프트 생성 | +| POST | `/api/music/upload-cover` | 외부 음원 AI Cover | +| POST | `/api/music/upload-extend` | 외부 음원 확장 | +| POST | `/api/music/add-vocals` | 인스트에 AI 보컬 추가 | +| POST | `/api/music/add-instrumental` | 보컬에 AI 반주 추가 | +| POST | `/api/music/video` | 뮤직비디오 MP4 생성 | +| GET | `/api/music/lyrics/library` | 저장된 가사 목록 | +| POST | `/api/music/lyrics/library` | 가사 저장 | +| PUT | `/api/music/lyrics/library/{id}` | 가사 수정 | +| DELETE | `/api/music/lyrics/library/{id}` | 가사 삭제 | **환경변수** - `SUNO_API_KEY`: Suno API 키 (미설정 시 Suno provider 비활성화) @@ -277,6 +295,11 @@ docker compose up -d - `lyrics`: Suno 생성 가사 텍스트 - `image_url`: Suno 생성 커버 이미지 URL - `suno_id`: Suno 곡 ID (CDN 참조용) +- `file_hash`: MD5 해시 (rename 감지용) +- `cover_images`: JSON 배열 — 커버 이미지 URL 목록 +- `wav_url`: WAV 변환 URL +- `video_url`: 뮤직비디오 URL +- `stem_urls`: JSON 객체 — 12스템 URL 맵 **Suno 생성 특이사항** - 1회 생성 시 2개 변형(variation) 반환 → 둘 다 라이브러리에 저장 diff --git a/music-lab/app/db.py b/music-lab/app/db.py index 3023872..c8504db 100644 --- a/music-lab/app/db.py +++ b/music-lab/app/db.py @@ -83,6 +83,18 @@ def init_db() -> None: except sqlite3.OperationalError: pass + # Phase 1~3 신규 컬럼 마이그레이션 + for col, default in [ + ("cover_images", "'[]'"), + ("wav_url", "''"), + ("video_url", "''"), + ("stem_urls", "'{}'"), + ]: + try: + conn.execute(f"ALTER TABLE music_library ADD COLUMN {col} TEXT NOT NULL DEFAULT {default}") + except sqlite3.OperationalError: + pass + # ── music_tasks CRUD ────────────────────────────────────────────────────────── @@ -161,6 +173,10 @@ def _track_row_to_dict(r) -> Dict[str, Any]: "image_url": r["image_url"] if "image_url" in keys else "", "suno_id": r["suno_id"] if "suno_id" in keys else "", "file_hash": r["file_hash"] if "file_hash" in keys else "", + "cover_images": json.loads(r["cover_images"]) if "cover_images" in keys and r["cover_images"] else [], + "wav_url": r["wav_url"] if "wav_url" in keys else "", + "video_url": r["video_url"] if "video_url" in keys else "", + "stem_urls": json.loads(r["stem_urls"]) if "stem_urls" in keys and r["stem_urls"] else {}, "created_at": r["created_at"], } @@ -300,6 +316,26 @@ def update_lyrics(lyrics_id: int, data: Dict[str, Any]) -> Optional[Dict[str, An return _lyrics_row_to_dict(row) if row else None +def update_track_cover_images(track_id: int, images: list) -> None: + with _conn() as conn: + conn.execute("UPDATE music_library SET cover_images=? WHERE id=?", (json.dumps(images), track_id)) + + +def update_track_wav_url(track_id: int, wav_url: str) -> None: + with _conn() as conn: + conn.execute("UPDATE music_library SET wav_url=? WHERE id=?", (wav_url, track_id)) + + +def update_track_video_url(track_id: int, video_url: str) -> None: + with _conn() as conn: + conn.execute("UPDATE music_library SET video_url=? WHERE id=?", (video_url, track_id)) + + +def update_track_stem_urls(track_id: int, stems: dict) -> None: + with _conn() as conn: + conn.execute("UPDATE music_library SET stem_urls=? WHERE id=?", (json.dumps(stems), track_id)) + + def delete_lyrics(lyrics_id: int) -> bool: with _conn() as conn: row = conn.execute("SELECT id FROM saved_lyrics WHERE id = ?", (lyrics_id,)).fetchone() diff --git a/music-lab/app/main.py b/music-lab/app/main.py index d5287aa..a3b057f 100644 --- a/music-lab/app/main.py +++ b/music-lab/app/main.py @@ -15,7 +15,9 @@ from .db import ( from .local_provider import run_local_generation from .suno_provider import ( run_suno_generation, run_suno_extend, run_vocal_removal, - generate_lyrics, get_credits, + run_cover_image, run_wav_convert, run_stem_split, + run_upload_cover, run_upload_extend, run_add_vocals, run_add_instrumental, run_video_generate, + generate_lyrics, get_credits, get_timestamped_lyrics, generate_style_boost, SUNO_API_KEY, SUNO_MODELS, ) @@ -102,6 +104,11 @@ class GenerateRequest(BaseModel): # Suno 전용 lyrics: str = "" # 커스텀 가사 ([Verse], [Chorus] 등) instrumental: bool = False # True면 보컬 없이 인스트루멘탈만 + # Phase 1 신규 + vocal_gender: Optional[str] = None # "m" | "f" + negative_tags: Optional[str] = None # 제외 스타일 + style_weight: Optional[float] = None # 0.0~1.0 + audio_weight: Optional[float] = None # 0.0~1.0 @app.post("/api/music/generate") @@ -402,6 +409,224 @@ def vocal_removal(req: VocalRemovalRequest, background_tasks: BackgroundTasks): return {"task_id": task_id, "provider": "suno"} +# ── 커버 이미지 생성 API ──────────────────────────────────────────────────── + +class CoverImageRequest(BaseModel): + suno_task_id: str # Suno 생성 task ID + track_id: Optional[int] = None # 라이브러리 트랙 ID (결과 저장용) + + +@app.post("/api/music/cover-image") +def cover_image(req: CoverImageRequest, background_tasks: BackgroundTasks): + """Suno 곡의 커버 이미지 2장 생성.""" + if not SUNO_API_KEY: + raise HTTPException(status_code=400, detail="Suno API 키가 설정되지 않았습니다") + + task_id = str(uuid.uuid4()) + params = req.model_dump() + create_task(task_id, params, provider="suno") + background_tasks.add_task(run_cover_image, task_id, params) + return {"task_id": task_id, "provider": "suno"} + + +# ── WAV 변환 API ──────────────────────────────────────────────────────────── + +class WavRequest(BaseModel): + suno_task_id: str + suno_id: str + track_id: Optional[int] = None + + +@app.post("/api/music/wav") +def wav_convert(req: WavRequest, background_tasks: BackgroundTasks): + """곡을 WAV 포맷으로 변환.""" + if not SUNO_API_KEY: + raise HTTPException(status_code=400, detail="Suno API 키가 설정되지 않았습니다") + task_id = str(uuid.uuid4()) + params = req.model_dump() + create_task(task_id, params, provider="suno") + background_tasks.add_task(run_wav_convert, task_id, params) + return {"task_id": task_id, "provider": "suno"} + + +# ── 12스템 분리 API ───────────────────────────────────────────────────────── + +class StemSplitRequest(BaseModel): + suno_task_id: str + suno_id: str + track_id: Optional[int] = None + + +@app.post("/api/music/stem-split") +def stem_split(req: StemSplitRequest, background_tasks: BackgroundTasks): + """곡을 12개 스템으로 분리 (50 크레딧). 보컬, 드럼, 베이스, 기타 등.""" + if not SUNO_API_KEY: + raise HTTPException(status_code=400, detail="Suno API 키가 설정되지 않았습니다") + task_id = str(uuid.uuid4()) + params = req.model_dump() + create_task(task_id, params, provider="suno") + background_tasks.add_task(run_stem_split, task_id, params) + return {"task_id": task_id, "provider": "suno"} + + +# ── 타임스탬프 가사 API ───────────────────────────────────────────────────── + +@app.get("/api/music/timestamped-lyrics") +def timestamped_lyrics(task_id: str, suno_id: str): + """타임스탬프 가사 조회 (가라오케 스타일 싱크용).""" + if not SUNO_API_KEY: + raise HTTPException(status_code=400, detail="Suno API 키가 설정되지 않았습니다") + result = get_timestamped_lyrics(task_id, suno_id) + if not result: + raise HTTPException(status_code=502, detail="타임스탬프 가사 조회 실패") + return result + + +# ── 스타일 부스트 API ─────────────────────────────────────────────────────── + +class StyleBoostRequest(BaseModel): + content: str + + +@app.post("/api/music/style-boost") +def style_boost(req: StyleBoostRequest): + """AI로 최적 스타일 프롬프트 생성.""" + if not SUNO_API_KEY: + raise HTTPException(status_code=400, detail="Suno API 키가 설정되지 않았습니다") + result = generate_style_boost(req.content) + if not result: + raise HTTPException(status_code=502, detail="스타일 부스트 생성 실패") + return result + + +# ── Phase 3: 업로드 + 커버 ────────────────────────────────────────────────── + +class UploadCoverRequest(BaseModel): + upload_url: str + model: str = "V4" + custom_mode: bool = True + instrumental: bool = False + prompt: str = "" + style: str = "" + title: str = "" + vocal_gender: Optional[str] = None + negative_tags: Optional[str] = None + style_weight: Optional[float] = None + audio_weight: Optional[float] = None + + +@app.post("/api/music/upload-cover") +def upload_cover(req: UploadCoverRequest, background_tasks: BackgroundTasks): + """외부 오디오를 Suno 스타일로 리메이크.""" + if not SUNO_API_KEY: + raise HTTPException(status_code=400, detail="Suno API 키가 설정되지 않았습니다") + task_id = str(uuid.uuid4()) + params = req.model_dump() + create_task(task_id, params, provider="suno") + background_tasks.add_task(run_upload_cover, task_id, params) + return {"task_id": task_id, "provider": "suno"} + + +# ── Phase 3: 업로드 + 확장 ────────────────────────────────────────────────── + +class UploadExtendRequest(BaseModel): + upload_url: str + model: str = "V4" + default_param_flag: bool = True + continue_at: Optional[float] = None + prompt: str = "" + style: str = "" + title: str = "" + instrumental: bool = False + vocal_gender: Optional[str] = None + negative_tags: Optional[str] = None + + +@app.post("/api/music/upload-extend") +def upload_extend(req: UploadExtendRequest, background_tasks: BackgroundTasks): + """외부 오디오를 이어서 확장.""" + if not SUNO_API_KEY: + raise HTTPException(status_code=400, detail="Suno API 키가 설정되지 않았습니다") + task_id = str(uuid.uuid4()) + params = req.model_dump() + create_task(task_id, params, provider="suno") + background_tasks.add_task(run_upload_extend, task_id, params) + return {"task_id": task_id, "provider": "suno"} + + +# ── Phase 3: 보컬 추가 ────────────────────────────────────────────────────── + +class AddVocalsRequest(BaseModel): + upload_url: str + prompt: str + title: str + style: str + negative_tags: str = "" + vocal_gender: Optional[str] = None + model: str = "V4_5PLUS" + style_weight: Optional[float] = None + audio_weight: Optional[float] = None + + +@app.post("/api/music/add-vocals") +def add_vocals(req: AddVocalsRequest, background_tasks: BackgroundTasks): + """인스트루멘탈에 AI 보컬 추가.""" + if not SUNO_API_KEY: + raise HTTPException(status_code=400, detail="Suno API 키가 설정되지 않았습니다") + task_id = str(uuid.uuid4()) + params = req.model_dump() + create_task(task_id, params, provider="suno") + background_tasks.add_task(run_add_vocals, task_id, params) + return {"task_id": task_id, "provider": "suno"} + + +# ── Phase 3: 인스트루멘탈 추가 ────────────────────────────────────────────── + +class AddInstrumentalRequest(BaseModel): + upload_url: str + title: str + tags: str + negative_tags: str = "" + vocal_gender: Optional[str] = None + model: str = "V4_5PLUS" + style_weight: Optional[float] = None + audio_weight: Optional[float] = None + + +@app.post("/api/music/add-instrumental") +def add_instrumental(req: AddInstrumentalRequest, background_tasks: BackgroundTasks): + """보컬에 AI 반주 추가.""" + if not SUNO_API_KEY: + raise HTTPException(status_code=400, detail="Suno API 키가 설정되지 않았습니다") + task_id = str(uuid.uuid4()) + params = req.model_dump() + create_task(task_id, params, provider="suno") + background_tasks.add_task(run_add_instrumental, task_id, params) + return {"task_id": task_id, "provider": "suno"} + + +# ── Phase 3: 뮤직비디오 생성 ──────────────────────────────────────────────── + +class VideoRequest(BaseModel): + suno_task_id: str + suno_id: str + author: str = "" + domain_name: str = "" + track_id: Optional[int] = None + + +@app.post("/api/music/video") +def video_generate(req: VideoRequest, background_tasks: BackgroundTasks): + """뮤직비디오(MP4) 생성.""" + if not SUNO_API_KEY: + raise HTTPException(status_code=400, detail="Suno API 키가 설정되지 않았습니다") + task_id = str(uuid.uuid4()) + params = req.model_dump() + create_task(task_id, params, provider="suno") + background_tasks.add_task(run_video_generate, task_id, params) + return {"task_id": task_id, "provider": "suno"} + + # ── 저장된 가사 CRUD API ──────────────────────────────────────────────────── class LyricsSave(BaseModel): diff --git a/music-lab/app/suno_provider.py b/music-lab/app/suno_provider.py index d2446c6..005924e 100644 --- a/music-lab/app/suno_provider.py +++ b/music-lab/app/suno_provider.py @@ -3,13 +3,18 @@ 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 +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__) @@ -29,6 +34,7 @@ SUNO_MODELS = [ {"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": "커스텀 모델, 최신 음악성"}, ] @@ -140,9 +146,13 @@ def run_suno_generation(task_id: str, params: dict) -> None: logger.info("Suno task created: %s (internal: %s)", suno_task_id, task_id) # ── 2단계: 상태 폴링 ── - completed_tracks = _poll_until_complete(task_id, suno_task_id) + 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: - return # 에러는 _poll_until_complete 내부에서 처리 + update_task(task_id, "failed", 0, "", error="Suno 생성 완료했으나 트랙 데이터 없음") + return update_task(task_id, "processing", 80, "오디오 파일 다운로드 중...") @@ -231,63 +241,73 @@ def _build_suno_payload(params: dict) -> dict: 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_until_complete(task_id: str, suno_task_id: str) -> Optional[list]: - """sunoapi.org record-info를 폴링하여 SUCCESS가 될 때까지 대기.""" +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(POLL_MAX_ATTEMPTS): - time.sleep(POLL_INTERVAL) - + for attempt in range(max_attempts): + time.sleep(interval) try: resp = requests.get( - f"{SUNO_BASE_URL}/generate/record-info", + 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 / POLL_MAX_ATTEMPTS) * 65), 79) + progress = min(15 + int((attempt / 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": - # data.response.sunoData 에 트랙 배열이 들어있음 - response_obj = data.get("response", {}) - tracks = response_obj.get("sunoData") or [] - if tracks: - return tracks - update_task(task_id, "failed", 0, "", error="Suno 생성 완료했으나 트랙 데이터 없음") - return None + 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})" + 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})") - + 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 생성 타임아웃 (5분 초과)") + update_task(task_id, "failed", 0, "", error="Suno 작업 타임아웃") return None @@ -351,20 +371,20 @@ def _download_and_register( # ── 크레딧 조회 ────────────────────────────────────────────────────────────── def get_credits() -> Optional[dict]: - """Suno API 잔여 크레딧 조회.""" + """Suno API 잔여 크레딧 조회. 두 엔드포인트 폴백.""" if not SUNO_API_KEY: return None - try: - resp = requests.get( - f"{SUNO_BASE_URL}/get-credits", - headers=_headers(), - timeout=15, - ) - if resp.status_code == 200: - body = resp.json() - return body.get("data", body) - except Exception as e: - logger.warning("Suno credits API error: %s", e) + 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 @@ -418,8 +438,12 @@ def run_suno_extend(task_id: str, params: dict) -> None: update_task(task_id, "processing", 15, "곡 연장 대기열에 등록됨...") - completed_tracks = _poll_until_complete(task_id, suno_task_id) + 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, "연장된 오디오 다운로드 중...") @@ -483,8 +507,12 @@ def run_vocal_removal(task_id: str, params: dict) -> None: update_task(task_id, "processing", 15, "보컬 분리 처리 중...") # 보컬 분리 결과 폴링 - completed_tracks = _poll_until_complete(task_id, suno_task_id) + 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, "분리된 오디오 다운로드 중...") @@ -512,3 +540,511 @@ def run_vocal_removal(task_id: str, params: dict) -> None: 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))