157 lines
5.0 KiB
Python
157 lines
5.0 KiB
Python
import os
|
|
import subprocess
|
|
import sys
|
|
import json
|
|
import tracker
|
|
from utils import verify_file_not_empty
|
|
|
|
def run_ffmpeg(args):
|
|
"""
|
|
Runs ffmpeg, automatically escaping Flatpak sandbox if necessary.
|
|
"""
|
|
in_flatpak = os.path.exists("/.flatpak-info")
|
|
cmd = ["flatpak-spawn", "--host", "ffmpeg"] + args if in_flatpak else ["ffmpeg"] + args
|
|
|
|
try:
|
|
subprocess.run(cmd, check=True)
|
|
return True
|
|
except subprocess.CalledProcessError as e:
|
|
tracker.logger.error(f"FFmpeg Error: {e}")
|
|
return False
|
|
except FileNotFoundError:
|
|
tracker.logger.error("Error: 'ffmpeg' command not found. Please ensure it is installed on your host system.")
|
|
return False
|
|
|
|
def get_video_duration(file_path):
|
|
"""
|
|
Gets video duration in seconds using ffprobe.
|
|
"""
|
|
in_flatpak = os.path.exists("/.flatpak-info")
|
|
cmd_base = ["flatpak-spawn", "--host"] if in_flatpak else []
|
|
|
|
cmd = cmd_base + [
|
|
"ffprobe",
|
|
"-v", "error",
|
|
"-show_entries", "format=duration",
|
|
"-of", "default=noprint_wrappers=1:nokey=1",
|
|
file_path
|
|
]
|
|
|
|
try:
|
|
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
|
|
return float(result.stdout.strip())
|
|
except Exception:
|
|
return 0.0
|
|
|
|
def extract_audio(video_path, output_path=None):
|
|
"""
|
|
Extracts audio from a video file using ffmpeg.
|
|
"""
|
|
if not os.path.exists(video_path):
|
|
raise FileNotFoundError(f"Video file not found: {video_path}")
|
|
|
|
if output_path is None:
|
|
base_name = os.path.splitext(video_path)[0]
|
|
output_path = f"{base_name}.wav"
|
|
|
|
if verify_file_not_empty(output_path):
|
|
tracker.logger.info(f"Audio file already exists: {output_path}")
|
|
return output_path
|
|
|
|
tracker.logger.info(f"Extracting audio from {video_path}...")
|
|
|
|
args = [
|
|
"-i", video_path,
|
|
"-ar", "16000",
|
|
"-ac", "1",
|
|
"-c:a", "pcm_s16le",
|
|
"-y",
|
|
"-v", "error",
|
|
output_path
|
|
]
|
|
|
|
if run_ffmpeg(args):
|
|
if not verify_file_not_empty(output_path):
|
|
raise Exception("FFmpeg succeeded but output is empty.")
|
|
tracker.logger.info(f"Audio extracted to: {output_path}")
|
|
return output_path
|
|
else:
|
|
sys.exit(1)
|
|
|
|
def embed_subtitles(video_path, srt_path, output_path=None):
|
|
"""
|
|
Embeds subtitles into the video file (Soft Subs) and sets them as primary.
|
|
Includes strict safety checks (Size & Duration) to prevent replacing videos with corrupted files.
|
|
"""
|
|
if not os.path.exists(video_path) or not os.path.exists(srt_path):
|
|
tracker.logger.error("Error: Video or SRT file not found for embedding.")
|
|
return False
|
|
|
|
# Capture original stats
|
|
original_size = os.path.getsize(video_path)
|
|
original_duration = get_video_duration(video_path)
|
|
|
|
if original_size == 0:
|
|
tracker.logger.error("Error: Source video is 0 bytes.")
|
|
return False
|
|
|
|
if output_path is None:
|
|
base, ext = os.path.splitext(video_path)
|
|
output_path = f"{base}.subbed{ext}"
|
|
|
|
tracker.logger.info(f"Embedding subtitles into: {output_path}...")
|
|
|
|
sub_codec = "mov_text" if video_path.lower().endswith(".mp4") else "srt"
|
|
|
|
args = [
|
|
"-ignore_editlist", "1",
|
|
"-i", video_path,
|
|
"-i", srt_path,
|
|
"-map", "0:v",
|
|
"-map", "0:a",
|
|
"-map", "1:0",
|
|
"-c", "copy",
|
|
"-c:s", sub_codec,
|
|
"-disposition:s:0", "default",
|
|
"-metadata:s:s:0", "language=eng",
|
|
"-metadata:s:s:0", "title=English (AI Translated)",
|
|
"-max_interleave_delta", "0",
|
|
"-avoid_negative_ts", "make_zero",
|
|
"-y",
|
|
"-v", "error",
|
|
output_path
|
|
]
|
|
|
|
if run_ffmpeg(args):
|
|
# --- Safety Checks ---
|
|
if not os.path.exists(output_path):
|
|
tracker.logger.error("Error: Output file was not created.")
|
|
return False
|
|
|
|
new_size = os.path.getsize(output_path)
|
|
new_duration = get_video_duration(output_path)
|
|
|
|
# 1. Zero Byte Check
|
|
if new_size == 0:
|
|
tracker.logger.error("❌ CRITICAL: Output file is 0 bytes. Deleting corrupted output.")
|
|
os.remove(output_path)
|
|
raise Exception("Embedding failed: Output is empty.")
|
|
|
|
# 2. Significant Size Drop Check
|
|
if new_size < (original_size * 0.8):
|
|
tracker.logger.error(f"❌ CRITICAL: Output file is significantly smaller than source!")
|
|
os.remove(output_path)
|
|
raise Exception("Embedding failed: Suspicious file size reduction.")
|
|
|
|
# 3. Duration Mismatch Check (New)
|
|
if abs(original_duration - new_duration) > 1.0:
|
|
tracker.logger.error(f"❌ CRITICAL: Duration mismatch detected!")
|
|
os.remove(output_path)
|
|
raise Exception("Embedding failed: Duration mismatch > 1s.")
|
|
|
|
tracker.logger.info(f"Subtitles embedded successfully: {output_path}")
|
|
return True
|
|
else:
|
|
tracker.logger.error("Error: Embedding failed.")
|
|
return False
|