feat(import): run import in background with per-phase progress + model override
This commit is contained in:
1 parent
e0e9377273
commit
fe355f034d
1 file changed
+119
-26
+119
-26
@@ -2211,21 +2211,49 @@ fn default_title_from_source(source: &str) -> Option<String> {
|
||||
.filter(|s| !s.trim().is_empty())
|
||||
}
|
||||
|
||||
/// Emit one `import://progress` tick for the Domino's-tracker UI. `state` is
|
||||
/// `active` (this phase is running), `done` (finished; `elapsed_ms` set), or
|
||||
/// `error` (`message` set). `phase` is one of prepare/transcribe/diarize/finalize.
|
||||
fn emit_import_progress(
|
||||
app: &AppHandle,
|
||||
meeting_id: &str,
|
||||
phase: &str,
|
||||
state: &str,
|
||||
elapsed_ms: Option<u128>,
|
||||
message: Option<&str>,
|
||||
) {
|
||||
let _ = app.emit(
|
||||
"import://progress",
|
||||
serde_json::json!({
|
||||
"meetingId": meeting_id,
|
||||
"phase": phase,
|
||||
"state": state,
|
||||
"elapsedMs": elapsed_ms.map(|m| m as u64),
|
||||
"error": message,
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
/// Manually add a meeting from an existing recording: a local audio/video file
|
||||
/// or a URL (YouTube/streaming page, or a direct media URL). Shells out to
|
||||
/// `ffmpeg` (transcode) and — for URLs — `yt-dlp` (both external, not bundled;
|
||||
/// a missing tool is a clear error). The produced 16kHz-mono WAV becomes the
|
||||
/// meeting's retained `audio.wav`, then goes through the same
|
||||
/// transcription + diarization + finalize path as a live recording.
|
||||
/// or a URL (YouTube/streaming page, or a direct media URL). Creates the meeting
|
||||
/// row (status `transcribing`) and returns its id **immediately**; the heavy
|
||||
/// transcode → transcribe → diarize → finalize work runs detached so the UI is
|
||||
/// never blocked (a 25-min video can take 10+ min). Progress streams via
|
||||
/// `import://progress` and completion via `transcript://finalized`. `model`
|
||||
/// overrides the Settings whisper model for this one import (so the user picks
|
||||
/// and can see what it was transcribed with); omit to use the Settings default.
|
||||
#[tauri::command]
|
||||
pub async fn import_media(
|
||||
app: AppHandle,
|
||||
state: State<'_, AppState>,
|
||||
source: String,
|
||||
title: Option<String>,
|
||||
model: Option<String>,
|
||||
) -> WaResult<MeetingId> {
|
||||
let settings = load_settings();
|
||||
let model_id = model_id_for(&settings);
|
||||
let model_id = model
|
||||
.filter(|m| !m.trim().is_empty())
|
||||
.unwrap_or_else(|| model_id_for(&settings));
|
||||
let backend = backend_for(&settings);
|
||||
let model_path = whisper_model_file(&model_id);
|
||||
if !model_path.exists() {
|
||||
@@ -2254,12 +2282,47 @@ pub async fn import_media(
|
||||
})
|
||||
.await
|
||||
.map_err(|e| WaError::new("storage", e.to_string()))?;
|
||||
// `create_meeting` starts rows as `recording`; mark this one `transcribing`
|
||||
// so the meetings-list badge reads as an import in flight, not a live mic.
|
||||
let _ = state
|
||||
.store
|
||||
.set_meeting_status(&meeting_id, "transcribing")
|
||||
.await;
|
||||
|
||||
// Run the pipeline detached and return now — the dialog closes and the
|
||||
// meeting appears in the list with a live tracker fed by the events below.
|
||||
let store = state.store.clone();
|
||||
let dir = meeting_dir(&meeting_id);
|
||||
let wav_path = dir.join("audio.wav");
|
||||
let mid = meeting_id.clone();
|
||||
tauri::async_runtime::spawn(run_import_pipeline(
|
||||
app, store, mid, source, wav_path, dir, backend, model_id, model_path, language,
|
||||
));
|
||||
Ok(meeting_id)
|
||||
}
|
||||
|
||||
// Transcode into the meeting's audio.wav off the async runtime (shells out
|
||||
// to ffmpeg/yt-dlp). On any failure, drop the empty meeting so a bad import
|
||||
// doesn't leave a husk row behind.
|
||||
/// The detached body of `import_media`: transcode, transcribe, diarize, finalize,
|
||||
/// emitting an `import://progress` tick at the start and end of each phase. On
|
||||
/// the first failure it marks the meeting `error` (kept in the list, not
|
||||
/// deleted, so the user sees the failed import) and stops.
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
async fn run_import_pipeline(
|
||||
app: AppHandle,
|
||||
store: Arc<dyn crate::storage::Store>,
|
||||
meeting_id: MeetingId,
|
||||
source: String,
|
||||
wav_path: std::path::PathBuf,
|
||||
dir: std::path::PathBuf,
|
||||
backend: BackendId,
|
||||
model_id: String,
|
||||
model_path: std::path::PathBuf,
|
||||
language: Option<String>,
|
||||
) {
|
||||
use std::time::Instant;
|
||||
|
||||
// Phase 1: prepare — transcode (and, for URLs, yt-dlp download) to audio.wav.
|
||||
emit_import_progress(&app, &meeting_id, "prepare", "active", None, None);
|
||||
let t = Instant::now();
|
||||
let transcode = tauri::async_runtime::spawn_blocking({
|
||||
let source = source.clone();
|
||||
let wav_path = wav_path.clone();
|
||||
@@ -2272,15 +2335,19 @@ pub async fn import_media(
|
||||
r
|
||||
}
|
||||
})
|
||||
.await
|
||||
.map_err(|e| WaError::new("import", e.to_string()))?;
|
||||
if let Err(e) = transcode {
|
||||
let _ = state.store.delete_meeting(&meeting_id).await;
|
||||
return Err(WaError::new("import", e.to_string()));
|
||||
.await;
|
||||
match transcode {
|
||||
Ok(Ok(())) => {
|
||||
emit_import_progress(&app, &meeting_id, "prepare", "done", Some(t.elapsed().as_millis()), None)
|
||||
}
|
||||
Ok(Err(e)) => return fail_import(&app, &store, &meeting_id, "prepare", &e.to_string()).await,
|
||||
Err(e) => return fail_import(&app, &store, &meeting_id, "prepare", &e.to_string()).await,
|
||||
}
|
||||
|
||||
// Transcribe the produced WAV (same batch path as reprocess_transcript).
|
||||
let (mut segments, resolved_language) = tauri::async_runtime::spawn_blocking({
|
||||
// Phase 2: transcribe (same batch path as reprocess_transcript).
|
||||
emit_import_progress(&app, &meeting_id, "transcribe", "active", None, None);
|
||||
let t = Instant::now();
|
||||
let transcribed = tauri::async_runtime::spawn_blocking({
|
||||
let wav_path = wav_path.clone();
|
||||
let model_path = model_path.clone();
|
||||
let requested_language = language.clone();
|
||||
@@ -2294,12 +2361,20 @@ pub async fn import_media(
|
||||
Ok::<_, crate::transcription::TrxError>((segments, resolved))
|
||||
}
|
||||
})
|
||||
.await
|
||||
.map_err(|e| WaError::new("transcription", e.to_string()))?
|
||||
.map_err(|e| WaError::new("transcription", e.to_string()))?;
|
||||
.await;
|
||||
let (mut segments, resolved_language) = match transcribed {
|
||||
Ok(Ok(v)) => {
|
||||
emit_import_progress(&app, &meeting_id, "transcribe", "done", Some(t.elapsed().as_millis()), None);
|
||||
v
|
||||
}
|
||||
Ok(Err(e)) => return fail_import(&app, &store, &meeting_id, "transcribe", &e.to_string()).await,
|
||||
Err(e) => return fail_import(&app, &store, &meeting_id, "transcribe", &e.to_string()).await,
|
||||
};
|
||||
|
||||
// One diarization pass if the models are installed, exactly like
|
||||
// stop_recording — otherwise every line stays the single "S1" placeholder.
|
||||
// Phase 3: diarize — one pass if the models are installed, else every line
|
||||
// stays the single "S1" placeholder (same as stop_recording).
|
||||
emit_import_progress(&app, &meeting_id, "diarize", "active", None, None);
|
||||
let t = Instant::now();
|
||||
let diarizer: Option<Arc<dyn Diarizer>> =
|
||||
tauri::async_runtime::spawn_blocking(diarizer_from_installed_models)
|
||||
.await
|
||||
@@ -2315,15 +2390,18 @@ pub async fn import_media(
|
||||
Err(e) => tracing::warn!("import diarization task failed: {e}"),
|
||||
}
|
||||
}
|
||||
emit_import_progress(&app, &meeting_id, "diarize", "done", Some(t.elapsed().as_millis()), None);
|
||||
|
||||
// Phase 4: finalize — persist segments, notes, seal audio.
|
||||
emit_import_progress(&app, &meeting_id, "finalize", "active", None, None);
|
||||
let t = Instant::now();
|
||||
let speakers = speaker_infos_from_segments(&segments, &HashMap::new());
|
||||
let duration_secs = segments
|
||||
.last()
|
||||
.map(|s| (s.end_ms / 1000) as i64)
|
||||
.unwrap_or(0);
|
||||
|
||||
state
|
||||
.store
|
||||
if let Err(e) = store
|
||||
.finalize_meeting(
|
||||
&meeting_id,
|
||||
FinalizeMeeting {
|
||||
@@ -2338,7 +2416,9 @@ pub async fn import_media(
|
||||
},
|
||||
)
|
||||
.await
|
||||
.map_err(|e| WaError::new("storage", e.to_string()))?;
|
||||
{
|
||||
return fail_import(&app, &store, &meeting_id, "finalize", &e.to_string()).await;
|
||||
}
|
||||
|
||||
let notes_md = crate::notes::MarkdownNotes.merge(
|
||||
&segments,
|
||||
@@ -2347,7 +2427,7 @@ pub async fn import_media(
|
||||
None,
|
||||
None,
|
||||
);
|
||||
let _ = state.store.update_notes(&meeting_id, ¬es_md).await;
|
||||
let _ = store.update_notes(&meeting_id, ¬es_md).await;
|
||||
|
||||
// Seal the retained recording at rest when the vault is unlocked (T8.8),
|
||||
// matching stop_recording so imports aren't left as plaintext outliers.
|
||||
@@ -2358,12 +2438,25 @@ pub async fn import_media(
|
||||
}
|
||||
}
|
||||
}
|
||||
emit_import_progress(&app, &meeting_id, "finalize", "done", Some(t.elapsed().as_millis()), None);
|
||||
|
||||
let _ = app.emit(
|
||||
"transcript://finalized",
|
||||
serde_json::json!({ "meetingId": meeting_id, "segmentCount": segments.len() }),
|
||||
);
|
||||
Ok(meeting_id)
|
||||
}
|
||||
|
||||
/// Mark a failed background import `error` (kept in the list) and emit the error
|
||||
/// tick for the phase that failed.
|
||||
async fn fail_import(
|
||||
app: &AppHandle,
|
||||
store: &Arc<dyn crate::storage::Store>,
|
||||
meeting_id: &str,
|
||||
phase: &str,
|
||||
message: &str,
|
||||
) {
|
||||
let _ = store.set_meeting_status(&meeting_id.to_string(), "error").await;
|
||||
emit_import_progress(app, meeting_id, phase, "error", None, Some(message));
|
||||
}
|
||||
|
||||
/// Re-run transcription from a `recovering` meeting's working `audio.wav`
|
||||
|
||||
Reference in new issue
Block a user