import sys from pathlib import Path from typing import Optional import os 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, 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) output_dir.mkdir(exist_ok=True) # final_output = output_dir / f"{Path(Quality_Video).stem}_ger.mkv" final_output = output_dir / f"{Path(Quality_Video).stem}.mkv" run([ "mkvmerge", "-o", str(final_output), str(Quality_Video), "--language", "0:deu", # assign German language "--track-name", "0:German (Opus Stereo)", str(audio_file) ]) return final_output def process_one(German_Video: Path, Quality_Video: Path, output_dir: Path, offset: Optional[int] = None) -> bool: steps = [ "Probe video", "Compute offset", "Extract pre beginning", "Extract german audio", "Extract post end", "Combine audio", "Mux", "Cleanup", "Final check", ] weights = [ 1, 3, 1, 2, 1, 2, 2, 1, 2, ] with tqdm(total=len(steps), desc=Quality_Video.name, leave=False) as pbar: for step, w in zip(steps, weights): pbar.set_description(step) if step == "Probe video": # --- Probe video --- fps, duration, _, vstart = probe_video(German_Video) fps2, duration2, _, vstart2 = probe_video(Quality_Video) if duration <= 0: raise ValueError("Duration of German_Video must be greater than 0") if duration2 <= 0: raise ValueError("Duration of Quality_Video must be greater than 0") if step == "Compute offset": # --- Offset --- if offset == None: offset0 = corelate_videos(Quality_Video, German_Video, fps2) else: offset0 = offset if offset0 > duration or offset0 > duration2: raise ValueError("Offset needs to be smaller than the video durations") pbar.set_postfix_str(f"Offset: {offset0}s") 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" # Clear old faulty files, ger is gearantied to be replaced if os.path.isfile(quality_audio_file1): os.remove(quality_audio_file1) if os.path.isfile(quality_audio_file2): os.remove(quality_audio_file2) ### 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: extract_audio_segment_file(Quality_Video, start2, end2, quality_audio_file1) except subprocess.CalledProcessError as e: raise BrokenPipeError( f"{e}\n" f"Trying to extract Audio from {Quality_Video} {start2}-{end2} " f"to {quality_audio_file2} to have a duration of {end2-start2}, " f"while the audio has a length of {get_audio_duration(Quality_Video)}" ) if step == "Extract german audio": ### Get the German Audio if offset0 < 0: start = vstart - offset0 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) except subprocess.CalledProcessError as e: raise BrokenPipeError( f"{e}\n" f"Trying to extract Audio from {German_Video} {start}-{end} " f"to {german_audio_file} to have a duration of {end-start}, " f"while the audio has a length of {get_audio_duration(German_Video)}" ) if step == "Extract post end": ### Get the end from the first original audio if it is longer than the german part 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 Exception as e: raise BrokenPipeError( f"{e}\n" f"Trying to extract Audio from {Quality_Video} {start2}-{end2} " f"to {quality_audio_file2} to have a duration of {end2-start2}, " f"while the audio has a length of {get_audio_duration(Quality_Video)}" ) if step == "Combine audio": # Combine the audio files if os.path.isfile(quality_audio_file1): 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): 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": # --- Mux trimmed audio --- final_file = mux_audio_into_output(Quality_Video, german_audio_file, output_dir) if step == "Cleanup": if os.path.isfile(quality_audio_file1): os.remove(quality_audio_file1) if os.path.isfile(quality_audio_file2): os.remove(quality_audio_file2) # pass if step == "Final check": try: files_acceptable, op_ed_offset, stream_num = check_intro_outro_sync(final_file) except: new_name = final_file.with_name(f"[NO CHAPTER]_{final_file.name}") final_file.rename(new_name) if os.path.isfile(german_audio_file): os.remove(path=german_audio_file) return False if files_acceptable and isinstance(op_ed_offset, float) and isinstance(stream_num, int): tqdm.write(f"✔ {final_file.name} | offset={op_ed_offset:.4f}") if np.abs(op_ed_offset) > (1/fps2) and os.path.isfile(german_audio_file): replace_with_shifted_mka(final_file, german_audio_file, stream_num, op_ed_offset) if os.path.isfile(german_audio_file): os.remove(path=german_audio_file) return True else: tqdm.write(f"✖ FAILED: {final_file.name}") if os.path.isfile(german_audio_file): os.remove(path=german_audio_file) new_name = final_file.with_name(f"[FAILED]_{final_file.name}") final_file.rename(new_name) return False pbar.update(w) def main(): output_dir = "./output_dir" 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"): # i =1 # if i: german_pattern = f"./dubbed_source/Season 01/Example Series S01E{i:02}.mkv" if os.path.isfile(german_pattern): German_Video = [german_pattern] else: German_Video = glob.glob(german_pattern) quality_pattern = f"./high_quality/S01/E{i:02}*" if os.path.isfile(quality_pattern): Quality_Video = [quality_pattern] else: Quality_Video = glob.glob(quality_pattern) # print(f"German_Video={German_Video}") # print(f"Quality_Video={Quality_Video}") if len(German_Video) != 1: raise ValueError(f"Error with German wildcard:\n{german_pattern}\n{German_Video}") if len(Quality_Video) != 1: raise ValueError(f"Error with Quality wildcard:\n{quality_pattern}\n{Quality_Video}") if not process_one(Path(German_Video[0]), Path(Quality_Video[0]), Path(output_dir)): fault_catch.append([Path(German_Video[0]), Path(Quality_Video[0])]) if fault_catch: tqdm.write("\nFailed Videos:") for ger, orig in fault_catch: tqdm.write(f"{ger} + {orig}") if __name__ == '__main__': if len(sys.argv) == 3: if not process_one(German_Video=Path(sys.argv[1]), Quality_Video=Path(sys.argv[2]), output_dir=Path("../output_dir")): print(f"Video could not be correctly combined") elif len(sys.argv) == 4: if not process_one(German_Video=Path(sys.argv[1]), Quality_Video=Path(sys.argv[2]), output_dir=Path(sys.argv[3])): print(f"Video could not be correctly combined") else: main()