233 lines
8.3 KiB
Python
233 lines
8.3 KiB
Python
#!/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())
|