from __future__ import annotations import json import shutil import subprocess from pathlib import Path def require_binary(name: str) -> str: path = shutil.which(name) if not path: raise RuntimeError(f"Required executable '{name}' was not found in PATH.") return path def run(command: list[str]) -> None: process = subprocess.run(command, text=True) if process.returncode != 0: raise RuntimeError(f"Command failed with exit code {process.returncode}: {' '.join(command)}") def extract_audio(media_path: Path, wav_path: Path, sample_rate: int = 16000) -> Path: require_binary("ffmpeg") wav_path.parent.mkdir(parents=True, exist_ok=True) run([ "ffmpeg", "-y", "-i", str(media_path), "-vn", "-ac", "1", "-ar", str(sample_rate), "-c:a", "pcm_s16le", str(wav_path), ]) return wav_path def probe_fps(media_path: Path) -> float: require_binary("ffprobe") result = subprocess.run([ "ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=avg_frame_rate", "-of", "json", str(media_path) ], capture_output=True, text=True, check=True) data = json.loads(result.stdout) rate = data["streams"][0]["avg_frame_rate"] num, den = rate.split("/") return float(num) / float(den) def render_ass(media_path: Path, ass_path: Path, output_path: Path) -> Path: require_binary("ffmpeg") escaped = str(ass_path.resolve()).replace("\\", "/").replace(":", "\\:") output_path.parent.mkdir(parents=True, exist_ok=True) run([ "ffmpeg", "-y", "-i", str(media_path), "-vf", f"ass='{escaped}'", "-c:v", "libx264", "-preset", "slow", "-crf", "16", "-c:a", "copy", str(output_path), ]) return output_path