# Batch Music Generation — Implementation Plan > **For agentic workers:** Use `superpowers:subagent-driven-development`. Steps use `- [ ]` checkboxes. **Goal:** 장르 1개로 N(1-10) 트랙 Suno 자동 순차 생성 + 자동 컴파일 + 영상 파이프라인 자동 시작. **Architecture:** music-lab 신규 `batch_generator` 모듈이 BackgroundTask로 N회 Suno 호출 → compile_job 자동 생성 → orchestrator.run_step("cover") 자동 호출. **Spec:** `docs/superpowers/specs/2026-05-10-batch-music-generation-design.md` --- ## File Structure | 경로 | 책임 | |------|------| | `music-lab/app/db.py` (modify) | `music_batch_jobs` 테이블 + 5 헬퍼 | | `music-lab/app/random_pools.py` (new) | 장르별 mood/instr/BPM/key/scale 랜덤 풀 + `randomize()` | | `music-lab/app/batch_generator.py` (new) | `run_batch(batch_id)` 순차 오케스트레이션 | | `music-lab/app/main.py` (modify) | 3개 endpoint (POST /generate-batch, GET /:id, GET 목록) | | `web-ui/src/api.js` (modify) | 3개 헬퍼 | | `web-ui/src/pages/music/components/BatchProgress.jsx` (new) | 진행 표시 컴포넌트 | | `web-ui/src/pages/music/MusicStudio.jsx` (modify) | Create 탭에 배치 섹션 + 폴링 | | `web-ui/src/pages/music/MusicStudio.css` (modify) | 배치 섹션 스타일 | --- ## Task 1: DB 테이블 + 헬퍼 + random_pools **Files:** - Modify: `music-lab/app/db.py` - Create: `music-lab/app/random_pools.py` - Test: `music-lab/tests/test_batch_db.py` - [ ] **Step 1: random_pools.py 작성** ```python """장르별 음악 파라미터 랜덤 풀.""" import random POOLS = { "lo-fi": { "moods": ["chill", "relaxing", "dreamy", "melancholic", "mellow", "nostalgic", "peaceful"], "instruments_pool": ["piano", "synth", "drums", "vinyl", "rhodes", "soft bass", "ambient pads"], "instruments_count": (3, 4), "bpm": (70, 90), "keys": ["C", "D", "F", "G", "A"], "scales": ["minor", "major"], "prompt_modifiers": ["cozy bedroom vibes", "rainy night", "late night study", "cafe ambience"], }, "phonk": { "moods": ["dark", "aggressive", "moody", "intense", "hypnotic"], "instruments_pool": ["808 bass", "hi-hat", "synth lead", "vocal chops", "bass drops", "trap drums"], "instruments_count": (3, 4), "bpm": (130, 160), "keys": ["C", "D", "F", "G"], "scales": ["minor"], "prompt_modifiers": ["drift atmosphere", "dark neon", "midnight drive"], }, "ambient": { "moods": ["peaceful", "meditative", "ethereal", "spacious", "dreamy"], "instruments_pool": ["pad synths", "atmospheric guitar", "soft strings", "field recordings", "drone bass"], "instruments_count": (2, 3), "bpm": (50, 75), "keys": ["C", "D", "E", "G", "A"], "scales": ["major", "minor"], "prompt_modifiers": ["misty mountain morning", "deep space", "still water", "forest dawn"], }, "pop": { "moods": ["uplifting", "happy", "energetic", "romantic", "catchy"], "instruments_pool": ["acoustic guitar", "piano", "drums", "bass", "synth", "vocals harmonies"], "instruments_count": (3, 5), "bpm": (95, 130), "keys": ["C", "D", "E", "F", "G", "A"], "scales": ["major"], "prompt_modifiers": ["radio-ready", "summer vibe", "feel-good"], }, "default": { "moods": ["chill", "relaxing", "uplifting", "mellow"], "instruments_pool": ["piano", "synth", "drums", "guitar", "bass", "strings"], "instruments_count": (3, 4), "bpm": (80, 110), "keys": ["C", "D", "F", "G", "A"], "scales": ["minor", "major"], "prompt_modifiers": [""], }, } def randomize(genre: str, rng=None) -> dict: rng = rng or random.Random() pool = POOLS.get(genre.lower(), POOLS["default"]) n_instr = rng.randint(*pool["instruments_count"]) instruments = rng.sample(pool["instruments_pool"], min(n_instr, len(pool["instruments_pool"]))) return { "moods": [rng.choice(pool["moods"])], "instruments": instruments, "bpm": rng.randint(*pool["bpm"]), "key": rng.choice(pool["keys"]), "scale": rng.choice(pool["scales"]), "prompt_modifier": rng.choice(pool["prompt_modifiers"]), } ``` - [ ] **Step 2: DB 테이블 + 헬퍼 추가** (db.py) `init_db()`에 추가: ```python cursor.execute(""" CREATE TABLE IF NOT EXISTS music_batch_jobs ( id INTEGER PRIMARY KEY AUTOINCREMENT, genre TEXT NOT NULL, count INTEGER NOT NULL, target_duration_sec INTEGER NOT NULL DEFAULT 180, auto_pipeline INTEGER NOT NULL DEFAULT 1, completed INTEGER NOT NULL DEFAULT 0, track_ids_json TEXT NOT NULL DEFAULT '[]', current_track_index INTEGER NOT NULL DEFAULT 0, current_track_status TEXT, status TEXT NOT NULL DEFAULT 'queued', error TEXT, compile_job_id INTEGER, pipeline_id INTEGER, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ) """) ``` `db.py` 끝에 헬퍼: ```python _BATCH_ALLOWED_COLS = frozenset([ "completed", "track_ids_json", "current_track_index", "current_track_status", "status", "error", "compile_job_id", "pipeline_id", ]) def create_batch_job(genre: str, count: int, target_duration_sec: int = 180, auto_pipeline: bool = True) -> int: with _conn() as conn: now = _now() cur = conn.cursor() cur.execute(""" INSERT INTO music_batch_jobs (genre, count, target_duration_sec, auto_pipeline, status, created_at, updated_at) VALUES (?, ?, ?, ?, 'queued', ?, ?) """, (genre, count, target_duration_sec, 1 if auto_pipeline else 0, now, now)) return cur.lastrowid def get_batch_job(batch_id: int) -> dict | None: with _conn() as conn: row = conn.execute( "SELECT * FROM music_batch_jobs WHERE id = ?", (batch_id,) ).fetchone() if not row: return None d = dict(row) d["track_ids"] = json.loads(d.get("track_ids_json") or "[]") return d def update_batch_job(batch_id: int, **fields) -> None: unknown = set(fields) - _BATCH_ALLOWED_COLS if unknown: raise ValueError(f"unknown batch job columns: {unknown}") cols = ", ".join(f"{k} = ?" for k in fields) vals = list(fields.values()) + [_now(), batch_id] with _conn() as conn: conn.execute( f"UPDATE music_batch_jobs SET {cols}, updated_at = ? WHERE id = ?", vals, ) def append_batch_track(batch_id: int, track_id: int) -> None: """track_ids_json에 새 track_id 추가 + completed += 1 (atomic).""" with _conn() as conn: row = conn.execute( "SELECT track_ids_json, completed FROM music_batch_jobs WHERE id = ?", (batch_id,), ).fetchone() if not row: return ids = json.loads(row["track_ids_json"] or "[]") ids.append(track_id) conn.execute( "UPDATE music_batch_jobs SET track_ids_json = ?, completed = ?, updated_at = ? WHERE id = ?", (json.dumps(ids), row["completed"] + 1, _now(), batch_id), ) def list_batch_jobs(active_only: bool = False) -> list[dict]: sql = "SELECT * FROM music_batch_jobs" if active_only: sql += " WHERE status NOT IN ('failed','cancelled','piped')" sql += " ORDER BY created_at DESC" with _conn() as conn: rows = conn.execute(sql).fetchall() out = [] for r in rows: d = dict(r) d["track_ids"] = json.loads(d.get("track_ids_json") or "[]") out.append(d) return out ``` - [ ] **Step 3: Test 작성** ```python # tests/test_batch_db.py import pytest from app import db @pytest.fixture def fresh_db(monkeypatch, tmp_path): monkeypatch.setattr(db, "DB_PATH", str(tmp_path / "music.db")) db.init_db() return db def test_create_batch_job(fresh_db): bid = db.create_batch_job(genre="lo-fi", count=10) j = db.get_batch_job(bid) assert j["genre"] == "lo-fi" assert j["count"] == 10 assert j["status"] == "queued" assert j["track_ids"] == [] assert j["auto_pipeline"] == 1 def test_update_batch_job(fresh_db): bid = db.create_batch_job(genre="phonk", count=5) db.update_batch_job(bid, status="generating", current_track_index=2) j = db.get_batch_job(bid) assert j["status"] == "generating" assert j["current_track_index"] == 2 def test_update_batch_rejects_unknown_col(fresh_db): bid = db.create_batch_job(genre="lo-fi", count=1) with pytest.raises(ValueError): db.update_batch_job(bid, evil_col="x") def test_append_batch_track(fresh_db): bid = db.create_batch_job(genre="lo-fi", count=3) db.append_batch_track(bid, 101) db.append_batch_track(bid, 102) j = db.get_batch_job(bid) assert j["track_ids"] == [101, 102] assert j["completed"] == 2 def test_list_batch_jobs_active_filter(fresh_db): b1 = db.create_batch_job(genre="lo-fi", count=1) b2 = db.create_batch_job(genre="phonk", count=1) db.update_batch_job(b1, status="failed") actives = db.list_batch_jobs(active_only=True) assert all(j["status"] not in ("failed",) for j in actives) assert any(j["id"] == b2 for j in actives) assert not any(j["id"] == b1 for j in actives) def test_random_pools_randomize(): from app.random_pools import randomize, POOLS import random rng = random.Random(42) result = randomize("lo-fi", rng) assert result["bpm"] in range(70, 91) assert result["key"] in POOLS["lo-fi"]["keys"] assert result["scale"] in POOLS["lo-fi"]["scales"] assert len(result["moods"]) == 1 assert result["moods"][0] in POOLS["lo-fi"]["moods"] assert 3 <= len(result["instruments"]) <= 4 def test_random_pools_unknown_genre_uses_default(): from app.random_pools import randomize, POOLS import random result = randomize("nonexistent", random.Random(0)) assert result["bpm"] in range(80, 111) # default range ``` - [ ] **Step 4: Run + commit** ```bash cd music-lab && python -m pytest tests/test_batch_db.py -v ``` Expected: 7 PASS. ```bash git add music-lab/app/db.py music-lab/app/random_pools.py music-lab/tests/test_batch_db.py git commit -m "feat(music-lab): music_batch_jobs 테이블 + 장르별 랜덤 풀" ``` --- ## Task 2: batch_generator + 3 엔드포인트 **Files:** - Create: `music-lab/app/batch_generator.py` - Modify: `music-lab/app/main.py` - Test: `music-lab/tests/test_batch_endpoints.py` - [ ] **Step 1: batch_generator.py 작성** ```python """배치 음악 생성 + 자동 컴파일·영상 파이프라인.""" import asyncio import logging from . import db from .random_pools import randomize logger = logging.getLogger("music-lab.batch") POLL_INTERVAL_S = 5 TRACK_GEN_TIMEOUT_S = 240 async def run_batch(batch_id: int) -> None: job = db.get_batch_job(batch_id) if not job: return genre = job["genre"] count = job["count"] duration = job["target_duration_sec"] auto_pipe = bool(job["auto_pipeline"]) db.update_batch_job(batch_id, status="generating") track_ids: list[int] = [] for i in range(1, count + 1): title = f"{genre.title()} Mix Track {i}" params = randomize(genre) db.update_batch_job(batch_id, current_track_index=i, current_track_status="generating") track_id = await _generate_one_track(title=title, genre=genre, duration_sec=duration, params=params) if track_id: track_ids.append(track_id) db.append_batch_track(batch_id, track_id) db.update_batch_job(batch_id, current_track_status="succeeded") else: db.update_batch_job(batch_id, current_track_status="failed") logger.warning("배치 %d 트랙 %d 실패 — 계속 진행", batch_id, i) if not track_ids: db.update_batch_job(batch_id, status="failed", error="모든 트랙 생성 실패") return db.update_batch_job(batch_id, status="generated") if not auto_pipe: return # 자동 컴파일 db.update_batch_job(batch_id, status="compiling") try: compile_id = db.create_compile_job( title=f"{genre.title()} Mix", track_ids=track_ids, crossfade_sec=3, ) db.update_batch_job(batch_id, compile_job_id=compile_id) except Exception as e: db.update_batch_job(batch_id, status="failed", error=f"compile create: {e}") return from . import compiler try: await asyncio.to_thread(compiler.run, compile_id) except Exception as e: db.update_batch_job(batch_id, status="failed", error=f"compile run: {e}") return job_after = db.get_compile_job(compile_id) if not job_after or job_after.get("status") not in ("done", "succeeded"): db.update_batch_job( batch_id, status="failed", error=f"compile not done (status={job_after.get('status') if job_after else 'unknown'})" ) return # 자동 영상 파이프라인 pipeline_id = db.create_pipeline(compile_job_id=compile_id) db.update_batch_job(batch_id, pipeline_id=pipeline_id, status="piped") from .pipeline import orchestrator await orchestrator.run_step(pipeline_id, "cover") async def _generate_one_track(*, title: str, genre: str, duration_sec: int, params: dict) -> int | None: """기존 Suno generate 호출 + 완료까지 polling. 성공 시 새 track id 반환.""" from .suno_provider import run_suno_generation from .db import create_task, get_task import uuid task_id = str(uuid.uuid4()) suno_params = { "title": title, "genre": genre, "moods": params["moods"], "instruments": params["instruments"], "duration_sec": duration_sec, "bpm": params["bpm"], "key": params["key"], "scale": params["scale"], "prompt": params.get("prompt_modifier", ""), } create_task(task_id, suno_params, provider="suno") # Suno background task 직접 호출 (BackgroundTasks 미사용 — 우리가 await) asyncio.create_task(asyncio.to_thread(run_suno_generation, task_id, suno_params)) # Polling waited = 0 while waited < TRACK_GEN_TIMEOUT_S: await asyncio.sleep(POLL_INTERVAL_S) waited += POLL_INTERVAL_S task = get_task(task_id) if not task: continue if task.get("status") == "succeeded": tr = task.get("track") return tr.get("id") if tr else None if task.get("status") == "failed": return None return None # timeout ``` NOTE: This assumes existing `db.create_task`, `db.get_task`, `suno_provider.run_suno_generation` are reusable. Read existing code to confirm function signatures, adjust if needed (especially `task["track"]["id"]` vs other format). - [ ] **Step 2: main.py에 3 endpoint 추가** ```python from app.batch_generator import run_batch as _run_batch 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(400, "count는 1-10 사이") if not (60 <= req.target_duration_sec <= 300): raise HTTPException(400, "target_duration_sec는 60-300 사이") if not req.genre: raise HTTPException(400, "genre 필수") if not SUNO_API_KEY: raise HTTPException(400, "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(404) # tracks 메타 LEFT JOIN (id, title, audio_url) if j["track_ids"]: ids_csv = ",".join(str(i) for i in j["track_ids"]) # 간단한 in-Python 매핑 (sqlite IN (...)) 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() j["tracks"] = [dict(r) for r in rows] 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"))} ``` (SUNO_API_KEY는 main.py에 이미 import돼있다고 가정. 없으면 `_db_module` 패턴처럼 처리.) - [ ] **Step 3: 테스트 작성** ```python # tests/test_batch_endpoints.py import pytest from unittest.mock import AsyncMock, patch, MagicMock from fastapi.testclient import TestClient from app.main import app from app import db @pytest.fixture def client(monkeypatch, tmp_path): monkeypatch.setattr(db, "DB_PATH", str(tmp_path / "music.db")) db.init_db() monkeypatch.setenv("SUNO_API_KEY", "test") return TestClient(app) def test_create_batch_201(client): with patch("app.main._run_batch", new=AsyncMock()): r = client.post("/api/music/generate-batch", json={"genre": "lo-fi", "count": 3}) assert r.status_code == 201 body = r.json() assert body["genre"] == "lo-fi" assert body["count"] == 3 assert body["status"] == "queued" def test_create_batch_rejects_count_too_high(client): r = client.post("/api/music/generate-batch", json={"genre": "lo-fi", "count": 11}) assert r.status_code == 400 def test_create_batch_rejects_count_zero(client): r = client.post("/api/music/generate-batch", json={"genre": "lo-fi", "count": 0}) assert r.status_code == 400 def test_create_batch_rejects_no_genre(client): r = client.post("/api/music/generate-batch", json={"count": 3}) # Pydantic missing 필드 → 422 (FastAPI default validation) assert r.status_code in (400, 422) def test_get_batch_returns_tracks(client): bid = db.create_batch_job(genre="lo-fi", count=2) db.append_batch_track(bid, 999) # phantom track id (not in library) r = client.get(f"/api/music/generate-batch/{bid}") assert r.status_code == 200 body = r.json() assert body["track_ids"] == [999] # tracks 배열은 비어있음 (해당 track 미존재) assert body["tracks"] == [] def test_list_batches(client): db.create_batch_job(genre="lo-fi", count=1) db.create_batch_job(genre="phonk", count=2) r = client.get("/api/music/generate-batch") assert len(r.json()["batches"]) == 2 ``` - [ ] **Step 4: Run + commit + push** ```bash cd music-lab && python -m pytest tests/ -v ``` Expected: 모두 PASS. ```bash git -C C:/Users/jaeoh/Desktop/workspace/web-backend add music-lab/app/batch_generator.py \ music-lab/app/main.py \ music-lab/tests/test_batch_endpoints.py git -C C:/Users/jaeoh/Desktop/workspace/web-backend commit -m "feat(music-lab): 배치 음악 생성 endpoint + orchestrator" git -C C:/Users/jaeoh/Desktop/workspace/web-backend push origin main ``` --- ## Task 3: Frontend Create 탭 배치 섹션 **Files:** - Modify: `web-ui/src/api.js` - Create: `web-ui/src/pages/music/components/BatchProgress.jsx` - Modify: `web-ui/src/pages/music/MusicStudio.jsx` - Modify: `web-ui/src/pages/music/MusicStudio.css` - [ ] **Step 1: api.js 헬퍼** ```javascript // === Batch generation === export const startBatchGen = (payload) => apiPost('/api/music/generate-batch', payload); export const getBatchJob = (id) => apiGet(`/api/music/generate-batch/${id}`); export const listBatchJobs = (status='all') => apiGet(`/api/music/generate-batch?status=${status}`); ``` - [ ] **Step 2: BatchProgress.jsx 신규** ```jsx const STATUS_LABELS = { queued: '대기 중', generating: '음악 생성 중', generated: '음악 완료, 컴파일 대기', compiling: '컴파일 중', piped: '영상 파이프라인 시작됨', failed: '실패', cancelled: '취소', }; export default function BatchProgress({ batch }) { if (!batch) return null; const trackList = Array.from({ length: batch.count }, (_, i) => i + 1); return (
예상: 약 {Math.ceil(batchCount * 1.5)}-{batchCount * 2}분 · 비용 ~${(batchCount * 0.005 + (batchAutoPipe ? 0.05 : 0)).toFixed(2)}