import subprocess import os import shutil from pathlib import Path from typing import List, Dict from ..config import EncodingConfig import logging logger = logging.getLogger(__name__) class FFmpegEncoder: def __init__(self, input_path: str, output_dir: str): # Verify FFmpeg is available if not shutil.which("ffmpeg"): raise RuntimeError("FFmpeg not found in PATH") self.input_path = input_path self.output_dir = output_dir self.base_name = Path(input_path).stem os.makedirs(output_dir, exist_ok=True) def generate_commands(self) -> List[List[str]]: """Generate FFmpeg commands for all resolutions""" commands = [] for res in EncodingConfig.RESOLUTIONS: output_path = os.path.join( self.output_dir, f"{self.base_name}_{res['name']}.m3u8" ) cmd = [ "ffmpeg", "-i", self.input_path, "-vf", f"scale={res['width']}:{res['height']}", "-c:v", res["codec"], "-profile:v", res["profile"], "-preset", res["preset"], "-b:v", res["video_bitrate"], "-c:a", "aac", "-b:a", res["audio_bitrate"], "-f", "hls", "-hls_time", "6", "-hls_playlist_type", "vod", "-hls_segment_filename", os.path.join(self.output_dir, f"{self.base_name}_{res['name']}_%03d.ts"), output_path ] commands.append(cmd) return commands def encode(self) -> str: """Execute encoding commands and return master playlist path""" master_playlist = os.path.join(self.output_dir, f"{self.base_name}_master.m3u8") with open(master_playlist, "w") as f: f.write("#EXTM3U\n") for res in reversed(EncodingConfig.RESOLUTIONS): f.write(f"#EXT-X-STREAM-INF:BANDWIDTH={res['video_bitrate'].replace('k', '000')}," f"RESOLUTION={res['width']}x{res['height']}\n") f.write(f"{self.base_name}_{res['name']}.m3u8\n") for cmd in self.generate_commands(): logger.info(f"Executing FFmpeg command: {' '.join(cmd)}") result = subprocess.run(cmd, capture_output=True) logger.debug(f"FFmpeg command: {' '.join(cmd)}") logger.debug(f"FFmpeg output: {result.stdout.decode()}") logger.debug(f"FFmpeg error: {result.stderr.decode()}") if result.returncode != 0: logger.error(f"FFmpeg command failed with return code {result.returncode}") raise subprocess.CalledProcessError(result.returncode, cmd) result.check_returncode() return master_playlist