no reencode in mix

utils/merge_audio.py

@@ -1,45 +1,87 @@
-from pydub import AudioSegment
+import subprocess
+import pathlib
+import tempfile
 import os
 
 def merge_mp3_files(file_paths, output_filename, pause_ms=500):
+    """
+    Concatenate MP3s *without re‑encoding*.
+    Adds a silent gap (`pause_ms`) between clips by generating a
+    temporary silent MP3 at the same encoding params as the sources.
+    """
     if not file_paths:
-        print("Warning: No file paths provided for merging.")
+        print("Warning: no input files.")
         return None
 
-    valid_segments = []
-    for file_path in file_paths:
-        if file_path and os.path.exists(file_path) and os.path.getsize(file_path) > 0:
-            try:
-                segment = AudioSegment.from_mp3(file_path)
-                valid_segments.append(segment)
-            except Exception as e:
-                print(f"Error loading audio segment from {file_path}: {e}. Skipping this file.")
-        elif file_path:
-            print(f"Warning: File {file_path} is missing or empty. Skipping.")
-
-    if not valid_segments:
-        print("No valid audio segments found to merge.")
+    # keep only files that exist and are non‑empty
+    valid = [p for p in file_paths if p and os.path.exists(p) and os.path.getsize(p) > 0]
+    if not valid:
+        print("No valid audio segments found.")
         return None
 
-    combined_audio = valid_segments[0]
-    if len(valid_segments) > 1:
-        pause_segment = AudioSegment.silent(duration=max(0, pause_ms))
-        for segment in valid_segments[1:]:
-            combined_audio += pause_segment
-            combined_audio += segment
-
-    try:
-        combined_audio.export(
-            output_filename,
-            format="mp3",
-            bitrate="128k",
-            parameters=["-ac", "2", "-ar", "44100"]
+    concat_list = []
+
+    # optional pause: create one silent mp3 we can reuse
+    if pause_ms > 0:
+        silent_path = _make_silent_mp3(duration_ms=pause_ms,
+                                       template_mp3=valid[0])
+    else:
+        silent_path = None
+
+    for i, p in enumerate(valid):
+        concat_list.append(p)
+        if silent_path and i < len(valid) - 1:
+            concat_list.append(silent_path)
+
+    # ffmpeg concat demuxer expects a text file listing the parts
+    with tempfile.NamedTemporaryFile("w+", delete=False) as tf:
+        for part in concat_list:
+            tf.write(f"file '{pathlib.Path(part).as_posix()}'\n")
+        tf.flush()
+
+        subprocess.run(
+            [
+                "ffmpeg", "-y",
+                "-f", "concat", "-safe", "0",
+                "-i", tf.name,
+                "-c", "copy",                    # ← no re‑encode!
+                output_filename
+            ],
+            check=True,
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
         )
-        if os.path.exists(output_filename) and os.path.getsize(output_filename) > 0:
-            return output_filename
-        else:
-            print(f"Error: Merged file {output_filename} was not created or is empty after export.")
-            return None
-    except Exception as e:
-        print(f"Error exporting merged MP3 to {output_filename}: {e}")
-        return None
+    return output_filename
+
+
+def _make_silent_mp3(duration_ms: int, template_mp3: str) -> str:
+    """
+    Create a silent MP3 (CBR, same sample‑rate & channels as template)
+    so we can insert a pause without changing codecs later.
+    """
+    silent_path = tempfile.mktemp(suffix=".mp3")
+    # Extract params from the template
+    probe = subprocess.check_output(
+        ["ffprobe", "-v", "error", "-select_streams", "a:0",
+         "-show_entries", "stream=sample_rate,channels,bit_rate",
+         "-of", "default=nw=1:nk=1", template_mp3],
+        text=True
+    ).strip().splitlines()
+    sr, ch, br = probe    # e.g. "44100", "2", "128000"
+
+    subprocess.run(
+        [
+            "ffmpeg", "-y",
+            "-f", "lavfi",
+            "-i", f"anullsrc=r={sr}:cl=mono",
+            "-t", str(duration_ms / 1000),
+            "-ac", ch,
+            "-ar", sr,
+            "-b:a", br,
+            silent_path
+        ],
+        check=True,
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    return silent_path