import subprocess from pathlib import Path from typing import List, Optional from app.config import COVER_DIR, HLS_DIR, PROCESSED_DIR class FFmpegError(RuntimeError): pass def run_ffmpeg(command: List[str]) -> None: completed = subprocess.run( command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False, ) if completed.returncode != 0: raise FFmpegError(completed.stderr.strip() or "ffmpeg command failed") def build_filter(watermark_text: Optional[str]) -> Optional[str]: if not watermark_text: return None escaped = watermark_text.replace(":", r"\:").replace("'", r"\\'") return ( "drawtext=text='" f"{escaped}" "':fontcolor=white:fontsize=28:box=1:boxcolor=black@0.45:boxborderw=6:x=w-tw-24:y=h-th-24" ) def transcode_to_mp4( source: Path, task_id: str, clip_seconds: Optional[int], watermark_text: Optional[str] ) -> Path: output = PROCESSED_DIR / f"{task_id}.mp4" command = ["ffmpeg", "-y", "-i", str(source)] if clip_seconds: command.extend(["-t", str(clip_seconds)]) video_filter = build_filter(watermark_text) if video_filter: command.extend(["-vf", video_filter]) command.extend( [ "-c:v", "libx264", "-preset", "veryfast", "-c:a", "aac", "-movflags", "+faststart", str(output), ] ) run_ffmpeg(command) return output def generate_cover(source: Path, task_id: str) -> Path: output = COVER_DIR / f"{task_id}.jpg" command = [ "ffmpeg", "-y", "-ss", "00:00:01", "-i", str(source), "-frames:v", "1", str(output), ] run_ffmpeg(command) return output def generate_hls(source_mp4: Path, task_id: str) -> Path: task_dir = HLS_DIR / task_id task_dir.mkdir(parents=True, exist_ok=True) playlist = task_dir / "index.m3u8" segment_pattern = task_dir / "segment_%03d.ts" command = [ "ffmpeg", "-y", "-i", str(source_mp4), "-codec:v", "libx264", "-codec:a", "aac", "-hls_time", "4", "-hls_playlist_type", "vod", "-hls_segment_filename", str(segment_pattern), str(playlist), ] run_ffmpeg(command) return playlist