V 1.0 - Only Prints
This commit is contained in:
Hannes committed
commit
5ac5b6615b
15 files changed
+1045
No files matched your search
@@ -0,0 +1,333 @@
|
||||
# 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}")
|
||||
Reference in new issue
Block a user