feat(music-lab): 배치 음악 생성 endpoint + 자동 compile·video 파이프라인 오케스트레이터

- batch_generator.py: 장르별 N트랙 순차 Suno 생성 → 자동 compile → 자동 video pipeline
- main.py: POST/GET /api/music/generate-batch, GET /api/music/generate-batch/{id} 추가
- tests: 10개 endpoint 테스트 (검증·필터·404)
This commit is contained in:
2026-05-10 18:57:23 +09:00
parent f0cb06268e
commit 77b8d05ad7
3 changed files with 294 additions and 0 deletions

View File

@@ -35,6 +35,7 @@ from .suno_provider import (
generate_lyrics, get_credits, get_timestamped_lyrics, generate_style_boost,
SUNO_API_KEY, SUNO_MODELS,
)
from .batch_generator import run_batch as _run_batch
app = FastAPI()
@@ -849,6 +850,62 @@ def export_compile(job_id: int):
}
# ── 배치 음악 생성 API ────────────────────────────────────────────────────────
class BatchGenerateRequest(BaseModel):
genre: str
count: int = 10
target_duration_sec: int = 180
auto_pipeline: bool = True
@app.post("/api/music/generate-batch", status_code=201)
async def generate_batch(req: BatchGenerateRequest, bg: BackgroundTasks):
if not (1 <= req.count <= 10):
raise HTTPException(status_code=400, detail="count는 1-10 사이")
if not (60 <= req.target_duration_sec <= 300):
raise HTTPException(status_code=400, detail="target_duration_sec는 60-300 사이")
if not req.genre:
raise HTTPException(status_code=400, detail="genre 필수")
if not SUNO_API_KEY:
raise HTTPException(status_code=400, detail="SUNO_API_KEY 미설정")
batch_id = _db_module.create_batch_job(
genre=req.genre, count=req.count,
target_duration_sec=req.target_duration_sec,
auto_pipeline=req.auto_pipeline,
)
bg.add_task(_run_batch, batch_id)
return _db_module.get_batch_job(batch_id)
@app.get("/api/music/generate-batch/{batch_id}")
def get_batch(batch_id: int):
j = _db_module.get_batch_job(batch_id)
if not j:
raise HTTPException(status_code=404, detail="Not found")
if j["track_ids"]:
ids_csv = ",".join(str(i) for i in j["track_ids"])
import sqlite3
conn = sqlite3.connect(_db_module.DB_PATH)
conn.row_factory = sqlite3.Row
rows = conn.execute(
f"SELECT id, title, audio_url, duration_sec FROM music_library WHERE id IN ({ids_csv})"
).fetchall()
conn.close()
# 트랙을 batch.track_ids 순서대로 정렬
by_id = {r["id"]: dict(r) for r in rows}
j["tracks"] = [by_id.get(tid) for tid in j["track_ids"] if tid in by_id]
else:
j["tracks"] = []
return j
@app.get("/api/music/generate-batch")
def list_batches(status: str = "all"):
return {"batches": _db_module.list_batch_jobs(active_only=(status == "active"))}
# ── 수익화 추적 API ───────────────────────────────────────────────────────────
@app.get("/api/music/revenue/dashboard")