78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
import os
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
from PIL import Image
|
|
|
|
from config import Config
|
|
from models import JobStore
|
|
|
|
|
|
def _ffmpeg_available() -> bool:
|
|
return shutil.which('ffmpeg') is not None
|
|
|
|
|
|
def _prepare_image(image_path: Path, output_path: Path, width: int, height: int) -> None:
|
|
img = Image.open(image_path)
|
|
img = img.convert('RGB')
|
|
|
|
# Fit inside target box, then letterbox/pillarbox with black to exact 16:9
|
|
img.thumbnail((width, height), Image.LANCZOS)
|
|
|
|
canvas = Image.new('RGB', (width, height), (0, 0, 0))
|
|
x = (width - img.width) // 2
|
|
y = (height - img.height) // 2
|
|
canvas.paste(img, (x, y))
|
|
canvas.save(output_path, 'JPEG', quality=95)
|
|
|
|
|
|
def _build_video(image_path: Path, audio_path: Path, output_path: Path, cfg: Config) -> None:
|
|
if not _ffmpeg_available():
|
|
raise RuntimeError('ffmpeg is not installed or not on PATH')
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
tmp = Path(tmpdir)
|
|
prepared_image = tmp / 'prepared.jpg'
|
|
_prepare_image(image_path, prepared_image, cfg.VIDEO_WIDTH, cfg.VIDEO_HEIGHT)
|
|
|
|
cmd = [
|
|
'ffmpeg',
|
|
'-y',
|
|
'-loop', '1',
|
|
'-framerate', str(cfg.VIDEO_FPS),
|
|
'-i', str(prepared_image),
|
|
'-i', str(audio_path),
|
|
'-c:v', cfg.VIDEO_CODEC,
|
|
'-pix_fmt', 'yuv420p',
|
|
'-preset', 'medium',
|
|
'-b:v', cfg.VIDEO_BITRATE,
|
|
'-c:a', cfg.AUDIO_CODEC,
|
|
'-b:a', cfg.AUDIO_BITRATE,
|
|
'-movflags', '+faststart',
|
|
'-shortest',
|
|
str(output_path),
|
|
]
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(f'ffmpeg failed: {result.stderr}')
|
|
|
|
|
|
def process_job(job_id: int, store: JobStore, cfg: Config) -> None:
|
|
job = store.get_job(job_id)
|
|
if job is None:
|
|
return
|
|
|
|
image_path = cfg.UPLOAD_FOLDER / job['image_file']
|
|
audio_path = cfg.UPLOAD_FOLDER / job['audio_file']
|
|
output_filename = f"job-{job_id}.{cfg.OUTPUT_FORMAT}"
|
|
output_path = cfg.OUTPUT_FOLDER / output_filename
|
|
|
|
try:
|
|
cfg.OUTPUT_FOLDER.mkdir(parents=True, exist_ok=True)
|
|
_build_video(image_path, audio_path, output_path, cfg)
|
|
store.update_status(job_id, 'complete', output_file=output_filename)
|
|
except Exception as exc:
|
|
store.update_status(job_id, 'error', error_message=str(exc))
|
|
raise
|