import tempfile import traceback from pathlib import Path import gradio as gr from pipeline import get_pipeline, to_json, to_csv, to_abab_text UPLOAD_DIR = Path(tempfile.gettempdir()) / "speech_annotation" UPLOAD_DIR.mkdir(parents=True, exist_ok=True) _last_segments = [] SPEAKER_COLORS = [ ("#4F46E5", "#EEF2FF"), ("#059669", "#ECFDF5"), ("#DC2626", "#FEF2F2"), ("#D97706", "#FFFBEB"), ("#7C3AED", "#F5F3FF"), ("#0891B2", "#ECFEFF"), ("#DB2777", "#FDF2F8"), ("#65A30D", "#F7FEE7"), ("#EA580C", "#FFF7ED"), ("#0284C7", "#F0F9FF"), ] def make_conversation_html(segments): if not segments: return "" speaker_list = list(dict.fromkeys(s.speaker for s in segments)) color_map = {spk: SPEAKER_COLORS[i % len(SPEAKER_COLORS)] for i, spk in enumerate(speaker_list)} legend_items = "".join( f"" f"" f"Speaker {spk}" for spk in speaker_list ) legend = f"
{legend_items}
" bubbles = "" for seg in segments: fg, bg = color_map[seg.speaker] align = "flex-end" if speaker_list.index(seg.speaker) % 2 == 1 else "flex-start" text_align = "text-align:right;" if align == "flex-end" else "" radius = "4px 16px 16px 16px" if align == "flex-start" else "16px 4px 16px 16px" bubbles += f"""
Speaker {seg.speaker}  · {seg.start_fmt} → {seg.end_fmt}
{seg.text}
""" return f"""
{legend}
{bubbles}
""" def make_table_html(segments): if not segments: return "" speaker_list = list(dict.fromkeys(s.speaker for s in segments)) color_map = {spk: SPEAKER_COLORS[i % len(SPEAKER_COLORS)] for i, spk in enumerate(speaker_list)} rows = "".join( f"" f"{s.speaker}" f"{s.start_fmt}" f"{s.end_fmt}" f"{s.text}" f"" for s in segments ) return f"""
{rows}
Speaker Start End Transcript
""" def process_audio(audio_path, num_speakers): global _last_segments if audio_path is None: return "⚠️ Please upload an audio file first.", "", "" try: pipeline = get_pipeline() n = int(num_speakers) if num_speakers and int(num_speakers) > 0 else 0 segments = pipeline.process(audio_path, num_speakers=n) except Exception as e: return f"❌ Error: {e}\n{traceback.format_exc()}", "", "" if not segments: return "⚠️ No speech detected.", "", "" _last_segments = segments unique = len(set(s.speaker for s in segments)) status = f"✅ Done — {len(segments)} segments · {unique} speaker(s) detected" return status, make_conversation_html(segments), make_table_html(segments) def export_json(): if not _last_segments: return None out = str(UPLOAD_DIR / "annotation.json") to_json(_last_segments, out) return out def export_csv(): if not _last_segments: return None out = str(UPLOAD_DIR / "annotation.csv") to_csv(_last_segments, out) return out css = """ .gradio-container { max-width: 1100px !important; margin: auto !important; } footer { display: none !important; } """ with gr.Blocks(title="Speech Annotation Pipeline", css=css) as demo: gr.Markdown(\"\"\"# 🎙️ Speech Annotation Pipeline *Upload audio · Detect speakers · Export transcript*\"\"\") with gr.Row(): with gr.Column(scale=1): audio_input = gr.Audio(label="Upload Audio (.wav / .mp3 / .flac)", type="filepath") num_speakers = gr.Slider(minimum=0, maximum=10, step=1, value=0, label="Number of speakers (0 = auto-detect)") run_btn = gr.Button("▶ Run Annotation", variant="primary", size="lg") status_box = gr.Textbox(label="Status", value="Ready.", interactive=False) gr.Markdown("### 📥 Export") with gr.Row(): json_btn = gr.Button("⬇ JSON", size="sm") csv_btn = gr.Button("⬇ CSV", size="sm") json_file = gr.File(label="JSON Download", visible=True) csv_file = gr.File(label="CSV Download", visible=True) with gr.Column(scale=2): gr.Markdown("### 💬 Conversation View") conversation_html = gr.HTML( value="
Transcript will appear here after processing…
" ) gr.Markdown("### 📋 Segment Table") table_html = gr.HTML(value="") run_btn.click( fn=process_audio, inputs=[audio_input, num_speakers], outputs=[status_box, conversation_html, table_html] ) json_btn.click(fn=export_json, inputs=[], outputs=[json_file]) csv_btn.click(fn=export_csv, inputs=[], outputs=[csv_file]) if __name__ == "__main__": demo.launch(server_name="0.0.0.0", server_port=7860, show_error=True)