diff --git a/music-lab/app/main.py b/music-lab/app/main.py index 438d8a9..a3b057f 100644 --- a/music-lab/app/main.py +++ b/music-lab/app/main.py @@ -16,6 +16,7 @@ from .local_provider import run_local_generation from .suno_provider import ( run_suno_generation, run_suno_extend, run_vocal_removal, 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, ) @@ -498,6 +499,134 @@ def style_boost(req: StyleBoostRequest): 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 f3e3d99..8f8ccb4 100644 --- a/music-lab/app/suno_provider.py +++ b/music-lab/app/suno_provider.py @@ -757,3 +757,280 @@ def generate_style_boost(content: str) -> Optional[dict]: 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) + + except Exception as e: + logger.exception("Video generate error for task %s", task_id) + update_task(task_id, "failed", 0, "", error=str(e))