add app.py

app.py

@@ -0,0 +1,179 @@
+import gradio as gr
+import io, os, uuid, zipfile, tempfile, subprocess
+from pydub import AudioSegment
+from pydub.silence import split_on_silence
+
+# ---------- helpers ----------
+def _load(file_or_bytes):
+    if isinstance(file_or_bytes, (bytes, bytearray)):
+        return AudioSegment.from_file(io.BytesIO(file_or_bytes))
+    if hasattr(file_or_bytes, "read"):
+        return AudioSegment.from_file(file_or_bytes)
+    return AudioSegment.from_file(file_or_bytes)
+
+def _export(seg: AudioSegment, fmt="mp3") -> io.BytesIO:
+    buf = io.BytesIO()
+    seg.export(buf, format=fmt)
+    buf.seek(0)
+    return buf
+
+def remove_silence(seg: AudioSegment, keep_ms=50, min_silence_ms=100, thresh_db=-45):
+    chunks = split_on_silence(
+        seg,
+        min_silence_len=int(min_silence_ms),
+        silence_thresh=float(thresh_db),
+        keep_silence=int(keep_ms),
+    )
+    return sum(chunks, AudioSegment.silent(duration=0)) if chunks else seg
+
+def trim_to_seconds(seg: AudioSegment, target_s: float):
+    t_ms = max(0, int(float(target_s) * 1000))
+    if len(seg) >= t_ms:
+        return seg[:t_ms]
+    return seg + AudioSegment.silent(duration=t_ms - len(seg))
+
+def _atempo_chain(factor: float) -> str:
+    # Build a chain so each step stays within [0.5, 2.0] for better quality.
+    steps = []
+    f = max(0.1, min(10.0, float(factor)))
+    while f < 0.5:
+        steps.append(0.5); f /= 0.5
+    while f > 2.0:
+        steps.append(2.0); f /= 2.0
+    steps.append(f)
+    return ",".join([f"atempo={s:.5f}" for s in steps])
+
+def fit_to_seconds(seg: AudioSegment, target_s: float, fmt_out="mp3") -> io.BytesIO:
+    """Pitch-preserving time stretch via FFmpeg atempo."""
+    with tempfile.TemporaryDirectory() as d:
+        inp = os.path.join(d, "in.wav")
+        outp = os.path.join(d, f"out.{fmt_out}")
+        seg.export(inp, format="wav")
+        orig = max(0.01, len(seg) / 1000)
+        factor = float(target_s) / orig
+        af = _atempo_chain(factor)
+        codec = ["-c:a", "libmp3lame", "-b:a", "128k"] if fmt_out == "mp3" else []
+        cmd = ["ffmpeg", "-y", "-i", inp, "-vn", "-af", af, *codec, outp]
+        subprocess.run(cmd, check=True)
+        with open(outp, "rb") as f:
+            return io.BytesIO(f.read())
+
+def normalize_lufs(seg: AudioSegment, target_lufs=-14.0):
+    # Lightweight perceived normalization using RMS (keeps deps minimal).
+    import math
+    rms = seg.rms or 1
+    current_db = 20 * math.log10(rms / (1 << 15))
+    gain_db = float(target_lufs) - current_db
+    return seg.apply_gain(gain_db)
+
+# ---------- processors ----------
+def process_single(file, mode, target_seconds, keep_silence_s,
+                   min_silence_ms, silence_thresh_db, do_normalize, fmt):
+    raw = file if isinstance(file, (bytes, bytearray)) else file.read()
+    original = _load(raw)
+
+    # 1) optional silence removal / pause control
+    cleaned = remove_silence(
+        original,
+        keep_ms=int(float(keep_silence_s) * 1000),
+        min_silence_ms=int(min_silence_ms),
+        thresh_db=float(silence_thresh_db),
+    )
+
+    # 2) optional loudness normalize
+    if do_normalize:
+        cleaned = normalize_lufs(cleaned, -14.0)
+
+    # 3) timing mode
+    if mode == "trim" and target_seconds:
+        final = trim_to_seconds(cleaned, target_seconds)
+        out = _export(final, fmt)
+    elif mode == "fit" and target_seconds:
+        out = fit_to_seconds(cleaned, target_seconds, fmt_out=fmt)
+    else:
+        out = _export(cleaned, fmt)
+
+    before = len(original) / 1000
+    after = len(_load(out.getvalue())) / 1000
+    report = f"Before: {before:.2f}s | After: {after:.2f}s"
+    return out, report
+
+def process_batch(files, **kwargs) -> io.BytesIO:
+    zbuf = io.BytesIO()
+    with zipfile.ZipFile(zbuf, "w", zipfile.ZIP_DEFLATED) as z:
+        for f in files:
+            single, _ = process_single(f, **kwargs)
+            name = getattr(f, "name", f"audio_{uuid.uuid4().hex}")
+            stem = os.path.splitext(name)[0]
+            z.writestr(f"{stem}_processed.{kwargs['fmt']}", single.getvalue())
+    zbuf.seek(0)
+    return zbuf
+
+def write_temp_for_preview(blob: io.BytesIO, fmt: str) -> str:
+    # Gradio Audio preview works great with a file path; write a temp file.
+    tf = tempfile.NamedTemporaryFile(delete=False, suffix=f".{fmt}")
+    tf.write(blob.getvalue())
+    tf.flush(); tf.close()
+    return tf.name
+
+# ---------- UI ----------
+with gr.Blocks(title="AI Voice Studio – Pause Control, Trim, Fit") as demo:
+    gr.Markdown(
+        "### Remove or normalize pauses, **set pause length**, **trim to exact time**, or **fit length (pitch preserved)**.\n"
+        "_Outputs: mp3 / wav / m4a / ogg. Single file → direct download. Multiple files → ZIP._"
+    )
+
+    with gr.Row():
+        with gr.Column():
+            files = gr.Files(label="Upload audio (one or many)", file_types=["audio"], type="filepath")
+            mode = gr.Radio(["none", "trim", "fit"], value="none", label="Timing mode")
+            target = gr.Number(value=30, label="Target seconds (for trim/fit)")
+            keep = gr.Number(value=0.25, label="Set pause length (seconds kept at cuts)")
+            min_sil = gr.Slider(50, 1000, 120, step=10, label="Count a pause if silence ≥ (ms)")
+            thresh = gr.Slider(-80, -10, -45, step=1, label="Silence threshold (dBFS)")
+            do_norm = gr.Checkbox(True, label="Normalize loudness (~-14 LUFS)")
+            fmt = gr.Dropdown(["mp3","wav","m4a","ogg"], value="mp3", label="Output format")
+            go = gr.Button("Process")
+
+        with gr.Column():
+            preview = gr.Audio(label="Preview (first file)", type="filepath", interactive=False)
+            direct = gr.File(label="Download processed file (single)")
+            zip_out = gr.File(label="Download ZIP (if multiple)")
+            rep = gr.Textbox(label="Report", lines=1)
+
+    def run(files, mode, target, keep, min_sil, thresh, do_norm, fmt):
+        files = files or []
+        if not files:
+            return None, None, None, "Please upload at least one audio file."
+
+        # Process first file for preview & (if single) for direct download
+        single_blob, report = process_single(
+            open(files[0], "rb"),
+            mode=mode, target_seconds=target, keep_silence_s=keep,
+            min_silence_ms=min_sil, silence_thresh_db=thresh,
+            do_normalize=do_norm, fmt=fmt
+        )
+        preview_path = write_temp_for_preview(single_blob, fmt)
+
+        if len(files) == 1:
+            # Direct download for single file
+            direct_file = single_blob
+            return preview_path, direct_file, None, report
+        else:
+            # ZIP for multiple files
+            opened = [open(p, "rb") for p in files]
+            zipped = process_batch(
+                opened, mode=mode, target_seconds=target, keep_silence_s=keep,
+                min_silence_ms=min_sil, silence_thresh_db=thresh,
+                do_normalize=do_norm, fmt=fmt
+            )
+            return preview_path, None, zipped, report
+
+    go.click(
+        run,
+        [files, mode, target, keep, min_sil, thresh, do_norm, fmt],
+        [preview, direct, zip_out, rep]
+    )
+
+if __name__ == "__main__":
+    demo.queue().launch()