feat(music-lab): AI 커버 생성 + 그라데이션 폴백

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-07 16:45:08 +09:00
parent fceca88db4
commit e33a2310af
6 changed files with 238 additions and 0 deletions

View File

@@ -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)

View File

@@ -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)

View File

@@ -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}"