honney/anime_dub_sync
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

A lot of bug fixes and edge cases

Browse Source
This commit is contained in:
Hannes committed 2026-03-22 03:20:45 +01:00
1 parent 58e66996cc
commit 004556c4a7
5 files changed
+232 -101
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

audio/__init__.py
+1
View File
@@ -8,6 +8,7 @@ from .audio_utils import (
get_audio_streams,
get_track_info,
shift_audio,
add_silence,
replace_with_shifted_mka
)
audio/audio_utils.py
+143 -60
View File
@@ -35,8 +35,9 @@ def extract_audio_segment_file(video: Path, start_time: float, end_time: float,
"-i", video,
"-ss", str(start_time),
"-to", str(end_time),
"-map", "0:a:0", # first audio track
"-c:a", "copy", # no re-encode
"-map", "0:a:0",
"-ac", "2", # force stereo
"-c:a", "aac", # re-encode
out_audio
])
@@ -152,32 +153,55 @@ def get_audio_duration(file_path: Path):
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.
"""
### V1
# 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 stick_together_audio(left_audio: Path, right_audio: Path, output_audio: Path) -> None:
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)
with tempfile.NamedTemporaryFile("w", delete=False, suffix=".txt") as f:
f.write(f"file '{left_audio.resolve()}'\n")
f.write(f"file '{right_audio.resolve()}'\n")
list_file = f.name
try:
run([
"ffmpeg",
"-y",
"-f", "concat",
"-safe", "0",
"-i", list_file,
"-c", "copy", # no re-encode
str(output_audio)
])
except Exception as e:
raise BrokenPipeError(f"{e}\n{f.readlines(-1)}")
def get_audio_streams(file: Path) -> List[int]:
"""
@@ -224,6 +248,9 @@ def get_track_info(file: Path, audioline: int):
return {"codec": "aac", "bitrate": "192k", "lang": "eng", "title": "Audio"}
def shift_audio(file: Path, audioline: int, offset: float):
"""
Shifts audio by offset sekonds to the right.
"""
if not file.is_file(): return
info = get_track_info(file, audioline)
@@ -258,7 +285,7 @@ def shift_audio(file: Path, audioline: int, offset: float):
"-b:a", str(info['bitrate']),
str(temp_mka)
]
subprocess.run(process_cmd, check=True)
run(process_cmd)
# --- STEP 2: Replace in MKV via mkvmerge ---
# --audio-tracks !ID excludes the old track
@@ -271,13 +298,71 @@ def shift_audio(file: Path, audioline: int, offset: float):
str(temp_mka)
]
subprocess.run(merge_cmd, check=True)
run(merge_cmd)
# Cleanup and Replace
shutil.move(str(output_mkv), str(file))
temp_mka.unlink()
print(f"Done: {file.name} updated physically.")
def add_silence(file: Path, offset: float) -> None:
"""
Hard-edits an MKA file by adding silence.
Positive offset: Adds silence at the beginning.
Negative offset: Adds silence at the end.
"""
if not file.is_file():
print(f"File not found: {file}")
return
if offset == 0:
print("Offset is 0, no silence to add.")
return
# Create a normal file path right next to the original file
new_file = file.parent / f"silenced_{file.name}"
# 1. Determine the filter
if offset > 0:
# Silence at the beginning: delay all channels
ms = int(offset * 1000)
audio_filter = f"adelay={ms}:all=1"
action_text = "beginning"
else:
# Silence at the end: pad with specific duration in seconds
sec = abs(offset)
audio_filter = f"apad=pad_dur={sec}"
action_text = "end"
# 2. Execute FFmpeg
cmd = [
"ffmpeg", "-y",
"-i", str(file),
"-af", audio_filter,
"-c:a", "libopus",
str(new_file)
]
# print(f"Adding {abs(offset)}s of silence to the {action_text} of {file.name}...")
result = subprocess.run(cmd, capture_output=True, text=True)
# 3. Verify and replace
if result.returncode == 0:
# Quick safety check to ensure FFmpeg didn't output an empty file
if new_file.exists() and new_file.stat().st_size > 0:
shutil.move(str(new_file), str(file))
# print(f"Successfully updated: {file.name}")
else:
print("Error: FFmpeg succeeded, but the output file is empty.")
if new_file.exists():
new_file.unlink()
else:
print(f"FFmpeg Error:\n{result.stderr}")
# Clean up the leftover file if the process failed
if new_file.exists():
new_file.unlink()
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.
@@ -291,43 +376,41 @@ def replace_with_shifted_mka(mkv_file: Path, mka_source: Path, audioline: int, o
# 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"
shifted_mka = mkv_file.parent / "shifted.mka"
output_mkv = mkv_file.parent / "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"
# --- 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)
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)
]
# --- 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)
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}")
# --- STEP 3: Cleanup and Overwrite ---
shutil.move(str(output_mkv), str(mkv_file))
# print(f"Success: Updated {mkv_file.name}")
main.py
+45 -9
View File
@@ -6,13 +6,15 @@ import glob
import numpy as np
import subprocess
from utils import run
from audio import extract_audio_segment_file, get_audio_duration, stick_together_audio, replace_with_shifted_mka
from audio import extract_audio_segment_file, get_audio_duration, stick_together_audio, replace_with_shifted_mka, add_silence
from video import probe_video, find_temporal_offset, corelate_videos
from mkv import check_intro_outro_sync
from tqdm import tqdm
DEBUG = False
MIN_END_DUR = 2.5
def mux_audio_into_output(Quality_Video, audio_file, output_dir):
"""Mux trimmed audio into Quality_Video and place final file in output_dir."""
output_dir = Path(output_dir)
@@ -82,6 +84,8 @@ def process_one(German_Video: Path, Quality_Video: Path, output_dir: Path, offse
if step == "Extract pre beginning":
german_audio_file = output_dir / f"{Path(Quality_Video).stem}_german_audio.mka"
# german_audio_file1 = output_dir / f"{Path(Quality_Video).stem}_german_audio1.mka"
# german_audio_file2 = output_dir / f"{Path(Quality_Video).stem}_german_audio2.mka"
quality_audio_file1 = output_dir / f"{Path(Quality_Video).stem}_original_audio1.mka"
quality_audio_file2 = output_dir / f"{Path(Quality_Video).stem}_original_audio2.mka"
@@ -94,6 +98,7 @@ def process_one(German_Video: Path, Quality_Video: Path, output_dir: Path, offse
### Get the first part till german starts (if german does not start before file)
start2 = vstart2
end2 = start2 + offset0
first_part = [start2, end2]
if end2 - start2 > 0.05:
if DEBUG: print(f"Cutting audio ({quality_audio_file1}) (from timestamp {start2:.6f}s) to timestamp {end2:.6f}s")
try:
@@ -114,6 +119,7 @@ def process_one(German_Video: Path, Quality_Video: Path, output_dir: Path, offse
else:
start = vstart
end = min(vstart + duration, start + duration2)
main_part = [start, end]
if DEBUG: print(f"Cutting audio ({german_audio_file}) (from timestamp {start:.6f}s) to timestamp {end:.6f}s")
try:
extract_audio_segment_file(German_Video, start, end, german_audio_file)
@@ -131,11 +137,12 @@ def process_one(German_Video: Path, Quality_Video: Path, output_dir: Path, offse
start2 = vstart2 + end
# end2 = start2 + (duration2 - duration - (n1 / fps))
end2 = vstart2 + get_audio_duration(Quality_Video)
last_part = [start2, end2]
if end2 - start2 > 0.5:
if DEBUG: print(f"Cutting audio ({quality_audio_file2}) (from timestamp {start2:.6f}s) to timestamp {end2:.6f}s")
try:
extract_audio_segment_file(Quality_Video, start2, end2, quality_audio_file2)
except subprocess.CalledProcessError as e:
except Exception as e:
raise BrokenPipeError(
f"{e}\n"
f"Trying to extract Audio from {Quality_Video} {start2}-{end2} "
@@ -147,9 +154,32 @@ def process_one(German_Video: Path, Quality_Video: Path, output_dir: Path, offse
# Combine the audio files
if os.path.isfile(quality_audio_file1):
stick_together_audio(quality_audio_file1, german_audio_file, german_audio_file)
try:
dur_q1 = get_audio_duration(quality_audio_file1)
if dur_q1:
stick_together_audio(quality_audio_file1, german_audio_file, german_audio_file)
else: raise
except:
try:
add_silence(german_audio_file, offset0)
except Exception as e:
raise BrokenPipeError(
f"{e}\n"
f"Trying to put together {quality_audio_file1} and {german_audio_file}"
f"into {german_audio_file}. First file start, end is {first_part} and second is {main_part}"
)
if os.path.isfile(quality_audio_file2):
stick_together_audio(german_audio_file, quality_audio_file2, german_audio_file)
try:
dur_q2 = get_audio_duration(quality_audio_file2)
if dur_q2 > MIN_END_DUR:
stick_together_audio(german_audio_file, quality_audio_file2, german_audio_file)
except:
raise BrokenPipeError(
f"{e}\n"
f"Trying to put together {german_audio_file} and {quality_audio_file2}"
f"into {german_audio_file}. First file start, end is {main_part} and second is {last_part}"
)
if step == "Mux":
@@ -162,6 +192,7 @@ def process_one(German_Video: Path, Quality_Video: Path, output_dir: Path, offse
os.remove(quality_audio_file1)
if os.path.isfile(quality_audio_file2):
os.remove(quality_audio_file2)
# pass
if step == "Final check":
try:
@@ -192,20 +223,25 @@ def process_one(German_Video: Path, Quality_Video: Path, output_dir: Path, offse
def main():
output_dir = "./output_dir"
start_num = 1
episode_num = 11
output_dir = "/home/honney/mount/Media/tmp/serien/ready/Classroom of the Elite/Season 01"
if not os.path.isdir(output_dir):
raise NotADirectoryError(f"Output directory is not a valid Directory: {output_dir}")
start_num = 2
episode_num = 24
fault_catch = []
episodes = range(start_num, episode_num + 1)
# episodes = [1]
for i in tqdm(episodes, desc="Episodes"):
german_pattern = f"Folder0/S01E{i:02}*.mkv"
# i =1
# if i:
german_pattern = f"/home/honney/mount/Media/tmp/serien/ger_low_quality/Classroom of the Elite (2017-2024) [imdbid-tt7263328]/Season 01/Classroom of the Elite S01E0{i:02}.mkv"
if os.path.isfile(german_pattern):
German_Video = [german_pattern]
else:
German_Video = glob.glob(german_pattern)
quality_pattern = f"Folder1/S01E{i:02}*.mkv"
quality_pattern = f"/home/honney/mount/Media/tmp/serien/high_quality/Classroom of the Elite/S01/E{i:02}*"
if os.path.isfile(quality_pattern):
Quality_Video = [quality_pattern]
else:
mkv/chapter.py
+23 -20
View File
@@ -35,32 +35,35 @@ from .intro_outro importme, duration) for each chapter."""
def get_intro_outro(file: Path, length: float = 90.0) -> List[Tuple[float, float]]:
try:
chapters = get_chapters(Path(file))
except:
raise AttributeError(f"{file} has no chapter Marks")
multiplier = 2
min_len = length / multiplier
max_len = length * multiplier
filtered = [
(start, dur) for start, dur in chapters
if min_len <= dur <= max_len
]
while len(filtered) > 2 and multiplier <= 1.1:
multiplier -= 0.1
multiplier = 2
min_len = length / multiplier
max_len = length * multiplier
filtered = [
(start, dur) for start, dur in chapters
if min_len <= dur <= max_len
]
if len(filtered) == 2:
return filtered
elif len(filtered) == 1:
### Found only a Intro or Outro
raise ValueError(f"Could only find one intro or outro:\n{filtered}")
elif len(filtered) == 0:
raise ValueError(f"Found no outro or intro:\n{filtered}")
else:
raise ValueError(f"Found too many intros/outros:\n{filtered}")
while len(filtered) > 2 and multiplier <= 1.1:
multiplier -= 0.1
min_len = length / multiplier
max_len = length * multiplier
filtered = [
(start, dur) for start, dur in chapters
if min_len <= dur <= max_len
]
if len(filtered) == 2:
return filtered
elif len(filtered) == 1:
### Found only a Intro or Outro
raise ValueError(f"Could only find one intro or outro:\n{filtered}")
elif len(filtered) == 0:
raise ValueError(f"Found no outro or intro:\n{filtered}")
else:
raise ValueError(f"Found too many intros/outros:\n{filtered}")
except:
raise AttributeError(f"{file} has no chapter Marks")
### TODO: Find more alternative ways to find intro and outro, possible: https://github.com/aniskip/aniskip-cli
### https://github.com/jonbarrow/open-anime-timestamps?utm_source=chatgpt.com
if __name__ == '__main__':
file = "/home/honney/projects/mkv/output_dir/08.mkv"
video/video_utils.py
+20 -12
View File
@@ -7,28 +7,36 @@ sys.path.append(str(Path(__file__).resolve().parent.parent))
from utils import run
### Normal Imports:
from subprocess import run, PIPE
from pathlib import Path
import json
def probe_video(path: Path) -> tuple[float, float, str, float]:
"""
Get video FPS, duration, codec, and stream start time.
Args:
path: To Video File
Returns:
fps, duration, codec, start_time
"""
import json
cmd = [
"ffprobe",
"-v", "error",
"-print_format", "json",
"-show_streams",
"-show_format",
path
str(path), # IMPORTANT
]
data = json.loads(run(cmd).stdout)
result = run(cmd, stdout=PIPE, stderr=PIPE, text=True)
if result.returncode != 0:
raise RuntimeError(f"ffprobe failed:\n{result.stderr}")
if not result.stdout.strip():
raise RuntimeError("ffprobe returned empty output")
try:
data = json.loads(result.stdout)
except json.JSONDecodeError as e:
print("RAW OUTPUT:\n", result.stdout) # debug
raise
duration = float(data["format"]["duration"])
for s in data["streams"]:
if s["codec_type"] == "video":
num, den = s["r_frame_rate"].split("/")