feat(music-lab): pipeline 영상·썸네일 생성
This commit is contained in:
51
music-lab/app/pipeline/thumb.py
Normal file
51
music-lab/app/pipeline/thumb.py
Normal file
@@ -0,0 +1,51 @@
|
|||||||
|
"""썸네일 생성 — 영상 5초 프레임 추출 + 텍스트 오버레이."""
|
||||||
|
import os
|
||||||
|
import subprocess
|
||||||
|
import logging
|
||||||
|
from PIL import Image, ImageDraw, ImageFont
|
||||||
|
|
||||||
|
from . import storage
|
||||||
|
|
||||||
|
logger = logging.getLogger("music-lab.thumb")
|
||||||
|
THUMB_TIMEOUT_S = 60
|
||||||
|
|
||||||
|
|
||||||
|
class ThumbGenerationError(Exception):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def generate(*, pipeline_id: int, video_path: str,
|
||||||
|
track_title: str = "", overlay_text: bool = True) -> dict:
|
||||||
|
out_path = os.path.join(storage.pipeline_dir(pipeline_id), "thumbnail.jpg")
|
||||||
|
cmd = ["ffmpeg", "-y", "-i", video_path,
|
||||||
|
"-ss", "00:00:05", "-vframes", "1", "-q:v", "2", out_path]
|
||||||
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=THUMB_TIMEOUT_S)
|
||||||
|
if result.returncode != 0:
|
||||||
|
raise ThumbGenerationError(f"ffmpeg 썸네일 실패: {result.stderr[:300]}")
|
||||||
|
|
||||||
|
if overlay_text and track_title:
|
||||||
|
_overlay_title(out_path, track_title)
|
||||||
|
|
||||||
|
return {"url": storage.media_url(pipeline_id, "thumbnail.jpg"), "used_fallback": False}
|
||||||
|
|
||||||
|
|
||||||
|
def _overlay_title(path: str, title: str) -> None:
|
||||||
|
try:
|
||||||
|
with Image.open(path) as src:
|
||||||
|
img = src.convert("RGB")
|
||||||
|
try:
|
||||||
|
font = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", 80)
|
||||||
|
except OSError:
|
||||||
|
font = ImageFont.load_default()
|
||||||
|
draw = ImageDraw.Draw(img)
|
||||||
|
# 하단 30% 영역에 검정 반투명 박스 + 흰 글씨
|
||||||
|
w, h = img.size
|
||||||
|
box_h = int(h * 0.3)
|
||||||
|
with Image.new("RGBA", (w, box_h), (0, 0, 0, 160)) as overlay:
|
||||||
|
img.paste(overlay, (0, h - box_h), overlay)
|
||||||
|
bbox = draw.textbbox((0, 0), title, font=font)
|
||||||
|
tw = bbox[2] - bbox[0]
|
||||||
|
draw.text(((w - tw) // 2, h - box_h + 30), title, fill=(255, 255, 255), font=font)
|
||||||
|
img.save(path, "JPEG", quality=92)
|
||||||
|
except Exception as e:
|
||||||
|
logger.warning("썸네일 오버레이 실패: %s", e)
|
||||||
55
music-lab/app/pipeline/video.py
Normal file
55
music-lab/app/pipeline/video.py
Normal file
@@ -0,0 +1,55 @@
|
|||||||
|
"""영상 비주얼 생성 — visualizer/슬라이드쇼 스타일."""
|
||||||
|
import os
|
||||||
|
import subprocess
|
||||||
|
import logging
|
||||||
|
|
||||||
|
from . import storage
|
||||||
|
|
||||||
|
logger = logging.getLogger("music-lab.video")
|
||||||
|
|
||||||
|
VIDEO_TIMEOUT_S = 300 # 5분
|
||||||
|
|
||||||
|
|
||||||
|
class VideoGenerationError(Exception):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def generate(*, pipeline_id: int, audio_path: str, cover_path: str,
|
||||||
|
genre: str, duration_sec: int, resolution: str = "1920x1080",
|
||||||
|
style: str = "visualizer") -> dict:
|
||||||
|
"""영상 생성. 성공 시 mp4 저장 + URL 반환. 실패 시 예외."""
|
||||||
|
w, h = resolution.split("x")
|
||||||
|
out_path = os.path.join(storage.pipeline_dir(pipeline_id), "video.mp4")
|
||||||
|
|
||||||
|
if style == "visualizer":
|
||||||
|
cmd = _build_visualizer_cmd(audio_path, cover_path, out_path, w, h)
|
||||||
|
else:
|
||||||
|
# 차후: 슬라이드쇼 등 다른 스타일 — 현재는 visualizer 폴백
|
||||||
|
cmd = _build_visualizer_cmd(audio_path, cover_path, out_path, w, h)
|
||||||
|
|
||||||
|
logger.info("ffmpeg 실행: %s", " ".join(cmd))
|
||||||
|
result = subprocess.run(cmd, capture_output=True, text=True, timeout=VIDEO_TIMEOUT_S)
|
||||||
|
if result.returncode != 0:
|
||||||
|
raise VideoGenerationError(f"ffmpeg 실패: {result.stderr[:500]}")
|
||||||
|
|
||||||
|
return {
|
||||||
|
"url": storage.media_url(pipeline_id, "video.mp4"),
|
||||||
|
"used_fallback": False,
|
||||||
|
"duration_sec": duration_sec,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _build_visualizer_cmd(audio: str, bg: str, out: str, w: str, h: str) -> list:
|
||||||
|
return [
|
||||||
|
"ffmpeg", "-y",
|
||||||
|
"-loop", "1", "-i", bg,
|
||||||
|
"-i", audio,
|
||||||
|
"-filter_complex",
|
||||||
|
f"[0:v]scale={w}:{h}[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", "libx264", "-preset", "fast", "-crf", "23",
|
||||||
|
"-c:a", "aac", "-b:a", "192k",
|
||||||
|
"-shortest", out,
|
||||||
|
]
|
||||||
66
music-lab/tests/test_video_thumb.py
Normal file
66
music-lab/tests/test_video_thumb.py
Normal file
@@ -0,0 +1,66 @@
|
|||||||
|
import os
|
||||||
|
import pytest
|
||||||
|
from unittest.mock import patch, MagicMock
|
||||||
|
from app.pipeline import video, thumb, storage
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def tmp_storage(monkeypatch, tmp_path):
|
||||||
|
monkeypatch.setattr(storage, "VIDEO_DATA_DIR", str(tmp_path))
|
||||||
|
# 더미 입력 파일들
|
||||||
|
audio = tmp_path / "audio.mp3"
|
||||||
|
audio.write_bytes(b"\x00" * 100)
|
||||||
|
cover_dir = tmp_path / "50"
|
||||||
|
cover_dir.mkdir()
|
||||||
|
cover = cover_dir / "cover.jpg"
|
||||||
|
cover.write_bytes(b"\x00" * 100)
|
||||||
|
return tmp_path
|
||||||
|
|
||||||
|
|
||||||
|
@patch("subprocess.run")
|
||||||
|
def test_generate_video_calls_ffmpeg(mock_run, tmp_storage):
|
||||||
|
mock_run.return_value = MagicMock(returncode=0, stderr="")
|
||||||
|
out = video.generate(pipeline_id=50, audio_path=str(tmp_storage / "audio.mp3"),
|
||||||
|
cover_path=str(tmp_storage / "50" / "cover.jpg"),
|
||||||
|
genre="lo-fi", duration_sec=120, resolution="1920x1080",
|
||||||
|
style="visualizer")
|
||||||
|
assert out["url"].endswith("/50/video.mp4")
|
||||||
|
assert out["used_fallback"] is False
|
||||||
|
args = mock_run.call_args[0][0]
|
||||||
|
assert args[0] == "ffmpeg"
|
||||||
|
assert "-i" in args
|
||||||
|
assert "showwaves" in " ".join(args)
|
||||||
|
|
||||||
|
|
||||||
|
@patch("subprocess.run")
|
||||||
|
def test_generate_video_failure_marks_failed(mock_run, tmp_storage):
|
||||||
|
mock_run.return_value = MagicMock(returncode=1, stderr="bad codec")
|
||||||
|
with pytest.raises(video.VideoGenerationError):
|
||||||
|
video.generate(pipeline_id=51, audio_path=str(tmp_storage / "audio.mp3"),
|
||||||
|
cover_path=str(tmp_storage / "50" / "cover.jpg"),
|
||||||
|
genre="lo-fi", duration_sec=120, resolution="1920x1080",
|
||||||
|
style="visualizer")
|
||||||
|
|
||||||
|
|
||||||
|
@patch("subprocess.run")
|
||||||
|
def test_thumb_extracts_frame(mock_run, tmp_storage):
|
||||||
|
mock_run.return_value = MagicMock(returncode=0, stderr="")
|
||||||
|
video_path = tmp_storage / "60" / "video.mp4"
|
||||||
|
video_path.parent.mkdir()
|
||||||
|
video_path.write_bytes(b"\x00" * 100)
|
||||||
|
out = thumb.generate(pipeline_id=60, video_path=str(video_path),
|
||||||
|
track_title="Midnight Drive", overlay_text=False)
|
||||||
|
assert out["url"].endswith("/60/thumbnail.jpg")
|
||||||
|
args = mock_run.call_args[0][0]
|
||||||
|
assert args[0] == "ffmpeg"
|
||||||
|
|
||||||
|
|
||||||
|
@patch("subprocess.run")
|
||||||
|
def test_thumb_failure_raises(mock_run, tmp_storage):
|
||||||
|
mock_run.return_value = MagicMock(returncode=1, stderr="bad input")
|
||||||
|
video_path = tmp_storage / "61" / "video.mp4"
|
||||||
|
video_path.parent.mkdir()
|
||||||
|
video_path.write_bytes(b"\x00" * 100)
|
||||||
|
with pytest.raises(thumb.ThumbGenerationError):
|
||||||
|
thumb.generate(pipeline_id=61, video_path=str(video_path),
|
||||||
|
track_title="X", overlay_text=False)
|
||||||
Reference in New Issue
Block a user