video_encoder/worker/ffmpeg.py

import subprocess
import os
from pathlib import Path
from typing import List, Dict
from ..config import EncodingConfig

class FFmpegEncoder:
    def __init__(self, input_path: str, output_dir: str):
        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():
            subprocess.run(cmd, check=True, capture_output=True)
            
        return master_playlist