164 lines
4.0 KiB
Python
164 lines
4.0 KiB
Python
"""
|
|
Utility functions for video processing.
|
|
|
|
Provides FFmpeg wrappers and common operations used by all strategies.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
|
|
def get_video_duration(video_path: str) -> float:
|
|
"""
|
|
Get video duration in seconds using ffprobe.
|
|
|
|
Args:
|
|
video_path: Path to video file
|
|
|
|
Returns:
|
|
Duration in seconds
|
|
|
|
Raises:
|
|
RuntimeError: If ffprobe fails or video is invalid
|
|
"""
|
|
cmd = [
|
|
"ffprobe",
|
|
"-v", "error",
|
|
"-show_entries", "format=duration",
|
|
"-of", "json",
|
|
video_path
|
|
]
|
|
|
|
try:
|
|
result = subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True
|
|
)
|
|
data = json.loads(result.stdout)
|
|
return float(data["format"]["duration"])
|
|
except subprocess.CalledProcessError as e:
|
|
raise RuntimeError(f"Failed to get video duration: {e.stderr}")
|
|
except (KeyError, ValueError) as e:
|
|
raise RuntimeError(f"Invalid ffprobe output: {e}")
|
|
|
|
|
|
def extract_audio(
|
|
video_path: str,
|
|
output_path: str | None = None,
|
|
sample_rate: int = 16000,
|
|
codec: str = "pcm_s16le"
|
|
) -> str:
|
|
"""
|
|
Extract audio from video file using FFmpeg.
|
|
|
|
Args:
|
|
video_path: Path to input video
|
|
output_path: Path for output audio (default: same name with .wav)
|
|
sample_rate: Audio sample rate in Hz (16000 recommended for Whisper)
|
|
codec: Audio codec (pcm_s16le for WAV)
|
|
|
|
Returns:
|
|
Path to extracted audio file
|
|
|
|
Raises:
|
|
RuntimeError: If FFmpeg fails
|
|
"""
|
|
if output_path is None:
|
|
video_name = Path(video_path).stem
|
|
output_path = str(Path(video_path).parent / f"{video_name}_audio.wav")
|
|
|
|
cmd = [
|
|
"ffmpeg",
|
|
"-i", video_path,
|
|
"-vn", # No video
|
|
"-acodec", codec,
|
|
"-ar", str(sample_rate),
|
|
"-ac", "1", # Mono
|
|
"-y", # Overwrite output file
|
|
output_path
|
|
]
|
|
|
|
try:
|
|
subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True
|
|
)
|
|
return output_path
|
|
except subprocess.CalledProcessError as e:
|
|
raise RuntimeError(f"Failed to extract audio: {e.stderr}")
|
|
|
|
|
|
def slice_video(
|
|
video_path: str,
|
|
start_time: float,
|
|
end_time: float,
|
|
output_path: str | None = None
|
|
) -> str:
|
|
"""
|
|
Slice a video segment using FFmpeg stream copy (fast, no re-encoding).
|
|
|
|
Args:
|
|
video_path: Path to input video
|
|
start_time: Start time in seconds
|
|
end_time: End time in seconds
|
|
output_path: Path for output segment
|
|
|
|
Returns:
|
|
Path to output segment
|
|
|
|
Raises:
|
|
RuntimeError: If FFmpeg fails
|
|
"""
|
|
if output_path is None:
|
|
video_name = Path(video_path).stem
|
|
output_path = str(
|
|
Path(video_path).parent / f"{video_name}_{start_time:.1f}-{end_time:.1f}.mp4"
|
|
)
|
|
|
|
duration = end_time - start_time
|
|
|
|
cmd = [
|
|
"ffmpeg",
|
|
"-ss", str(start_time),
|
|
"-i", video_path,
|
|
"-t", str(duration),
|
|
"-c", "copy", # Stream copy (no re-encoding)
|
|
"-y",
|
|
output_path
|
|
]
|
|
|
|
try:
|
|
subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
check=True
|
|
)
|
|
return output_path
|
|
except subprocess.CalledProcessError as e:
|
|
raise RuntimeError(f"Failed to slice video: {e.stderr}")
|
|
|
|
|
|
def check_ffmpeg_installed() -> bool:
|
|
"""Check if FFmpeg and FFprobe are available in PATH."""
|
|
try:
|
|
subprocess.run(
|
|
["ffmpeg", "-version"],
|
|
capture_output=True,
|
|
check=True
|
|
)
|
|
subprocess.run(
|
|
["ffprobe", "-version"],
|
|
capture_output=True,
|
|
check=True
|
|
)
|
|
return True
|
|
except (subprocess.CalledProcessError, FileNotFoundError):
|
|
return False
|