From 842b4d75eb46eca203639e035f9b2274cd428498 Mon Sep 17 00:00:00 2001 From: oimwiodev Date: Wed, 17 Jun 2026 19:32:31 +0100 Subject: [PATCH] Initial: WhisperX Gradio GUI with 5-min chunking --- .gitignore | 4 + app.py | 232 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 236 insertions(+) create mode 100644 .gitignore create mode 100644 app.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..bdd06af --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +venv/ +__pycache__/ +*.pyc +.DS_Store diff --git a/app.py b/app.py new file mode 100644 index 0000000..1ec03e9 --- /dev/null +++ b/app.py @@ -0,0 +1,232 @@ +#!/usr/bin/env python3 +""" +WhisperX Transcription GUI — Gradio app +Real 5-min chunk splitting to avoid hallucination. +No Gitea bloat — just transcribe & download. +""" + +import os, json, subprocess, tempfile, shutil, re, pathlib, uuid, io +import gradio as gr + +# ── Config ── +WHISPERX = os.path.expanduser("~/.local/bin/whisperx") +OUTPUT_DIR = os.path.expanduser("~/workspace/transcripts/gradio_uploads") +CHUNK_SECS = 300 # 5 minutes +os.makedirs(OUTPUT_DIR, exist_ok=True) + +LANGUAGES = [ + ("French", "fr"), ("Arabic / Darija", "ar"), ("English", "en"), + ("German", "de"), ("Spanish", "es"), ("Italian", "it"), + ("Portuguese", "pt"), ("Dutch", "nl"), ("Japanese", "ja"), + ("Chinese", "zh"), ("Russian", "ru"), ("Turkish", "tr"), + ("Polish", "pl"), ("Swedish", "sv"), ("Norwegian", "no"), + ("Danish", "da"), ("Finnish", "fi"), ("Czech", "cs"), + ("Romanian", "ro"), ("Hungarian", "hu"), ("Greek", "el"), + ("Hebrew", "he"), ("Hindi", "hi"), ("Korean", "ko"), + ("Thai", "th"), ("Vietnamese", "vi"), ("Indonesian", "id"), + ("Malay", "ms"), ("Auto-detect", ""), +] + +MODELS = [ + ("large-v3 (best, ~8GB VRAM)", "large-v3"), + ("large-v2 (~8GB VRAM)", "large-v2"), + ("medium (~5GB VRAM)", "medium"), + ("small (~2GB VRAM)", "small"), + ("base (~1GB VRAM)", "base"), + ("tiny (~0.3GB VRAM)", "tiny"), +] + + +def fmt_ts(secs: float) -> str: + h = int(secs // 3600) + m = int((secs % 3600) // 60) + s = secs % 60 + if h: + return f"{h:02d}:{m:02d}:{s:06.3f}" + return f"{m:02d}:{s:06.3f}" + + +def fmt_srt(secs: float) -> str: + h = int(secs // 3600) + m = int((secs % 3600) // 60) + s = int(secs % 60) + ms = int((secs - int(secs)) * 1000) + return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}" + + +def transcribe(audio_path, lang, model): + """Run WhisperX on one file. Returns {segments, error, out_dir}.""" + out_dir = pathlib.Path(OUTPUT_DIR) / f"{pathlib.Path(audio_path).stem}_{uuid.uuid4().hex[:8]}" + out_dir.mkdir(parents=True, exist_ok=True) + + cmd = [ + WHISPERX, audio_path, + "--model", model, + "--task", "transcribe", + "--device", "cuda", + "--compute_type", "float16", + "--batch_size", "4", + "--chunk_size", "30", + "--vad_method", "pyannote", + "--condition_on_previous_text", "False", + "--temperature", "0", + "--compression_ratio_threshold", "2.4", + "--logprob_threshold", "-1.0", + "--no_speech_threshold", "0.6", + "--output_dir", str(out_dir), + "--output_format", "all", + ] + if lang: + cmd.extend(["--language", lang]) + + result = subprocess.run(cmd, capture_output=True, text=True, timeout=1800) + if result.returncode != 0: + return {"error": f"WhisperX failed:\n{result.stderr[:2000]}", "segments": [], "out_dir": str(out_dir)} + + # Parse JSON output for segments + json_files = list(out_dir.glob("*.json")) + segments = [] + if json_files: + with open(json_files[0], encoding="utf-8") as f: + data = json.load(f) + segments = data.get("segments", []) + + return {"error": None, "segments": segments, "out_dir": str(out_dir)} + + +# ── Gradio function ── +def do_transcribe(audio_file, lang_code, model_code): + if audio_file is None: + yield "", gr.Dropdown(), gr.Dropdown(), "⚠️ Upload an audio file first." + return + + yield "", gr.Dropdown(), gr.Dropdown(), "⏳ Converting audio to WAV..." + + # Convert to 16kHz mono WAV + temp_dir = tempfile.mkdtemp() + wav_path = os.path.join(temp_dir, "full.wav") + subprocess.run( + ["ffmpeg", "-y", "-i", audio_file, "-ar", "16000", "-ac", "1", wav_path], + capture_output=True, timeout=120, + ) + + # Get duration + probe = subprocess.run( + ["ffprobe", "-v", "error", "-show_entries", "format=duration", + "-of", "csv=p=0", wav_path], + capture_output=True, text=True, + ) + total_secs = float(probe.stdout.strip() or 0) + + # Split into 5-min chunks + yield "", gr.Dropdown(), gr.Dropdown(), f"⏳ Splitting {fmt_ts(total_secs)} into {int(total_secs // CHUNK_SECS) + 1} chunks..." + chunks_dir = os.path.join(temp_dir, "chunks") + os.makedirs(chunks_dir, exist_ok=True) + subprocess.run( + ["ffmpeg", "-y", "-i", wav_path, + "-f", "segment", "-segment_time", str(CHUNK_SECS), + "-reset_timestamps", "1", "-c", "copy", + f"{chunks_dir}/chunk_%03d.wav"], + capture_output=True, timeout=600, + ) + + chunk_files = sorted(pathlib.Path(chunks_dir).glob("chunk_*.wav")) + if not chunk_files: + # File shorter than 5 min — use original + chunk_files = [pathlib.Path(wav_path)] + + n_chunks = len(chunk_files) + all_segments = [] + + for idx, chunk_path in enumerate(chunk_files): + yield "", gr.Dropdown(), gr.Dropdown(), ( + f"⏳ Transcribing chunk {idx + 1}/{n_chunks}..." + ) + + result = transcribe(str(chunk_path), lang_code, model_code) + if result["error"]: + yield "", gr.Dropdown(), gr.Dropdown(), f"❌ Chunk {idx + 1} failed: {result['error']}" + shutil.rmtree(temp_dir, ignore_errors=True) + return + + offset = idx * CHUNK_SECS + for seg in result["segments"]: + seg["start"] += offset + seg["end"] += offset + all_segments.append(seg) + + all_segments.sort(key=lambda s: s["start"]) + shutil.rmtree(temp_dir, ignore_errors=True) + + # Build outputs + dur = all_segments[-1]["end"] if all_segments else 0 + + # Timestamped TXT + txt_lines = [] + for seg in all_segments: + txt_lines.append(f"[{fmt_ts(seg['start'])} --> {fmt_ts(seg['end'])}] {seg['text'].strip()}") + txt_content = "\n".join(txt_lines) + + # SRT + srt_lines = [] + for i, seg in enumerate(all_segments, 1): + srt_lines.append(f"{i}") + srt_lines.append(f"{fmt_srt(seg['start'])} --> {fmt_srt(seg['end'])}") + srt_lines.append(seg['text'].strip()) + srt_lines.append("") + srt_content = "\n".join(srt_lines) + + # Write files + session_dir = pathlib.Path(OUTPUT_DIR) / f"transcript_{uuid.uuid4().hex[:8]}" + session_dir.mkdir(parents=True, exist_ok=True) + txt_path = str(session_dir / "transcript.txt") + srt_path = str(session_dir / "transcript.srt") + with open(txt_path, "w") as f: + f.write(txt_content) + with open(srt_path, "w") as f: + f.write(srt_content) + + status = ( + f"✅ **Done** — {lang_code.upper() if lang_code else 'Auto'} | {model_code} | " + f"{fmt_ts(dur)} | {len(all_segments)} segments | {n_chunks} chunk{'s' if n_chunks > 1 else ''}" + ) + + yield txt_content, gr.Dropdown(value=txt_path, visible=True), gr.Dropdown(value=srt_path, visible=True), status + + +# ── UI ── +with gr.Blocks(title="WhisperX Transcription") as app: + gr.Markdown( + "# 🎙️ WhisperX Transcription\n" + "Upload audio → auto-splits into **5-minute chunks** (anti-hallucination) → downloads TXT + SRT." + ) + + with gr.Row(): + with gr.Column(scale=1): + audio_input = gr.Audio(type="filepath", label="Upload Audio", sources=["upload", "microphone"]) + lang_dropdown = gr.Dropdown(choices=LANGUAGES, value="fr", label="Language", info="Auto-detect if unsure") + model_dropdown = gr.Dropdown(choices=MODELS, value="large-v3", label="Model Size") + transcribe_btn = gr.Button("🚀 Transcribe", variant="primary", size="lg") + + with gr.Column(scale=2): + status_text = gr.Markdown("Ready to transcribe.") + output_preview = gr.Textbox(label="Transcript Preview", lines=15, max_lines=30, interactive=False) + with gr.Row(): + txt_download = gr.File(label="📄 Download TXT", visible=False) + srt_download = gr.File(label="📝 Download SRT", visible=False) + + transcribe_btn.click( + fn=do_transcribe, + inputs=[audio_input, lang_dropdown, model_dropdown], + outputs=[output_preview, txt_download, srt_download, status_text], + ) + + +if __name__ == "__main__": + import argparse + parser = argparse.ArgumentParser() + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--port", type=int, default=7860) + args = parser.parse_args() + print(f"🚀 WhisperX Gradio on http://{args.host}:{args.port}") + app.launch(server_name=args.host, server_port=args.port, theme=gr.themes.Soft())