gahusb/web-page-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

music-lab: 서비스 고도화 — duration 수정 + 모델/크레딧/연장/분리 API 추가

Browse Source 
Phase 1A:
- mutagen으로 MP3 실제 재생시간 추출 (sync + startup backfill)
- update_track_duration() DB 헬퍼 추가

Phase 2:
- GET /api/music/models — Suno 모델 목록 (V4~V5)
- GET /api/music/credits — 잔여 크레딧 조회
- POST /api/music/extend — 곡 연장 (continueAt 지점부터)
- POST /api/music/vocal-removal — 보컬/인스트루멘탈 분리
- GenerateRequest에 model 필드 추가 (하드코딩 V4 제거)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
gahusb
2026-04-04 14:36:52 +09:00
parent 4b339d9d4f
commit 649b99d143
4 changed files with 281 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
music-lab/app/db.py
View File

@@ -207,6 +207,14 @@ def get_track_by_task_id(task_id: str) -> Optional[Dict[str, Any]]:
return _track_row_to_dict(row) if row else None
def update_track_duration(track_id: int, duration_sec: int) -> None:
with _conn() as conn:
conn.execute(
"UPDATE music_library SET duration_sec = ? WHERE id = ? AND duration_sec IS NULL",
(duration_sec, track_id),
)
def get_track_file_path(track_id: int) -> Optional[str]:
with _conn() as conn:
row = conn.execute("SELECT file_path FROM music_library WHERE id = ?", (track_id,)).fetchone()

99
music-lab/app/main.py
View File

@@ -9,9 +9,14 @@ from .db import (
init_db,
create_task, get_task,
get_all_tracks, add_track, delete_track, get_track_file_path, get_track_by_task_id,
update_track_duration,
)
from .local_provider import run_local_generation
from .suno_provider import run_suno_generation, generate_lyrics, SUNO_API_KEY
from .suno_provider import (
run_suno_generation, run_suno_extend, run_vocal_removal,
generate_lyrics, get_credits,
SUNO_API_KEY, SUNO_MODELS,
)
app = FastAPI()
@@ -27,10 +32,30 @@ app.add_middleware(
MUSIC_DATA_DIR = "/app/data"
def _get_mp3_duration(file_path: str) -> Optional[int]:
"""MP3 파일에서 실제 재생 시간(초) 추출."""
try:
from mutagen.mp3 import MP3
audio = MP3(file_path)
return int(audio.info.length)
except Exception:
return None
def _backfill_durations():
"""duration_sec이 없는 기존 트랙에 MP3 메타데이터에서 길이 채우기."""
for t in get_all_tracks():
if t["duration_sec"] is None and t.get("file_path"):
dur = _get_mp3_duration(t["file_path"])
if dur:
update_track_duration(t["id"], dur)
@app.on_event("startup")
def on_startup():
init_db()
os.makedirs(MUSIC_DATA_DIR, exist_ok=True)
_backfill_durations()
@app.get("/health")
@@ -63,6 +88,7 @@ def get_providers():
class GenerateRequest(BaseModel):
provider: str = "suno" # "suno" | "local"
model: str = "V4" # Suno 모델 (V4, V4_5, V5 등)
title: str = ""
genre: str = ""
moods: List[str] = []
@@ -205,15 +231,17 @@ def _sync_library_with_disk():
if fname not in disk_files:
delete_track(t["id"])
# 디스크에는 있지만 DB에 없는 → 추가
# 디스크에는 있지만 DB에 없는 → 추가 (duration 자동 추출)
for f in disk_files:
if f not in db_filenames:
file_path = os.path.join(MUSIC_DATA_DIR, f)
title = os.path.splitext(f)[0].replace("-", " ").replace("_", " ")
add_track({
"title": title,
"audio_url": f"{media_base}/{f}",
"file_path": os.path.join(MUSIC_DATA_DIR, f),
"file_path": file_path,
"provider": "suno",
"duration_sec": _get_mp3_duration(file_path),
})
@@ -239,3 +267,68 @@ def remove_from_library(track_id: int):
pass
return {"ok": True}
# ── 모델 목록 API ────────────────────────────────────────────────────────────
@app.get("/api/music/models")
def get_models():
"""사용 가능한 Suno AI 모델 목록."""
return {"models": SUNO_MODELS}
# ── 크레딧 조회 API ──────────────────────────────────────────────────────────
@app.get("/api/music/credits")
def check_credits():
"""Suno 잔여 크레딧 조회."""
if not SUNO_API_KEY:
raise HTTPException(status_code=400, detail="Suno API 키가 설정되지 않았습니다")
result = get_credits()
if result is None:
raise HTTPException(status_code=502, detail="크레딧 조회 실패")
return result
# ── 곡 연장 API ──────────────────────────────────────────────────────────────
class ExtendRequest(BaseModel):
suno_id: str # 원본 Suno 곡 ID
continue_at: int = 0 # 연장 시작 지점 (초)
prompt: str = "" # 추가 가사/프롬프트
style: str = "" # 스타일 오버라이드
title: str = ""
model: str = "V4"
@app.post("/api/music/extend")
def extend_music(req: ExtendRequest, background_tasks: BackgroundTasks):
"""기존 곡을 특정 지점부터 연장 (Suno Extend API)."""
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_suno_extend, task_id, params)
return {"task_id": task_id, "provider": "suno"}
# ── 보컬 분리 API ────────────────────────────────────────────────────────────
class VocalRemovalRequest(BaseModel):
suno_id: str # Suno 곡 ID
title: str = "" # 원본 트랙 제목
@app.post("/api/music/vocal-removal")
def vocal_removal(req: VocalRemovalRequest, background_tasks: BackgroundTasks):
"""트랙에서 보컬과 인스트루멘탈을 분리 (Suno Vocal Removal API)."""
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_vocal_removal, task_id, params)
return {"task_id": task_id, "provider": "suno"}

177
music-lab/app/suno_provider.py
View File

@@ -22,6 +22,15 @@ 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": "최신, 빠른 생성 + 뛰어난 음악성"},
]
def _headers() -> dict:
return {
@@ -180,7 +189,7 @@ def _build_suno_payload(params: dict) -> dict:
payload = {
"customMode": custom_mode,
"instrumental": instrumental,
"model": "V4", # 안정적인 기본 모델
"model": params.get("model", "V4"),
"callBackUrl": "https://example.com/noop", # 필수 파라미터, 폴링 방식이므로 더미 URL
}
@@ -337,3 +346,169 @@ def _download_and_register(
}
return add_track(track_data)
# ── 크레딧 조회 ──────────────────────────────────────────────────────────────
def get_credits() -> Optional[dict]:
"""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)
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, "곡 연장 대기열에 등록됨...")
completed_tracks = _poll_until_complete(task_id, suno_task_id)
if not completed_tracks:
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, "보컬 분리 처리 중...")
# 보컬 분리 결과 폴링
completed_tracks = _poll_until_complete(task_id, suno_task_id)
if not completed_tracks:
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))

1
music-lab/requirements.txt
View File

@@ -2,3 +2,4 @@ fastapi==0.115.6
uvicorn[standard]==0.30.6
requests==2.32.3
python-multipart==0.0.12
mutagen==1.47.0