feat(music-lab): Phase 3 백엔드 — 업로드커버, 업로드확장, 보컬추가, 인스트추가, 뮤직비디오

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-04-08 09:10:07 +09:00
parent 94969f97a8
commit 26997a7dc7
2 changed files with 406 additions and 0 deletions
--- 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):
--- 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))