From ee8fb2557a0a6418ffba44330cd2b5f18ccd34ab Mon Sep 17 00:00:00 2001 From: Grzegorz <19194188+farce1@users.noreply.github.com> Date: Sun, 14 Jun 2026 16:50:04 +0200 Subject: [PATCH] feat(transcription): implement retranscribe_meeting Re-run ASR on a meeting's saved recording and replace its transcript. transcribe_samples_48k drives the existing streaming worker over the decoded file (reusing the proven VAD/resampler/engine pipeline); retranscribe_audio_file resolves the engine, checks readiness via the shared model::check_engine_ready helper (also used by the live worker), decodes the Ogg/Opus recording, and refuses to wipe an existing transcript when no speech is found. The command replaces transcript rows transactionally and reindexes FTS. --- src-tauri/src/commands.rs | 65 +++++++++++++++++- src-tauri/src/transcription/mod.rs | 97 +++++++++++++++++++++++++-- src-tauri/src/transcription/model.rs | 32 +++++++++ src-tauri/src/transcription/worker.rs | 46 +++++++++++++ 4 files changed, 232 insertions(+), 8 deletions(-) diff --git a/src-tauri/src/commands.rs b/src-tauri/src/commands.rs index 986b0d0..b7aed36 100644 --- a/src-tauri/src/commands.rs +++ b/src-tauri/src/commands.rs @@ -419,9 +419,68 @@ pub async fn get_transcript_page( } #[tauri::command] -pub async fn retranscribe_meeting(meeting_id: i64) -> Result<(), String> { - eprintln!("[retranscribe] requested for meeting_id={meeting_id} — not yet implemented"); - Err("Re-transcription is not yet available. Coming in a future update.".to_string()) +pub async fn retranscribe_meeting( + pool: tauri::State<'_, SqlitePool>, + data_dir: tauri::State<'_, DataDir>, + meeting_id: i64, +) -> Result<(), String> { + let db = pool.inner(); + + let audio_path = sqlx::query_scalar::<_, Option>( + "SELECT audio_path FROM meetings WHERE id = ?", + ) + .bind(meeting_id) + .fetch_optional(db) + .await + .map_err(|err| format!("failed to load meeting audio path: {err}"))? + .flatten() + .ok_or_else(|| "meeting has no recording to re-transcribe".to_string())?; + + let data_dir = data_dir.inner().0.clone(); + let segments = tokio::task::spawn_blocking(move || { + transcription::retranscribe_audio_file(&data_dir, Path::new(&audio_path)) + }) + .await + .map_err(|err| format!("re-transcription task failed: {err}"))??; + + let rows = transcription::transcript_rows(&segments); + + let mut tx = db + .begin() + .await + .map_err(|err| format!("failed to begin transcript transaction: {err}"))?; + + sqlx::query("DELETE FROM transcripts WHERE meeting_id = ?") + .bind(meeting_id) + .execute(&mut *tx) + .await + .map_err(|err| format!("failed to clear old transcript: {err}"))?; + + for row in &rows { + sqlx::query( + "INSERT INTO transcripts (meeting_id, segment_index, text, start_time_ms, end_time_ms, is_final) + VALUES (?, ?, ?, ?, ?, 1)", + ) + .bind(meeting_id) + .bind(row.segment_index) + .bind(&row.text) + .bind(row.start_time_ms) + .bind(row.end_time_ms) + .execute(&mut *tx) + .await + .map_err(|err| format!("failed to insert transcript segment: {err}"))?; + } + + tx.commit() + .await + .map_err(|err| format!("failed to commit re-transcription: {err}"))?; + + // The transcript is already replaced; a stale search index must not fail the command. + if let Err(err) = fts_upsert(db, meeting_id).await { + eprintln!("[fts] upsert after re-transcription failed: {err}"); + } + + Ok(()) } #[tauri::command] diff --git a/src-tauri/src/transcription/mod.rs b/src-tauri/src/transcription/mod.rs index 062fa1f..6e1f086 100644 --- a/src-tauri/src/transcription/mod.rs +++ b/src-tauri/src/transcription/mod.rs @@ -45,6 +45,34 @@ pub struct SegmentResult { pub detected_language: Option, } +/// A transcript row ready for insertion, derived from transcription segments. +#[derive(Debug, PartialEq)] +pub struct TranscriptInsert { + pub segment_index: i64, + pub text: String, + pub start_time_ms: i64, + pub end_time_ms: i64, +} + +/// Approximate spoken span assigned to each segment (matches the live worker). +const SEGMENT_SPAN_MS: i64 = 1_000; + +pub fn transcript_rows(segments: &[SegmentResult]) -> Vec { + segments + .iter() + .enumerate() + .map(|(index, segment)| { + let start_time_ms = segment.elapsed_ms as i64; + TranscriptInsert { + segment_index: index as i64, + text: segment.text.clone(), + start_time_ms, + end_time_ms: start_time_ms + SEGMENT_SPAN_MS, + } + }) + .collect() +} + #[derive(Clone, Serialize)] #[serde(rename_all = "camelCase", rename_all_fields = "camelCase", tag = "event", content = "data")] pub enum TranscriptEvent { @@ -92,6 +120,33 @@ fn resolve_asr_engine(preference: Option<&str>, data_dir: &Path) -> (String, Pat } } +/// Re-transcribe a saved recording end-to-end and return its segments. Returns an +/// error (leaving any existing transcript untouched) when the model is missing or +/// the run yields no speech, so a re-transcription never silently wipes a transcript. +pub fn retranscribe_audio_file( + data_dir: &Path, + audio_path: &Path, +) -> Result, String> { + let (asr_engine, model_dir) = resolve_asr_engine(None, data_dir); + + if !model::check_engine_ready(&asr_engine, data_dir) { + return Err("transcription model is not ready; download required model files first".to_string()); + } + + let samples_48k = crate::diarization::decode::decode_ogg_opus_to_f32(audio_path)?; + let vad_model = model::vad_model_path(data_dir).to_string_lossy().to_string(); + let segments = worker::transcribe_samples_48k(&samples_48k, model_dir, asr_engine, vad_model); + + if segments.is_empty() { + return Err( + "re-transcription produced no transcript; the existing transcript was left unchanged" + .to_string(), + ); + } + + Ok(segments) +} + pub fn start_transcription_worker( state: &mut TranscriptionState, args: StartWorkerArgs, @@ -110,11 +165,7 @@ pub fn start_transcription_worker( let (asr_engine, model_dir) = resolve_asr_engine(asr_engine_preference.as_deref(), data_dir.as_path()); - let assets_ready = match asr_engine.as_str() { - "parakeet" => model::check_parakeet_assets_ready(data_dir.as_path()), - _ => model::check_transcription_assets_ready(data_dir.as_path()), - }; - if !assets_ready || !model::vad_model_path(data_dir.as_path()).exists() { + if !model::check_engine_ready(&asr_engine, data_dir.as_path()) { return Err("transcription model is not ready; download required model files first".to_string()); } @@ -361,4 +412,40 @@ mod tests { std::fs::remove_dir_all(&tmp).ok(); } + + #[test] + fn transcript_rows_assigns_sequential_indices_and_padded_end_times() { + let segments = vec![ + SegmentResult { + text: "hello".to_string(), + elapsed_ms: 0, + detected_language: None, + }, + SegmentResult { + text: "world".to_string(), + elapsed_ms: 2_500, + detected_language: Some("en".to_string()), + }, + ]; + + let rows = transcript_rows(&segments); + + assert_eq!( + rows, + vec![ + TranscriptInsert { + segment_index: 0, + text: "hello".to_string(), + start_time_ms: 0, + end_time_ms: 1_000, + }, + TranscriptInsert { + segment_index: 1, + text: "world".to_string(), + start_time_ms: 2_500, + end_time_ms: 3_500, + }, + ] + ); + } } diff --git a/src-tauri/src/transcription/model.rs b/src-tauri/src/transcription/model.rs index 22af5de..969aae1 100644 --- a/src-tauri/src/transcription/model.rs +++ b/src-tauri/src/transcription/model.rs @@ -52,6 +52,13 @@ pub fn check_parakeet_model_ready(data_dir: &Path) -> bool { check_parakeet_assets_ready(data_dir) && vad_model_path(data_dir).exists() } +pub fn check_engine_ready(engine: &str, data_dir: &Path) -> bool { + match engine { + "parakeet" => check_parakeet_model_ready(data_dir), + _ => check_model_ready(data_dir), + } +} + #[cfg(test)] mod tests { use super::*; @@ -103,4 +110,29 @@ mod tests { std::fs::remove_dir_all(&tmp).ok(); } + + #[test] + fn check_engine_ready_dispatches_by_engine() { + let tmp = std::env::temp_dir().join(format!("on-engine-ready-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&tmp); + + assert!(!check_engine_ready("whisper", &tmp)); + assert!(!check_engine_ready("parakeet", &tmp)); + + let whisper_dir = whisper_turbo_model_dir(&tmp); + std::fs::create_dir_all(&whisper_dir).unwrap(); + for f in [ + "turbo-encoder.int8.onnx", + "turbo-decoder.int8.onnx", + "turbo-tokens.txt", + ] { + std::fs::write(whisper_dir.join(f), b"x").unwrap(); + } + std::fs::write(vad_model_path(&tmp), b"x").unwrap(); + + assert!(check_engine_ready("whisper", &tmp)); + assert!(!check_engine_ready("parakeet", &tmp)); + + std::fs::remove_dir_all(&tmp).ok(); + } } diff --git a/src-tauri/src/transcription/worker.rs b/src-tauri/src/transcription/worker.rs index 890ed9d..29bc1cd 100644 --- a/src-tauri/src/transcription/worker.rs +++ b/src-tauri/src/transcription/worker.rs @@ -180,3 +180,49 @@ pub fn run_worker( config.recording_start_ms, ); } + +/// Transcribe a complete 48 kHz mono buffer by driving `run_worker` to completion, +/// reusing the exact streaming VAD -> chunk -> transcribe pipeline. Returns the +/// collected segments (empty if the model fails to load or no speech is found). +pub fn transcribe_samples_48k( + samples_48k: &[f32], + model_dir: PathBuf, + asr_engine: String, + vad_model: String, +) -> Vec { + // Bounded audio channel for backpressure; unbounded result channel so the + // worker never blocks while we are still feeding audio. + let (audio_tx, audio_rx) = mpsc::sync_channel::>(64); + // Held (never sent on) for the worker's lifetime so its command_rx never sees a + // disconnect and breaks early; dropped at scope end after the worker has joined. + let (_command_tx, command_rx) = mpsc::channel::(); + let (result_tx, result_rx) = mpsc::channel::(); + let shutdown = Arc::new(AtomicBool::new(false)); + + let config = WorkerConfig { + model_dir, + asr_engine, + vad_model, + recording_start_ms: 0, + result_tx, + }; + + let worker = std::thread::spawn(move || run_worker(audio_rx, command_rx, config, shutdown)); + + for chunk in samples_48k.chunks(ASR_SAMPLE_RATE as usize) { + if audio_tx.send(chunk.to_vec()).is_err() { + break; + } + } + // Closing the audio channel makes the worker drain, do its final flush, and exit. + drop(audio_tx); + + let segments: Vec = result_rx.iter().collect(); + // A panicked worker yields only partial segments; discard them so the caller's + // empty-result guard preserves the existing transcript instead of replacing it. + if worker.join().is_ok() { + segments + } else { + Vec::new() + } +}