"""영상 비주얼 생성 — visualizer/슬라이드쇼 스타일.""" import os import subprocess import logging from . import storage logger = logging.getLogger("music-lab.video") VIDEO_TIMEOUT_S = 600 # 10분 — Celeron J4025 같은 저성능 CPU 여유 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)) try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=VIDEO_TIMEOUT_S) except subprocess.TimeoutExpired: raise VideoGenerationError(f"ffmpeg 타임아웃 ({VIDEO_TIMEOUT_S}s) — CPU 부하 또는 입력 파일 문제") if result.returncode != 0: raise VideoGenerationError(f"ffmpeg 실패: {result.stderr[-800:]}") 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", "ultrafast", # was "fast" — ultrafast is 5-10x faster on weak CPUs "-tune", "stillimage", # 정적 배경 + 파형 오버레이는 stillimage 튜닝이 적합 "-threads", "0", # use all available CPU cores "-crf", "23", "-c:a", "aac", "-b:a", "192k", "-shortest", out, ]