333 lines
11 KiB
Python
333 lines
11 KiB
Python
# for import needed
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.append(str(Path(__file__).resolve().parent.parent))
|
|
|
|
from utils import run
|
|
### Normal Imports:
|
|
import subprocess
|
|
import shutil
|
|
import tempfile
|
|
from typing import Tuple, Optional, overload, Union, List
|
|
import json
|
|
|
|
def extract_audio_segment_file(video: Path, start_time: float, end_time: float, out_audio: Path):
|
|
"""
|
|
Extract a segment of the first audio track from a video file.
|
|
|
|
Uses FFmpeg to copy the selected audio range without re-encoding,
|
|
which makes the operation fast and preserves the original audio quality.
|
|
|
|
Args:
|
|
German_Videout_mkv: Path to the source video file.
|
|
start_time: Start time of the audio segment (seconds or FFmpeg time format).
|
|
end_time: End time of the audio segment.
|
|
out_audio: Path to the output audio file.
|
|
|
|
Notes:
|
|
The function selects the first audio stream (`0:a:0`) and copies it
|
|
directly (`-c:a copy`), meaning the codec and quality remain unchanged.
|
|
"""
|
|
run([
|
|
"ffmpeg",
|
|
"-y",
|
|
"-i", video,
|
|
"-ss", str(start_time),
|
|
"-to", str(end_time),
|
|
"-map", "0:a:0", # first audio track
|
|
"-c:a", "copy", # no re-encode
|
|
out_audio
|
|
])
|
|
|
|
@overload
|
|
def extract_audio_segment_tmp(
|
|
file: Path,
|
|
start_or_section: float,
|
|
end: float,
|
|
audio_stream: Optional[int] = None
|
|
) -> Path: ...
|
|
|
|
# Overload 2: Using a tuple for the section
|
|
@overload
|
|
def extract_audio_segment_tmp(
|
|
file: Path,
|
|
start_or_section: tuple[float, float],
|
|
end: None = None,
|
|
audio_stream: Optional[int] = None
|
|
) -> Path: ...
|
|
|
|
# Implementation
|
|
def extract_audio_segment_tmp(
|
|
file: Path,
|
|
start_or_section: float | tuple[float, float],
|
|
end: Optional[float] = None,
|
|
audio_stream: Optional[int] = None
|
|
) -> Path:
|
|
"""
|
|
Extract audio using (start, duration) from get_intro.
|
|
|
|
Parameters:
|
|
file: input video file
|
|
intro: (start, duration)
|
|
audio_stream: index of audio stream (0 = first audio track, etc.)
|
|
|
|
Returns:
|
|
Path to temporary audio file
|
|
"""
|
|
# 1. Handle Tuple Input: (start, end)
|
|
if isinstance(start_or_section, tuple):
|
|
if len(start_or_section) != 2:
|
|
raise ValueError("Tuple must strictly be (start, end)")
|
|
start = float(start_or_section[0])
|
|
duration = float(start_or_section[1])
|
|
|
|
# 2. Handle Float Input: start, end
|
|
elif isinstance(start_or_section, (int, float)):
|
|
if end is None:
|
|
raise ValueError("The 'end' parameter must be provided when using a separate start float.")
|
|
start = float(start_or_section)
|
|
end_time = float(end)
|
|
duration = end_time - start
|
|
|
|
else:
|
|
raise TypeError("Invalid type for start_or_section. Expected float or tuple.")
|
|
|
|
if duration <= 0:
|
|
raise ValueError("Duration must be above 0")
|
|
|
|
tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
|
|
tmp_path = Path(tmp.name)
|
|
tmp.close()
|
|
|
|
cmd = [
|
|
"ffmpeg",
|
|
"-y",
|
|
"-i", str(file),
|
|
"-ss", str(start),
|
|
"-t", str(duration),
|
|
]
|
|
|
|
# Select specific audio stream if provided
|
|
if audio_stream is not None:
|
|
cmd += ["-map", f"0:a:{audio_stream}"]
|
|
else:
|
|
cmd += ["-map", "0:a:0"] # default: first audio track
|
|
|
|
cmd += [
|
|
"-vn",
|
|
"-acodec", "pcm_s16le",
|
|
"-ar", "44100",
|
|
"-ac", "2",
|
|
str(tmp_path)
|
|
]
|
|
|
|
run(cmd)
|
|
return tmp_path
|
|
|
|
def get_audio_duration(file_path: Path):
|
|
"""
|
|
Get the actual duration of the first audio stream by asking ffprobe
|
|
to count packets/frames if the stream metadata is missing.
|
|
Returns duration in seconds.
|
|
"""
|
|
|
|
# As a last resort, use the format duration
|
|
result = subprocess.run([
|
|
"ffprobe",
|
|
"-v", "error",
|
|
"-select_streams", "a:0",
|
|
"-show_entries", "stream_tags=DURATION",
|
|
"-of", "default=noprint_wrappers=1:nokey=1",
|
|
file_path
|
|
],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True
|
|
)
|
|
result = result.stdout.strip().split(":")[::-1]
|
|
# if DEBUG: print(f"audio duration: {result}")
|
|
res = 0
|
|
for i in range(len(result)):
|
|
res += float(result[i])*(60**i)
|
|
return res
|
|
|
|
def stick_together_audio(left_audio: Path, right_audio: Path, output_audio: Path) -> None:
|
|
"""
|
|
Concatenate two audio files so that left_audio plays directly
|
|
before right_audio in the output file.
|
|
"""
|
|
|
|
left_audio = Path(left_audio)
|
|
right_audio = Path(right_audio)
|
|
output_audio = Path(output_audio)
|
|
|
|
# Use stem instead of prefix to create a temporary file
|
|
output_audio_tmp = output_audio.parent / (output_audio.stem + "_temp" + output_audio.suffix)
|
|
|
|
run([
|
|
"ffmpeg",
|
|
"-y",
|
|
"-i", str(left_audio),
|
|
"-i", str(right_audio),
|
|
"-filter_complex", "[0:a][1:a]concat=n=2:v=0:a=1[a]",
|
|
"-map", "[a]",
|
|
"-c:a", "opus", # or "aac" if you prefer
|
|
"-strict", "-2",
|
|
str(output_audio_tmp)
|
|
])
|
|
|
|
shutil.move(output_audio_tmp, output_audio)
|
|
|
|
def get_audio_streams(file: Path) -> List[int]:
|
|
"""
|
|
Return a list of available audio stream indices (0-based for ffmpeg mapping).
|
|
|
|
Example return:
|
|
[0, 1] -> corresponds to 0:a:0 and 0:a:1
|
|
"""
|
|
cmd = [
|
|
"ffprobe",
|
|
"-i", str(file),
|
|
"-select_streams", "a",
|
|
"-show_entries", "stream=index",
|
|
"-of", "json",
|
|
"-loglevel", "error"
|
|
]
|
|
|
|
result = run(cmd)
|
|
data = json.loads(result.stdout)
|
|
|
|
# ffprobe gives absolute stream indices, but we want relative audio indices
|
|
streams = data.get("streams", [])
|
|
|
|
# Convert to 0-based audio stream numbering
|
|
return list(range(len(streams)))
|
|
|
|
def get_track_info(file: Path, audioline: int):
|
|
"""Gets codec, bitrate, language, and title."""
|
|
cmd = [
|
|
"ffprobe", "-v", "error", "-select_streams", f"a:{audioline}",
|
|
"-show_entries", "stream=codec_name,bit_rate:stream_tags=language,title",
|
|
"-of", "json", str(file)
|
|
]
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
data = json.loads(result.stdout)
|
|
if "streams" in data and len(data["streams"]) > 0:
|
|
s = data["streams"][0]
|
|
return {
|
|
"codec": s.get("codec_name", "aac"),
|
|
"bitrate": s.get("bit_rate", "192k"),
|
|
"lang": s.get("tags", {}).get("language", "eng"),
|
|
"title": s.get("tags", {}).get("title", f"Track {audioline}")
|
|
}
|
|
return {"codec": "aac", "bitrate": "192k", "lang": "eng", "title": "Audio"}
|
|
|
|
def shift_audio(file: Path, audioline: int, offset: float):
|
|
if not file.is_file(): return
|
|
|
|
info = get_track_info(file, audioline)
|
|
# mkvmerge uses 0 for video, usually 1+ for audio.
|
|
# We need the MKV Toolnix Track ID.
|
|
mkv_id = audioline + 1
|
|
|
|
temp_mka = file.parent / "temp_shifted.mka"
|
|
output_mkv = file.parent / "temp_output.mkv"
|
|
|
|
# --- STEP 1: Physical Shift via Filter ---
|
|
# We convert offset to milliseconds for the 'adelay' filter
|
|
ms = int(abs(offset) * 1000)
|
|
|
|
if offset > 0:
|
|
# RIGHT SHIFT: Delay start, then trim the end to keep original length
|
|
# 'all' applies delay to all channels
|
|
filter_str = f"adelay={ms}|{ms}|{ms}|{ms}|{ms}|{ms},atrim=0,atrim=duration=base" # Simplified logic below
|
|
# For reliability, we use a simpler pad/trim approach:
|
|
filter_chain = f"adelay={ms}:all=1"
|
|
else:
|
|
# LEFT SHIFT: Skip the beginning
|
|
filter_chain = f"atrim=start={abs(offset)}"
|
|
|
|
# Process to MKA
|
|
# We use the same codec and bitrate detected earlier
|
|
process_cmd = [
|
|
"ffmpeg", "-y", "-i", str(file),
|
|
"-map", f"0:a:{audioline}",
|
|
"-af", filter_chain,
|
|
"-c:a", info['codec'],
|
|
"-b:a", str(info['bitrate']),
|
|
str(temp_mka)
|
|
]
|
|
subprocess.run(process_cmd, check=True)
|
|
|
|
# --- STEP 2: Replace in MKV via mkvmerge ---
|
|
# --audio-tracks !ID excludes the old track
|
|
# The new MKA is added at the end
|
|
merge_cmd = [
|
|
"mkvmerge", "-o", str(output_mkv),
|
|
"--audio-tracks", f"!{mkv_id}", str(file),
|
|
"--language", f"0:{info['lang']}",
|
|
"--track-name", f"0:{info['title']}",
|
|
str(temp_mka)
|
|
]
|
|
|
|
subprocess.run(merge_cmd, check=True)
|
|
|
|
# Cleanup and Replace
|
|
shutil.move(str(output_mkv), str(file))
|
|
temp_mka.unlink()
|
|
print(f"Done: {file.name} updated physically.")
|
|
|
|
def replace_with_shifted_mka(mkv_file: Path, mka_source: Path, audioline: int, offset: float):
|
|
"""
|
|
Physically shifts an existing MKA and replaces a track in the MKV.
|
|
- Codec: Re-encoded to Opus (to ensure physical shift compatibility)
|
|
- Metadata: Hardcoded to German / German (Opus Stereo)
|
|
"""
|
|
if not mkv_file.is_file() or not mka_source.is_file():
|
|
print("Error: Input files not found.")
|
|
return
|
|
|
|
# mkvmerge IDs: 0 is usually video, so audioline 0 is track 1, etc.
|
|
mkv_track_id = audioline + 1
|
|
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
temp_path = Path(temp_dir)
|
|
shifted_mka = temp_path / "shifted.mka"
|
|
output_mkv = temp_path / "final_output.mkv"
|
|
|
|
# --- STEP 1: Physical Shift ---
|
|
# Positive offset: adelay (adds real silence packets)
|
|
# Negative offset: atrim (removes packets) + asetpts (resets timestamps)
|
|
ms = int(abs(offset) * 1000)
|
|
if offset > 0:
|
|
filter_chain = f"adelay={ms}:all=1"
|
|
else:
|
|
filter_chain = f"atrim=start={abs(offset)},asetpts=PTS-STARTPTS"
|
|
|
|
print(f"Physically shifting {mka_source.name}...")
|
|
process_cmd = [
|
|
"ffmpeg", "-y", "-i", str(mka_source),
|
|
"-af", filter_chain,
|
|
"-c:a", "libopus", # Set to Opus as per your track name
|
|
"-b:a", "160k", # Standard high-quality bitrate for stereo Opus
|
|
str(shifted_mka)
|
|
]
|
|
subprocess.run(process_cmd, check=True, capture_output=True)
|
|
|
|
# --- STEP 2: Replace track in MKV ---
|
|
# --audio-tracks !ID excludes the old line from the original MKV
|
|
print(f"Replacing track {audioline} in {mkv_file.name}...")
|
|
merge_cmd = [
|
|
"mkvmerge", "-o", str(output_mkv),
|
|
"--audio-tracks", f"!{mkv_track_id}", str(mkv_file),
|
|
"--language", "0:ger",
|
|
"--track-name", "0:German (Opus Stereo)",
|
|
str(shifted_mka)
|
|
]
|
|
|
|
subprocess.run(merge_cmd, check=True, capture_output=True)
|
|
|
|
# --- STEP 3: Cleanup and Overwrite ---
|
|
shutil.move(str(output_mkv), str(mkv_file))
|
|
print(f"Success: Updated {mkv_file.name}") |