Files

gahusb ebbfa6299a docs(plan): Essential Mix 파이프라인 — 17 task 구현 계획

DB 마이그레이션 → orchestrator _resolve_input → cover Pexels 분기 →
background.py 신규 → metadata tracks → video.py 파라미터 확장 →
main.py compile_job_id → Windows essential filter (showfreqs+ring+drawtext) →
server.py schema → 통합 테스트 → 배포 → 프론트(api.js, CompileTab,
PipelineStartModal, PipelineCard+DetailModal, SetupTab) → 프론트 푸시 → E2E.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

2026-05-09 12:44:02 +09:00

94 KiB

Raw Blame History

Essential Mix Pipeline Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 1시간+ 음악 mix를 컴파일 결과물에서 바로 영상화하고, essential 채널 스타일(배경 + 중앙 비주얼라이저 + 곡명 자막)로 발행. 진행 탭에서 산출물을 미리보기 + 상세 모달.

Architecture: 기존 파이프라인 확장 — track_id XOR compile_job_id 입력 받아 orchestrator가 분기. Windows GPU 인코더에 새 essential filter graph 추가. 프론트는 새 모달 + 기존 카드/탭 보강.

Tech Stack: SQLite (마이그레이션), httpx, FastAPI, ffmpeg showfreqs + drawtext + Pexels image/video API, React 18.

Spec: docs/superpowers/specs/2026-05-09-essential-mix-pipeline-design.md

File Structure

NAS music-lab (`web-backend/music-lab/`)

경로	책임
`app/db.py` (modify)	`video_pipelines` 4개 컬럼 추가 + `_add_column_if_missing` 헬퍼 + 컴파일 헬퍼 강화
`app/pipeline/orchestrator.py` (modify)	`_resolve_input(p)` 함수 + 모든 step runner에 적용
`app/pipeline/cover.py` (modify)	`image_source` 파라미터 (`ai`/`pexels`) + Pexels 이미지 검색·다운로드
`app/pipeline/background.py` (new)	`fetch_video_loop(pid, keyword)` — Pexels Video API 호출 + mp4 저장
`app/pipeline/metadata.py` (modify)	`tracks` 파라미터 + Claude prompt에 트랙 리스트 + 챕터 형식
`app/pipeline/video.py` (modify)	`style`/`background_mode`/`background_path`/`tracks` Windows로 전달
`app/main.py` (modify)	`PipelineCreate` 모델 확장, `compile_job_id` 분기, 응답 확장

Windows music_ai (`C:/Users/jaeoh/Desktop/workspace/music_ai/`)

경로	책임
`video_encoder.py` (modify)	`style`/`background_mode`/`tracks` 분기. essential filter_complex 빌더. drawtext 자막 빌더
`assets/visualizer_ring.png` (auto-generate)	서버 시작 시 PIL로 1회 생성 (없으면)
`server.py` (modify)	`EncodeVideoRequest` 모델 확장

Frontend (`web-ui/`)

경로	책임
`src/api.js` (modify)	`createPipeline`을 `(payload)` 받게 변경, `getCompileJobs`/`triggerVideoFromCompile` 헬퍼
`src/pages/music/components/CompileTab.jsx` (modify)	완료된 job 카드에 "🎬 영상 만들기" 버튼
`src/pages/music/components/PipelineStartModal.jsx` (modify)	입력 라디오 (단일/Mix) + compile job 드롭다운 + 고급 옵션
`src/pages/music/components/PipelineCard.jsx` (modify)	mini 미리보기 inline + 카드 클릭 → 모달
`src/pages/music/components/PipelineDetailModal.jsx` (new)	큰 미리보기 + 영상 플레이어 + 메타·검토·트랙·피드백
`src/pages/music/components/SetupTab.jsx` (modify)	visual_defaults 5개 옵션 확장
`src/pages/music/MusicStudio.css` (modify)	새 미리보기/모달 스타일
`src/pages/music/MusicStudio.jsx` (modify)	CompileTab 핸들러에 onVideoFromCompile 추가, YoutubeTab으로 라우팅

Task 1: DB 마이그레이션 — `video_pipelines` 4개 컬럼

Files:

Modify: music-lab/app/db.py
Test: music-lab/tests/test_pipeline_db.py
Step 1: Write failing tests (append to existing test file)

# tests/test_pipeline_db.py — append at end

def test_create_pipeline_with_compile_job(fresh_db):
    pid = db.create_pipeline(track_id=None, compile_job_id=42,
                              visual_style="essential", background_mode="static",
                              background_keyword="rainy cafe")
    row = db.get_pipeline(pid)
    assert row["track_id"] is None
    assert row["compile_job_id"] == 42
    assert row["visual_style"] == "essential"
    assert row["background_mode"] == "static"
    assert row["background_keyword"] == "rainy cafe"


def test_create_pipeline_with_track_keeps_defaults(fresh_db):
    pid = db.create_pipeline(track_id=1)
    row = db.get_pipeline(pid)
    assert row["track_id"] == 1
    assert row["compile_job_id"] is None
    assert row["visual_style"] == "essential"  # default
    assert row["background_mode"] == "static"  # default
    assert row["background_keyword"] is None


def test_migration_idempotent(monkeypatch, tmp_path):
    """init_db 두 번 호출해도 ALTER TABLE 에러 없이 통과."""
    db_path = tmp_path / "music.db"
    monkeypatch.setattr(db, "DB_PATH", str(db_path))
    db.init_db()
    db.init_db()  # 두 번째 — 컬럼 이미 존재해도 OK여야
    # 새 컬럼 모두 존재 확인
    import sqlite3
    conn = sqlite3.connect(str(db_path))
    cols = [r[1] for r in conn.execute("PRAGMA table_info(video_pipelines)").fetchall()]
    assert "compile_job_id" in cols
    assert "visual_style" in cols
    assert "background_mode" in cols
    assert "background_keyword" in cols
    conn.close()

Step 2: Run, verify fail

cd music-lab && python -m pytest tests/test_pipeline_db.py -v

Expected: FAIL on create_pipeline(... compile_job_id ...) (시그니처 불일치) + 새 컬럼 미존재.

Step 3: Add migration helper + ALTER TABLE in init_db()

db.py의 init_db() 안 (기존 video_pipelines CREATE TABLE 직후):

# After the CREATE TABLE IF NOT EXISTS video_pipelines block,
# add idempotent column migrations:
_add_column_if_missing(cursor, "video_pipelines", "compile_job_id", "INTEGER")
_add_column_if_missing(cursor, "video_pipelines", "visual_style", "TEXT NOT NULL DEFAULT 'essential'")
_add_column_if_missing(cursor, "video_pipelines", "background_mode", "TEXT NOT NULL DEFAULT 'static'")
_add_column_if_missing(cursor, "video_pipelines", "background_keyword", "TEXT")

db.py 모듈 레벨 헬퍼 추가 (init_db 함수 앞):

def _add_column_if_missing(cursor, table: str, column: str, ddl: str) -> None:
    """SQLite-safe ALTER TABLE ADD COLUMN — idempotent.

    SQLite의 ALTER TABLE은 컬럼 존재 시 에러 발생. PRAGMA로 미리 확인.
    """
    cursor.execute(f"PRAGMA table_info({table})")
    existing = {row[1] for row in cursor.fetchall()}
    if column not in existing:
        cursor.execute(f"ALTER TABLE {table} ADD COLUMN {column} {ddl}")

Step 4: Update create_pipeline signature

def create_pipeline(track_id: int | None = None, *,
                    compile_job_id: int | None = None,
                    visual_style: str = "essential",
                    background_mode: str = "static",
                    background_keyword: str | None = None) -> int:
    """track_id XOR compile_job_id 검증."""
    if (track_id is None) == (compile_job_id is None):
        raise ValueError("track_id와 compile_job_id 중 정확히 하나만 지정")
    with _conn() as conn:
        cur = conn.cursor()
        now = _now()
        cur.execute("""
            INSERT INTO video_pipelines
              (track_id, compile_job_id, visual_style, background_mode, background_keyword,
               state, state_started_at, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?, 'created', ?, ?, ?)
        """, (track_id, compile_job_id, visual_style, background_mode,
              background_keyword, now, now, now))
        return cur.lastrowid

Step 5: Run tests

cd music-lab && python -m pytest tests/test_pipeline_db.py -v

Expected: 모두 PASS (기존 + 신규 3개 = 12+개).

Step 6: Commit (DO NOT PUSH)

git add music-lab/app/db.py music-lab/tests/test_pipeline_db.py
git commit -m "feat(music-lab): video_pipelines 4 컬럼 추가 (compile_job_id, visual_style, background_mode, background_keyword)"

Task 2: Orchestrator — `_resolve_input` 함수

Files:

Modify: music-lab/app/pipeline/orchestrator.py
Test: music-lab/tests/test_orchestrator_resolve.py (new)
Step 1: Write failing test

# tests/test_orchestrator_resolve.py
import pytest
from unittest.mock import patch, MagicMock
from app.pipeline.orchestrator import _resolve_input


def test_resolve_input_track():
    pipeline = {"id": 1, "track_id": 13, "compile_job_id": None}
    track = {
        "id": 13, "title": "Lo-Fi Drive", "genre": "lo-fi",
        "moods": ["chill"], "duration_sec": 176,
        "file_path": "/app/data/x.mp3", "audio_url": "/media/music/x.mp3",
    }
    with patch("app.pipeline.orchestrator.db.get_track_by_id", return_value=track):
        result = _resolve_input(pipeline)
    assert result["audio_path"] == "/app/data/x.mp3"
    assert result["duration_sec"] == 176
    assert len(result["tracks"]) == 1
    assert result["tracks"][0]["start_offset_sec"] == 0
    assert result["title"] == "Lo-Fi Drive"
    assert result["genre"] == "lo-fi"


def test_resolve_input_compile_job():
    pipeline = {"id": 2, "track_id": None, "compile_job_id": 5}
    job = {
        "id": 5, "status": "succeeded", "title": "Chill Mix",
        "audio_path": "/app/data/compiles/5.mp3",
        "track_ids": [13, 14, 15],
        "crossfade_sec": 3,
    }
    tracks = {
        13: {"id": 13, "title": "T1", "duration_sec": 180},
        14: {"id": 14, "title": "T2", "duration_sec": 200},
        15: {"id": 15, "title": "T3", "duration_sec": 150},
    }
    with patch("app.pipeline.orchestrator.db.get_compile_job", return_value=job), \
         patch("app.pipeline.orchestrator.db.get_track_by_id", side_effect=lambda i: tracks[i]):
        result = _resolve_input(pipeline)
    assert result["audio_path"] == "/app/data/compiles/5.mp3"
    # 누적 = 180+200+150 - 2*3(crossfade) = 524
    assert result["duration_sec"] == 524
    assert len(result["tracks"]) == 3
    assert result["tracks"][0]["start_offset_sec"] == 0
    assert result["tracks"][1]["start_offset_sec"] == 177  # 180 - 3
    assert result["tracks"][2]["start_offset_sec"] == 374  # 177 + 200 - 3
    assert result["title"] == "Chill Mix"
    assert result["genre"] == "mix"


def test_resolve_input_compile_not_ready():
    pipeline = {"id": 3, "track_id": None, "compile_job_id": 6}
    job = {"id": 6, "status": "rendering"}
    with patch("app.pipeline.orchestrator.db.get_compile_job", return_value=job):
        with pytest.raises(ValueError, match="not ready"):
            _resolve_input(pipeline)


def test_resolve_input_neither():
    pipeline = {"id": 4, "track_id": None, "compile_job_id": None}
    with pytest.raises(ValueError, match="track_id"):
        _resolve_input(pipeline)

Step 2: Run, verify fail

cd music-lab && python -m pytest tests/test_orchestrator_resolve.py -v

Expected: ImportError on _resolve_input.

Step 3: Implement _resolve_input in orchestrator.py

orchestrator.py에 추가 (file 상단 import 정리 + 함수 추가):

def _resolve_input(p: dict) -> dict:
    """파이프라인 입력 = 단일 트랙 또는 컴파일 결과.

    반환: {
        "audio_path": str,           # 컨테이너 절대경로
        "duration_sec": int,
        "tracks": list[{"id", "title", "start_offset_sec", "duration_sec"}],
        "title": str,
        "genre": str,                 # mix는 "mix"
        "moods": list[str],
    }
    """
    track_id = p.get("track_id")
    compile_id = p.get("compile_job_id")

    if track_id is None and compile_id is None:
        raise ValueError("track_id 또는 compile_job_id 중 하나는 필요")

    if compile_id is not None:
        job = db.get_compile_job(compile_id)
        if not job or job.get("status") != "succeeded":
            raise ValueError(f"compile job {compile_id} not ready (status={job.get('status') if job else None})")

        tracks = []
        offset = 0.0
        crossfade = job.get("crossfade_sec", 0) or 0
        track_ids = job.get("track_ids") or []
        last_dur = 0
        for tid in track_ids:
            t = db.get_track_by_id(tid)
            if not t:
                continue
            dur = t.get("duration_sec", 0)
            tracks.append({
                "id": tid, "title": t.get("title", ""),
                "start_offset_sec": int(offset),
                "duration_sec": dur,
            })
            offset += dur - crossfade  # acrossfade overlap 차감
            last_dur = dur
        # 마지막 트랙은 풀 길이 반영 (crossfade 빼기 한 것 복구)
        total = int(offset + crossfade) if tracks else 0
        return {
            "audio_path": job["audio_path"],
            "duration_sec": total,
            "tracks": tracks,
            "title": job.get("title") or "Mix",
            "genre": "mix",
            "moods": [],
        }

    # 단일 트랙
    t = db.get_track_by_id(track_id)
    if not t:
        raise ValueError(f"track {track_id} 없음")
    return {
        "audio_path": t.get("file_path") or _local_path(t.get("audio_url", "")),
        "duration_sec": t.get("duration_sec", 0),
        "tracks": [{
            "id": t["id"], "title": t.get("title", ""),
            "start_offset_sec": 0,
            "duration_sec": t.get("duration_sec", 0),
        }],
        "title": t.get("title", ""),
        "genre": t.get("genre", "default"),
        "moods": t.get("moods", []) or [],
    }

Step 4: Run tests

cd music-lab && python -m pytest tests/test_orchestrator_resolve.py -v

Expected: 4 PASS.

Step 5: Update step runners to use _resolve_input

orchestrator.py의 run_step 안 + 각 _run_* 함수 변경. run_step에서 _resolve_input을 한 번만 호출하고 결과를 step runner에 전달:

async def run_step(pipeline_id: int, step: str, feedback: str = "") -> None:
    job_id = db.create_pipeline_job(pipeline_id, step)
    db.update_pipeline_job(job_id, status="running")
    p = db.get_pipeline(pipeline_id)
    try:
        ctx = _resolve_input(p)
    except ValueError as e:
        db.update_pipeline_job(job_id, status="failed", error=str(e))
        db.update_pipeline_state(pipeline_id, "failed", failed_reason=f"{step}: {e}")
        return

    try:
        if step == "cover":
            result = await _run_cover(p, ctx, feedback)
        elif step == "video":
            result = await _run_video(p, ctx)
        elif step == "thumb":
            result = await _run_thumb(p, ctx, feedback)
        elif step == "meta":
            result = await _run_meta(p, ctx, feedback)
        elif step == "review":
            result = await _run_review(p, ctx)
        elif step == "publish":
            result = await _run_publish(p, ctx)
        else:
            raise ValueError(f"unknown step: {step}")
        db.update_pipeline_job(job_id, status="succeeded")
        db.update_pipeline_state(pipeline_id, result["next_state"], **result.get("fields", {}))
    except Exception as e:
        logger.exception("step %s failed for pipeline %s", step, pipeline_id)
        db.update_pipeline_job(job_id, status="failed", error=str(e))
        db.update_pipeline_state(pipeline_id, "failed", failed_reason=f"{step}: {e}")

각 _run_* 함수의 track 파라미터를 ctx로 변경, 안에서 ctx["title"] ctx["genre"] 등 사용:

async def _run_cover(p, ctx, feedback):
    setup = db.get_youtube_setup()
    prompts = setup["cover_prompts"]
    template = prompts.get(ctx["genre"].lower(), prompts.get("default", ""))
    image_source = setup.get("background_image_source", "ai")
    out = await cover.generate(
        pipeline_id=p["id"], genre=ctx["genre"],
        prompt_template=template,
        mood=", ".join(ctx["moods"] or []),
        track_title=ctx["title"], feedback=feedback,
        image_source=image_source,  # NEW
    )
    return {"next_state": "cover_pending", "fields": {"cover_url": out["url"]}}


async def _run_video(p, ctx):
    setup = db.get_youtube_setup()
    vd = setup["visual_defaults"]
    audio_path = ctx["audio_path"]
    cover_path = _local_path(p["cover_url"])
    bg_mode = p.get("background_mode") or vd.get("default_background_mode", "static")
    bg_path = None
    if bg_mode == "video_loop":
        # background.py가 fetch한 결과 — _run_cover가 호출 시점에 같이 받았어야
        # 하나 cover step과 분리해도 OK. 여기선 cover_url과 같은 위치에 loop.mp4 있다 가정.
        loop_local = os.path.join(storage.pipeline_dir(p["id"]), "loop.mp4")
        bg_path = loop_local if os.path.isfile(loop_local) else None
    out = video.generate(
        pipeline_id=p["id"], audio_path=audio_path, cover_path=cover_path,
        genre=ctx["genre"], duration_sec=ctx["duration_sec"],
        resolution=vd.get("resolution", "1920x1080"),
        style=p.get("visual_style") or vd.get("default_visual_style", "essential"),
        background_mode=bg_mode, background_path=bg_path,
        tracks=ctx["tracks"] if len(ctx["tracks"]) > 1 else None,  # 1곡이면 자막 X
    )
    return {"next_state": "video_pending", "fields": {"video_url": out["url"]}}


async def _run_thumb(p, ctx, feedback):
    video_path = _local_path(p["video_url"])
    out = await asyncio.to_thread(
        thumb.generate,
        pipeline_id=p["id"], video_path=video_path,
        track_title=ctx["title"], overlay_text=True,
    )
    return {"next_state": "thumb_pending", "fields": {"thumbnail_url": out["url"]}}


async def _run_meta(p, ctx, feedback):
    setup = db.get_youtube_setup()
    trend_top = _get_trend_top()
    out = await metadata.generate(
        track={"title": ctx["title"], "genre": ctx["genre"],
               "duration_sec": ctx["duration_sec"], "moods": ctx["moods"]},
        template=setup["metadata_template"],
        trend_keywords=trend_top, feedback=feedback,
        tracks=ctx["tracks"] if len(ctx["tracks"]) > 1 else None,  # NEW
    )
    return {"next_state": "meta_pending",
            "fields": {"metadata_json": json.dumps(out, ensure_ascii=False)}}

(나머지 _run_review, _run_publish도 ctx 받게 시그니처만 바꾸고 내용 그대로 — 필요 시 ctx에서 정보 추출)

Step 6: Run all music-lab tests

cd music-lab && python -m pytest tests/ -v

Expected: 모두 PASS (기존 통합 테스트 영향 없는지 확인).

Step 7: Commit

git add music-lab/app/pipeline/orchestrator.py music-lab/tests/test_orchestrator_resolve.py
git commit -m "feat(music-lab): orchestrator _resolve_input — track/compile_job 통합 입력"

Task 3: cover.py — Pexels 이미지 검색 분기

Files:

Modify: music-lab/app/pipeline/cover.py
Test: music-lab/tests/test_cover_generation.py
Step 1: Write failing test (append)

# 기존 imports
import os
from pathlib import Path

# 기존 테스트 끝에 append:

@pytest.mark.asyncio
@respx.mock
async def test_pexels_image_source(tmp_storage, monkeypatch):
    monkeypatch.setenv("PEXELS_API_KEY", "test-pexels-key")
    img_url = "https://images.pexels.com/photos/123/photo.jpg"
    respx.get("https://api.pexels.com/v1/search").mock(
        return_value=Response(200, json={
            "photos": [{
                "id": 123,
                "src": {"large2x": img_url, "original": img_url},
            }],
        })
    )
    png_bytes = bytes.fromhex(
        "89504e470d0a1a0a0000000d49484452000000010000000108020000009077"
        "53de0000000c4944415478da6300010000050001"
        "0d0a2db40000000049454e44ae426082"
    )
    respx.get(img_url).mock(return_value=Response(200, content=png_bytes))

    out = await cover.generate(
        pipeline_id=99, genre="lo-fi", prompt_template="ignored",
        mood="chill", track_title="Mix",
        image_source="pexels",
    )
    assert out["used_fallback"] is False
    assert out["url"].endswith("/cover.jpg")
    assert (tmp_storage / "99" / "cover.jpg").exists()


@pytest.mark.asyncio
async def test_pexels_no_api_key_falls_back(tmp_storage, monkeypatch):
    monkeypatch.delenv("PEXELS_API_KEY", raising=False)
    out = await cover.generate(
        pipeline_id=98, genre="lo-fi", prompt_template="x",
        mood="", track_title="Test",
        image_source="pexels",
    )
    assert out["used_fallback"] is True


@pytest.mark.asyncio
@respx.mock
async def test_pexels_zero_results_falls_back(tmp_storage, monkeypatch):
    monkeypatch.setenv("PEXELS_API_KEY", "test-key")
    respx.get("https://api.pexels.com/v1/search").mock(
        return_value=Response(200, json={"photos": []})
    )
    out = await cover.generate(
        pipeline_id=97, genre="lo-fi", prompt_template="x",
        mood="", track_title="Test",
        image_source="pexels",
    )
    assert out["used_fallback"] is True

Step 2: Run, verify fail

python -m pytest tests/test_cover_generation.py -v Expected: FAIL on image_source 파라미터 미지원.

Step 3: Add Pexels branch to cover.py

cover.py 끝에 헬퍼 함수 + generate()에 image_source 파라미터:

PEXELS_API_KEY = lambda: os.getenv("PEXELS_API_KEY", "")
PEXELS_IMG_TIMEOUT_S = 30


async def _generate_with_pexels(genre: str, mood: str, track_title: str,
                                  out_path: str, keyword_override: str = "") -> bool:
    """Pexels 이미지 검색·다운로드. 성공 시 True. API key 없거나 0 결과면 False."""
    api_key = PEXELS_API_KEY()
    if not api_key:
        return False
    keyword = keyword_override or f"{genre} aesthetic background"
    try:
        async with httpx.AsyncClient(timeout=PEXELS_IMG_TIMEOUT_S) as client:
            resp = await client.get(
                "https://api.pexels.com/v1/search",
                headers={"Authorization": api_key},
                params={"query": keyword, "per_page": 5, "orientation": "landscape"},
            )
            resp.raise_for_status()
            data = resp.json()
            photos = data.get("photos", [])
            if not photos:
                return False
            img_url = photos[0]["src"].get("large2x") or photos[0]["src"].get("original")
            img_resp = await client.get(img_url)
            img_resp.raise_for_status()
            from io import BytesIO
            with Image.open(BytesIO(img_resp.content)) as src:
                img = src.convert("RGB")
            img.save(out_path, "JPEG", quality=92)
            return True
    except (httpx.HTTPError, httpx.TimeoutException, KeyError, ValueError, OSError) as e:
        logger.warning("Pexels 이미지 검색 실패: %s", e)
        return False

generate() 시그니처 + 분기:

async def generate(*, pipeline_id: int, genre: str, prompt_template: str,
                   mood: str = "", track_title: str = "", feedback: str = "",
                   image_source: str = "ai",
                   background_keyword: str = "") -> dict:
    """image_source: 'ai' (DALL·E 기본) | 'pexels' (스톡 사진)."""
    out_path = os.path.join(storage.pipeline_dir(pipeline_id), "cover.jpg")
    used_fallback = False
    error = None

    if image_source == "pexels":
        ok = await _generate_with_pexels(genre, mood, track_title, out_path, background_keyword)
        if ok:
            return {"url": storage.media_url(pipeline_id, "cover.jpg"),
                    "used_fallback": False, "error": None}
        # Pexels 실패 → 그라데이션 폴백
        used_fallback = True
        error = "Pexels 검색 실패 또는 API 키 없음"
        from .gradient import make_gradient_with_title
        make_gradient_with_title(genre, track_title, out_path)
        return {"url": storage.media_url(pipeline_id, "cover.jpg"),
                "used_fallback": True, "error": error}

    # 기존 AI 흐름 그대로 (이미 구현됨)
    ...

(기존 코드는 그대로 두고, 위 분기를 함수 시작부에 추가)

Step 4: Run tests

python -m pytest tests/test_cover_generation.py -v Expected: 모두 PASS (기존 4 + 신규 3 = 7).

Step 5: Commit

git add music-lab/app/pipeline/cover.py music-lab/tests/test_cover_generation.py
git commit -m "feat(music-lab): cover.py Pexels 이미지 검색 분기 (image_source=pexels)"

Task 4: background.py — video_loop 모드 (Pexels 영상)

Files:

Create: music-lab/app/pipeline/background.py
Create: music-lab/tests/test_background.py
Step 1: Write failing test

# tests/test_background.py
import os
import pytest
import respx
from httpx import Response
from app.pipeline import background, storage


@pytest.fixture
def tmp_storage(monkeypatch, tmp_path):
    monkeypatch.setattr(storage, "VIDEO_DATA_DIR", str(tmp_path))
    return tmp_path


@pytest.mark.asyncio
@respx.mock
async def test_fetch_video_loop_success(tmp_storage, monkeypatch):
    monkeypatch.setenv("PEXELS_API_KEY", "k")
    video_url = "https://videos.pexels.com/video-files/123/sample.mp4"
    respx.get("https://api.pexels.com/videos/search").mock(
        return_value=Response(200, json={
            "videos": [{
                "id": 123, "duration": 10,
                "video_files": [
                    {"quality": "hd", "width": 1920, "link": video_url},
                ],
            }],
        })
    )
    respx.get(video_url).mock(return_value=Response(200, content=b"\x00" * 4096))

    result = await background.fetch_video_loop(pipeline_id=10, keyword="rainy window")
    assert result["used_fallback"] is False
    assert (tmp_storage / "10" / "loop.mp4").exists()


@pytest.mark.asyncio
async def test_fetch_video_loop_no_api_key(tmp_storage, monkeypatch):
    monkeypatch.delenv("PEXELS_API_KEY", raising=False)
    result = await background.fetch_video_loop(pipeline_id=11, keyword="rain")
    assert result["used_fallback"] is True


@pytest.mark.asyncio
@respx.mock
async def test_fetch_video_loop_zero_results(tmp_storage, monkeypatch):
    monkeypatch.setenv("PEXELS_API_KEY", "k")
    respx.get("https://api.pexels.com/videos/search").mock(
        return_value=Response(200, json={"videos": []})
    )
    result = await background.fetch_video_loop(pipeline_id=12, keyword="impossible-keyword")
    assert result["used_fallback"] is True

Step 2: Run, verify fail

python -m pytest tests/test_background.py -v Expected: ImportError.

Step 3: Implement background.py

"""Pexels Video API로 background loop 영상 받아오기."""
import os
import logging
import httpx

from . import storage

logger = logging.getLogger("music-lab.background")
TIMEOUT_S = 60


async def fetch_video_loop(pipeline_id: int, keyword: str) -> dict:
    """Pexels Video API → 720p HD mp4 다운로드 → /app/data/videos/{id}/loop.mp4 저장.

    반환: {"path": str | None, "used_fallback": bool, "error": str | None}
    """
    api_key = os.getenv("PEXELS_API_KEY", "")
    if not api_key:
        return {"path": None, "used_fallback": True, "error": "PEXELS_API_KEY 미설정"}

    out_dir = storage.pipeline_dir(pipeline_id)
    out_path = os.path.join(out_dir, "loop.mp4")

    try:
        async with httpx.AsyncClient(timeout=TIMEOUT_S) as client:
            resp = await client.get(
                "https://api.pexels.com/videos/search",
                headers={"Authorization": api_key},
                params={"query": keyword or "ambient calm", "per_page": 5,
                        "orientation": "landscape"},
            )
            resp.raise_for_status()
            data = resp.json()
            videos = data.get("videos", [])
            if not videos:
                return {"path": None, "used_fallback": True,
                        "error": f"Pexels 결과 없음: {keyword}"}

            # 720p HD 우선, 없으면 첫 번째
            chosen = None
            for v in videos:
                for f in v.get("video_files", []):
                    if f.get("quality") == "hd" and f.get("width") in (1280, 1920):
                        chosen = f
                        break
                if chosen:
                    break
            if not chosen:
                chosen = videos[0]["video_files"][0]

            video_url = chosen["link"]
            vid_resp = await client.get(video_url)
            vid_resp.raise_for_status()
            with open(out_path, "wb") as f:
                f.write(vid_resp.content)

            return {"path": out_path, "used_fallback": False, "error": None}
    except (httpx.HTTPError, httpx.TimeoutException, KeyError, ValueError, OSError) as e:
        logger.warning("Pexels video fetch 실패: %s", e)
        return {"path": None, "used_fallback": True, "error": str(e)}

Step 4: Run tests

python -m pytest tests/test_background.py -v Expected: 3 PASS.

Step 5: Hook into orchestrator's _run_cover for video_loop mode

orchestrator.py _run_cover 시작부:

from . import background  # add at top

async def _run_cover(p, ctx, feedback):
    setup = db.get_youtube_setup()
    bg_mode = p.get("background_mode") or setup["visual_defaults"].get("default_background_mode", "static")
    keyword = p.get("background_keyword") or setup["visual_defaults"].get("default_background_keyword", "")

    # video_loop 모드면 Pexels video 받아오기 (병렬로 cover.jpg도 받기 — visualizer overlay용 단색 배경 폴백)
    if bg_mode == "video_loop":
        bg_result = await background.fetch_video_loop(p["id"], keyword)
        # video_loop 받기 성공 여부와 무관하게 cover.jpg는 또 만들어야 (썸네일 추출 + 폴백용)
        # 그냥 그라데이션 cover로 충분 (video_loop 받기 성공 시 시각적으로 안 보이고, 실패 시 cover로 fallback)
        from .gradient import make_gradient_with_title
        out_path = os.path.join(storage.pipeline_dir(p["id"]), "cover.jpg")
        make_gradient_with_title(ctx["genre"], ctx["title"], out_path)
        return {"next_state": "cover_pending",
                "fields": {"cover_url": storage.media_url(p["id"], "cover.jpg")}}

    # 정적 모드 — 기존 cover.generate 흐름 그대로
    prompts = setup["cover_prompts"]
    template = prompts.get(ctx["genre"].lower(), prompts.get("default", ""))
    image_source = setup["visual_defaults"].get("background_image_source", "ai")
    out = await cover.generate(
        pipeline_id=p["id"], genre=ctx["genre"],
        prompt_template=template,
        mood=", ".join(ctx["moods"] or []),
        track_title=ctx["title"], feedback=feedback,
        image_source=image_source,
        background_keyword=keyword,
    )
    return {"next_state": "cover_pending", "fields": {"cover_url": out["url"]}}

Step 6: Run all tests

python -m pytest tests/ -v Expected: 모두 PASS.

Step 7: Commit

git add music-lab/app/pipeline/background.py music-lab/app/pipeline/orchestrator.py \
        music-lab/tests/test_background.py
git commit -m "feat(music-lab): background.py — video_loop 모드 Pexels 영상 다운로드"

Task 5: metadata.py — tracks 옵션 + 챕터 prompt

Files:

Modify: music-lab/app/pipeline/metadata.py
Test: music-lab/tests/test_metadata_generation.py
Step 1: Write failing test (append)

@pytest.mark.asyncio
@respx.mock
async def test_metadata_with_tracks_includes_chapter_format(monkeypatch):
    monkeypatch.setenv("ANTHROPIC_API_KEY", "k")
    captured = {}

    def hook(req):
        import json as _json
        captured["body"] = _json.loads(req.content)
        return Response(200, json={"content": [{"type": "text", "text":
            '{"title":"Lo-Fi Mix 3 Tracks","description":"Track 1: [00:00] T1\\nTrack 2: [03:00] T2",'
            '"tags":["lofi","mix"],"category_id":10}'}]})

    respx.post("https://api.anthropic.com/v1/messages").mock(side_effect=hook)
    result = await metadata.generate(
        track={"title": "Mix", "genre": "mix", "duration_sec": 600,
               "moods": []},
        template={"title": "{title}", "description": "{title}",
                  "tags": [], "category_id": 10},
        trend_keywords=[],
        tracks=[
            {"id": 1, "title": "T1", "start_offset_sec": 0, "duration_sec": 180},
            {"id": 2, "title": "T2", "start_offset_sec": 180, "duration_sec": 200},
            {"id": 3, "title": "T3", "start_offset_sec": 380, "duration_sec": 220},
        ],
    )
    body_str = str(captured["body"])
    assert "T1" in body_str and "T2" in body_str and "T3" in body_str
    assert "00:00" in body_str
    # 챕터 형식 포함 검증 (description에)
    assert result["used_fallback"] is False


@pytest.mark.asyncio
async def test_metadata_fallback_with_tracks(monkeypatch):
    """API 키 없을 때 폴백에서도 트랙 리스트 포함되는지."""
    monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
    result = await metadata.generate(
        track={"title": "Mix", "genre": "mix", "duration_sec": 600, "moods": []},
        template={"title": "{title}", "description": "{title}",
                  "tags": [], "category_id": 10},
        trend_keywords=[],
        tracks=[
            {"id": 1, "title": "T1", "start_offset_sec": 0, "duration_sec": 180},
            {"id": 2, "title": "T2", "start_offset_sec": 180, "duration_sec": 200},
        ],
    )
    assert result["used_fallback"] is True
    # 폴백 description에도 트랙 챕터 자동 포함
    assert "00:00" in result["description"]
    assert "T1" in result["description"]
    assert "T2" in result["description"]

Step 2: Run, verify fail

python -m pytest tests/test_metadata_generation.py -v Expected: FAIL on tracks 파라미터 + 폴백 description에 트랙 미포함.

Step 3: Update metadata.generate signature + prompt + fallback

async def generate(*, track: dict, template: dict, trend_keywords: list[str],
                   feedback: str = "", tracks: list[dict] | None = None) -> dict:
    api_key = _get_api_key()
    if not api_key:
        return {**_fallback_template(track, template, tracks), "used_fallback": True, "error": "no api key"}

    try:
        result = await _call_claude(track, template, trend_keywords, feedback, tracks,
                                    api_key=api_key, model=_get_model())
        return {**result, "used_fallback": False, "error": None}
    except (httpx.HTTPError, httpx.TimeoutException, KeyError, ValueError, json.JSONDecodeError) as e:
        logger.warning("메타데이터 LLM 실패 — 폴백: %s", e)
        return {**_fallback_template(track, template, tracks), "used_fallback": True, "error": str(e)}


def _format_chapters(tracks: list[dict]) -> str:
    """YouTube 챕터 자동 인식 형식: '[mm:ss] 제목' 한 줄씩."""
    if not tracks:
        return ""
    lines = []
    for t in tracks:
        offset = t.get("start_offset_sec", 0)
        m, s = divmod(int(offset), 60)
        h, m = divmod(m, 60)
        if h > 0:
            ts = f"{h:02d}:{m:02d}:{s:02d}"
        else:
            ts = f"{m:02d}:{s:02d}"
        lines.append(f"{ts} {t.get('title', '')}")
    return "\n".join(lines)


def _fallback_template(track: dict, template: dict, tracks: list[dict] | None = None) -> dict:
    fmt_vars = {
        "title": track.get("title", ""),
        "genre": track.get("genre", ""),
        "bpm": track.get("bpm", ""),
        "key": track.get("key", ""),
        "scale": track.get("scale", ""),
    }
    title = template.get("title", "{title}").format(**fmt_vars)
    description = template.get("description", "{title}").format(**fmt_vars)
    if tracks and len(tracks) > 1:
        description = description + "\n\n" + _format_chapters(tracks)
    return {
        "title": title[:100],
        "description": description[:5000],
        "tags": (template.get("tags") or [])[:15],
        "category_id": template.get("category_id", 10),
    }


async def _call_claude(track, template, trend_keywords, feedback, tracks,
                        *, api_key, model):
    user_prompt = (
        "다음 트랙의 YouTube 메타데이터를 생성하세요. JSON으로만 응답.\n\n"
        f"트랙: {json.dumps(track, ensure_ascii=False)}\n"
        f"템플릿: {json.dumps(template, ensure_ascii=False)}\n"
        f"트렌드 키워드: {', '.join(trend_keywords)}\n"
    )
    if tracks and len(tracks) > 1:
        chapters = _format_chapters(tracks)
        user_prompt += (
            f"\n이 영상은 {len(tracks)}개 트랙의 mix입니다. "
            f"description에 다음 챕터 리스트를 그대로 포함하세요 (YouTube 자동 챕터 인식용):\n{chapters}\n"
        )
    if feedback:
        user_prompt += f"\n사용자 피드백: {feedback}\n"
    user_prompt += (
        '\n출력 JSON: {"title": "60자 이내", "description": "1000자 이내",'
        ' "tags": ["15개 이내"], "category_id": 10}'
    )

    async with httpx.AsyncClient(timeout=TIMEOUT_S) as client:
        resp = await client.post(
            "https://api.anthropic.com/v1/messages",
            headers={
                "x-api-key": api_key,
                "anthropic-version": "2023-06-01",
                "content-type": "application/json",
            },
            json={
                "model": model, "max_tokens": 2048,  # mix 더 길어서
                "messages": [{"role": "user", "content": user_prompt}],
            },
        )
        resp.raise_for_status()
        text = resp.json()["content"][0]["text"]
        start = text.find("{")
        end = text.rfind("}") + 1
        if start < 0 or end <= start:
            raise ValueError("Claude 응답에 JSON 블록 없음")
        return json.loads(text[start:end])

Step 4: Run tests

python -m pytest tests/test_metadata_generation.py -v Expected: 5 PASS (기존 3 + 신규 2).

Step 5: Commit

git add music-lab/app/pipeline/metadata.py music-lab/tests/test_metadata_generation.py
git commit -m "feat(music-lab): metadata tracks 옵션 + YouTube 챕터 자동 형식"

Task 6: NAS pipeline/video.py — Windows 추가 파라미터 전달

Files:

Modify: music-lab/app/pipeline/video.py
Test: music-lab/tests/test_video_thumb.py
Step 1: Write failing test (append)

@respx.mock
def test_generate_video_passes_essential_params(encoder_env, tmp_path, monkeypatch):
    monkeypatch.setattr(storage, "VIDEO_DATA_DIR", str(tmp_path))
    captured = {}

    def hook(req):
        import json as _json
        captured["body"] = _json.loads(req.content)
        return Response(200, json={"ok": True, "duration_ms": 5000,
                                    "output_path_nas": "/v/3/video.mp4",
                                    "output_bytes": 10_000_000,
                                    "encoder": "h264_nvenc", "preset": "p4"})

    respx.post("http://192.168.45.59:8765/encode_video").mock(side_effect=hook)
    out = video.generate(
        pipeline_id=3, audio_path="/app/data/x.mp3",
        cover_path="/app/data/videos/3/cover.jpg",
        genre="mix", duration_sec=3600, resolution="1920x1080",
        style="essential", background_mode="video_loop",
        background_path="/app/data/videos/3/loop.mp4",
        tracks=[{"id": 1, "title": "T1", "start_offset_sec": 0}],
    )
    body = captured["body"]
    assert body["style"] == "essential"
    assert body["background_mode"] == "video_loop"
    assert body["background_path_nas"] == "/volume1/docker/webpage/data/videos/3/loop.mp4"
    assert body["tracks"][0]["title"] == "T1"

Step 2: Run, verify fail

python -m pytest tests/test_video_thumb.py -v Expected: FAIL — video.generate 새 파라미터 미지원.

Step 3: Update video.generate signature + payload

pipeline/video.py:

def generate(*, pipeline_id: int, audio_path: str, cover_path: str,
             genre: str, duration_sec: int, resolution: str = "1920x1080",
             style: str = "essential",
             background_mode: str = "static",
             background_path: str | None = None,
             tracks: list[dict] | None = None) -> dict:
    if not ENCODER_URL:
        raise VideoGenerationError(
            "WINDOWS_VIDEO_ENCODER_URL 미설정 — Windows 인코더 서버 주소 필요"
        )

    out_path = os.path.join(storage.pipeline_dir(pipeline_id), "video.mp4")
    nas_audio = _container_to_nas(audio_path)
    nas_cover = _container_to_nas(cover_path)
    nas_output = _container_to_nas(out_path)
    nas_bg = _container_to_nas(background_path) if background_path else None

    payload = {
        "cover_path_nas": nas_cover,
        "audio_path_nas": nas_audio,
        "output_path_nas": nas_output,
        "resolution": resolution,
        "duration_sec": duration_sec,
        "style": style,
        "background_mode": background_mode,
        "background_path_nas": nas_bg,
        "tracks": tracks or [],
    }
    # ... 기존 httpx 호출 그대로 ...

(기존 함수 내용에서 payload만 확장.)

Step 4: Run tests

python -m pytest tests/test_video_thumb.py -v Expected: 모두 PASS.

Step 5: Commit

git add music-lab/app/pipeline/video.py music-lab/tests/test_video_thumb.py
git commit -m "feat(music-lab): video.py — Windows에 style/background_mode/tracks 전달"

Task 7: main.py — POST /pipeline body 확장

Files:

Modify: music-lab/app/main.py
Test: music-lab/tests/test_pipeline_endpoints.py
Step 1: Write failing test (append)

def test_create_pipeline_with_compile_job(client, monkeypatch):
    # compile_jobs 1개 신규 + status=succeeded
    import sqlite3
    conn = sqlite3.connect(db.DB_PATH)
    cur = conn.cursor()
    try:
        cur.execute("""
            INSERT INTO compile_jobs (title, track_ids_json, crossfade_sec,
                                       audio_path, status, created_at)
            VALUES ('Test Mix', '[1,2,3]', 3, '/app/data/compiles/9.mp3',
                     'succeeded', datetime())
        """)
    except sqlite3.OperationalError:
        # 스키마 다를 수 있음 — 실 컬럼명에 맞게 INSERT를 conftest 패턴 따라 조정
        pytest.skip("compile_jobs schema mismatch")
    conn.commit()
    cid = cur.lastrowid
    conn.close()

    r = client.post("/api/music/pipeline", json={"compile_job_id": cid})
    assert r.status_code == 201
    body = r.json()
    assert body["track_id"] is None
    assert body["compile_job_id"] == cid
    assert body["visual_style"] == "essential"


def test_create_pipeline_rejects_both_inputs(client):
    r = client.post("/api/music/pipeline", json={"track_id": 1, "compile_job_id": 1})
    assert r.status_code == 400


def test_create_pipeline_rejects_neither(client):
    r = client.post("/api/music/pipeline", json={})
    assert r.status_code == 400

Step 2: Run, verify fail

python -m pytest tests/test_pipeline_endpoints.py -v Expected: FAIL on compile_job_id 미지원.

Step 3: Update PipelineCreate model + endpoint

main.py:

class PipelineCreate(BaseModel):
    track_id: int | None = None
    compile_job_id: int | None = None
    visual_style: str | None = None       # single | essential
    background_mode: str | None = None    # static | video_loop
    background_keyword: str | None = None


@app.post("/api/music/pipeline", status_code=201)
def create_pipeline(req: PipelineCreate):
    # XOR 검증
    if (req.track_id is None) == (req.compile_job_id is None):
        raise HTTPException(400, "track_id 또는 compile_job_id 중 정확히 하나를 지정")

    # compile_job 상태 확인
    if req.compile_job_id is not None:
        job = db.get_compile_job(req.compile_job_id)
        if not job:
            raise HTTPException(404, f"compile job {req.compile_job_id} 없음")
        if job.get("status") != "succeeded":
            raise HTTPException(400, f"compile job {req.compile_job_id} not ready (status={job.get('status')})")

    # 동일 입력으로 이미 active 파이프라인 있으면 409
    actives = db.list_pipelines(active_only=True)
    for p in actives:
        if (req.track_id and p.get("track_id") == req.track_id) or \
           (req.compile_job_id and p.get("compile_job_id") == req.compile_job_id):
            raise HTTPException(409, "이미 진행 중인 파이프라인이 있습니다")

    setup = db.get_youtube_setup()
    vd = setup["visual_defaults"]
    pid = db.create_pipeline(
        track_id=req.track_id,
        compile_job_id=req.compile_job_id,
        visual_style=req.visual_style or vd.get("default_visual_style", "essential"),
        background_mode=req.background_mode or vd.get("default_background_mode", "static"),
        background_keyword=req.background_keyword or vd.get("default_background_keyword", ""),
    )
    return db.get_pipeline(pid)

Step 4: Run tests

python -m pytest tests/test_pipeline_endpoints.py -v Expected: 모두 PASS (기존 + 신규 3).

Step 5: Commit

git add music-lab/app/main.py music-lab/tests/test_pipeline_endpoints.py
git commit -m "feat(music-lab): POST /pipeline에 compile_job_id 입력 + visual_style 등 옵션"

Task 8: Windows video_encoder.py — essential filter 분기

Files:

Modify: C:/Users/jaeoh/Desktop/workspace/music_ai/video_encoder.py
Modify: C:/Users/jaeoh/Desktop/workspace/music_ai/tests/test_video_encoder.py
Step 1: Write failing tests

# tests/test_video_encoder.py — append

@patch("video_encoder.subprocess.run")
def test_encode_essential_static_branch(mock_run, env, tmp_path, monkeypatch):
    cover = tmp_path / "cover.jpg"; cover.write_bytes(b"\x00")
    audio = tmp_path / "audio.mp3"; audio.write_bytes(b"\x00")
    out_path = tmp_path / "out.mp4"

    captured_cmd = []
    def fake_run(cmd, **kwargs):
        captured_cmd.extend(cmd)
        out_path.write_bytes(b"\x00" * (2 * 1024 * 1024))
        return MagicMock(returncode=0, stderr="")
    mock_run.side_effect = fake_run

    def fake_translate(p):
        return str(tmp_path / p.split("/")[-1])
    monkeypatch.setattr(video_encoder, "translate_path", fake_translate)

    result = video_encoder.encode_video(
        cover_path_nas="/volume1/cover.jpg",
        audio_path_nas="/volume1/audio.mp3",
        output_path_nas="/volume1/out.mp4",
        resolution="1920x1080",
        duration_sec=120,
        style="essential",
        background_mode="static",
    )
    assert result["ok"] is True
    cmd_str = " ".join(captured_cmd)
    assert "showfreqs" in cmd_str
    assert "h264_nvenc" in cmd_str


@patch("video_encoder.subprocess.run")
def test_encode_essential_video_loop_branch(mock_run, env, tmp_path, monkeypatch):
    cover = tmp_path / "cover.jpg"; cover.write_bytes(b"\x00")
    audio = tmp_path / "audio.mp3"; audio.write_bytes(b"\x00")
    loop = tmp_path / "loop.mp4"; loop.write_bytes(b"\x00" * 1024)
    out = tmp_path / "out.mp4"

    captured_cmd = []
    def fake_run(cmd, **kwargs):
        captured_cmd.extend(cmd)
        out.write_bytes(b"\x00" * (2 * 1024 * 1024))
        return MagicMock(returncode=0, stderr="")
    mock_run.side_effect = fake_run

    def fake_translate(p):
        return str(tmp_path / p.split("/")[-1])
    monkeypatch.setattr(video_encoder, "translate_path", fake_translate)

    result = video_encoder.encode_video(
        cover_path_nas="/volume1/cover.jpg",
        audio_path_nas="/volume1/audio.mp3",
        output_path_nas="/volume1/out.mp4",
        resolution="1920x1080",
        duration_sec=120,
        style="essential",
        background_mode="video_loop",
        background_path_nas="/volume1/loop.mp4",
    )
    assert result["ok"] is True
    cmd_str = " ".join(captured_cmd)
    assert "stream_loop" in cmd_str
    assert "h264_nvenc" in cmd_str


@patch("video_encoder.subprocess.run")
def test_encode_essential_with_track_subtitles(mock_run, env, tmp_path, monkeypatch):
    cover = tmp_path / "cover.jpg"; cover.write_bytes(b"\x00")
    audio = tmp_path / "audio.mp3"; audio.write_bytes(b"\x00")
    out = tmp_path / "out.mp4"

    captured_cmd = []
    def fake_run(cmd, **kwargs):
        captured_cmd.extend(cmd)
        out.write_bytes(b"\x00" * (2 * 1024 * 1024))
        return MagicMock(returncode=0, stderr="")
    mock_run.side_effect = fake_run

    def fake_translate(p):
        return str(tmp_path / p.split("/")[-1])
    monkeypatch.setattr(video_encoder, "translate_path", fake_translate)

    result = video_encoder.encode_video(
        cover_path_nas="/volume1/cover.jpg",
        audio_path_nas="/volume1/audio.mp3",
        output_path_nas="/volume1/out.mp4",
        resolution="1920x1080",
        duration_sec=600,
        style="essential",
        background_mode="static",
        tracks=[
            {"start_offset_sec": 0, "title": "T1"},
            {"start_offset_sec": 180, "title": "T2"},
            {"start_offset_sec": 380, "title": "T3"},
        ],
    )
    assert result["ok"] is True
    cmd_str = " ".join(captured_cmd)
    assert "drawtext" in cmd_str
    assert "T1" in cmd_str and "T2" in cmd_str and "T3" in cmd_str


@patch("video_encoder.subprocess.run")
def test_encode_single_branch_unchanged(mock_run, env, tmp_path, monkeypatch):
    """style=single은 기존 visualizer cmd 유지 (showwaves)."""
    cover = tmp_path / "cover.jpg"; cover.write_bytes(b"\x00")
    audio = tmp_path / "audio.mp3"; audio.write_bytes(b"\x00")
    out = tmp_path / "out.mp4"
    captured_cmd = []
    def fake_run(cmd, **kwargs):
        captured_cmd.extend(cmd)
        out.write_bytes(b"\x00" * (2 * 1024 * 1024))
        return MagicMock(returncode=0, stderr="")
    mock_run.side_effect = fake_run

    def fake_translate(p):
        return str(tmp_path / p.split("/")[-1])
    monkeypatch.setattr(video_encoder, "translate_path", fake_translate)

    video_encoder.encode_video(
        cover_path_nas="/volume1/cover.jpg",
        audio_path_nas="/volume1/audio.mp3",
        output_path_nas="/volume1/out.mp4",
        resolution="1920x1080",
        duration_sec=120,
        style="single",
    )
    cmd_str = " ".join(captured_cmd)
    assert "showwaves" in cmd_str
    assert "showfreqs" not in cmd_str

Step 2: Run, verify fail

cd music_ai && venv/Scripts/python -m pytest tests/test_video_encoder.py -v

Expected: FAIL — style/background_mode/tracks 파라미터 미지원.

Step 3: Update encode_video + add filter builders

video_encoder.py 변경:

def encode_video(*, cover_path_nas: str, audio_path_nas: str,
                  output_path_nas: str, resolution: str,
                  duration_sec: int = 0, style: str = "essential",
                  background_mode: str = "static",
                  background_path_nas: str | None = None,
                  tracks: list | None = None) -> dict:
    # 1) Resolution 검증 + 경로 변환 + 입력 존재 확인 (기존 로직 유지)
    if not RESOLUTION_RE.match(resolution):
        raise EncodeError("input_validation", f"invalid resolution: {resolution}")
    w, h = resolution.split("x")
    try:
        cover_win = translate_path(cover_path_nas)
        audio_win = translate_path(audio_path_nas)
        out_win = translate_path(output_path_nas)
        bg_win = translate_path(background_path_nas) if background_path_nas else None
    except ValueError as e:
        raise EncodeError("path_translate", str(e))

    if not os.path.isfile(cover_win):
        raise EncodeError("input_validation", f"cover not found: {cover_win}")
    if not os.path.isfile(audio_win):
        raise EncodeError("input_validation", f"audio not found: {audio_win}")
    if bg_win and not os.path.isfile(bg_win):
        raise EncodeError("input_validation", f"background video not found: {bg_win}")
    os.makedirs(os.path.dirname(out_win), exist_ok=True)

    # 2) Filter builder 분기
    if style == "single":
        cmd = _build_single_cmd(cover_win, audio_win, out_win, w, h)
    elif style == "essential":
        if background_mode == "video_loop" and bg_win:
            cmd = _build_essential_video_loop_cmd(bg_win, audio_win, out_win, w, h, tracks)
        else:
            cmd = _build_essential_static_cmd(cover_win, audio_win, out_win, w, h, tracks)
    else:
        raise EncodeError("input_validation", f"unknown style: {style}")

    logger.info("ffmpeg: %s", " ".join(cmd))

    # 3) 실행 + 검증 (기존 로직 유지)
    import time
    t0 = time.time()
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=FFMPEG_TIMEOUT_S)
    except subprocess.TimeoutExpired:
        raise EncodeError("ffmpeg", f"timeout after {FFMPEG_TIMEOUT_S}s")
    duration_ms = int((time.time() - t0) * 1000)
    if result.returncode != 0:
        raise EncodeError("ffmpeg", f"returncode={result.returncode}: {result.stderr[-800:]}")
    if not os.path.isfile(out_win):
        raise EncodeError("output_check", "output file not created")
    output_bytes = os.path.getsize(out_win)
    if output_bytes < MIN_OUTPUT_BYTES:
        raise EncodeError("output_check", f"output too small: {output_bytes} bytes")
    return {
        "ok": True, "duration_ms": duration_ms,
        "output_path_nas": output_path_nas, "output_bytes": output_bytes,
        "encoder": "h264_nvenc", "preset": "p4",
    }

신규 builder 함수들:

def _build_single_cmd(cover_win, audio_win, out_win, w, h):
    """기존 단일 트랙 — showwaves 가장자리 파형."""
    return [
        FFMPEG_PATH, "-y",
        "-hwaccel", "cuda",
        "-loop", "1", "-i", cover_win,
        "-i", audio_win,
        "-filter_complex",
        f"[0:v]scale={w}:{h},format=yuv420p[bg];"
        f"[1:a]showwaves=s={w}x200:mode=cline:colors=0xFF4444@0.8[wave];"
        f"[bg][wave]overlay=0:({h}-200)[out]",
        "-map", "[out]", "-map", "1:a",
        "-c:v", "h264_nvenc", "-preset", "p4",
        "-rc", "vbr", "-cq", "23", "-b:v", "0",
        "-pix_fmt", "yuv420p",
        "-c:a", "aac", "-b:a", "192k",
        "-shortest", out_win,
    ]


def _build_essential_static_cmd(cover_win, audio_win, out_win, w, h, tracks):
    """essential 정적 — 풀스크린 cover + 중앙 showfreqs + ring + drawtext 자막."""
    ring_path = _ensure_ring_asset()
    base_filter = (
        f"[0:v]scale={w}:{h},format=yuv420p[bg];"
        f"[1:a]showfreqs=s=400x200:mode=bar:cmode=combined:colors=0xFFFFFF@0.9[bars];"
        f"[2:v]format=rgba[ring];"
        f"[bg][bars]overlay=({w}-400)/2:({h}-200)/2[mid];"
        f"[mid][ring]overlay=0:0[viz]"
    )
    drawtext = _build_drawtext_filter(tracks, w, h)
    if drawtext:
        full_filter = base_filter + ";" + f"[viz]{drawtext}[out]"
    else:
        full_filter = base_filter.replace("[viz]", "[out]", 1)

    return [
        FFMPEG_PATH, "-y",
        "-hwaccel", "cuda",
        "-loop", "1", "-i", cover_win,
        "-i", audio_win,
        "-loop", "1", "-i", ring_path,
        "-filter_complex", full_filter,
        "-map", "[out]", "-map", "1:a",
        "-c:v", "h264_nvenc", "-preset", "p4",
        "-rc", "vbr", "-cq", "23", "-b:v", "0",
        "-pix_fmt", "yuv420p",
        "-c:a", "aac", "-b:a", "192k",
        "-shortest", out_win,
    ]


def _build_essential_video_loop_cmd(bg_win, audio_win, out_win, w, h, tracks):
    """essential 영상 루프 — bg 영상 무한 반복 + 중앙 showfreqs + ring + drawtext."""
    ring_path = _ensure_ring_asset()
    base_filter = (
        f"[0:v]scale={w}:{h},setpts=PTS-STARTPTS,format=yuv420p[bg];"
        f"[1:a]showfreqs=s=400x200:mode=bar:cmode=combined:colors=0xFFFFFF@0.9[bars];"
        f"[2:v]format=rgba[ring];"
        f"[bg][bars]overlay=({w}-400)/2:({h}-200)/2[mid];"
        f"[mid][ring]overlay=0:0[viz]"
    )
    drawtext = _build_drawtext_filter(tracks, w, h)
    if drawtext:
        full_filter = base_filter + ";" + f"[viz]{drawtext}[out]"
    else:
        full_filter = base_filter.replace("[viz]", "[out]", 1)

    return [
        FFMPEG_PATH, "-y",
        "-hwaccel", "cuda",
        "-stream_loop", "-1", "-i", bg_win,
        "-i", audio_win,
        "-loop", "1", "-i", ring_path,
        "-filter_complex", full_filter,
        "-map", "[out]", "-map", "1:a",
        "-c:v", "h264_nvenc", "-preset", "p4",
        "-rc", "vbr", "-cq", "23", "-b:v", "0",
        "-pix_fmt", "yuv420p",
        "-c:a", "aac", "-b:a", "192k",
        "-shortest", out_win,
    ]


def _build_drawtext_filter(tracks, w, h) -> str:
    """곡명 자막 — 트랙 시작 시점에 5초 표시 (alpha fade in/out)."""
    if not tracks or len(tracks) <= 1:
        return ""
    font_path = os.getenv("SUBTITLE_FONT", "C:/Windows/Fonts/malgun.ttf")
    font_path_esc = font_path.replace(":", r"\:").replace("\\", "/")
    parts = []
    for tr in tracks:
        start = int(tr.get("start_offset_sec", 0))
        end = start + 5
        title = tr.get("title", "")
        # ffmpeg drawtext escape: : ' \ %
        title_esc = title.replace("\\", r"\\").replace("'", r"\'").replace(":", r"\:").replace("%", r"\%")
        # alpha: 0~1초 fade in, 4~5초 fade out
        alpha_expr = (
            f"if(between(t,{start},{end}),"
            f"  if(lt(t-{start},1), t-{start},"
            f"     if(gt(t-{start},4), {end}-t, 1)), 0)"
        ).replace(" ", "").replace("\n", "")
        parts.append(
            f"drawtext=fontfile='{font_path_esc}':text='{title_esc}'"
            f":fontcolor=white:fontsize=36:x=(w-text_w)/2:y=h-100"
            f":alpha='{alpha_expr}':box=1:boxcolor=black@0.5:boxborderw=10"
        )
    return ",".join(parts)


def _ensure_ring_asset() -> str:
    """assets/visualizer_ring.png 없으면 PIL로 자동 생성."""
    asset_dir = os.path.join(os.path.dirname(__file__), "assets")
    os.makedirs(asset_dir, exist_ok=True)
    ring_path = os.path.join(asset_dir, "visualizer_ring.png")
    if os.path.isfile(ring_path):
        return ring_path
    from PIL import Image, ImageDraw
    W, H = 1920, 1080
    img = Image.new("RGBA", (W, H), (0, 0, 0, 0))
    draw = ImageDraw.Draw(img)
    cx, cy = W // 2, H // 2
    # 외곽 원 (반투명 흰색)
    for r in (250, 240):
        draw.ellipse((cx - r, cy - r, cx + r, cy + r),
                      outline=(255, 255, 255, 90), width=2)
    # 점선 데코 (12개 점)
    import math
    for i in range(24):
        angle = i * (360 / 24) * math.pi / 180
        x = cx + 280 * math.cos(angle)
        y = cy + 280 * math.sin(angle)
        draw.ellipse((x - 3, y - 3, x + 3, y + 3), fill=(255, 255, 255, 120))
    img.save(ring_path, "PNG")
    return ring_path

Step 4: Run tests

cd music_ai && venv/Scripts/python -m pytest tests/test_video_encoder.py -v

Expected: 12 PASS (기존 8 + 신규 4).

Step 5: No commit (music_ai is local-only)

Task 9: Windows server.py — schema 확장

Files:

Modify: C:/Users/jaeoh/Desktop/workspace/music_ai/server.py
Step 1: Update EncodeVideoRequest

class TrackInfo(BaseModel):
    start_offset_sec: int
    title: str


class EncodeVideoRequest(BaseModel):
    cover_path_nas: str
    audio_path_nas: str
    output_path_nas: str
    resolution: str = "1920x1080"
    duration_sec: int = 0
    style: str = "essential"
    background_mode: str = "static"
    background_path_nas: str | None = None
    tracks: list[TrackInfo] = []


@app.post("/encode_video")
def encode_video_endpoint(req: EncodeVideoRequest):
    try:
        result = video_encoder.encode_video(
            cover_path_nas=req.cover_path_nas,
            audio_path_nas=req.audio_path_nas,
            output_path_nas=req.output_path_nas,
            resolution=req.resolution,
            duration_sec=req.duration_sec,
            style=req.style,
            background_mode=req.background_mode,
            background_path_nas=req.background_path_nas,
            tracks=[t.dict() for t in req.tracks] if req.tracks else None,
        )
        return result
    except video_encoder.EncodeError as e:
        status_code = 400 if e.stage in ("input_validation", "path_translate") else 500
        raise HTTPException(
            status_code=status_code,
            detail={"ok": False, "stage": e.stage, "error": e.message},
        )

Step 2: Verify imports

cd music_ai && venv/Scripts/python -c "from server import app; print(len([r for r in app.routes if hasattr(r,'path')]))"

Expected: 임포트 성공, 라우트 그대로.

Step 3: Run all tests

cd music_ai && venv/Scripts/python -m pytest tests/ -v

Expected: 모두 PASS.

Step 4: No commit (local-only)

Task 10: NAS 통합 테스트 — compile_job 기반 happy path

Files:

Modify: music-lab/tests/test_pipeline_flow.py
Step 1: Append new test

@patch("app.pipeline.youtube.upload_video", return_value={"video_id": "MIX_VID"})
@patch("app.pipeline.review.run_4_axis", new=AsyncMock(return_value={
    "metadata_quality": {"score": 90, "notes": ""},
    "policy_compliance": {"score": 95, "issues": []},
    "viewer_experience": {"score": 85, "notes": ""},
    "trend_alignment": {"score": 70, "matched_keywords": []},
    "weighted_total": 87.0, "verdict": "pass", "summary": "ok",
    "used_fallback": False,
}))
@patch("app.pipeline.metadata.generate", new=AsyncMock(return_value={
    "title": "Mix", "description": "Track desc",
    "tags": ["lofi"], "category_id": 10,
    "used_fallback": False, "error": None,
}))
@patch("app.pipeline.thumb.generate", return_value={
    "url": "/media/videos/X/thumbnail.jpg", "used_fallback": False,
})
@patch("app.pipeline.video.generate", return_value={
    "url": "/media/videos/X/video.mp4", "used_fallback": False, "duration_sec": 600,
})
@patch("app.pipeline.cover.generate", new=AsyncMock(return_value={
    "url": "/media/videos/X/cover.jpg", "used_fallback": False, "error": None,
}))
def test_full_pipeline_compile_job_happy_path(*_, client):
    # compile_job 1개 추가 (succeeded)
    import sqlite3
    conn = sqlite3.connect(db.DB_PATH)
    cur = conn.cursor()
    try:
        cur.execute("""
            INSERT INTO compile_jobs (title, track_ids_json, crossfade_sec, audio_path,
                                       status, created_at)
            VALUES ('Test Mix', '[1]', 3, '/app/data/compiles/1.mp3', 'succeeded', datetime())
        """)
    except sqlite3.OperationalError:
        pytest.skip("compile_jobs schema mismatch")
    conn.commit()
    cid = cur.lastrowid
    conn.close()

    pid = client.post("/api/music/pipeline", json={"compile_job_id": cid}).json()["id"]
    assert db.get_pipeline(pid)["state"] == "created"
    assert db.get_pipeline(pid)["compile_job_id"] == cid
    assert db.get_pipeline(pid)["track_id"] is None

    client.post(f"/api/music/pipeline/{pid}/start")
    p = db.get_pipeline(pid)
    assert p["state"] == "cover_pending"

    for step in ["cover", "video", "thumb", "meta"]:
        r = client.post(f"/api/music/pipeline/{pid}/feedback",
                         json={"step": step, "intent": "approve"})
        assert r.status_code == 202

    p = db.get_pipeline(pid)
    assert p["state"] == "publish_pending"

    client.post(f"/api/music/pipeline/{pid}/publish")
    p = db.get_pipeline(pid)
    assert p["state"] == "published"
    assert p["youtube_video_id"] == "MIX_VID"

Step 2: Run

python -m pytest tests/test_pipeline_flow.py -v Expected: 3 PASS (기존 2 + 신규 1).

Step 3: Run full suite

python -m pytest tests/ -v Expected: 전부 PASS.

Step 4: Commit

git add music-lab/tests/test_pipeline_flow.py
git commit -m "test(music-lab): compile_job 기반 happy path 통합 테스트"

Task 11: NAS 푸시 → webhook 자동 배포 + Windows 수동 배포

Files: (배포 단계, 코드 변경 없음)

Step 1: 백엔드 푸시

git -C C:/Users/jaeoh/Desktop/workspace/web-backend push origin main

(여러 commit 누적된 상태에서 한 번에 push. Gitea 인증 첫 시도 실패 시 재시도.)

Step 2: NAS 자동 배포 확인

ssh user@gahusb.synology.me
cd /volume1/docker/webpage
docker compose logs music-lab --tail=20

Application startup complete + 새 컬럼 마이그레이션 메시지 확인.

Step 3: Windows music_ai 수동 재시작

Windows PowerShell:

cd C:\Users\jaeoh\Desktop\workspace\music_ai
# 기존 실행 중이면 Ctrl+C 종료
.\start.bat

Step 4: 헬스 체크

Invoke-RestMethod http://localhost:8765/health

ffmpeg_nvenc: true 확인.

Step 5: Windows 신규 어셋 자동 생성 확인

ls C:\Users\jaeoh\Desktop\workspace\music_ai\assets\

visualizer_ring.png 자동 생성됐는지 확인 (없으면 첫 encode 호출 시 생성).

Task 12: 프론트엔드 — api.js 헬퍼 + CompileTab 버튼

Files:

Modify: web-ui/src/api.js
Modify: web-ui/src/pages/music/components/CompileTab.jsx
Step 1: api.js — createPipeline 시그니처 변경 + 영상 트리거

기존 createPipeline(track_id) → createPipeline(payload). 호환을 위해 track_id만 받는 옛 호출도 동작하게:

// 기존 createPipeline 교체:
export const createPipeline = (payload) => {
    // 옛 호출 호환: createPipeline(13) → { track_id: 13 }
    if (typeof payload === 'number') payload = { track_id: payload };
    return apiPost('/api/music/pipeline', payload);
};

Step 2: CompileTab.jsx — "🎬 영상 만들기" 버튼

CompileTab.jsx의 완료된 job 카드 액션 영역에 추가:

import { createPipeline, startPipeline } from '../../../api';

// 컴포넌트 내부 핸들러:
const handleVideoFromCompile = async (jobId) => {
    if (!confirm('이 mix로 영상 파이프라인을 시작할까요?')) return;
    try {
        const p = await createPipeline({ compile_job_id: jobId });
        await startPipeline(p.id);
        // YoutubeTab > 진행 서브탭으로 이동 (props onSwitchToPipeline 활용)
        if (props.onSwitchToPipeline) {
            props.onSwitchToPipeline(p.id);
        }
    } catch (e) {
        alert(`파이프라인 시작 실패: ${e.message || e}`);
    }
};

// 카드 액션 영역 (job.status === 'succeeded' 분기):
{job.status === 'succeeded' && (
    <button onClick={() => handleVideoFromCompile(job.id)} className="cmp-btn-video">
        🎬 영상 만들기
    </button>
)}

props.onSwitchToPipeline prop 받아옴 — CompileTab({ library, onSwitchToPipeline }).

Step 3: MusicStudio.jsx에서 onSwitchToPipeline prop 연결

기존 <YoutubeTab>에 prop 전달:

const handleVideoFromCompile = (pipelineId) => {
    setTab('youtube');
    setOpenPipelineFor(pipelineId);  // 진행 서브탭 자동 전환
};

// YoutubeTab에서 CompileTab으로 onSwitchToPipeline prop 전달

YoutubeTab.jsx 수정 — CompileTab에 prop 전달:

{subtab === 'compile' && <CompileTab library={library} onSwitchToPipeline={(pid) => {
    setSubtab('pipeline');
    // openPipelineFor 갱신해 PipelineTab이 해당 pipeline 표시
}} />}

Step 4: Build + manual verify

cd web-ui && npm run build 2>&1 | tail -5

빌드 성공 확인.

Step 5: Commit (DO NOT PUSH)

git -C C:/Users/jaeoh/Desktop/workspace/web-ui add src/api.js \
                                                   src/pages/music/components/CompileTab.jsx \
                                                   src/pages/music/components/YoutubeTab.jsx \
                                                   src/pages/music/MusicStudio.jsx
git -C C:/Users/jaeoh/Desktop/workspace/web-ui commit -m "feat(web-ui): CompileTab에 영상 만들기 버튼 + createPipeline payload 시그니처"

Task 13: PipelineStartModal — Mix 라디오 + 고급 옵션

Files:

Modify: web-ui/src/pages/music/components/PipelineStartModal.jsx
Step 1: 모달 개편

import { useState, useEffect } from 'react';
import { createPipeline, startPipeline, getCompileJobs } from '../../../api';

const fmtDur = (s) => {
    const m = Math.floor(s / 60);
    const sec = Math.round(s % 60);
    return `${m}:${String(sec).padStart(2, '0')}`;
};

export default function PipelineStartModal({ library, initialTrackId, onClose, onCreated }) {
    const [inputType, setInputType] = useState('track');  // 'track' | 'compile'
    const [tid, setTid] = useState(initialTrackId || library?.[0]?.id || '');
    const [cid, setCid] = useState('');
    const [compileJobs, setCompileJobs] = useState([]);
    const [advanced, setAdvanced] = useState(false);
    const [visualStyle, setVisualStyle] = useState('');  // '' = use default
    const [bgMode, setBgMode] = useState('');
    const [bgKeyword, setBgKeyword] = useState('');
    const [error, setError] = useState('');

    useEffect(() => {
        if (inputType === 'compile') {
            getCompileJobs().then(r => {
                const completed = (r.jobs || []).filter(j => j.status === 'succeeded');
                setCompileJobs(completed);
                if (completed.length && !cid) setCid(completed[0].id);
            }).catch(e => setError(String(e)));
        }
    }, [inputType]);

    const submit = async () => {
        try {
            const payload = {};
            if (inputType === 'track') payload.track_id = parseInt(tid);
            else payload.compile_job_id = parseInt(cid);
            if (visualStyle) payload.visual_style = visualStyle;
            if (bgMode) payload.background_mode = bgMode;
            if (bgKeyword) payload.background_keyword = bgKeyword;

            const p = await createPipeline(payload);
            await startPipeline(p.id);
            onCreated(p);
        } catch (e) { setError(e.message || String(e)); }
    };

    return (
        <div className="modal-overlay" onClick={onClose}>
            <div className="modal-body" onClick={e => e.stopPropagation()}>
                <h3>새 파이프라인 시작</h3>

                <fieldset className="psm-input-radio">
                    <legend>입력</legend>
                    <label>
                        <input type="radio" checked={inputType==='track'}
                            onChange={() => setInputType('track')}/> 단일 트랙
                    </label>
                    <label>
                        <input type="radio" checked={inputType==='compile'}
                            onChange={() => setInputType('compile')}/> Mix (컴파일 결과)
                    </label>
                </fieldset>

                {inputType === 'track' ? (
                    <select value={tid} onChange={e => setTid(e.target.value)}>
                        {(library || []).map(t => (
                            <option key={t.id} value={t.id}>{t.title} ({t.genre})</option>
                        ))}
                    </select>
                ) : (
                    <select value={cid} onChange={e => setCid(e.target.value)}>
                        {compileJobs.length === 0 && <option value="">완료된 Mix 없음</option>}
                        {compileJobs.map(j => (
                            <option key={j.id} value={j.id}>
                                {j.title || `Mix #${j.id}`} ({fmtDur(j.duration_sec || 0)}, {j.tracks_count || j.track_ids?.length || '?'}곡)
                            </option>
                        ))}
                    </select>
                )}

                <details className="psm-advanced" open={advanced}>
                    <summary onClick={(e) => { e.preventDefault(); setAdvanced(!advanced); }}>
                        고급 옵션
                    </summary>
                    <label>시각 스타일
                        <select value={visualStyle} onChange={e => setVisualStyle(e.target.value)}>
                            <option value="">기본 (구성 탭 default)</option>
                            <option value="essential">essential (배경 + 중앙 비주얼)</option>
                            <option value="single">single (커버 + 가장자리 파형)</option>
                        </select>
                    </label>
                    <label>배경 모드
                        <select value={bgMode} onChange={e => setBgMode(e.target.value)}>
                            <option value="">기본 (구성 탭 default)</option>
                            <option value="static">정적 사진</option>
                            <option value="video_loop">영상 루프 (Pexels)</option>
                        </select>
                    </label>
                    <label>배경 키워드 (Pexels 검색용)
                        <input value={bgKeyword} onChange={e => setBgKeyword(e.target.value)}
                            placeholder="rainy window, lofi cafe ..." />
                    </label>
                </details>

                {error && <div className="ms-error">{error}</div>}
                <div className="modal-actions">
                    <button onClick={onClose}>취소</button>
                    <button className="button primary" onClick={submit}
                        disabled={inputType === 'compile' && !cid}>
                        시작
                    </button>
                </div>
            </div>
        </div>
    );
}

Step 2: api.js — getCompileJobs 헬퍼 (기존에 있는지 확인)

api.js에 이미 있으면 skip. 없으면 추가:

export const getCompileJobs = () => apiGet('/api/music/compile/jobs');
// (실제 endpoint 경로는 기존 CompileTab이 호출하는 것 그대로)

Step 3: Build + verify

npm run build 깔끔히 통과.

Step 4: Commit

git add src/pages/music/components/PipelineStartModal.jsx src/api.js
git commit -m "feat(web-ui): PipelineStartModal Mix 입력 라디오 + 고급 옵션"

Task 14: PipelineCard — mini 미리보기 + 클릭 → 상세 모달

Files:

Modify: web-ui/src/pages/music/components/PipelineCard.jsx
Create: web-ui/src/pages/music/components/PipelineDetailModal.jsx
Step 1: PipelineDetailModal.jsx (신규)

const fmtTimestamp = (sec) => {
    if (sec == null) return '';
    const m = Math.floor(sec / 60);
    const s = Math.round(sec % 60);
    const h = Math.floor(m / 60);
    if (h) return `${h}:${String(m % 60).padStart(2,'0')}:${String(s).padStart(2,'0')}`;
    return `${m}:${String(s).padStart(2,'0')}`;
};

const fmtDuration = fmtTimestamp;

export default function PipelineDetailModal({ pipeline, onClose }) {
    if (!pipeline) return null;
    const meta = pipeline.metadata || {};
    const review = pipeline.review || {};

    return (
        <div className="modal-overlay" onClick={onClose}>
            <div className="modal-body modal-body--lg" onClick={e => e.stopPropagation()}>
                <header className="pdm-header">
                    <h3>{pipeline.compile_title || pipeline.track_title || `Pipeline #${pipeline.id}`}</h3>
                    <span className="pdm-badge">{pipeline.visual_style || 'essential'}</span>
                    <button onClick={onClose} className="pdm-close">×</button>
                </header>

                <div className="pdm-grid">
                    {pipeline.cover_url && (
                        <figure className="pdm-figure">
                            <img src={pipeline.cover_url} alt="cover" />
                            <figcaption>커버 (배경)</figcaption>
                        </figure>
                    )}
                    {pipeline.thumbnail_url && (
                        <figure className="pdm-figure">
                            <img src={pipeline.thumbnail_url} alt="thumbnail" />
                            <figcaption>썸네일</figcaption>
                        </figure>
                    )}
                </div>

                {pipeline.video_url && (
                    <div className="pdm-video">
                        <video src={pipeline.video_url} controls preload="metadata" width="100%" />
                    </div>
                )}

                {meta.title && (
                    <section className="pdm-section">
                        <h4>메타데이터</h4>
                        <p><strong>제목:</strong> {meta.title}</p>
                        <details>
                            <summary>설명 ({(meta.description || '').length}자)</summary>
                            <pre className="pdm-pre">{meta.description}</pre>
                        </details>
                        <p><strong>태그:</strong> {(meta.tags || []).join(', ')}</p>
                    </section>
                )}

                {review.weighted_total != null && (
                    <section className="pdm-section">
                        <h4>AI 검토
                            <span className={`pdm-verdict pdm-verdict--${review.verdict}`}>{review.verdict}</span>
                            <span className="pdm-score">({review.weighted_total}/100)</span>
                        </h4>
                        <table className="pdm-review-table">
                            <tbody>
                                <tr><td>메타데이터 품질</td><td>{review.metadata_quality?.score}</td></tr>
                                <tr><td>콘텐츠 정책</td><td>{review.policy_compliance?.score}</td></tr>
                                <tr><td>시청 경험</td><td>{review.viewer_experience?.score}</td></tr>
                                <tr><td>트렌드 정렬</td><td>{review.trend_alignment?.score}</td></tr>
                            </tbody>
                        </table>
                        {review.summary && <p className="pdm-summary"><em>{review.summary}</em></p>}
                    </section>
                )}

                {pipeline.tracks && pipeline.tracks.length > 1 && (
                    <section className="pdm-section">
                        <h4>트랙 리스트 ({pipeline.tracks.length})</h4>
                        <ol className="pdm-tracks">
                            {pipeline.tracks.map(t => (
                                <li key={t.id}>
                                    <span className="pdm-track-time">[{fmtTimestamp(t.start_offset_sec)}]</span>
                                    {' '}{t.title}
                                    <span className="pdm-track-dur"> ({fmtDuration(t.duration_sec)})</span>
                                </li>
                            ))}
                        </ol>
                    </section>
                )}

                {pipeline.feedback && pipeline.feedback.length > 0 && (
                    <section className="pdm-section">
                        <h4>피드백 히스토리 ({pipeline.feedback.length})</h4>
                        <ul className="pdm-feedback">
                            {pipeline.feedback.map(f => (
                                <li key={f.id}>
                                    <code>[{f.step}]</code> {f.feedback_text}
                                    <small> {f.received_at?.replace('T', ' ')}</small>
                                </li>
                            ))}
                        </ul>
                    </section>
                )}

                {pipeline.youtube_video_id && (
                    <a href={`https://youtu.be/${pipeline.youtube_video_id}`}
                       target="_blank" rel="noreferrer" className="pdm-youtube">
                        🎬 YouTube에서 보기
                    </a>
                )}
            </div>
        </div>
    );
}

Step 2: PipelineCard.jsx — 미리보기 + 클릭 핸들러

import { useState } from 'react';
import { cancelPipeline, publishPipeline } from '../../../api';
import PipelineDetailModal from './PipelineDetailModal';

const STEP_LABELS = ['커버','영상','썸네','메타','검토','발행'];

function stepIndex(state) {
    if (!state) return -1;
    if (state.startsWith('cover'))   return 0;
    if (state.startsWith('video'))   return 1;
    if (state.startsWith('thumb'))   return 2;
    if (state.startsWith('meta'))    return 3;
    if (state.startsWith('ai_review') || state === 'publish_pending') return 4;
    if (state.startsWith('publish')) return 5;
    if (state === 'published')       return 6;
    return -1;
}

export default function PipelineCard({ pipeline, onChanged }) {
    const [showDetail, setShowDetail] = useState(false);
    const i = stepIndex(pipeline.state);
    const title = pipeline.compile_title || pipeline.track_title || `Pipeline #${pipeline.id}`;

    const handleCardClick = (e) => {
        // 버튼/링크 클릭은 무시
        if (e.target.closest('button') || e.target.closest('a')) return;
        setShowDetail(true);
    };

    return (
        <>
            <div className="pipeline-card" onClick={handleCardClick}>
                <div className="pipeline-card__head">
                    <h4>{title}</h4>
                    {pipeline.visual_style && (
                        <span className="pipeline-style-badge">{pipeline.visual_style}</span>
                    )}
                    {!['published','cancelled','failed'].includes(pipeline.state) && (
                        <button onClick={async () => { await cancelPipeline(pipeline.id); onChanged(); }}>
                            취소
                        </button>
                    )}
                </div>

                {/* 미니 미리보기 */}
                <div className="pipeline-previews">
                    {pipeline.cover_url && (
                        <img src={pipeline.cover_url} alt="" className="pipeline-preview-mini" />
                    )}
                    {pipeline.thumbnail_url && (
                        <img src={pipeline.thumbnail_url} alt="" className="pipeline-preview-mini" />
                    )}
                    {pipeline.video_url && <span className="pipeline-video-icon">▶</span>}
                </div>

                <div className="pipeline-progress">
                    {STEP_LABELS.map((lbl, idx) => (
                        <div key={lbl}
                            className={`pipeline-dot ${idx <= i ? 'is-done' : ''} ${idx === i ? 'is-current' : ''}`}>
                            <span>{lbl}</span>
                        </div>
                    ))}
                </div>

                <div className="pipeline-state">현재: {pipeline.state}</div>

                {pipeline.review && (
                    <div className="pipeline-review">
                        AI 검토: <strong>{pipeline.review.verdict}</strong>
                        ({pipeline.review.weighted_total}/100)
                    </div>
                )}

                {pipeline.state === 'publish_pending' && (
                    <button className="button primary"
                        onClick={async () => { await publishPipeline(pipeline.id); onChanged(); }}>
                        YouTube 업로드
                    </button>
                )}

                {pipeline.youtube_video_id && (
                    <a href={`https://youtu.be/${pipeline.youtube_video_id}`}
                       target="_blank" rel="noreferrer">
                        유튜브에서 보기
                    </a>
                )}
            </div>

            {showDetail && <PipelineDetailModal pipeline={pipeline} onClose={() => setShowDetail(false)} />}
        </>
    );
}

Step 3: CSS 추가 (MusicStudio.css append)

/* === Pipeline Detail Modal === */
.modal-body--lg { max-width: 720px; max-height: 90vh; overflow-y: auto; }
.pdm-header { display:flex; align-items:center; gap:12px; margin-bottom:16px; }
.pdm-header h3 { flex:1; margin:0; }
.pdm-badge { padding:2px 8px; background:rgba(56,189,248,.2); color:#bae6fd; border-radius:6px; font-size:11px; }
.pdm-close { background:none; border:none; font-size:24px; cursor:pointer; color:var(--ms-muted); }

.pdm-grid { display:grid; grid-template-columns:1fr 1fr; gap:12px; margin-bottom:16px; }
.pdm-figure { margin:0; }
.pdm-figure img { width:100%; border-radius:8px; }
.pdm-figure figcaption { font-size:11px; color:var(--ms-muted); text-align:center; margin-top:4px; }

.pdm-video { margin-bottom:16px; }
.pdm-video video { border-radius:8px; }

.pdm-section { margin-bottom:16px; padding-bottom:12px; border-bottom:1px solid var(--ms-line, #2a2a3a); }
.pdm-section h4 { margin:0 0 8px; font-size:14px; }
.pdm-pre { background:rgba(0,0,0,.3); padding:8px; border-radius:6px; font-size:12px; white-space:pre-wrap; overflow-x:auto; }

.pdm-verdict { padding:2px 8px; margin-left:8px; border-radius:6px; font-size:12px; font-weight:bold; }
.pdm-verdict--pass { background:rgba(34,197,94,.2); color:#86efac; }
.pdm-verdict--fail { background:rgba(248,113,113,.2); color:#fca5a5; }
.pdm-score { color:var(--ms-muted); font-size:12px; margin-left:8px; }
.pdm-review-table { width:100%; border-collapse:collapse; font-size:13px; }
.pdm-review-table td { padding:4px 8px; border-bottom:1px solid var(--ms-line, #2a2a3a); }
.pdm-review-table td:nth-child(2) { text-align:right; font-weight:bold; }
.pdm-summary { font-size:12px; color:var(--ms-muted); margin-top:8px; }

.pdm-tracks { padding-left:24px; }
.pdm-tracks li { margin-bottom:4px; font-size:13px; }
.pdm-track-time { color:var(--ms-accent, #38bdf8); font-family: var(--ms-ff-mono, monospace); }
.pdm-track-dur { color:var(--ms-muted); font-size:11px; }

.pdm-feedback { padding-left:0; list-style:none; }
.pdm-feedback li { padding:6px 8px; background:rgba(0,0,0,.2); border-radius:6px; margin-bottom:4px; font-size:12px; }
.pdm-feedback code { color:#fb923c; font-size:11px; }
.pdm-feedback small { display:block; color:var(--ms-muted); margin-top:2px; }

.pdm-youtube { display:inline-block; padding:8px 16px; background:#ff0000; color:white;
    border-radius:8px; text-decoration:none; font-weight:bold; }

/* PipelineCard mini previews */
.pipeline-previews { display:flex; gap:8px; margin:8px 0; align-items:center; }
.pipeline-preview-mini { width:64px; height:64px; object-fit:cover; border-radius:6px;
    border:1px solid var(--ms-line, #2a2a3a); }
.pipeline-video-icon { font-size:24px; color:var(--ms-accent, #38bdf8); margin-left:4px; }
.pipeline-style-badge { padding:1px 6px; background:rgba(56,189,248,.15); color:#bae6fd;
    border-radius:4px; font-size:10px; }

Step 4: Build verify

npm run build 통과.

Step 5: Commit

git add src/pages/music/components/PipelineCard.jsx \
        src/pages/music/components/PipelineDetailModal.jsx \
        src/pages/music/MusicStudio.css
git commit -m "feat(web-ui): PipelineDetailModal + 카드 mini 미리보기"

Task 15: SetupTab — visual_defaults 5개 옵션 확장

Files:

Modify: web-ui/src/pages/music/components/SetupTab.jsx
Step 1: 영상 비주얼 카드 확장

기존 setup-card 중 "영상 비주얼 기본값"을 다음으로 교체:

<section className="setup-card">
    <h3>영상 비주얼 기본값</h3>

    <label>해상도
        <select value={setup.visual_defaults.resolution || '1920x1080'}
            onChange={e => setSetup(s => ({...s, visual_defaults: {...s.visual_defaults, resolution: e.target.value}}))}>
            <option value="1920x1080">1920×1080 (가로)</option>
            <option value="1080x1920">1080×1920 (세로/Shorts)</option>
        </select>
    </label>

    <label>기본 시각 스타일
        <select value={setup.visual_defaults.default_visual_style || 'essential'}
            onChange={e => setSetup(s => ({...s, visual_defaults: {...s.visual_defaults, default_visual_style: e.target.value}}))}>
            <option value="essential">essential (배경 + 중앙 비주얼)</option>
            <option value="single">single (커버 + 가장자리 파형)</option>
        </select>
    </label>

    <label>기본 배경 모드
        <select value={setup.visual_defaults.default_background_mode || 'static'}
            onChange={e => setSetup(s => ({...s, visual_defaults: {...s.visual_defaults, default_background_mode: e.target.value}}))}>
            <option value="static">정적 사진</option>
            <option value="video_loop">영상 루프 (Pexels)</option>
        </select>
    </label>

    <label>기본 배경 키워드 (비우면 장르 기반 자동)
        <input value={setup.visual_defaults.default_background_keyword || ''}
            onChange={e => setSetup(s => ({...s, visual_defaults: {...s.visual_defaults, default_background_keyword: e.target.value}}))}
            placeholder="lofi cafe, rainy window, mountain ..." />
    </label>

    <label>배경 이미지 소스 (정적 모드)
        <select value={setup.visual_defaults.background_image_source || 'ai'}
            onChange={e => setSetup(s => ({...s, visual_defaults: {...s.visual_defaults, background_image_source: e.target.value}}))}>
            <option value="ai">AI 생성 (DALL·E)</option>
            <option value="pexels">Pexels 스톡 사진</option>
        </select>
    </label>

    <label>
        <input type="checkbox"
            checked={setup.visual_defaults.subtitle_track_titles ?? true}
            onChange={e => setSetup(s => ({...s, visual_defaults: {...s.visual_defaults, subtitle_track_titles: e.target.checked}}))}/>
        Mix에서 곡명 자막 표시 (트랙 시작 시 5초)
    </label>

    <button onClick={() => save({ visual_defaults: setup.visual_defaults })}>저장</button>
</section>

Step 2: Build verify + commit

npm run build
git add src/pages/music/components/SetupTab.jsx
git commit -m "feat(web-ui): SetupTab visual_defaults 6개 옵션 확장"

Task 16: 프론트 푸시 + 배포

Step 1: 모든 web-ui commit 한 번에 push

git -C C:/Users/jaeoh/Desktop/workspace/web-ui push origin main

Step 2: NAS에 빌드 + robocopy

로컬에서:

cd C:/Users/jaeoh/Desktop/workspace/web-ui
npm run release:nas

Step 3: 캐시 무력화 — 브라우저 강력 새로고침

Ctrl+Shift+R.

Task 17: 수동 E2E 검증 체크리스트

단일 트랙 essential 영상
- 진행 탭 → "+ 새 파이프라인" → 단일 트랙 라디오 → 트랙 선택 → 시작
- 카드에 cover 미니 썸네일 표시 ✓
- 카드 클릭 → 상세 모달 cover 큰 이미지 표시 ✓
- 텔레그램 "승인" → 영상 단계
- 영상 단계 후 카드에 video ▶ 아이콘 표시
- 상세 모달의 <video controls> 재생 가능
- essential 시각 스타일: 풀스크린 cover + 중앙 막대 + ring 데코 확인
- YouTube에 비공개 업로드 ✓
Mix essential 영상 (1시간+)
- 컴파일 탭 → 트랙 5–10개 + crossfade 3s → 컴파일 (1시간+)
- 완료된 컴파일 카드의 "🎬 영상 만들기" 클릭
- 진행 탭 자동 이동 + 새 파이프라인 카드
- cover 단계: 컴파일 mix는 genre="mix" → cover_prompts의 "mix" 또는 "default" 사용
- 영상 단계: ~3-5분 NVENC 인코딩
- 메타데이터 단계: description에 챕터 ([00:00] T1) 자동 포함 ✓
- 발행 후 YouTube에서 챕터 마커 자동 인식 ✓
- 상세 모달에 트랙 리스트 표시 ✓
video_loop 모드
- 구성 탭 default_background_mode = video_loop, keyword = rainy window cafe 저장
- 또는 파이프라인 시작 모달의 고급 옵션에서 override
- 새 파이프라인 시작 → cover 단계에서 Pexels 영상 다운로드 (loop.mp4)
- 영상 단계: 루프 영상 무한 반복 + 중앙 비주얼라이저
- YouTube 발행 후 시각 확인
AI 검토 verdict 표시
- 상세 모달에 4축 점수 표 + verdict 배지 (pass: 초록, fail: 빨강)
- summary 텍스트 표시
반려 + 피드백 흐름
- 어느 단계에서 텔레그램 "반려, 더 어둡게" 답장
- 카드의 진행도가 같은 단계에 머무르고 재생성
- 새 알림 도착 (feedback_count++ 덕분)
- 상세 모달의 피드백 히스토리에 새 항목 추가

Self-Review

Spec coverage:

§3 사용자 흐름 → Task 12 (Compile 버튼), 13 (모달), 14 (카드/모달), 17 (E2E)
§4 데이터 모델 → Task 1
§5 API 변경 → Task 7
§6 NAS 백엔드 → Task 2-7
§7 Windows 백엔드 → Task 8-9
§8 프론트엔드 → Task 12-15
§9 환경변수 → 변경 없음 (기존 PEXELS_API_KEY 재활용)
§10 에러 처리 → Task 7 (XOR 검증), Task 3-4 (Pexels 폴백)
§11 테스트 → Task 1-10 단위, Task 17 E2E
§12 마이그레이션 → Task 1 (idempotent ALTER), Task 8 (visualizer_ring 자동 생성)

Placeholder scan: 없음.

Type consistency:

_resolve_input 반환 dict 키: audio_path, duration_sec, tracks, title, genre, moods → Task 2 정의, Task 4-5 사용 일치
tracks 항목: id, title, start_offset_sec, duration_sec → Task 2/5/8 일치
EncodeError(stage, message) Windows측 일관 (이전 task에서 정의됨)
pipeline.compile_title / pipeline.track_title 백엔드 응답 필드 → Task 1 (db.get_pipeline JOIN), Task 14 (UI 표시) 일치 — note: db.py JOIN에서 compile_title도 추가해야 함

Spec gap 발견: db._parse_pipeline_row가 video_pipelines + music_library JOIN만 함. compile_title도 JOIN으로 같이 가져와야 PipelineCard/Modal에서 사용 가능. Task 1에 추가:

→ Task 1 Step 3에 추가: _parse_pipeline_row도 compile_title 컬럼 인식하게 수정. 마찬가지로 get_pipeline / list_pipelines 쿼리에 compile_jobs LEFT JOIN 추가.

Type consistency check 끝. 위 gap은 spec에 명시되어 있지만 실제 마이그레이션 task의 step 안에 명시 안 됐음. 보완:

Task 1 Step 3 마지막에 다음 추가:

# get_pipeline 쿼리 확장 — compile_title도 JOIN으로 가져오기
def get_pipeline(pid: int) -> dict | None:
    with _conn() as conn:
        row = conn.execute("""
            SELECT vp.*, ml.title AS track_title, cj.title AS compile_title
            FROM video_pipelines vp
            LEFT JOIN music_library ml ON ml.id = vp.track_id
            LEFT JOIN compile_jobs cj ON cj.id = vp.compile_job_id
            WHERE vp.id = ?
        """, (pid,)).fetchone()
    if not row:
        return None
    return _parse_pipeline_row(row)


def list_pipelines(active_only: bool = False) -> list[dict]:
    sql = """
        SELECT vp.*, ml.title AS track_title, cj.title AS compile_title
        FROM video_pipelines vp
        LEFT JOIN music_library ml ON ml.id = vp.track_id
        LEFT JOIN compile_jobs cj ON cj.id = vp.compile_job_id
    """
    if active_only:
        sql += " WHERE vp.state NOT IN ('published','cancelled','failed','awaiting_manual')"
    sql += " ORDER BY vp.created_at DESC"
    with _conn() as conn:
        rows = conn.execute(sql).fetchall()
    return [_parse_pipeline_row(r) for r in rows]

_parse_pipeline_row는 dict(row)이므로 추가 컬럼 자동 포함.

추가 테스트도 Task 1에 포함:

def test_pipeline_response_includes_compile_title(fresh_db):
    # compile_jobs 1행 + pipeline 1행 (compile_job_id=1)
    import sqlite3
    conn = sqlite3.connect(db.DB_PATH)
    cur = conn.cursor()
    cur.execute("""CREATE TABLE IF NOT EXISTS compile_jobs (
        id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT, status TEXT,
        track_ids_json TEXT, crossfade_sec INTEGER, audio_path TEXT, created_at TEXT)""")
    cur.execute("INSERT INTO compile_jobs (id, title, status) VALUES (1, 'My Mix', 'succeeded')")
    conn.commit()
    conn.close()
    pid = db.create_pipeline(compile_job_id=1)
    p = db.get_pipeline(pid)
    assert p["compile_title"] == "My Mix"

Plan complete. 17 task로 분해 — 각 task TDD + commit. Subagent-driven 실행 권장.

94 KiB Raw Blame History Unescape Escape

Essential Mix Pipeline Implementation Plan

File Structure

NAS music-lab (web-backend/music-lab/)

Windows music_ai (C:/Users/jaeoh/Desktop/workspace/music_ai/)

Frontend (web-ui/)

Task 1: DB 마이그레이션 — video_pipelines 4개 컬럼

Task 2: Orchestrator — _resolve_input 함수

Task 3: cover.py — Pexels 이미지 검색 분기

Task 4: background.py — video_loop 모드 (Pexels 영상)

Task 5: metadata.py — tracks 옵션 + 챕터 prompt

Task 6: NAS pipeline/video.py — Windows 추가 파라미터 전달

Task 7: main.py — POST /pipeline body 확장

Task 8: Windows video_encoder.py — essential filter 분기

Task 9: Windows server.py — schema 확장

Task 10: NAS 통합 테스트 — compile_job 기반 happy path

Task 11: NAS 푸시 → webhook 자동 배포 + Windows 수동 배포

Task 12: 프론트엔드 — api.js 헬퍼 + CompileTab 버튼

Task 13: PipelineStartModal — Mix 라디오 + 고급 옵션

Task 14: PipelineCard — mini 미리보기 + 클릭 → 상세 모달

Task 15: SetupTab — visual_defaults 5개 옵션 확장

Task 16: 프론트 푸시 + 배포

Task 17: 수동 E2E 검증 체크리스트

Self-Review

94 KiB

Raw Blame History

NAS music-lab (`web-backend/music-lab/`)

Windows music_ai (`C:/Users/jaeoh/Desktop/workspace/music_ai/`)

Frontend (`web-ui/`)

Task 1: DB 마이그레이션 — `video_pipelines` 4개 컬럼

Task 2: Orchestrator — `_resolve_input` 함수