diff --git a/music-lab/app/pipeline/cover.py b/music-lab/app/pipeline/cover.py new file mode 100644 index 0000000..51f3394 --- /dev/null +++ b/music-lab/app/pipeline/cover.py @@ -0,0 +1,88 @@ +"""AI 커버 아트 생성 — DALL·E 3 / gpt-image-1 + 그라데이션 폴백.""" +import base64 +import logging +import os +from io import BytesIO + +import httpx +from PIL import Image + +from . import storage +from .gradient import make_gradient_with_title + +logger = logging.getLogger("music-lab.cover") + +DALLE_TIMEOUT_S = 90 + + +def _get_api_key() -> str: + return os.getenv("OPENAI_API_KEY", "") + + +def _get_model() -> str: + return os.getenv("OPENAI_IMAGE_MODEL", "gpt-image-1") + + +async def generate(*, pipeline_id: int, genre: str, prompt_template: str, + mood: str = "", track_title: str = "", feedback: str = "") -> dict: + """커버 아트 생성. 성공 시 jpg 저장 + URL 반환. 실패 시 그라데이션 폴백. + + 반환: {"url": str, "used_fallback": bool, "error": str | None} + """ + out_path = os.path.join(storage.pipeline_dir(pipeline_id), "cover.jpg") + used_fallback = False + error = None + + api_key = _get_api_key() + model = _get_model() + if api_key: + try: + await _generate_with_dalle(prompt_template, mood, feedback, out_path, + api_key=api_key, model=model) + except (httpx.HTTPError, httpx.TimeoutException, KeyError, ValueError, OSError) as e: + logger.warning("DALL·E 실패 — 폴백: %s", e) + error = str(e) + used_fallback = True + make_gradient_with_title(genre, track_title, out_path) + else: + used_fallback = True + error = "OPENAI_API_KEY 미설정" + make_gradient_with_title(genre, track_title, out_path) + + return { + "url": storage.media_url(pipeline_id, "cover.jpg"), + "used_fallback": used_fallback, + "error": error, + } + + +async def _generate_with_dalle(prompt_template: str, mood: str, + feedback: str, out_path: str, + *, api_key: str, model: str) -> None: + prompt = prompt_template + if mood: + prompt = f"{prompt}, {mood} mood" + if feedback: + prompt = f"{prompt}. 추가 지시: {feedback}" + prompt = f"{prompt}, no text, high quality" + + async with httpx.AsyncClient(timeout=DALLE_TIMEOUT_S) as client: + resp = await client.post( + "https://api.openai.com/v1/images/generations", + headers={"Authorization": f"Bearer {api_key}"}, + json={"model": model, "prompt": prompt, "size": "1024x1024", "n": 1}, + ) + resp.raise_for_status() + data = resp.json()["data"][0] + if "url" in data: + img_resp = await client.get(data["url"]) + img_resp.raise_for_status() + img_bytes = img_resp.content + elif "b64_json" in data: + img_bytes = base64.b64decode(data["b64_json"]) + else: + raise ValueError("DALL·E response has neither url nor b64_json") + # PNG → JPG 변환 + with Image.open(BytesIO(img_bytes)) as src: + img = src.convert("RGB") + img.save(out_path, "JPEG", quality=92) diff --git a/music-lab/app/pipeline/gradient.py b/music-lab/app/pipeline/gradient.py new file mode 100644 index 0000000..7d67cb4 --- /dev/null +++ b/music-lab/app/pipeline/gradient.py @@ -0,0 +1,38 @@ +"""장르별 그라데이션 배경 + 텍스트 오버레이 — cover/video 공용.""" +from PIL import Image, ImageDraw, ImageFont + +GENRE_COLORS = { + "lo-fi": ((26, 26, 46), (22, 33, 62)), + "phonk": ((26, 10, 10), (45, 0, 0)), + "ambient": ((13, 33, 55), (10, 22, 40)), + "pop": ((26, 10, 46), (45, 27, 78)), + "default": ((17, 24, 39), (31, 41, 55)), +} + + +def make_gradient_with_title(genre: str, title: str, out_path: str, + size: tuple[int, int] = (1024, 1024), + quality: int = 92) -> None: + w, h = size + top, bot = GENRE_COLORS.get(genre.lower(), GENRE_COLORS["default"]) + with Image.new("RGB", (w, h)) as img: + px = img.load() + for y in range(h): + t = y / h + r = int(top[0] + (bot[0] - top[0]) * t) + g = int(top[1] + (bot[1] - top[1]) * t) + b = int(top[2] + (bot[2] - top[2]) * t) + for x in range(w): + px[x, y] = (r, g, b) + + if title: + try: + font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", 64) + except OSError: + font = ImageFont.load_default() + draw = ImageDraw.Draw(img) + bbox = draw.textbbox((0, 0), title, font=font) + tw, th = bbox[2] - bbox[0], bbox[3] - bbox[1] + draw.text(((w - tw) // 2, (h - th) // 2), title, fill=(255, 255, 255), font=font) + + img.save(out_path, "JPEG", quality=quality) diff --git a/music-lab/app/pipeline/storage.py b/music-lab/app/pipeline/storage.py new file mode 100644 index 0000000..16ad077 --- /dev/null +++ b/music-lab/app/pipeline/storage.py @@ -0,0 +1,15 @@ +"""파이프라인 산출물 디렉토리 관리.""" +import os + +VIDEO_DATA_DIR = os.getenv("VIDEO_DATA_DIR", "/app/data/videos") +VIDEO_MEDIA_BASE = os.getenv("VIDEO_MEDIA_BASE", "/media/videos") + + +def pipeline_dir(pipeline_id: int) -> str: + path = os.path.join(VIDEO_DATA_DIR, str(pipeline_id)) + os.makedirs(path, exist_ok=True) + return path + + +def media_url(pipeline_id: int, filename: str) -> str: + return f"{VIDEO_MEDIA_BASE}/{pipeline_id}/{filename}" diff --git a/music-lab/pytest.ini b/music-lab/pytest.ini index 4584de7..654b7eb 100644 --- a/music-lab/pytest.ini +++ b/music-lab/pytest.ini @@ -1,3 +1,4 @@ [pytest] testpaths = tests pythonpath = . +asyncio_mode = auto diff --git a/music-lab/requirements.txt b/music-lab/requirements.txt index 4ce3db6..6a8efd9 100644 --- a/music-lab/requirements.txt +++ b/music-lab/requirements.txt @@ -4,7 +4,10 @@ requests==2.32.3 python-multipart==0.0.12 mutagen==1.47.0 anthropic>=0.40.0 +openai>=1.20.0 Pillow>=11.0.0 pytest>=8.0.0 +pytest-asyncio>=0.21 httpx>=0.27.0 +respx>=0.21 freezegun>=1.4 diff --git a/music-lab/tests/test_cover_generation.py b/music-lab/tests/test_cover_generation.py new file mode 100644 index 0000000..d2ba335 --- /dev/null +++ b/music-lab/tests/test_cover_generation.py @@ -0,0 +1,93 @@ +import base64 + +import pytest +import respx +from httpx import Response +from app.pipeline import cover, storage + + +# Real PNG bytes (1x1 red pixel) so PIL can open +_TINY_PNG = bytes.fromhex( + "89504e470d0a1a0a0000000d49484452000000010000000108020000009077" + "53de0000000c4944415478da6300010000050001" + "0d0a2db40000000049454e44ae426082" +) + + +@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_dalle_success_saves_jpg(tmp_storage, monkeypatch): + monkeypatch.setenv("OPENAI_API_KEY", "test-key") + image_url = "https://oaidalleapiprodscus.blob.core.windows.net/x.png" + respx.post("https://api.openai.com/v1/images/generations").mock( + return_value=Response(200, json={"data": [{"url": image_url}]}) + ) + respx.get(image_url).mock(return_value=Response(200, content=_TINY_PNG)) + + out = await cover.generate(pipeline_id=42, genre="lo-fi", + prompt_template="moody anime", mood="chill", + track_title="Test") + assert out["used_fallback"] is False + assert out["url"].startswith("/media/videos/42/cover") + assert (tmp_storage / "42" / "cover.jpg").exists() + + +@pytest.mark.asyncio +@respx.mock +async def test_dalle_http_error_falls_back_to_gradient(tmp_storage, monkeypatch): + monkeypatch.setenv("OPENAI_API_KEY", "test-key") + respx.post("https://api.openai.com/v1/images/generations").mock( + return_value=Response(504) + ) + out = await cover.generate(pipeline_id=43, genre="phonk", + prompt_template="dark drift", mood="aggressive", + track_title="Midnight Drive") + assert out["used_fallback"] is True + assert (tmp_storage / "43" / "cover.jpg").exists() + + +@pytest.mark.asyncio +async def test_no_api_key_falls_back(tmp_storage, monkeypatch): + monkeypatch.delenv("OPENAI_API_KEY", raising=False) + out = await cover.generate(pipeline_id=44, genre="ambient", + prompt_template="x", mood="calm", + track_title="Calm") + assert out["used_fallback"] is True + + +@pytest.mark.asyncio +@respx.mock +async def test_dalle_with_feedback_appends_to_prompt(tmp_storage, monkeypatch): + import json as _json + monkeypatch.setenv("OPENAI_API_KEY", "test-key") + captured = {} + def hook(req): + captured["body"] = _json.loads(req.content) + return Response(200, json={"data": [{"url": "https://x"}]}) + respx.post("https://api.openai.com/v1/images/generations").mock(side_effect=hook) + respx.get("https://x").mock(return_value=Response(200, content=_TINY_PNG)) + out = await cover.generate(pipeline_id=45, genre="lo-fi", + prompt_template="moody anime", mood="chill", + track_title="X", feedback="더 어둡게") + assert "더 어둡게" in captured["body"]["prompt"] + assert out["used_fallback"] is False + + +@pytest.mark.asyncio +@respx.mock +async def test_dalle_b64_response_handled(tmp_storage, monkeypatch): + monkeypatch.setenv("OPENAI_API_KEY", "test-key") + b64 = base64.b64encode(_TINY_PNG).decode() + respx.post("https://api.openai.com/v1/images/generations").mock( + return_value=Response(200, json={"data": [{"b64_json": b64}]}) + ) + out = await cover.generate(pipeline_id=46, genre="lo-fi", + prompt_template="x", mood="", track_title="X") + assert out["used_fallback"] is False + assert (tmp_storage / "46" / "cover.jpg").exists()