diff --git a/docs/superpowers/plans/2026-05-09-essential-mix-pipeline.md b/docs/superpowers/plans/2026-05-09-essential-mix-pipeline.md new file mode 100644 index 0000000..35e69bb --- /dev/null +++ b/docs/superpowers/plans/2026-05-09-essential-mix-pipeline.md @@ -0,0 +1,2513 @@ +# 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)** + +```python +# 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** + +```bash +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 직후): + +```python +# 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 함수 앞): + +```python +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** + +```python +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** + +```bash +cd music-lab && python -m pytest tests/test_pipeline_db.py -v +``` +Expected: 모두 PASS (기존 + 신규 3개 = 12+개). + +- [ ] **Step 6: Commit (DO NOT PUSH)** + +```bash +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** + +```python +# 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** + +```bash +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 정리 + 함수 추가): + +```python +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** + +```bash +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에 전달: + +```python +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"]` 등 사용: + +```python +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** + +```bash +cd music-lab && python -m pytest tests/ -v +``` +Expected: 모두 PASS (기존 통합 테스트 영향 없는지 확인). + +- [ ] **Step 7: Commit** + +```bash +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)** + +```python +# 기존 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` 파라미터: + +```python +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()` 시그니처 + 분기: + +```python +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** + +```bash +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** + +```python +# 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`** + +```python +"""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` 시작부: + +```python +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** + +```bash +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)** + +```python +@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** + +```python +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** + +```bash +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)** + +```python +@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`: + +```python +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** + +```bash +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)** + +```python +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`: + +```python +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** + +```bash +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** + +```python +# 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** + +```bash +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` 변경: + +```python +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 함수들: + +```python +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** + +```bash +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`** + +```python +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** + +```bash +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** + +```bash +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** + +```python +@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** + +```bash +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: 백엔드 푸시** + +```bash +git -C C:/Users/jaeoh/Desktop/workspace/web-backend push origin main +``` +(여러 commit 누적된 상태에서 한 번에 push. Gitea 인증 첫 시도 실패 시 재시도.) + +- [ ] **Step 2: NAS 자동 배포 확인** + +```bash +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: +```powershell +cd C:\Users\jaeoh\Desktop\workspace\music_ai +# 기존 실행 중이면 Ctrl+C 종료 +.\start.bat +``` + +- [ ] **Step 4: 헬스 체크** + +```powershell +Invoke-RestMethod http://localhost:8765/health +``` +`ffmpeg_nvenc: true` 확인. + +- [ ] **Step 5: Windows 신규 어셋 자동 생성 확인** + +```powershell +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`만 받는 옛 호출도 동작하게: + +```javascript +// 기존 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 카드 액션 영역에 추가: + +```jsx +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' && ( + +)} +``` + +`props.onSwitchToPipeline` prop 받아옴 — `CompileTab({ library, onSwitchToPipeline })`. + +- [ ] **Step 3: MusicStudio.jsx에서 onSwitchToPipeline prop 연결** + +기존 ``에 prop 전달: +```jsx +const handleVideoFromCompile = (pipelineId) => { + setTab('youtube'); + setOpenPipelineFor(pipelineId); // 진행 서브탭 자동 전환 +}; + +// YoutubeTab에서 CompileTab으로 onSwitchToPipeline prop 전달 +``` + +`YoutubeTab.jsx` 수정 — CompileTab에 prop 전달: +```jsx +{subtab === 'compile' && { + setSubtab('pipeline'); + // openPipelineFor 갱신해 PipelineTab이 해당 pipeline 표시 +}} />} +``` + +- [ ] **Step 4: Build + manual verify** + +```bash +cd web-ui && npm run build 2>&1 | tail -5 +``` +빌드 성공 확인. + +- [ ] **Step 5: Commit (DO NOT PUSH)** + +```bash +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: 모달 개편** + +```jsx +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 ( +
+
e.stopPropagation()}> +

새 파이프라인 시작

+ +
+ 입력 + + +
+ + {inputType === 'track' ? ( + + ) : ( + + )} + +
+ { e.preventDefault(); setAdvanced(!advanced); }}> + 고급 옵션 + + + + +
+ + {error &&
{error}
} +
+ + +
+
+
+ ); +} +``` + +- [ ] **Step 2: api.js — `getCompileJobs` 헬퍼 (기존에 있는지 확인)** + +`api.js`에 이미 있으면 skip. 없으면 추가: +```javascript +export const getCompileJobs = () => apiGet('/api/music/compile/jobs'); +// (실제 endpoint 경로는 기존 CompileTab이 호출하는 것 그대로) +``` + +- [ ] **Step 3: Build + verify** + +`npm run build` 깔끔히 통과. + +- [ ] **Step 4: Commit** + +```bash +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 (신규)** + +```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 ( +
+
e.stopPropagation()}> +
+

{pipeline.compile_title || pipeline.track_title || `Pipeline #${pipeline.id}`}

+ {pipeline.visual_style || 'essential'} + +
+ +
+ {pipeline.cover_url && ( +
+ cover +
커버 (배경)
+
+ )} + {pipeline.thumbnail_url && ( +
+ thumbnail +
썸네일
+
+ )} +
+ + {pipeline.video_url && ( +
+
+ )} + + {meta.title && ( +
+

메타데이터

+

제목: {meta.title}

+
+ 설명 ({(meta.description || '').length}자) + 
{meta.description}
+
+

태그: {(meta.tags || []).join(', ')}

+
+ )} + + {review.weighted_total != null && ( +
+

AI 검토 + {review.verdict} + ({review.weighted_total}/100) +

+ + + + + + + +
메타데이터 품질{review.metadata_quality?.score}
콘텐츠 정책{review.policy_compliance?.score}
시청 경험{review.viewer_experience?.score}
트렌드 정렬{review.trend_alignment?.score}
+ {review.summary &&

{review.summary}

} +
+ )} + + {pipeline.tracks && pipeline.tracks.length > 1 && ( +
+

트랙 리스트 ({pipeline.tracks.length})

+
    + {pipeline.tracks.map(t => ( +
  1. + [{fmtTimestamp(t.start_offset_sec)}] + {' '}{t.title} + ({fmtDuration(t.duration_sec)}) +
    2. + ))} +
+
+ )} + + {pipeline.feedback && pipeline.feedback.length > 0 && ( +
+

피드백 히스토리 ({pipeline.feedback.length})

+
    + {pipeline.feedback.map(f => ( +
  • + [{f.step}] {f.feedback_text} + {f.received_at?.replace('T', ' ')} +
    • + ))} +
+
+ )} + + {pipeline.youtube_video_id && ( + + 🎬 YouTube에서 보기 + + )} +
+
+ ); +} +``` + +- [ ] **Step 2: PipelineCard.jsx — 미리보기 + 클릭 핸들러** + +```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 ( + <> +
+
+

{title}

+ {pipeline.visual_style && ( + {pipeline.visual_style} + )} + {!['published','cancelled','failed'].includes(pipeline.state) && ( + + )} +
+ + {/* 미니 미리보기 */} +
+ {pipeline.cover_url && ( + + )} + {pipeline.thumbnail_url && ( + + )} + {pipeline.video_url && ▶} +
+ +
+ {STEP_LABELS.map((lbl, idx) => ( +
+ {lbl} +
+ ))} +
+ +
현재: {pipeline.state}
+ + {pipeline.review && ( +
+ AI 검토: {pipeline.review.verdict} + ({pipeline.review.weighted_total}/100) +
+ )} + + {pipeline.state === 'publish_pending' && ( + + )} + + {pipeline.youtube_video_id && ( + + 유튜브에서 보기 + + )} +
+ + {showDetail && setShowDetail(false)} />} + + ); +} +``` + +- [ ] **Step 3: CSS 추가 (`MusicStudio.css` append)** + +```css +/* === 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** + +```bash +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` 중 "영상 비주얼 기본값"을 다음으로 교체: + +```jsx +
+

영상 비주얼 기본값

+ + + + + + + + + + + + + + +
+``` + +- [ ] **Step 2: Build verify + commit** + +```bash +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** + +```bash +git -C C:/Users/jaeoh/Desktop/workspace/web-ui push origin main +``` + +- [ ] **Step 2: NAS에 빌드 + robocopy** + +로컬에서: +```bash +cd C:/Users/jaeoh/Desktop/workspace/web-ui +npm run release:nas +``` + +- [ ] **Step 3: 캐시 무력화 — 브라우저 강력 새로고침** + +`Ctrl+Shift+R`. + +--- + +## Task 17: 수동 E2E 검증 체크리스트 + +- [ ] **단일 트랙 essential 영상** + - 진행 탭 → "+ 새 파이프라인" → 단일 트랙 라디오 → 트랙 선택 → 시작 + - 카드에 cover 미니 썸네일 표시 ✓ + - 카드 클릭 → 상세 모달 cover 큰 이미지 표시 ✓ + - 텔레그램 "승인" → 영상 단계 + - 영상 단계 후 카드에 video ▶ 아이콘 표시 + - 상세 모달의 `