diff --git a/cc_audio_transfer.py b/cc_audio_transfer.py new file mode 100644 index 0000000..8e2c82a --- /dev/null +++ b/cc_audio_transfer.py @@ -0,0 +1,269 @@ +from pathlib import Path +import subprocess +import numpy as np +from scipy.signal import correlate +import tempfile +import soundfile as sf +import matplotlib.pyplot as plt +import subprocess +import sys +import os +import glob +import shutil + +DEBUG = False + +def run(cmd): + """ + Run a subprocess command. + + Args: + cmd: Command and arguments as a list. + + Raises: + subprocess.CalledProcessError: If the command exits with a non-zero status. + """ + try: + result = subprocess.run( + cmd, + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True + ) + return result + except subprocess.CalledProcessError as e: + print(e.stdout) + raise + + +def extract_audio(file: Path, duration: float): + tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False) + + subprocess.run([ + "ffmpeg", + "-y", + "-i", str(file), + "-map", "0:a:0", # first audio stream + "-ac", "1", # mono + "-ar", "44100", # fixed sample rate + "-t", str(duration), # limit duration + tmp.name + ], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + + data, sr = sf.read(tmp.name) + return data, sr + + +def get_audio_offset(file1: Path, file2: Path, important_time: float): + sig1, sr1 = extract_audio(file1, important_time) + sig2, sr2 = extract_audio(file2, important_time) + + if sr1 != sr2: + raise ValueError("Sample rates do not match") + + # normalize signals + sig1 = sig1 - np.mean(sig1) + sig2 = sig2 - np.mean(sig2) + + corr = correlate(sig1, sig2, mode="full") + lag = np.argmax(corr) - (len(sig2) - 1) + + offset_seconds = lag / sr1 + return offset_seconds + +def probe_video(path: Path): + """Get video FPS, duration, codec, and stream start time.""" + import json + cmd = [ + "ffprobe", + "-v", "error", + "-print_format", "json", + "-show_streams", + "-show_format", + path + ] + data = json.loads(subprocess.run(cmd, capture_output=True, text=True, check=True).stdout) + + duration = float(data["format"]["duration"]) + for s in data["streams"]: + if s["codec_type"] == "video": + num, den = s["r_frame_rate"].split("/") + fps = float(num) / float(den) + codec = s["codec_name"] + start_time = float(s.get("start_time", 0)) + return fps, duration, codec, start_time + + raise RuntimeError("No video stream found") + +def trim_audio(German_Videout_mkv: 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", German_Videout_mkv, + "-ss", str(start_time), + "-to", str(end_time), + "-map", "0:a:0", # first audio track + "-c:a", "copy", # no re-encode + out_audio + ]) + +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 (2.0 aac)", + str(audio_file) + ]) + + return final_output + +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 process_one(German_Video: Path, Quality_Video: Path, output_dir: Path): + # --- 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") + + offset = get_audio_offset(Quality_Video, German_Video, 30) + print(f"Offset: {offset}s") + + if offset > duration or offset > duration2: + raise ValueError("Offset needs to be smaller than the video durations") + + # print(f"fps: {fps}") + # print(f"Cutting audio from frame {n1} (timestamp {start:.6f}s) to timestamp {end:.6f}s") + + # --- Trim only audio --- + german_audio_file = output_dir / f"{Path(Quality_Video).stem}_german_audio.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" + if os.path.isfile(quality_audio_file1): + os.remove(quality_audio_file1) + if os.path.isfile(quality_audio_file2): + os.remove(quality_audio_file2) + + start2 = vstart2 + end2 = start2 + offset + if end2 > start2: + if DEBUG: print(f"Cutting audio ({quality_audio_file1}) (from timestamp {start2:.6f}s) to timestamp {end2:.6f}s") + trim_audio(Quality_Video, start2, end2, quality_audio_file1) + + start = vstart + end = start + duration + if DEBUG: print(f"Cutting audio ({german_audio_file}) (from timestamp {start:.6f}s) to timestamp {end:.6f}s") + trim_audio(German_Video, start, end, german_audio_file) + # Get the end from the first original audio + # start2 = vstart2 + end + # end2 = start2 + (duration2 - duration - offset) + # if end2 > start2: + # if DEBUG: print(f"Cutting audio ({quality_audio_file2}) (from timestamp {start2:.6f}s) to timestamp {end2:.6f}s") + # trim_audio(Quality_Video, start2, end2, quality_audio_file2) + + if os.path.isfile(quality_audio_file1): + stick_together_audio(quality_audio_file1, german_audio_file, german_audio_file) + if os.path.isfile(quality_audio_file2): + stick_together_audio(german_audio_file, quality_audio_file2, german_audio_file) + + # --- Mux trimmed audio --- + final_file = mux_audio_into_output(Quality_Video, german_audio_file, output_dir) + + print(f"Final file with trimmed audio created at: {final_file}") + + print(f"Removing tmp Audiofiles") + if os.path.isfile(german_audio_file): + os.remove(german_audio_file) + if os.path.isfile(quality_audio_file1): + os.remove(quality_audio_file1) + if os.path.isfile(quality_audio_file2): + os.remove(quality_audio_file2) + +def main(): + output_dir = "/home/honney/projects/mkv/output_dir" + # for i in range(1, 14): + for i in [7, 8, 10, 11, 13, 16, 17, 19, 20, 22, 25]: + # --- German_Videouts --- + pattern = f"AnoHana The Flower We Saw That Day (2011-2011) [imdbid-tt1913273]/Season 01/AnoHana The Flower We Saw That Day S01E0{i:02}.mkv" + German_Video = glob.glob(pattern) + pattern = f"Anohana - The Flower We Saw That Day/S01/AnoHana The Flower We Saw That Day S01E0{i:02}.mkv" + Quality_Video = glob.glob(pattern) + + if len(German_Video) > 1: + raise ValueError("There are to many videos matching the German wildcard") + if len(Quality_Video) > 1: + raise ValueError("There are to many videos matching the Quality wildcard") + + process_one(Path(German_Video[0]), Path(Quality_Video[0]), Path(output_dir)) + sys.exit() + + +if __name__ == "__main__": + if len(sys.argv) == 3: + print(sys.argv) + print(f"German_Video={sys.argv[1]}") + print(f"Quality_Video={sys.argv[2]}") + process_one(German_Video=Path(sys.argv[1]), Quality_Video=Path(sys.argv[2]), output_dir=Path("./output_dir")) + elif len(sys.argv) == 4: + print(sys.argv) + print(f"German_Video={sys.argv[1]}") + print(f"Quality_Video={sys.argv[2]}") + process_one(German_Video=Path(sys.argv[1]), Quality_Video=Path(sys.argv[2]), output_dir=Path(sys.argv[3])) + else: + # print(len(sys.argv), sys.argv) + # sys.exit() + main() diff --git a/chapter_OP_ED_detection.py b/chapter_OP_ED_detection.py new file mode 100644 index 0000000..1341b0e --- /dev/null +++ b/chapter_OP_ED_detection.py @@ -0,0 +1,62 @@ +from pathlib import Path +from typing import Tuple, List, Optional +import subprocess +import json + +def run(cmd): + """ + Run a subprocess command. + + Args: + cmd: Command and arguments as a list. + + Raises: + subprocess.CalledProcessError: If the command exits with a non-zero status. + """ + try: + result = subprocess.run( + cmd, + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True + ) + return result + except subprocess.CalledProcessError as e: + print(e.stdout) + raise + + +def _get_chapters(file: Path) -> List[Tuple[float, float]]: + """Return list of (start_time, duration) for each chapter.""" + cmd = [ + "ffprobe", + "-i", str(file), + "-print_format", "json", + "-show_chapters", + "-loglevel", "error" + ] + + result = run(cmd) + data = json.loads(result.stdout) + + chapters = [] + for ch in data.get("chapters", []): + start = float(ch["start_time"]) + end = float(ch["end_time"]) + duration = end - start + chapters.append((start, duration)) + + return chapters + +if __name__ == '__main__': + file = "/home/honney/projects/mkv/output_dir/08.mkv" + chapters = _get_chapters(Path(file)) + time = 90 + min_len = time / 2 + max_len = time * 2 + filtered = [ + (start, dur) for start, dur in chapters + if min_len <= dur <= max_len + ] + print(filtered) \ No newline at end of file diff --git a/check_final_file.py b/check_final_file.py new file mode 100644 index 0000000..3bd0886 --- /dev/null +++ b/check_final_file.py @@ -0,0 +1,210 @@ +from pathlib import Path +from typing import Tuple, List, Optional +import json +import subprocess +import tempfile +import numpy as np +import soundfile as sf +from scipy.signal import correlate + +def run(cmd): + """ + Run a subprocess command. + + Args: + cmd: Command and arguments as a list. + + Raises: + subprocess.CalledProcessError: If the command exits with a non-zero status. + """ + try: + result = subprocess.run( + cmd, + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True + ) + return result + except subprocess.CalledProcessError as e: + print(e.stdout) + raise + +def _get_chapters(file: Path) -> List[Tuple[float, float]]: + """Return list of (start_time, duration) for each chapter.""" + cmd = [ + "ffprobe", + "-i", str(file), + "-print_format", "json", + "-show_chapters", + "-loglevel", "error" + ] + + result = run(cmd) + data = json.loads(result.stdout) + + chapters = [] + for ch in data.get("chapters", []): + start = float(ch["start_time"]) + end = float(ch["end_time"]) + duration = end - start + chapters.append((start, duration)) + + return chapters + + +def get_intro(file: Path, time: float) -> Tuple[float, float]: + """ + Find the chapter whose duration is closest to `time`, + considering only chapters with duration between time/2 and time*2. + Returns (start_time, duration). + """ + chapters = _get_chapters(file) + + if not chapters: + raise ValueError("No chapters found") + + # Filter by allowed duration range + min_len = time / 2 + max_len = time * 2 + filtered = [ + (start, dur) for start, dur in chapters + if min_len <= dur <= max_len + ] + + if not filtered: + raise ValueError("No chapters match duration constraints") + + # Sort by closeness to target time + filtered.sort(key=lambda x: abs(x[1] - time)) + + # Take the two closest + closest_two = filtered[:2] + + # Choose the better one (closest duration) + best = min(closest_two, key=lambda x: abs(x[1] - time)) + + return best + +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 extract_audio_segment( + file: Path, + intro: Tuple[float, float], + 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 + """ + start, duration = intro + + 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 extract_audio(file: Path, duration: float): + tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False) + + run([ + "ffmpeg", + "-y", + "-i", str(file), + "-map", "0:a:0", # first audio stream + "-ac", "1", # mono + "-ar", "44100", # fixed sample rate + "-t", str(duration), # limit duration + tmp.name + ]) + + data, sr = sf.read(tmp.name) + return data, sr + +def get_audio_offset(file1: Path, file2: Path, important_time: float): + sig1, sr1 = extract_audio(file1, important_time) + sig2, sr2 = extract_audio(file2, important_time) + + if sr1 != sr2: + raise ValueError("Sample rates do not match") + + # normalize signals + sig1 = sig1 - np.mean(sig1) + sig2 = sig2 - np.mean(sig2) + + corr = correlate(sig1, sig2, mode="full") + lag = np.argmax(corr) - (len(sig2) - 1) + + offset_seconds = lag / sr1 + return offset_seconds + +def check_final_file(video: Path, fps: float) -> bool: + try: + intro_duration = 90 + intro = get_intro(video, 90.0) + audio_streams = get_audio_streams(video) + extract_path1 = extract_audio_segment(video, intro, audio_streams[0]) + extract_path2 = extract_audio_segment(video, intro, audio_streams[-1]) + offset = get_audio_offset(extract_path1, extract_path2, intro[-1]) + if np.abs(offset) < (1/fps): + return True + else: + print(f"NOT CORRECT!!!!!!!!!!!!!!!!!!!!!!!!: {offset}, {1/fps}") + return False + except: + print("Chapter detection not working") + return False \ No newline at end of file diff --git a/detect_chapter.py b/detect_chapter.py new file mode 100644 index 0000000..d71121a --- /dev/null +++ b/detect_chapter.py @@ -0,0 +1,298 @@ +from pathlib import Path +from typing import Tuple, List, Optional +from scipy.signal import correlate +import soundfile as sf +import numpy as np +import subprocess +import json +import tempfile +import os + + +def run(cmd): + """ + Run a subprocess command. + + Args: + cmd: Command and arguments as a list. + + Raises: + subprocess.CalledProcessError: If the command exits with a non-zero status. + """ + try: + result = subprocess.run( + cmd, + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True + ) + return result + except subprocess.CalledProcessError as e: + print(e.stdout) + raise + +def probe_video(path: Path): + """Get video FPS, duration, codec, and stream start time.""" + import json + cmd = [ + "ffprobe", + "-v", "error", + "-print_format", "json", + "-show_streams", + "-show_format", + path + ] + data = json.loads(run(cmd).stdout) + + duration = float(data["format"]["duration"]) + for s in data["streams"]: + if s["codec_type"] == "video": + num, den = s["r_frame_rate"].split("/") + fps = float(num) / float(den) + codec = s["codec_name"] + start_time = float(s.get("start_time", 0)) + return fps, duration, codec, start_time + + raise RuntimeError("No video stream found") + +def _get_chapters(file: Path) -> List[Tuple[float, float]]: + """Return list of (start_time, duration) for each chapter.""" + cmd = [ + "ffprobe", + "-i", str(file), + "-print_format", "json", + "-show_chapters", + "-loglevel", "error" + ] + + result = run(cmd) + data = json.loads(result.stdout) + + chapters = [] + for ch in data.get("chapters", []): + start = float(ch["start_time"]) + end = float(ch["end_time"]) + duration = end - start + chapters.append((start, duration)) + + return chapters + + +def get_intro(file: Path, time: float) -> Tuple[float, float]: + """ + Find the chapter whose duration is closest to `time`, + considering only chapters with duration between time/2 and time*2. + Returns (start_time, duration). + """ + chapters = _get_chapters(file) + + if not chapters: + raise ValueError("No chapters found") + + # Filter by allowed duration range + min_len = time / 2 + max_len = time * 2 + filtered = [ + (start, dur) for start, dur in chapters + if min_len <= dur <= max_len + ] + + if not filtered: + raise ValueError("No chapters match duration constraints") + + # Sort by closeness to target time + filtered.sort(key=lambda x: abs(x[1] - time)) + + # Take the two closest + closest_two = filtered[:2] + + # Choose the better one (closest duration) + best = min(closest_two, key=lambda x: abs(x[1] - time)) + + return best + + +def extract_audio_segment( + file: Path, + intro: Tuple[float, float], + 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 + """ + start, duration = intro + + 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_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 play_audio_segment(file: Path, intro: Tuple[float, float]) -> None: + """ + Extract and play audio from (start, duration). + """ + audio_file = extract_audio_segment(file, intro) + + try: + run( + ["ffplay", "-nodisp", "-autoexit", str(audio_file)] + ) + finally: + os.remove(audio_file) + +def extract_audio(file: Path, duration: float): + tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False) + + run([ + "ffmpeg", + "-y", + "-i", str(file), + "-map", "0:a:0", # first audio stream + "-ac", "1", # mono + "-ar", "44100", # fixed sample rate + "-t", str(duration), # limit duration + tmp.name + ]) + + data, sr = sf.read(tmp.name) + return data, sr + +def get_audio_offset(file1: Path, file2: Path, important_time: float): + sig1, sr1 = extract_audio(file1, important_time) + sig2, sr2 = extract_audio(file2, important_time) + + if sr1 != sr2: + raise ValueError("Sample rates do not match") + + # normalize signals + sig1 = sig1 - np.mean(sig1) + sig2 = sig2 - np.mean(sig2) + + corr = correlate(sig1, sig2, mode="full") + lag = np.argmax(corr) - (len(sig2) - 1) + + offset_seconds = lag / sr1 + return offset_seconds + +def main(video: Path) -> bool: + + intro_duration = 90 + print("finding intro") + intro = get_intro(video, 90.0) + print("geting audio streams") + audio_streams = get_audio_streams(video) + print("extract first intro") + extract_path1 = extract_audio_segment(video, intro, audio_streams[0]) + print("extract second intro") + extract_path2 = extract_audio_segment(video, intro, audio_streams[-1]) + + # # or directly play: + # time = (0, 20) + # play_audio_segment(extract_path1, time) + # play_audio_segment(extract_path2, time) + + fps, _, _, _ = probe_video(video) + + print("Calculate") + + offset = get_audio_offset(extract_path1, extract_path2, intro[-1]-10) + + if np.abs(offset) < (1/fps): + print(f"no offset: {offset}") + return True + else: + print(f"NOT CORRECT!!!!!!!!!!!!!!!!!!!!!!!!: {offset}, {1/fps}") + return False + +extensions=("mp4", "mkv", "avi", "mov", "wmv", "flv", "webm", "lrv", "gif") + +def check_video_ext(video: Path) -> bool: + if video.suffix.strip(".").lower() in extensions: + return True + return False + +if __name__ == '__main__': + import sys + # sys.argv.remove(sys.argv[0]) + if len(sys.argv) == 2: + print("One file") + if os.path.isfile(sys.argv[1]): + main(Path(sys.argv[1])) + else: raise + elif len(sys.argv) > 2: + sys.argv.remove(sys.argv[0]) + error_files = [] + for str_path in sys.argv: + video = Path(str_path) + if (check_video_ext(video)): + print(f"{video}") + working = main(video) + if not working: + error_files.append(video) + else: print(f"Not a compatabile Video: {video}") + if error_files != []: + print(f"These files are broken:") + for file in error_files: + print(file) + else: + print(f"Use this script like this:\npython detect_chapter.py /path/to/file") \ No newline at end of file diff --git a/test1.py b/test1.py new file mode 100644 index 0000000..4155d4a --- /dev/null +++ b/test1.py @@ -0,0 +1,135 @@ +import subprocess +import json +import shutil +from pathlib import Path +from typing import List + +def run(cmd: subprocess._CMD) -> subprocess.CompletedProcess[str]: + """ + Run a subprocess command. + + Args: + cmd: Command and arguments as a list. + + Raises: + subprocess.CalledProcessError: If the command exits with a non-zero status. + """ + try: + result = subprocess.run( + cmd, + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True + ) + return result + except subprocess.CalledProcessError as e: + print(e.stdout) + raise + +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): + """ + Shifts audio by offset sekonds to the right. + """ + 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.") + +# Example usage: +file_path = Path("/home/honney/mount/Media/tmp/serien/ready/Charlotte/Season 01/[NO CHAPTER]_E12.mkv") +audio_streams = get_audio_streams(file_path) +shift_audio(file_path, audioline=audio_streams[-1], offset=(1/23.98)) # Shifts the 2nd audio track right by 1.5 seconds \ No newline at end of file diff --git a/test2.py b/test2.py new file mode 100644 index 0000000..0da1196 --- /dev/null +++ b/test2.py @@ -0,0 +1,185 @@ +import subprocess +import tempfile +import os +from pathlib import Path +from typing import List, Tuple, Optional + +import numpy as np +from scipy.io import wavfile + +class ConfidenceError(Exception): + """Raised when an area is found but the energy difference is too low to be confident.""" + pass + +class MissingDetectionError(Exception): + """Raised when the script fails to find both an intro and an outro.""" + pass + +def get_video_duration(file: Path) -> float: + """Uses ffprobe to get the total duration of the video.""" + cmd = [ + "ffprobe", + "-v", "error", + "-show_entries", "format=duration", + "-of", "default=noprint_wrappers=1:nokey=1", + str(file) + ] + result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) + try: + return float(result.stdout.strip()) + except ValueError: + raise RuntimeError(f"Could not determine duration for {file}") + +def extract_audio_segment_tmp( + file: Path, + start_or_section: float | tuple[float, float], + end: Optional[float] = None, + audio_stream: Optional[int] = None +) -> Path: + """Extracts audio using FFmpeg and saves it to a temporary WAV file.""" + 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]) + 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) + ] + + if audio_stream is not None: + cmd += ["-map", f"0:a:{audio_stream}"] + else: + cmd += ["-map", "0:a:0"] + + cmd += [ + "-vn", + "-acodec", "pcm_s16le", + "-ar", "44100", + "-ac", "2", + str(tmp_path) + ] + + subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=True) + return tmp_path + +def find_high_power_segment(wav_path: Path, expected_length: float, confidence_threshold: float = 1.0) -> float: + rate, data = wavfile.read(wav_path) + + if len(data.shape) > 1: + data = data.mean(axis=1) + + data = data.astype(np.float32) / (np.max(np.abs(data)) + 1e-9) + + chunk_size = rate + num_chunks = len(data) // chunk_size + + # Calculate RMS (Volume) and Standard Deviation (Variation) for every second + powers = np.zeros(num_chunks) + variations = np.zeros(num_chunks) + + for i in range(num_chunks): + chunk = data[i * chunk_size : (i + 1) * chunk_size] + powers[i] = np.sqrt(np.mean(chunk**2)) + variations[i] = np.std(chunk) + + window_size = int(expected_length) + if len(powers) < window_size: + raise ValueError("Audio segment too short.") + + # SCORING ALGORITHM: + # We want High Power and Low Variation (that "solid block" look) + # We invert variation so that a small variation results in a higher score. + scores = [] + for i in range(len(powers) - window_size + 1): + win_power = np.mean(powers[i : i + window_size]) + win_var = np.mean(variations[i : i + window_size]) + + # The score increases with power and decreases as the variation increases + # We add a small constant to win_var to avoid division by zero + score = win_power / (win_var + 0.05) + scores.append(score) + + best_idx = np.argmax(scores) + + # Confidence check: is our best score significantly better than the average score? + avg_score = np.mean(scores) + if scores[best_idx] < avg_score * confidence_threshold: + raise ConfidenceError("Detection unreliable: no distinct high-power constant block found.") + + return float(best_idx) + +def get_intro_outro(file: Path, length: float = 90.0) -> List[Tuple[float, float]]: + """ + Finds the intro and outro of a video file by analyzing audio power peaks. + Searches the first 8 minutes and the last 8 minutes. + """ + results = [] + total_duration = get_video_duration(file) + search_window = 8 * 60.0 # 8 minutes + + # --- 1. FIND INTRO --- + intro_search_dur = min(search_window, total_duration) + intro_wav = extract_audio_segment_tmp(file, 0.0, intro_search_dur) + + try: + intro_start_rel = find_high_power_segment(intro_wav, length) + results.append((intro_start_rel, length)) + except ConfidenceError: + intro_wav.unlink() + raise ConfidenceError("Could not confidently detect the Intro.") + finally: + if intro_wav.exists(): + intro_wav.unlink() # Cleanup temp file + + # --- 2. FIND OUTRO --- + outro_search_start = max(0.0, total_duration - search_window) + outro_search_dur = total_duration - outro_search_start + outro_wav = extract_audio_segment_tmp(file, outro_search_start, total_duration) + + try: + outro_start_rel = find_high_power_segment(outro_wav, length) + # Add the 8-minute offset back to get the absolute timestamp in the video + outro_start_abs = outro_search_start + outro_start_rel + results.append((outro_start_abs, length)) + except ConfidenceError: + outro_wav.unlink() + raise ConfidenceError("Could not confidently detect the Outro.") + finally: + if outro_wav.exists(): + outro_wav.unlink() # Cleanup temp file + + # Ensure we got both + if len(results) < 2: + raise MissingDetectionError("Failed to find both an intro and an outro.") + + return results + +# --- Example Usage --- +if __name__ == "__main__": + try: + # Requires: pip install numpy scipy + file_path = Path("/home/honney/mount/Media/tmp/serien/ready/Charlotte/Season 01/[NO CHAPTER]_E04.mkv") + if file_path.exists(): + timestamps = get_intro_outro(file_path, length=90.0) + print("Found Intro and Outro [Start, Duration]:", timestamps) + except Exception as e: + print(f"Error: {e}") \ No newline at end of file diff --git a/test3.py b/test3.py new file mode 100644 index 0000000..41b81ed --- /dev/null +++ b/test3.py @@ -0,0 +1,81 @@ +import subprocess +from typing import Optional, List +from pathlib import Path +import shutil + +def run(cmd: subprocess._CMD) -> subprocess.CompletedProcess[str]: + """ + Run a subprocess command. + + Args: + cmd: Command and arguments as a list. + + Raises: + subprocess.CalledProcessError: If the command exits with a non-zero status. + """ + try: + result = subprocess.run( + cmd, + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True + ) + return result + except subprocess.CalledProcessError as e: + print(e.stdout) + raise + +# 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) +import tempfile + +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) + + 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 + + run([ + "ffmpeg", + "-y", + "-f", "concat", + "-safe", "0", + "-i", list_file, + "-c", "copy", # no re-encode + str(output_audio) + ]) + +file1 = Path("/home/honney/mount/Media/tmp/serien/ready/Clannad (2007) [tmdb-24835]/Season 02/Clannad Season 02 E01_german_audio.mka") +file2 = Path("/home/honney/mount/Media/tmp/serien/ready/Clannad (2007) [tmdb-24835]/Season 02/Clannad Season 02 E01_original_audio2.mka") +out = Path("/home/honney/mount/Media/tmp/serien/ready/Clannad (2007) [tmdb-24835]/Season 02/Clannad Season 02 E01_german_audio2.mka") + +stick_together_audio(file1, file2, out) \ No newline at end of file diff --git a/test4.py b/test4.py new file mode 100644 index 0000000..49175ff --- /dev/null +++ b/test4.py @@ -0,0 +1,33 @@ +from pathlib import Path +import subprocess + +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 + if not result: + raise ValueError(f"Could not extract length") + for i in range(len(result)): + res += float(result[i])*(60**i) + return res + +printget_audio_duration(Path("/home/honney/mount/Media/tmp/serien/ready/Classroom of the Elite/Season 01/E01_original_audio2.mka")) \ No newline at end of file diff --git a/test5.py b/test5.py new file mode 100644 index 0000000..977b3cf --- /dev/null +++ b/test5.py @@ -0,0 +1,54 @@ +import subprocess +import shutil +from pathlib import Path + +def shift_mka(file: Path, offset: float) -> None: + """ + Hard-edits an MKA file by shifting audio data. + Positive offset: Adds silence at start (Delay). + Negative offset: Trims the beginning (Advance). + """ + if not file.is_file(): + print(f"File not found: {file}") + return + + # Create a normal file path right next to the original file + # If the file is "/folder/audio.mka", this becomes "/folder/shifted_audio.mka" + new_file = file.parent / f"shifted_{file.name}" + + # 1. Determine the filter + if offset > 0: + # Delay all channels by X milliseconds + ms = int(offset * 1000) + audio_filter = f"adelay={ms}:all=1" + elif offset < 0: + # Trim the start and reset timestamps to 0 + audio_filter = f"atrim=start={abs(offset)},asetpts=PTS-STARTPTS" + else: + print("Offset is 0, no shift needed.") + return + + # 2. Execute FFmpeg + cmd = [ + "ffmpeg", "-y", + "-i", str(file), + "-af", audio_filter, + "-c:a", "libopus", + str(new_file) + ] + + print(f"Processing shift: {offset}s...") + + result = subprocess.run(cmd, capture_output=True, text=True) + + if result.returncode == 0: + # Move the new file over the original file (overwriting it) + shutil.move(str(new_file), str(file)) + print(f"Successfully updated: {file.name}") + else: + print(f"FFmpeg Error:\n{result.stderr}") + # Clean up the leftover file if the process failed + if new_file.exists(): + new_file.unlink() + +shift_mka(Path("/home/honney/mount/Media/tmp/serien/ready/Classroom of the Elite/Season 01/E01_german_audio.mka"), 0.9592916666666667) \ No newline at end of file