gahusb
bb76e62774
- saved_lyrics 테이블 (id, title, text, prompt, created_at, updated_at) - GET /api/music/lyrics/library — 저장된 가사 목록 조회 - POST /api/music/lyrics/library — 가사 저장 - PUT /api/music/lyrics/library/:id — 가사 수정 - DELETE /api/music/lyrics/library/:id — 가사 삭제 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
380 lines
13 KiB
Python
380 lines
13 KiB
Python
import os
|
|
import uuid
|
|
from typing import List, Optional
|
|
from fastapi import FastAPI, HTTPException, BackgroundTasks
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from pydantic import BaseModel
|
|
|
|
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,
|
|
get_all_lyrics, add_lyrics, update_lyrics, delete_lyrics,
|
|
)
|
|
from .local_provider import run_local_generation
|
|
from .suno_provider import (
|
|
run_suno_generation, run_suno_extend, run_vocal_removal,
|
|
generate_lyrics, get_credits,
|
|
SUNO_API_KEY, SUNO_MODELS,
|
|
)
|
|
|
|
app = FastAPI()
|
|
|
|
_cors_origins = os.getenv("CORS_ALLOW_ORIGINS", "http://localhost:3007,http://localhost:8080").split(",")
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=[o.strip() for o in _cors_origins],
|
|
allow_credentials=False,
|
|
allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
|
|
allow_headers=["Content-Type"],
|
|
)
|
|
|
|
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")
|
|
def health():
|
|
return {"ok": True}
|
|
|
|
|
|
@app.get("/api/music/providers")
|
|
def get_providers():
|
|
"""사용 가능한 음악 생성 프로바이더 목록 반환."""
|
|
providers = []
|
|
if os.getenv("MUSIC_AI_SERVER_URL"):
|
|
providers.append({
|
|
"id": "local",
|
|
"name": "MusicGen",
|
|
"description": "로컬 AI 서버 (인스트루멘탈 전용)",
|
|
"features": ["instrumental"],
|
|
})
|
|
if SUNO_API_KEY:
|
|
providers.append({
|
|
"id": "suno",
|
|
"name": "Suno",
|
|
"description": "Suno AI (보컬·가사·인스트루멘탈)",
|
|
"features": ["vocals", "lyrics", "instrumental"],
|
|
})
|
|
return {"providers": providers}
|
|
|
|
|
|
# ── 음악 생성 API ─────────────────────────────────────────────────────────────
|
|
|
|
class GenerateRequest(BaseModel):
|
|
provider: str = "suno" # "suno" | "local"
|
|
model: str = "V4" # Suno 모델 (V4, V4_5, V5 등)
|
|
title: str = ""
|
|
genre: str = ""
|
|
moods: List[str] = []
|
|
instruments: List[str] = []
|
|
duration_sec: Optional[int] = None
|
|
bpm: Optional[int] = None
|
|
key: str = ""
|
|
scale: str = ""
|
|
prompt: str = ""
|
|
# Suno 전용
|
|
lyrics: str = "" # 커스텀 가사 ([Verse], [Chorus] 등)
|
|
instrumental: bool = False # True면 보컬 없이 인스트루멘탈만
|
|
|
|
|
|
@app.post("/api/music/generate")
|
|
def generate_music(req: GenerateRequest, background_tasks: BackgroundTasks):
|
|
"""
|
|
음악 생성 작업 시작. task_id 즉시 반환 후 백그라운드에서 AI 서버 호출.
|
|
provider: "suno" (Suno API) 또는 "local" (MusicGen)
|
|
"""
|
|
provider = req.provider
|
|
if provider == "suno" and not SUNO_API_KEY:
|
|
raise HTTPException(status_code=400, detail="Suno API 키가 설정되지 않았습니다")
|
|
if provider == "local" and not os.getenv("MUSIC_AI_SERVER_URL"):
|
|
raise HTTPException(status_code=400, detail="로컬 AI 서버 URL이 설정되지 않았습니다")
|
|
if provider not in ("suno", "local"):
|
|
raise HTTPException(status_code=400, detail=f"지원하지 않는 provider: {provider}")
|
|
|
|
task_id = str(uuid.uuid4())
|
|
params = req.model_dump()
|
|
create_task(task_id, params, provider=provider)
|
|
|
|
if provider == "suno":
|
|
background_tasks.add_task(run_suno_generation, task_id, params)
|
|
else:
|
|
background_tasks.add_task(run_local_generation, task_id, params)
|
|
|
|
return {"task_id": task_id, "provider": provider}
|
|
|
|
|
|
@app.get("/api/music/status/{task_id}")
|
|
def get_status(task_id: str):
|
|
"""
|
|
생성 작업 상태 조회. 프론트는 succeeded 또는 failed가 될 때까지 폴링.
|
|
status: queued | processing | succeeded | failed
|
|
succeeded 시 track 메타데이터 포함.
|
|
"""
|
|
task = get_task(task_id)
|
|
if not task:
|
|
raise HTTPException(status_code=404, detail="Task not found")
|
|
|
|
resp = {
|
|
"status": task["status"],
|
|
"progress": task["progress"],
|
|
"message": task["message"],
|
|
"audio_url": task["audio_url"],
|
|
"error": task["error"],
|
|
"provider": task["provider"],
|
|
}
|
|
|
|
if task["status"] == "succeeded":
|
|
track = get_track_by_task_id(task_id)
|
|
resp["track"] = track
|
|
|
|
return resp
|
|
|
|
|
|
# ── 가사 생성 API (Suno 전용) ────────────────────────────────────────────────
|
|
|
|
class LyricsRequest(BaseModel):
|
|
prompt: str
|
|
|
|
|
|
@app.post("/api/music/lyrics")
|
|
def gen_lyrics(req: LyricsRequest):
|
|
"""Suno AI로 가사를 생성합니다. 곡 생성 전 가사 미리보기용."""
|
|
if not SUNO_API_KEY:
|
|
raise HTTPException(status_code=400, detail="Suno API 키가 설정되지 않았습니다")
|
|
result = generate_lyrics(req.prompt)
|
|
if not result:
|
|
raise HTTPException(status_code=502, detail="가사 생성에 실패했습니다")
|
|
return result
|
|
|
|
|
|
# ── 라이브러리 API ────────────────────────────────────────────────────────────
|
|
|
|
class TrackCreate(BaseModel):
|
|
title: str = ""
|
|
genre: str = ""
|
|
moods: List[str] = []
|
|
instruments: List[str] = []
|
|
duration_sec: Optional[int] = None
|
|
bpm: Optional[int] = None
|
|
key: str = ""
|
|
scale: str = ""
|
|
prompt: str = ""
|
|
audio_url: str = ""
|
|
file_path: str = ""
|
|
task_id: Optional[str] = None
|
|
tags: List[str] = []
|
|
provider: str = "local"
|
|
lyrics: str = ""
|
|
image_url: str = ""
|
|
suno_id: str = ""
|
|
|
|
|
|
@app.get("/api/music/library")
|
|
def list_library():
|
|
"""저장된 트랙 목록 전체 조회 (생성일 내림차순). 파일시스템과 자동 동기화."""
|
|
_sync_library_with_disk()
|
|
return {"tracks": get_all_tracks()}
|
|
|
|
|
|
def _sync_library_with_disk():
|
|
"""파일시스템의 .mp3 파일과 DB를 동기화.
|
|
- 디스크에 없는 트랙 → DB에서 삭제
|
|
- DB에 없는 .mp3 파일 → 새 트랙으로 추가
|
|
"""
|
|
tracks = get_all_tracks()
|
|
media_base = os.getenv("MUSIC_MEDIA_BASE", "/media/music")
|
|
|
|
# 디스크의 .mp3 파일 목록
|
|
disk_files = set()
|
|
try:
|
|
for f in os.listdir(MUSIC_DATA_DIR):
|
|
if f.lower().endswith(".mp3"):
|
|
disk_files.add(f)
|
|
except OSError:
|
|
return # 디렉토리 접근 불가 시 동기화 스킵
|
|
|
|
# DB 트랙의 파일명 매핑
|
|
db_filenames = {} # filename → track
|
|
for t in tracks:
|
|
if t.get("audio_url"):
|
|
fname = t["audio_url"].split("/")[-1]
|
|
db_filenames[fname] = t
|
|
|
|
# DB에는 있지만 디스크에 없는 → 삭제
|
|
for fname, t in db_filenames.items():
|
|
if fname not in disk_files:
|
|
delete_track(t["id"])
|
|
|
|
# 디스크에는 있지만 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": file_path,
|
|
"provider": "suno",
|
|
"duration_sec": _get_mp3_duration(file_path),
|
|
})
|
|
|
|
|
|
@app.post("/api/music/library", status_code=201)
|
|
def save_to_library(req: TrackCreate):
|
|
"""트랙 수동 추가 (외부 파일 등록 또는 프론트 직접 저장용)"""
|
|
track = add_track(req.model_dump())
|
|
return track
|
|
|
|
|
|
@app.delete("/api/music/library/{track_id}")
|
|
def remove_from_library(track_id: int):
|
|
"""라이브러리에서 트랙 삭제. 로컬 파일도 함께 삭제."""
|
|
file_path = get_track_file_path(track_id)
|
|
ok = delete_track(track_id)
|
|
if not ok:
|
|
raise HTTPException(status_code=404, detail="Track not found")
|
|
|
|
if file_path and os.path.isfile(file_path):
|
|
try:
|
|
os.remove(file_path)
|
|
except OSError:
|
|
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"}
|
|
|
|
|
|
# ── 저장된 가사 CRUD API ────────────────────────────────────────────────────
|
|
|
|
class LyricsSave(BaseModel):
|
|
title: str = ""
|
|
text: str = ""
|
|
prompt: str = ""
|
|
|
|
|
|
class LyricsUpdate(BaseModel):
|
|
title: Optional[str] = None
|
|
text: Optional[str] = None
|
|
prompt: Optional[str] = None
|
|
|
|
|
|
@app.get("/api/music/lyrics/library")
|
|
def list_saved_lyrics():
|
|
"""저장된 가사 목록 전체 조회 (생성일 내림차순)."""
|
|
return {"lyrics": get_all_lyrics()}
|
|
|
|
|
|
@app.post("/api/music/lyrics/library", status_code=201)
|
|
def save_lyrics(req: LyricsSave):
|
|
"""가사 저장."""
|
|
return add_lyrics(req.model_dump())
|
|
|
|
|
|
@app.put("/api/music/lyrics/library/{lyrics_id}")
|
|
def edit_lyrics(lyrics_id: int, req: LyricsUpdate):
|
|
"""가사 수정."""
|
|
data = {k: v for k, v in req.model_dump().items() if v is not None}
|
|
result = update_lyrics(lyrics_id, data)
|
|
if not result:
|
|
raise HTTPException(status_code=404, detail="Lyrics not found")
|
|
return result
|
|
|
|
|
|
@app.delete("/api/music/lyrics/library/{lyrics_id}")
|
|
def remove_lyrics(lyrics_id: int):
|
|
"""가사 삭제."""
|
|
if not delete_lyrics(lyrics_id):
|
|
raise HTTPException(status_code=404, detail="Lyrics not found")
|
|
return {"ok": True}
|