diff --git a/music-lab/app/pipeline/background.py b/music-lab/app/pipeline/background.py new file mode 100644 index 0000000..47831f7 --- /dev/null +++ b/music-lab/app/pipeline/background.py @@ -0,0 +1,60 @@ +"""Pexels Video API로 background loop 영상 받아오기 (video_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/1080p HD 우선, 없으면 첫 번째 video file + 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)} diff --git a/music-lab/app/pipeline/orchestrator.py b/music-lab/app/pipeline/orchestrator.py index cb15e27..ce2a5ff 100644 --- a/music-lab/app/pipeline/orchestrator.py +++ b/music-lab/app/pipeline/orchestrator.py @@ -6,7 +6,8 @@ import os import sqlite3 from app import db -from . import cover, video, thumb, metadata, review, youtube +from . import cover, video, thumb, metadata, review, youtube, background, storage +from .gradient import make_gradient_with_title logger = logging.getLogger("music-lab.orchestrator") @@ -167,13 +168,31 @@ def _fetch_track_fallback(track_id: int) -> dict | None: async def _run_cover(p, ctx, feedback): setup = db.get_youtube_setup() + vd = setup["visual_defaults"] + bg_mode = p.get("background_mode") or vd.get("default_background_mode", "static") + keyword = p.get("background_keyword") or vd.get("default_background_keyword", "") + + if bg_mode == "video_loop": + # Pexels 영상 다운로드 시도 — 성공 여부와 무관하게 cover.jpg는 그라데이션으로 별도 생성 + # (실패 시 video.py가 cover.jpg를 fallback 배경으로 사용 가능) + await background.fetch_video_loop(p["id"], keyword) + + 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 = vd.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"]}} diff --git a/music-lab/tests/test_background.py b/music-lab/tests/test_background.py new file mode 100644 index 0000000..1982261 --- /dev/null +++ b/music-lab/tests/test_background.py @@ -0,0 +1,51 @@ +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