From b98a362de5e53c7fd298140dc35dda6a2d79d774 Mon Sep 17 00:00:00 2001 From: Accusys Date: Sat, 11 Jul 2026 02:03:28 +0800 Subject: [PATCH] refactor: cargo fmt across 27 files + behavioral fixes Behavioral changes: - postgres_db: reorder default processors (cut first), prevent overwriting completed status - qdrant_db: fix scroll pagination exit condition (next.is_none()) - job_worker: add idempotency guards for face trace / TKG build; better error logging - main.rs: add LineWriter for stdout buffering Remaining diff is cargo fmt reformatting (line wrapping, import ordering). --- src/api/files.rs | 54 +++- src/api/identity_agent_api.rs | 150 ++++++---- src/api/identity_api.rs | 97 ++++--- src/api/identity_binding.rs | 41 ++- src/api/media_api.rs | 119 +++++--- src/api/processing.rs | 12 +- src/api/scan.rs | 335 ++++++++++++++++------ src/api/search.rs | 6 +- src/core/agent/tools.rs | 60 ++-- src/core/chunk/rule1_ingest.rs | 5 +- src/core/chunk/rule2_ingest.rs | 11 +- src/core/chunk/trace_ingest.rs | 70 +++-- src/core/db/postgres_db.rs | 35 ++- src/core/db/qdrant_db.rs | 2 +- src/core/mod.rs | 2 +- src/core/pipeline/mod.rs | 8 +- src/core/processor/heuristic_scene.rs | 5 +- src/core/progress.rs | 149 ++++++---- src/core/tkg/log.rs | 94 +++++-- src/core/tkg/mod.rs | 4 +- src/core/tkg/models.rs | 23 +- src/core/tkg/service.rs | 66 +++-- src/core/tmdb/face_agent.rs | 20 +- src/main.rs | 1 + src/playground.rs | 3 +- src/worker/job_worker.rs | 382 +++++++++++++++++--------- src/worker/processor.rs | 8 +- 27 files changed, 1216 insertions(+), 546 deletions(-) diff --git a/src/api/files.rs b/src/api/files.rs index bc4566e..fe3df14 100644 --- a/src/api/files.rs +++ b/src/api/files.rs @@ -339,9 +339,16 @@ async fn register_single_file( existing_uuid ); if let Err(e) = unregister_internal(&state, &existing_uuid).await { - tracing::error!("[REGISTER] Force unregister failed for {}: {:?}", existing_uuid, e); + tracing::error!( + "[REGISTER] Force unregister failed for {}: {:?}", + existing_uuid, + e + ); } else { - tracing::info!("[REGISTER] Force unregister completed for {}", existing_uuid); + tracing::info!( + "[REGISTER] Force unregister completed for {}", + existing_uuid + ); } } else { let existing_info: Option<(String, String, f64, i32, i32, f64, i64, Option)> = sqlx::query_as( @@ -726,7 +733,11 @@ async fn register_file( existing_uuid ); if let Err(e) = unregister_internal(&state, &existing_uuid).await { - tracing::error!("[REGISTER] Force unregister failed for {}: {:?}", existing_uuid, e); + tracing::error!( + "[REGISTER] Force unregister failed for {}: {:?}", + existing_uuid, + e + ); } } // Also check by content_hash if provided @@ -744,7 +755,11 @@ async fn register_file( existing_uuid ); if let Err(e) = unregister_internal(&state, &existing_uuid).await { - tracing::error!("[REGISTER] Force unregister failed for {}: {:?}", existing_uuid, e); + tracing::error!( + "[REGISTER] Force unregister failed for {}: {:?}", + existing_uuid, + e + ); } } } @@ -1241,25 +1256,31 @@ async fn unregister( match RedisClient::new() { Ok(redis) => { let mut deleted = 0; - + // Delete worker job keys if redis.delete_worker_job(&uuid).await.is_ok() { tracing::info!("[UNREGISTER] Deleted Redis worker job keys for {}", uuid); deleted += 1; } - + // Delete PipelineProgress key - let progress_key = format!("{}progress:{}:pipeline", - crate::core::config::REDIS_KEY_PREFIX.as_str(), uuid); + let progress_key = format!( + "{}progress:{}:pipeline", + crate::core::config::REDIS_KEY_PREFIX.as_str(), + uuid + ); if let Ok(mut conn) = redis.get_conn().await { - let _: Option = redis::cmd("DEL").arg(&progress_key) - .query_async(&mut conn).await.ok(); + let _: Option = redis::cmd("DEL") + .arg(&progress_key) + .query_async(&mut conn) + .await + .ok(); tracing::info!("[UNREGISTER] Deleted Redis PipelineProgress for {}", uuid); deleted += 1; } - + Some(deleted) - }, + } Err(e) => { tracing::warn!("[UNREGISTER] Failed to create Redis client: {}", e); None @@ -1440,11 +1461,18 @@ async fn unregister_internal(state: &AppState, uuid: &str) -> Result<(), StatusC uuid, ) .await; + // Delete from speaker collection + let schema = std::env::var("DATABASE_SCHEMA").unwrap_or_else(|_| "dev".to_string()); + let speaker_collection = if schema == "public" { + "momentry_public_speaker" + } else { + &format!("momentry_{}_speaker", schema) + }; let _ = QdrantDb::delete_by_uuid_from_collection( &qdrant.client, &qdrant.base_url, &qdrant.api_key, - &format!("{}_voice", uuid), + speaker_collection, uuid, ) .await; diff --git a/src/api/identity_agent_api.rs b/src/api/identity_agent_api.rs index 196e15e..9dd2b72 100644 --- a/src/api/identity_agent_api.rs +++ b/src/api/identity_agent_api.rs @@ -11,11 +11,11 @@ use std::path::PathBuf; use std::sync::Arc; use crate::api::types::AppState; +use crate::core::db::redis_client::RedisClient; use crate::core::db::schema; use crate::core::db::PostgresDb; use crate::core::db::QdrantDb; -use crate::core::progress::{AgentPhase, AgentProgress, AgentStats, publish_agent_progress}; -use crate::core::db::redis_client::RedisClient; +use crate::core::progress::{publish_agent_progress, AgentPhase, AgentProgress, AgentStats}; pub fn identity_agent_routes() -> Router { Router::new() @@ -213,17 +213,23 @@ async fn match_from_photo( // 4. Find best matching trace via Qdrant _faces search let qdrant = QdrantDb::new(); - let best_match: Option<(i32, f64)> = match qdrant.search_face_collection( - "_faces", - &embedding_f32, - 1, - "file_uuid", - "", - Some(&file_uuid), - ).await { + let best_match: Option<(i32, f64)> = match qdrant + .search_face_collection( + "_faces", + &embedding_f32, + 1, + "file_uuid", + "", + Some(&file_uuid), + ) + .await + { Ok(hits) if !hits.is_empty() => { let (score, payload) = &hits[0]; - let trace_id = payload.get("trace_id").and_then(|v| v.as_i64()).unwrap_or(0) as i32; + let trace_id = payload + .get("trace_id") + .and_then(|v| v.as_i64()) + .unwrap_or(0) as i32; Some((trace_id, *score)) } _ => None, @@ -295,18 +301,27 @@ async fn match_from_trace( {"key": "trace_id", "match": {"value": req.trace_id}} ] }); - let points = qdrant.scroll_all_points("_faces", trace_filter, 500).await.unwrap_or_default(); + let points = qdrant + .scroll_all_points("_faces", trace_filter, 500) + .await + .unwrap_or_default(); - let all_faces: Vec<(Vec, i64)> = points.iter().filter_map(|p| { - let vector = p.get("vector").and_then(|v| v.as_array())?; - let embedding: Vec = vector.iter().filter_map(|v| v.as_f64().map(|f| f as f32)).collect(); - let frame = p["payload"]["frame"].as_i64()?; - if embedding.len() == 512 { - Some((embedding, frame)) - } else { - None - } - }).collect(); + let all_faces: Vec<(Vec, i64)> = points + .iter() + .filter_map(|p| { + let vector = p.get("vector").and_then(|v| v.as_array())?; + let embedding: Vec = vector + .iter() + .filter_map(|v| v.as_f64().map(|f| f as f32)) + .collect(); + let frame = p["payload"]["frame"].as_i64()?; + if embedding.len() == 512 { + Some((embedding, frame)) + } else { + None + } + }) + .collect(); if all_faces.is_empty() { return Err(( @@ -328,13 +343,16 @@ async fn match_from_trace( let mut query_embeddings: Vec> = Vec::new(); // Get bbox size info from Qdrant payload - let face_sizes: Vec<(i64, i32)> = points.iter().filter_map(|p| { - let frame = p["payload"]["frame"].as_i64()?; - let bbox = &p["payload"]["bbox"]; - let w = bbox["width"].as_f64().unwrap_or(0.0) as i32; - let h = bbox["height"].as_f64().unwrap_or(0.0) as i32; - Some((frame, w * h)) - }).collect(); + let face_sizes: Vec<(i64, i32)> = points + .iter() + .filter_map(|p| { + let frame = p["payload"]["frame"].as_i64()?; + let bbox = &p["payload"]["bbox"]; + let w = bbox["width"].as_f64().unwrap_or(0.0) as i32; + let h = bbox["height"].as_f64().unwrap_or(0.0) as i32; + Some((frame, w * h)) + }) + .collect(); let face_sizes_map: std::collections::HashMap = face_sizes.into_iter().collect(); @@ -373,14 +391,17 @@ async fn match_from_trace( ] }); - let hits = match qdrant.search_face_collection( - "_faces", - qemb, - 1, - "trace_id", - &req.trace_id.to_string(), - Some(&req.file_uuid), - ).await { + let hits = match qdrant + .search_face_collection( + "_faces", + qemb, + 1, + "trace_id", + &req.trace_id.to_string(), + Some(&req.file_uuid), + ) + .await + { Ok(h) => h, Err(e) => { tracing::warn!("[match_from_trace] Qdrant search failed: {}", e); @@ -389,7 +410,10 @@ async fn match_from_trace( }; if let Some((score, payload)) = hits.first() { - let trace_id = payload.get("trace_id").and_then(|v| v.as_i64()).unwrap_or(0) as i32; + let trace_id = payload + .get("trace_id") + .and_then(|v| v.as_i64()) + .unwrap_or(0) as i32; if seen_trace_ids.insert(trace_id) { validated.push((0, trace_id, *score)); } @@ -787,7 +811,10 @@ pub async fn bind_speakers(pool: &sqlx::PgPool, file_uuid: &str) -> anyhow::Resu {"key": "trace_id", "match": {"value": 1}} ] }); - let points = qdrant.scroll_all_points("_faces", trace_filter, 500).await.unwrap_or_default(); + let points = qdrant + .scroll_all_points("_faces", trace_filter, 500) + .await + .unwrap_or_default(); // Group by trace_id, collect frames let mut traces: HashMap> = HashMap::new(); @@ -855,15 +882,14 @@ pub async fn bind_speakers(pool: &sqlx::PgPool, file_uuid: &str) -> anyhow::Resu } // Compute fps from video table - let fps: f64 = sqlx::query_scalar::<_, f64>( - "SELECT COALESCE(fps, 25.0) FROM videos WHERE file_uuid=$1" - ) - .bind(file_uuid) - .fetch_optional(pool) - .await - .ok() - .flatten() - .unwrap_or(25.0); + let fps: f64 = + sqlx::query_scalar::<_, f64>("SELECT COALESCE(fps, 25.0) FROM videos WHERE file_uuid=$1") + .bind(file_uuid) + .fetch_optional(pool) + .await + .ok() + .flatten() + .unwrap_or(25.0); tracing::info!( "[SpeakerBind] Using fps={:.3} for {} ({} traces)", @@ -880,7 +906,8 @@ pub async fn bind_speakers(pool: &sqlx::PgPool, file_uuid: &str) -> anyhow::Resu } // Get identity_id for this trace from Qdrant payload - let identity_id: Option = points.iter() + let identity_id: Option = points + .iter() .find(|p| { p["payload"]["trace_id"].as_i64() == Some(*trace_id as i64) && p["payload"]["identity_id"].as_i64().is_some() @@ -993,7 +1020,11 @@ pub async fn run_identity_agent( let matched = match_faces_iterative(pool, file_uuid).await.unwrap_or(0); progress.stats.tmdb_matches = matched as i64; - progress.update_phase(AgentPhase::TmdbMatching, 1.0, &format!("Face matching: {} matches", matched)); + progress.update_phase( + AgentPhase::TmdbMatching, + 1.0, + &format!("Face matching: {} matches", matched), + ); if let Some(r) = redis.as_ref() { publish_agent_progress(&r, file_uuid, &progress).await; } @@ -1051,9 +1082,11 @@ pub async fn run_identity_agent( .await; } progress.stats.identities_created = identities.len() as i64; - progress.update_phase(AgentPhase::IdentityCreation, 1.0, &format!( - "Created {} identities from clusters", identities.len() - )); + progress.update_phase( + AgentPhase::IdentityCreation, + 1.0, + &format!("Created {} identities from clusters", identities.len()), + ); if let Some(r) = redis.as_ref() { publish_agent_progress(&r, file_uuid, &progress).await; } @@ -1081,7 +1114,11 @@ pub async fn run_identity_agent( let bound = bind_speakers(pool, file_uuid).await.unwrap_or(0); progress.stats.speaker_bindings = bound as i64; - progress.update_phase(AgentPhase::SpeakerBinding, 1.0, &format!("Speaker binding: {} bound", bound)); + progress.update_phase( + AgentPhase::SpeakerBinding, + 1.0, + &format!("Speaker binding: {} bound", bound), + ); if let Some(r) = redis.as_ref() { publish_agent_progress(&r, file_uuid, &progress).await; } @@ -1162,7 +1199,10 @@ async fn find_ready_files(pool: &sqlx::PgPool) -> anyhow::Result> { {"key": "identity_id", "match": {"value": null}} ] }); - let points = qdrant.scroll_all_points("_faces", filter, 1000).await.unwrap_or_default(); + let points = qdrant + .scroll_all_points("_faces", filter, 1000) + .await + .unwrap_or_default(); let mut file_uuids: std::collections::HashSet = std::collections::HashSet::new(); for point in &points { diff --git a/src/api/identity_api.rs b/src/api/identity_api.rs index 757a364..f9b7f12 100644 --- a/src/api/identity_api.rs +++ b/src/api/identity_api.rs @@ -164,8 +164,7 @@ async fn list_files( .await .unwrap_or_default(); - let count_map: std::collections::HashMap = - chunk_counts.into_iter().collect(); + let count_map: std::collections::HashMap = chunk_counts.into_iter().collect(); for item in &mut data { if let Some(cnt) = count_map.get(&item.file_uuid) { @@ -483,13 +482,20 @@ async fn delete_identity( {"key": "identity_id", "match": {"value": identity_id}} ] }); - let points = qdrant.scroll_all_points("_faces", face_filter, 1000).await.unwrap_or_default(); + let points = qdrant + .scroll_all_points("_faces", face_filter, 1000) + .await + .unwrap_or_default(); - let unbound_faces: Vec<(String, Option, Option)> = points.iter() + let unbound_faces: Vec<(String, Option, Option)> = points + .iter() .filter_map(|p| { let payload = &p["payload"]; let file_uuid = payload["file_uuid"].as_str()?.to_string(); - let face_id = payload.get("face_id").and_then(|v| v.as_str()).map(|s| s.to_string()); + let face_id = payload + .get("face_id") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()); let trace_id = payload["trace_id"].as_i64().map(|t| t as i32); Some((file_uuid, face_id, trace_id)) }) @@ -629,8 +635,12 @@ async fn get_identity_files( {"key": "identity_id", "match": {"value": identity_id}} ] }); - let points = qdrant.scroll_all_points("_faces", face_filter, 1000).await.unwrap_or_default(); - let unique_files: std::collections::HashSet = points.iter() + let points = qdrant + .scroll_all_points("_faces", face_filter, 1000) + .await + .unwrap_or_default(); + let unique_files: std::collections::HashSet = points + .iter() .filter_map(|p| p["payload"]["file_uuid"].as_str().map(|s| s.to_string())) .collect(); let total = unique_files.len() as i64; @@ -730,7 +740,10 @@ async fn get_identity_faces( {"key": "identity_id", "match": {"value": identity_id}} ] }); - let points2 = qdrant2.scroll_all_points("_faces", face_filter2, 2000).await.unwrap_or_default(); + let points2 = qdrant2 + .scroll_all_points("_faces", face_filter2, 2000) + .await + .unwrap_or_default(); let total = points2.len() as i64; Ok(Json(IdentityFacesResponse { @@ -827,9 +840,7 @@ async fn get_file_faces( use serde_json::json; let qdrant = QdrantDb::new(); - let mut filter_conditions = vec![ - json!({"key": "file_uuid", "match": {"value": file_uuid}}) - ]; + let mut filter_conditions = vec![json!({"key": "file_uuid", "match": {"value": file_uuid}})]; if let Some(ref binding) = params.binding { match binding.as_str() { @@ -851,30 +862,46 @@ async fn get_file_faces( } let face_filter = json!({"must": filter_conditions}); - let points = qdrant.scroll_all_points("_faces", face_filter, 2000).await.unwrap_or_default(); + let points = qdrant + .scroll_all_points("_faces", face_filter, 2000) + .await + .unwrap_or_default(); // Apply additional filters in Rust - let filtered: Vec<_> = points.into_iter().filter(|p| { - let payload = &p["payload"]; - let confidence = payload["confidence"].as_f64().unwrap_or(0.0); - let frame = payload["frame"].as_i64().unwrap_or(0); + let filtered: Vec<_> = points + .into_iter() + .filter(|p| { + let payload = &p["payload"]; + let confidence = payload["confidence"].as_f64().unwrap_or(0.0); + let frame = payload["frame"].as_i64().unwrap_or(0); - if let Some(mc) = params.min_confidence { - if confidence < mc { return false; } - } - if let Some(sf) = params.start_frame { - if frame < sf { return false; } - } - if let Some(ef) = params.end_frame { - if frame > ef { return false; } - } - true - }).collect(); + if let Some(mc) = params.min_confidence { + if confidence < mc { + return false; + } + } + if let Some(sf) = params.start_frame { + if frame < sf { + return false; + } + } + if let Some(ef) = params.end_frame { + if frame > ef { + return false; + } + } + true + }) + .collect(); let total = filtered.len() as i64; // Apply pagination - let paged: Vec<_> = filtered.into_iter().skip(offset as usize).take(page_size as usize).collect(); + let paged: Vec<_> = filtered + .into_iter() + .skip(offset as usize) + .take(page_size as usize) + .collect(); // Build response items let mut data = Vec::new(); @@ -889,7 +916,10 @@ async fn get_file_faces( file_uuid: file_uuid.clone(), frame_number: frame, timestamp_secs: Some(frame as f64 / fps), - face_id: payload.get("face_id").and_then(|v| v.as_str()).map(|s| s.to_string()), + face_id: payload + .get("face_id") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()), trace_id: payload["trace_id"].as_i64().map(|t| t as i32), bbox: BBox { x: bbox["x"].as_f64().unwrap_or(0.0), @@ -1315,8 +1345,8 @@ async fn set_profile_from_face( Path(identity_uuid): Path, Json(req): Json, ) -> Result, (StatusCode, Json)> { - use crate::core::db::schema; use crate::core::db::qdrant_db::QdrantDb; + use crate::core::db::schema; use serde_json::json; let videos_table = schema::table_name("videos"); @@ -1341,13 +1371,16 @@ async fn set_profile_from_face( let row: Option<(i64, i32, i32, i32, i32, f64)> = if let Some(trace_id) = use_trace { let mut filter_conds = vec![ json!({"key": "file_uuid", "match": {"value": req.file_uuid}}), - json!({"key": "trace_id", "match": {"value": trace_id}}) + json!({"key": "trace_id", "match": {"value": trace_id}}), ]; if let Some(frame) = use_frame { filter_conds.push(json!({"key": "frame", "match": {"value": frame}})); } let face_filter = json!({"must": filter_conds}); - let points = qdrant.scroll_all_points("_faces", face_filter, 10).await.unwrap_or_default(); + let points = qdrant + .scroll_all_points("_faces", face_filter, 10) + .await + .unwrap_or_default(); points.first().map(|p| { let payload = &p["payload"]; let bbox = &payload["bbox"]; diff --git a/src/api/identity_binding.rs b/src/api/identity_binding.rs index eac846c..d46e20a 100644 --- a/src/api/identity_binding.rs +++ b/src/api/identity_binding.rs @@ -249,7 +249,10 @@ pub async fn unbind_identity( {"key": "trace_id", "match": {"value": tid}} ] }); - let points = qdrant.scroll_points("_faces", filter, 1, None).await.unwrap_or_default(); + let points = qdrant + .scroll_points("_faces", filter, 1, None) + .await + .unwrap_or_default(); points.0.first().map(|p| { let identity_id = p["payload"]["identity_id"].as_i64().map(|i| i as i32); (identity_id, Some(tid)) @@ -260,7 +263,9 @@ pub async fn unbind_identity( None } else { // face_id lookup not supported in Qdrant - tracing::warn!("[unbind_identity] face_id field lookup not supported, use trace_id instead"); + tracing::warn!( + "[unbind_identity] face_id field lookup not supported, use trace_id instead" + ); None }; @@ -440,7 +445,6 @@ pub async fn merge_identities( Path(identity_uuid): Path, Json(req): Json, ) -> Result>, (StatusCode, Json)> { - let id_table = crate::core::db::schema::table_name("identities"); let db = sqlx::PgPool::connect(&crate::core::config::DATABASE_URL) @@ -523,10 +527,17 @@ pub async fn merge_identities( {"key": "identity_id", "match": {"value": from_id}} ] }); - let points = qdrant.scroll_all_points("_faces", face_filter, 1000).await.unwrap_or_default(); - let face_ids: Vec<(Option, Option)> = points.iter() + let points = qdrant + .scroll_all_points("_faces", face_filter, 1000) + .await + .unwrap_or_default(); + let face_ids: Vec<(Option, Option)> = points + .iter() .map(|p| { - let face_id = p["payload"].get("face_id").and_then(|v| v.as_str()).map(|s| s.to_string()); + let face_id = p["payload"] + .get("face_id") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()); let trace_id = p["payload"]["trace_id"].as_i64().map(|t| t as i32); (face_id, trace_id) }) @@ -537,7 +548,8 @@ pub async fn merge_identities( let faces_count = face_id_list.len() as i64; // Get file_uuid for these faces (from Qdrant) - let file_uuid: Option = points.first() + let file_uuid: Option = points + .first() .and_then(|p| p["payload"]["file_uuid"].as_str()) .map(|s| s.to_string()); @@ -1651,7 +1663,9 @@ async fn apply_bind_snapshot( // Legacy face_id path not supported - no trace_id in snapshot Err(( StatusCode::INTERNAL_SERVER_ERROR, - Json(serde_json::json!({"error": "Snapshot has no trace_id, legacy face_id path not supported"})), + Json( + serde_json::json!({"error": "Snapshot has no trace_id, legacy face_id path not supported"}), + ), )) } } @@ -1789,8 +1803,7 @@ pub async fn bind_redo( let after = &records.last().unwrap().2; let identity_id_after = after.get("identity_id_after").cloned(); - let affected_rows = - apply_bind_snapshot(state.db.pool(), "", after, identity_id_after).await?; + let affected_rows = apply_bind_snapshot(state.db.pool(), "", after, identity_id_after).await?; // Mark all as redone for (history_id, _, _) in &records { @@ -2098,8 +2111,12 @@ pub async fn list_pending_persons( {"key": "identity_id", "match": {"value": id}} ] }); - let points = qdrant.scroll_all_points("_faces", face_filter, 500).await.unwrap_or_default(); - let unique_traces: std::collections::HashSet = points.iter() + let points = qdrant + .scroll_all_points("_faces", face_filter, 500) + .await + .unwrap_or_default(); + let unique_traces: std::collections::HashSet = points + .iter() .filter_map(|p| p["payload"]["trace_id"].as_i64()) .collect(); let trace_count = unique_traces.len() as i64; diff --git a/src/api/media_api.rs b/src/api/media_api.rs index 82105bf..7799979 100644 --- a/src/api/media_api.rs +++ b/src/api/media_api.rs @@ -232,20 +232,29 @@ async fn bbox_overlay_video( {"key": "trace_id", "match": {"value": 1}} ] }); - let points = qdrant.scroll_all_points("_faces", face_filter, 500).await.unwrap_or_default(); + let points = qdrant + .scroll_all_points("_faces", face_filter, 500) + .await + .unwrap_or_default(); - let rows: Vec<(i64, i32, i32, i32, i32, Option, Option)> = points.iter().filter_map(|p| { - let payload = &p["payload"]; - let frame = payload["frame"].as_i64()?; - let bbox = &payload["bbox"]; - let x = bbox["x"].as_f64()? as i32; - let y = bbox["y"].as_f64()? as i32; - let w = bbox["width"].as_f64()? as i32; - let h = bbox["height"].as_f64()? as i32; - let trace_id = payload["trace_id"].as_i64().map(|t| t as i32); - let face_id = payload.get("face_id").and_then(|v| v.as_str()).map(|s| s.to_string()); - Some((frame, x, y, w, h, trace_id, face_id)) - }).collect(); + let rows: Vec<(i64, i32, i32, i32, i32, Option, Option)> = points + .iter() + .filter_map(|p| { + let payload = &p["payload"]; + let frame = payload["frame"].as_i64()?; + let bbox = &payload["bbox"]; + let x = bbox["x"].as_f64()? as i32; + let y = bbox["y"].as_f64()? as i32; + let w = bbox["width"].as_f64()? as i32; + let h = bbox["height"].as_f64()? as i32; + let trace_id = payload["trace_id"].as_i64().map(|t| t as i32); + let face_id = payload + .get("face_id") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()); + Some((frame, x, y, w, h, trace_id, face_id)) + }) + .collect(); // Build filters — each bbox enabled only on its frame let mut parts: Vec = Vec::new(); @@ -362,18 +371,24 @@ async fn trace_video_inner( {"key": "trace_id", "match": {"value": trace_id}} ] }); - let points = qdrant.scroll_all_points("_faces", trace_filter, 500).await.unwrap_or_default(); + let points = qdrant + .scroll_all_points("_faces", trace_filter, 500) + .await + .unwrap_or_default(); - let rows: Vec<(i64, i32, i32, i32, i32)> = points.iter().filter_map(|p| { - let payload = &p["payload"]; - let frame = payload["frame"].as_i64()?; - let bbox = &payload["bbox"]; - let x = bbox["x"].as_f64()? as i32; - let y = bbox["y"].as_f64()? as i32; - let w = bbox["width"].as_f64()? as i32; - let h = bbox["height"].as_f64()? as i32; - Some((frame, x, y, w, h)) - }).collect(); + let rows: Vec<(i64, i32, i32, i32, i32)> = points + .iter() + .filter_map(|p| { + let payload = &p["payload"]; + let frame = payload["frame"].as_i64()?; + let bbox = &payload["bbox"]; + let x = bbox["x"].as_f64()? as i32; + let y = bbox["y"].as_f64()? as i32; + let w = bbox["width"].as_f64()? as i32; + let h = bbox["height"].as_f64()? as i32; + Some((frame, x, y, w, h)) + }) + .collect(); if rows.is_empty() { return Err(StatusCode::NOT_FOUND); @@ -424,13 +439,20 @@ async fn trace_video_inner( // Query all traces with identity names and bbox positions in the visible frame range let identities_table = schema::table_name("identities"); - let all_points = qdrant.scroll_all_points("_faces", json!({ - "must": [ - {"key": "file_uuid", "match": {"value": file_uuid}}, - {"key": "frame", "range": {"gte": start_fn, "lte": end_fn}}, - {"key": "trace_id", "match": {"value": 1}} - ] - }), 1000).await.unwrap_or_default(); + let all_points = qdrant + .scroll_all_points( + "_faces", + json!({ + "must": [ + {"key": "file_uuid", "match": {"value": file_uuid}}, + {"key": "frame", "range": {"gte": start_fn, "lte": end_fn}}, + {"key": "trace_id", "match": {"value": 1}} + ] + }), + 1000, + ) + .await + .unwrap_or_default(); // Get identity names for traces that have identity_id let mut identity_names: HashMap = HashMap::new(); @@ -455,18 +477,21 @@ async fn trace_video_inner( } } - let all_rows: Vec<(i32, i64, i32, i32, i32, i32, Option)> = all_points.iter().filter_map(|p| { - let payload = &p["payload"]; - let trace_id = payload["trace_id"].as_i64()? as i32; - let frame = payload["frame"].as_i64()?; - let bbox = &payload["bbox"]; - let x = bbox["x"].as_f64()? as i32; - let y = bbox["y"].as_f64()? as i32; - let w = bbox["width"].as_f64()? as i32; - let h = bbox["height"].as_f64()? as i32; - let name = identity_names.get(&trace_id).cloned(); - Some((trace_id, frame, x, y, w, h, name)) - }).collect(); + let all_rows: Vec<(i32, i64, i32, i32, i32, i32, Option)> = all_points + .iter() + .filter_map(|p| { + let payload = &p["payload"]; + let trace_id = payload["trace_id"].as_i64()? as i32; + let frame = payload["frame"].as_i64()?; + let bbox = &payload["bbox"]; + let x = bbox["x"].as_f64()? as i32; + let y = bbox["y"].as_f64()? as i32; + let w = bbox["width"].as_f64()? as i32; + let h = bbox["height"].as_f64()? as i32; + let name = identity_names.get(&trace_id).cloned(); + Some((trace_id, frame, x, y, w, h, name)) + }) + .collect(); // Group frames by trace_id, compute start_frame per trace; collect bbox per frame // frame_number is i64 (BIGINT), so HashMaps need i64 for frame values @@ -1169,9 +1194,13 @@ async fn stranger_video_inner( {"key": "stranger_id", "match": {"value": stranger_id}} ] }); - let points = qdrant.scroll_all_points("_faces", face_filter, 1000).await.unwrap_or_default(); + let points = qdrant + .scroll_all_points("_faces", face_filter, 1000) + .await + .unwrap_or_default(); - let rows: Vec<(i64, i32, i32, i32, i32)> = points.iter() + let rows: Vec<(i64, i32, i32, i32, i32)> = points + .iter() .filter_map(|p| { let payload = &p["payload"]; let frame = payload["frame"].as_i64()?; diff --git a/src/api/processing.rs b/src/api/processing.rs index 093170f..e2f197e 100644 --- a/src/api/processing.rs +++ b/src/api/processing.rs @@ -471,14 +471,22 @@ async fn get_progress(file_uuid: Path) -> Result let agent_key = format!("{}progress:{}:agent", REDIS_KEY_PREFIX.as_str(), file_uuid); let tkg_progress: Option = if let Ok(mut c) = redis.get_conn().await { - let val: Option = redis::cmd("GET").arg(&tkg_key).query_async(&mut c).await.ok(); + let val: Option = redis::cmd("GET") + .arg(&tkg_key) + .query_async(&mut c) + .await + .ok(); val.and_then(|s| serde_json::from_str(&s).ok()) } else { None }; let agent_progress: Option = if let Ok(mut c) = redis.get_conn().await { - let val: Option = redis::cmd("GET").arg(&agent_key).query_async(&mut c).await.ok(); + let val: Option = redis::cmd("GET") + .arg(&agent_key) + .query_async(&mut c) + .await + .ok(); val.and_then(|s| serde_json::from_str(&s).ok()) } else { None diff --git a/src/api/scan.rs b/src/api/scan.rs index 5ed68b9..6820066 100644 --- a/src/api/scan.rs +++ b/src/api/scan.rs @@ -460,7 +460,10 @@ async fn get_ingestion_status( {"key": "file_uuid", "match": {"value": file_uuid}} ] }); - let points = qdrant.scroll_all_points("_faces", face_filter, 1000).await.unwrap_or_default(); + let points = qdrant + .scroll_all_points("_faces", face_filter, 1000) + .await + .unwrap_or_default(); let face_total = points.len() as i64; let mut trace_ids: std::collections::HashSet = std::collections::HashSet::new(); @@ -564,7 +567,11 @@ async fn get_ingestion_status( // Get related identities from Qdrant _faces let related_identity_ids: Vec = identity_ids.into_iter().collect(); let related_identities: Vec = if !related_identity_ids.is_empty() { - let id_list: String = related_identity_ids.iter().map(|id| id.to_string()).collect::>().join(","); + let id_list: String = related_identity_ids + .iter() + .map(|id| id.to_string()) + .collect::>() + .join(","); match sqlx::query_as::<_, (String, String)>(&format!( "SELECT DISTINCT uuid::text, name FROM {identities} \ WHERE id IN ({id_list}) ORDER BY name" @@ -592,21 +599,25 @@ async fn get_ingestion_status( // Check if job is completed - if so, all ingestion steps are considered done let mj_table = schema::table_name("monitor_jobs"); - let job_completed: bool = sqlx::query_scalar::<_, String>(&format!( - "SELECT status FROM {mj_table} WHERE uuid = $1" - )) - .bind(&file_uuid) - .fetch_optional(pool) - .await - .unwrap_or(None) - .map(|s| s == "completed") - .unwrap_or(false); + let job_completed: bool = + sqlx::query_scalar::<_, String>(&format!("SELECT status FROM {mj_table} WHERE uuid = $1")) + .bind(&file_uuid) + .fetch_optional(pool) + .await + .unwrap_or(None) + .map(|s| s == "completed") + .unwrap_or(false); macro_rules! step { ($name:expr, $done:expr, $detail:expr) => { IngestionStep { name: $name.into(), - status: if $done || job_completed { "done" } else { "pending" }.into(), + status: if $done || job_completed { + "done" + } else { + "pending" + } + .into(), detail: $detail, } }; @@ -639,24 +650,92 @@ async fn get_ingestion_status( Some(format!("{trace_count} traces / {face_total} detections")) ), // TKG Nodes - step!("tkg_face_track", face_track_nodes > 0, Some(format!("{face_track_nodes} nodes"))), - step!("tkg_gaze_track", gaze_track_nodes > 0, Some(format!("{gaze_track_nodes} nodes"))), - step!("tkg_lip_track", lip_track_nodes > 0, Some(format!("{lip_track_nodes} nodes"))), - step!("tkg_text_region", text_region_nodes > 0, Some(format!("{text_region_nodes} nodes"))), - step!("tkg_appearance", appearance_nodes > 0, Some(format!("{appearance_nodes} nodes"))), - step!("tkg_accessory", accessory_nodes > 0, Some(format!("{accessory_nodes} nodes"))), - step!("tkg_object", object_nodes > 0, Some(format!("{object_nodes} nodes"))), - step!("tkg_hand", hand_nodes > 0, Some(format!("{hand_nodes} nodes"))), - step!("tkg_speaker", speaker_nodes > 0, Some(format!("{speaker_nodes} nodes"))), + step!( + "tkg_face_track", + face_track_nodes > 0, + Some(format!("{face_track_nodes} nodes")) + ), + step!( + "tkg_gaze_track", + gaze_track_nodes > 0, + Some(format!("{gaze_track_nodes} nodes")) + ), + step!( + "tkg_lip_track", + lip_track_nodes > 0, + Some(format!("{lip_track_nodes} nodes")) + ), + step!( + "tkg_text_region", + text_region_nodes > 0, + Some(format!("{text_region_nodes} nodes")) + ), + step!( + "tkg_appearance", + appearance_nodes > 0, + Some(format!("{appearance_nodes} nodes")) + ), + step!( + "tkg_accessory", + accessory_nodes > 0, + Some(format!("{accessory_nodes} nodes")) + ), + step!( + "tkg_object", + object_nodes > 0, + Some(format!("{object_nodes} nodes")) + ), + step!( + "tkg_hand", + hand_nodes > 0, + Some(format!("{hand_nodes} nodes")) + ), + step!( + "tkg_speaker", + speaker_nodes > 0, + Some(format!("{speaker_nodes} nodes")) + ), // TKG Edges - step!("tkg_co_occurrence", co_occurrence_edges > 0, Some(format!("{co_occurrence_edges} edges"))), - step!("tkg_speaker_face", speaker_face_edges > 0, Some(format!("{speaker_face_edges} edges"))), - step!("tkg_face_face", face_face_edges > 0, Some(format!("{face_face_edges} edges"))), - step!("tkg_mutual_gaze", mutual_gaze_edges > 0, Some(format!("{mutual_gaze_edges} edges"))), - step!("tkg_lip_sync", lip_sync_edges > 0, Some(format!("{lip_sync_edges} edges"))), - step!("tkg_has_appearance", has_appearance_edges > 0, Some(format!("{has_appearance_edges} edges"))), - step!("tkg_wears", wears_edges > 0, Some(format!("{wears_edges} edges"))), - step!("tkg_hand_object", hand_object_edges > 0, Some(format!("{hand_object_edges} edges"))), + step!( + "tkg_co_occurrence", + co_occurrence_edges > 0, + Some(format!("{co_occurrence_edges} edges")) + ), + step!( + "tkg_speaker_face", + speaker_face_edges > 0, + Some(format!("{speaker_face_edges} edges")) + ), + step!( + "tkg_face_face", + face_face_edges > 0, + Some(format!("{face_face_edges} edges")) + ), + step!( + "tkg_mutual_gaze", + mutual_gaze_edges > 0, + Some(format!("{mutual_gaze_edges} edges")) + ), + step!( + "tkg_lip_sync", + lip_sync_edges > 0, + Some(format!("{lip_sync_edges} edges")) + ), + step!( + "tkg_has_appearance", + has_appearance_edges > 0, + Some(format!("{has_appearance_edges} edges")) + ), + step!( + "tkg_wears", + wears_edges > 0, + Some(format!("{wears_edges} edges")) + ), + step!( + "tkg_hand_object", + hand_object_edges > 0, + Some(format!("{hand_object_edges} edges")) + ), // Rule 2 step!( "rule2_relationship", @@ -703,34 +782,45 @@ async fn get_file_stats( .unwrap_or_else(|| (file_uuid.clone(), None, None)); // 2. Get processor status from processing_status JSONB - let processing_status: serde_json::Value = - sqlx::query_scalar(&format!( - "SELECT processing_status FROM {} WHERE file_uuid = $1", - videos_table - )) - .bind(&file_uuid) - .fetch_optional(pool) - .await - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? - .unwrap_or(serde_json::json!({})); + let processing_status: serde_json::Value = sqlx::query_scalar(&format!( + "SELECT processing_status FROM {} WHERE file_uuid = $1", + videos_table + )) + .bind(&file_uuid) + .fetch_optional(pool) + .await + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? + .unwrap_or(serde_json::json!({})); - let processors: Vec = processing_status + let progress_map = processing_status .get("progress") .and_then(|p| p.as_object()) - .map(|progress| { - progress - .iter() - .filter_map(|(name, info)| { + .cloned() + .unwrap_or_default(); + + let active_order: Vec = processing_status + .get("active_processors") + .and_then(|a| a.as_array()) + .map(|arr| { + arr.iter() + .filter_map(|v| v.as_str().map(|s| s.to_uppercase())) + .collect() + }) + .unwrap_or_default(); + + let processors: Vec = if !active_order.is_empty() { + active_order + .iter() + .filter_map(|name| { + progress_map.get(name).and_then(|info| { info.as_object().map(|obj| { let status = obj .get("status") .and_then(|s| s.as_str()) .unwrap_or("pending") .to_string(); - let progress_val = obj - .get("percentage") - .and_then(|p| p.as_u64()) - .unwrap_or(0) as u32; + let progress_val = + obj.get("percentage").and_then(|p| p.as_u64()).unwrap_or(0) as u32; let message = obj .get("message") .and_then(|m| m.as_str()) @@ -743,9 +833,34 @@ async fn get_file_stats( } }) }) - .collect() - }) - .unwrap_or_default(); + }) + .collect() + } else { + progress_map + .iter() + .filter_map(|(name, info)| { + info.as_object().map(|obj| { + let status = obj + .get("status") + .and_then(|s| s.as_str()) + .unwrap_or("pending") + .to_string(); + let progress_val = + obj.get("percentage").and_then(|p| p.as_u64()).unwrap_or(0) as u32; + let message = obj + .get("message") + .and_then(|m| m.as_str()) + .map(|s| s.to_string()); + ProcessorStatus { + name: name.clone(), + status, + progress: progress_val, + message, + } + }) + }) + .collect() + }; // 3. Get PostgreSQL counts let chunk_table = schema::table_name("chunk"); @@ -853,10 +968,22 @@ async fn get_file_stats( let tkg_nodes_table = schema::table_name("tkg_nodes"); let tkg_edges_table = schema::table_name("tkg_edges"); - let tkg_nodes_total: i64 = sqlx::query_scalar::<_, i64>(&format!("SELECT COUNT(*) FROM {} WHERE file_uuid = $1", tkg_nodes_table)) - .bind(&file_uuid).fetch_one(pool).await.unwrap_or(0); - let tkg_edges_total: i64 = sqlx::query_scalar::<_, i64>(&format!("SELECT COUNT(*) FROM {} WHERE file_uuid = $1", tkg_edges_table)) - .bind(&file_uuid).fetch_one(pool).await.unwrap_or(0); + let tkg_nodes_total: i64 = sqlx::query_scalar::<_, i64>(&format!( + "SELECT COUNT(*) FROM {} WHERE file_uuid = $1", + tkg_nodes_table + )) + .bind(&file_uuid) + .fetch_one(pool) + .await + .unwrap_or(0); + let tkg_edges_total: i64 = sqlx::query_scalar::<_, i64>(&format!( + "SELECT COUNT(*) FROM {} WHERE file_uuid = $1", + tkg_edges_table + )) + .bind(&file_uuid) + .fetch_one(pool) + .await + .unwrap_or(0); let tkg = TkgFileStats { total_nodes: tkg_nodes_total, @@ -870,12 +997,14 @@ async fn get_file_stats( object_nodes: count_nodes(pool, &tkg_nodes_table, &file_uuid, "yolo_object").await, hand_nodes: count_nodes(pool, &tkg_nodes_table, &file_uuid, "hand").await, speaker_nodes: count_nodes(pool, &tkg_nodes_table, &file_uuid, "speaker").await, - co_occurrence_edges: count_edges(pool, &tkg_edges_table, &file_uuid, "CO_OCCURS_WITH").await, + co_occurrence_edges: count_edges(pool, &tkg_edges_table, &file_uuid, "CO_OCCURS_WITH") + .await, speaker_face_edges: count_edges(pool, &tkg_edges_table, &file_uuid, "SPEAKS_AS").await, face_face_edges: count_edges(pool, &tkg_edges_table, &file_uuid, "FACE_TO_FACE").await, mutual_gaze_edges: count_edges(pool, &tkg_edges_table, &file_uuid, "MUTUAL_GAZE").await, lip_sync_edges: count_edges(pool, &tkg_edges_table, &file_uuid, "LIP_SYNC").await, - has_appearance_edges: count_edges(pool, &tkg_edges_table, &file_uuid, "HAS_APPEARANCE").await, + has_appearance_edges: count_edges(pool, &tkg_edges_table, &file_uuid, "HAS_APPEARANCE") + .await, wears_edges: count_edges(pool, &tkg_edges_table, &file_uuid, "WEARS").await, hand_object_edges: count_edges(pool, &tkg_edges_table, &file_uuid, "HAND_OBJECT").await, ..Default::default() @@ -895,7 +1024,8 @@ async fn get_file_stats( let identity_agent = IdentityAgentStats { clusters: 0, // From face_clustered.json if available identities_created: face_identities.len() as i64, - tmdb_matches: seed_points.iter() + tmdb_matches: seed_points + .iter() .filter(|p| p["payload"]["source"].as_str() == Some("tmdb")) .count() as i64, speaker_bindings: speaker_points.len() as i64, @@ -946,10 +1076,7 @@ pub fn scan_routes() -> Router { "/api/v1/stats/ingestion-status/:file_uuid", get(get_ingestion_status), ) - .route( - "/api/v1/stats/file/:file_uuid", - get(get_file_stats), - ) + .route("/api/v1/stats/file/:file_uuid", get(get_file_stats)) .route( "/api/v1/stats/pipeline/:file_uuid", get(get_pipeline_progress_handler), @@ -969,9 +1096,13 @@ async fn get_pipeline_progress_handler( let mj_table = schema::table_name("monitor_jobs"); // Compute actual progress from DB state - let sentence_count: i64 = sqlx::query_scalar::<_, i64>( - &format!("SELECT COUNT(*) FROM {chunk_table} WHERE file_uuid = $1 AND chunk_type = 'sentence'") - ).bind(&file_uuid).fetch_one(pool).await.unwrap_or(0); + let sentence_count: i64 = sqlx::query_scalar::<_, i64>(&format!( + "SELECT COUNT(*) FROM {chunk_table} WHERE file_uuid = $1 AND chunk_type = 'sentence'" + )) + .bind(&file_uuid) + .fetch_one(pool) + .await + .unwrap_or(0); let sentence_embedded: i64 = sqlx::query_scalar::<_, i64>( &format!("SELECT COUNT(*) FROM {chunk_table} WHERE file_uuid = $1 AND chunk_type = 'sentence' AND embedding IS NOT NULL") @@ -981,17 +1112,29 @@ async fn get_pipeline_progress_handler( &format!("SELECT COUNT(*) FROM {pr_table} pr JOIN {mj_table} mj ON pr.job_id = mj.id WHERE mj.uuid = $1 AND pr.processor = 'face' AND pr.status = 'completed'") ).bind(&file_uuid).fetch_one(pool).await.unwrap_or(0); - let tkg_node_count: i64 = sqlx::query_scalar::<_, i64>( - &format!("SELECT COUNT(*) FROM {tkg_nodes_table} WHERE file_uuid = $1") - ).bind(&file_uuid).fetch_one(pool).await.unwrap_or(0); + let tkg_node_count: i64 = sqlx::query_scalar::<_, i64>(&format!( + "SELECT COUNT(*) FROM {tkg_nodes_table} WHERE file_uuid = $1" + )) + .bind(&file_uuid) + .fetch_one(pool) + .await + .unwrap_or(0); - let tkg_edge_count: i64 = sqlx::query_scalar::<_, i64>( - &format!("SELECT COUNT(*) FROM {tkg_edges_table} WHERE file_uuid = $1") - ).bind(&file_uuid).fetch_one(pool).await.unwrap_or(0); + let tkg_edge_count: i64 = sqlx::query_scalar::<_, i64>(&format!( + "SELECT COUNT(*) FROM {tkg_edges_table} WHERE file_uuid = $1" + )) + .bind(&file_uuid) + .fetch_one(pool) + .await + .unwrap_or(0); - let relationship_count: i64 = sqlx::query_scalar::<_, i64>( - &format!("SELECT COUNT(*) FROM {chunk_table} WHERE file_uuid = $1 AND chunk_type = 'relationship'") - ).bind(&file_uuid).fetch_one(pool).await.unwrap_or(0); + let relationship_count: i64 = sqlx::query_scalar::<_, i64>(&format!( + "SELECT COUNT(*) FROM {chunk_table} WHERE file_uuid = $1 AND chunk_type = 'relationship'" + )) + .bind(&file_uuid) + .fetch_one(pool) + .await + .unwrap_or(0); let asrx_completed: i64 = sqlx::query_scalar::<_, i64>( &format!("SELECT COUNT(*) FROM {pr_table} pr JOIN {mj_table} mj ON pr.job_id = mj.id WHERE mj.uuid = $1 AND pr.processor = 'asrx' AND pr.status = 'completed'") @@ -1007,7 +1150,10 @@ async fn get_pipeline_progress_handler( } if sentence_count > 0 { let detail = if sentence_embedded > 0 { - Some(format!("{} chunks, {} embedded", sentence_count, sentence_embedded)) + Some(format!( + "{} chunks, {} embedded", + sentence_count, sentence_embedded + )) } else { Some(format!("{} chunks", sentence_count)) }; @@ -1017,13 +1163,28 @@ async fn get_pipeline_progress_handler( pp.update_stage("face_tracing", 1.0, "completed", None); } if tkg_node_count > 0 { - pp.update_stage("tkg_nodes", 1.0, "completed", Some(format!("{} nodes", tkg_node_count))); + pp.update_stage( + "tkg_nodes", + 1.0, + "completed", + Some(format!("{} nodes", tkg_node_count)), + ); } if tkg_edge_count > 0 { - pp.update_stage("tkg_edges", 1.0, "completed", Some(format!("{} edges", tkg_edge_count))); + pp.update_stage( + "tkg_edges", + 1.0, + "completed", + Some(format!("{} edges", tkg_edge_count)), + ); } if relationship_count > 0 { - pp.update_stage("rule2_ingestion", 1.0, "completed", Some(format!("{} chunks", relationship_count))); + pp.update_stage( + "rule2_ingestion", + 1.0, + "completed", + Some(format!("{} chunks", relationship_count)), + ); } // Check identity agent from _seeds @@ -1031,11 +1192,23 @@ async fn get_pipeline_progress_handler( use serde_json::json; let qdrant = QdrantDb::new(); let schema = std::env::var("DATABASE_SCHEMA").unwrap_or_else(|_| "dev".to_string()); - let seeds_collection = if schema == "public" { "momentry_public_speaker" } else { &format!("momentry_{}_speaker", schema) }; + let seeds_collection = if schema == "public" { + "momentry_public_speaker" + } else { + &format!("momentry_{}_speaker", schema) + }; let seeds_filter = json!({"must": [{"key": "file_uuid", "match": {"value": &file_uuid}}]}); - let seed_points = qdrant.scroll_all_points("_seeds", seeds_filter, 100).await.unwrap_or_default(); + let seed_points = qdrant + .scroll_all_points("_seeds", seeds_filter, 100) + .await + .unwrap_or_default(); if !seed_points.is_empty() { - pp.update_stage("identity_agent", 1.0, "completed", Some(format!("{} seeds", seed_points.len()))); + pp.update_stage( + "identity_agent", + 1.0, + "completed", + Some(format!("{} seeds", seed_points.len())), + ); } Ok(Json(pp)) diff --git a/src/api/search.rs b/src/api/search.rs index 5dd9896..d9abbbe 100644 --- a/src/api/search.rs +++ b/src/api/search.rs @@ -386,9 +386,11 @@ pub async fn smart_search( // 取更多結果以便過濾 // Use no_embedding version for keyword results, regular for semantic let pg_opt = if mr.keyword_score.is_some() && mr.semantic_score.is_none() { - db.get_chunk_by_id_no_embedding(&mr.file_uuid, &mr.chunk_id).await + db.get_chunk_by_id_no_embedding(&mr.file_uuid, &mr.chunk_id) + .await } else { - db.get_chunk_by_file_and_chunk_id(&mr.file_uuid, &mr.chunk_id).await + db.get_chunk_by_file_and_chunk_id(&mr.file_uuid, &mr.chunk_id) + .await }; if let Some(pg) = pg_opt.ok().flatten() { // 關鍵字結果跳過 text_match 過濾（search_bm25 已經匹配過） diff --git a/src/core/agent/tools.rs b/src/core/agent/tools.rs index d9cfffd..55a9e7c 100644 --- a/src/core/agent/tools.rs +++ b/src/core/agent/tools.rs @@ -1094,13 +1094,18 @@ pub async fn exec_tkg_node_detail( } /// Search for people by clothing color using appearance data -pub async fn exec_search_by_appearance(pool: &sqlx::PgPool, args: &serde_json::Value) -> Result { - let file_uuid = args.get("file_uuid") +pub async fn exec_search_by_appearance( + pool: &sqlx::PgPool, + args: &serde_json::Value, +) -> Result { + let file_uuid = args + .get("file_uuid") .and_then(|v| v.as_str()) .ok_or("file_uuid is required".to_string())?; - let color = args.get("color") - .and_then(|v| v.as_str()) - .ok_or("color is required (red, blue, green, yellow, orange, cyan, purple, white, black)".to_string())?; + let color = args.get("color").and_then(|v| v.as_str()).ok_or( + "color is required (red, blue, green, yellow, orange, cyan, purple, white, black)" + .to_string(), + )?; let output_dir = std::env::var("MOMENTRY_OUTPUT_DIR") .unwrap_or_else(|_| "/Users/accusys/momentry/output".to_string()); @@ -1117,7 +1122,8 @@ pub async fn exec_search_by_appearance(pool: &sqlx::PgPool, args: &serde_json::V // Get video path from videos table let videos_table = schema::table_name("videos"); let video_path: Option = sqlx::query_scalar(&format!( - "SELECT file_path FROM {} WHERE file_uuid = $1", videos_table + "SELECT file_path FROM {} WHERE file_uuid = $1", + videos_table )) .bind(file_uuid) .fetch_optional(pool) @@ -1129,29 +1135,33 @@ pub async fn exec_search_by_appearance(pool: &sqlx::PgPool, args: &serde_json::V return Err("Video path not found".to_string()); } - let executor = crate::core::processor::PythonExecutor::new() + let executor = crate::core::processor::PythonExecutor::new().map_err(|e| e.to_string())?; + + executor + .run( + &script_path, + &[ + "--file-uuid", + file_uuid, + "--color", + color, + "--video-path", + &video_path, + "--appearance-path", + &appearance_path, + "--output", + &output_path, + ], + None, + "CLOTHING_COLOR_SEARCH", + Some(std::time::Duration::from_secs(300)), + ) + .await .map_err(|e| e.to_string())?; - - executor.run( - &script_path, - &[ - "--file-uuid", file_uuid, - "--color", color, - "--video-path", &video_path, - "--appearance-path", &appearance_path, - "--output", &output_path, - ], - None, - "CLOTHING_COLOR_SEARCH", - Some(std::time::Duration::from_secs(300)), - ) - .await - .map_err(|e| e.to_string())?; // Read results if std::path::Path::new(&output_path).exists() { - let content = std::fs::read_to_string(&output_path) - .map_err(|e| e.to_string())?; + let content = std::fs::read_to_string(&output_path).map_err(|e| e.to_string())?; Ok(content) } else { Err("Color search output not found".to_string()) diff --git a/src/core/chunk/rule1_ingest.rs b/src/core/chunk/rule1_ingest.rs index f987a16..847b499 100644 --- a/src/core/chunk/rule1_ingest.rs +++ b/src/core/chunk/rule1_ingest.rs @@ -290,10 +290,7 @@ fn collect_ocr_text( /// Group ALL OCR frames by proximity into chunks /// Returns vec of (start_frame, end_frame, combined_ocr_text) -fn group_ocr_frames( - ocr_map: &BTreeMap>, - _fps: f64, -) -> Vec<(i64, i64, String)> { +fn group_ocr_frames(ocr_map: &BTreeMap>, _fps: f64) -> Vec<(i64, i64, String)> { const MAX_FRAME_GAP: i64 = 5; // ~0.2s at 24fps let mut result = Vec::new(); diff --git a/src/core/chunk/rule2_ingest.rs b/src/core/chunk/rule2_ingest.rs index 230c21a..ffe67b1 100644 --- a/src/core/chunk/rule2_ingest.rs +++ b/src/core/chunk/rule2_ingest.rs @@ -1,10 +1,10 @@ +use crate::core::db::redis_client::RedisClient; use crate::core::db::schema; use anyhow::{Context, Result}; use serde_json::Value; use sqlx::PgPool; -use tracing::{info, warn}; use std::sync::Arc; -use crate::core::db::redis_client::RedisClient; +use tracing::{info, warn}; fn t(name: &str) -> String { let schema = std::env::var("DATABASE_SCHEMA").unwrap_or_else(|_| "dev".to_string()); @@ -24,7 +24,12 @@ pub type Rule2ProgressFn = Box; /// 2. Resolve source/target nodes and identities. /// 3. Generate natural language description (template-based). /// 4. Insert chunks with chunk_type='relationship'. -pub async fn ingest_rule2(pool: &PgPool, file_uuid: &str, redis: Option>, progress_fn: Option) -> Result { +pub async fn ingest_rule2( + pool: &PgPool, + file_uuid: &str, + redis: Option>, + progress_fn: Option, +) -> Result { let edges_table = t("tkg_edges"); let nodes_table = t("tkg_nodes"); let chunk_table = t("chunk"); diff --git a/src/core/chunk/trace_ingest.rs b/src/core/chunk/trace_ingest.rs index d4115a3..230197e 100644 --- a/src/core/chunk/trace_ingest.rs +++ b/src/core/chunk/trace_ingest.rs @@ -1,12 +1,12 @@ use crate::core::chunk::types::{Chunk, ChunkRule, ChunkType}; +use crate::core::db::qdrant_db::QdrantDb; use crate::core::db::schema; use crate::core::db::PostgresDb; -use crate::core::db::qdrant_db::QdrantDb; use anyhow::{Context, Result}; use serde_json::json; use sqlx::Row; -use tracing::{error, info}; use std::collections::HashMap; +use tracing::{error, info}; pub async fn ingest_traces(db: &PostgresDb, file_uuid: &str) -> Result { let pool = db.pool(); @@ -27,21 +27,38 @@ pub async fn ingest_traces(db: &PostgresDb, file_uuid: &str) -> Result { {"key": "trace_id", "match": {"value": 1}} ] }); - let points = qdrant.scroll_all_points("_faces", face_filter, 500).await.unwrap_or_default(); + let points = qdrant + .scroll_all_points("_faces", face_filter, 500) + .await + .unwrap_or_default(); - let mut trace_data: HashMap = HashMap::new(); + let mut trace_data: HashMap = + HashMap::new(); for point in &points { let payload = &point["payload"]; let trace_id = payload["trace_id"].as_i64().unwrap_or(0) as i32; let frame = payload["frame"].as_i64().unwrap_or(0); - let timestamp = payload.get("timestamp_secs").and_then(|v| v.as_f64()).unwrap_or(0.0); + let timestamp = payload + .get("timestamp_secs") + .and_then(|v| v.as_f64()) + .unwrap_or(0.0); let bbox = &payload["bbox"]; let x = bbox["x"].as_f64().unwrap_or(0.0); let y = bbox["y"].as_f64().unwrap_or(0.0); let w = bbox["width"].as_f64().unwrap_or(0.0); let h = bbox["height"].as_f64().unwrap_or(0.0); - let entry = trace_data.entry(trace_id).or_insert((i64::MAX, i64::MIN, f64::MAX, f64::MIN, 0, 0.0, 0.0, 0.0, 0.0)); + let entry = trace_data.entry(trace_id).or_insert(( + i64::MAX, + i64::MIN, + f64::MAX, + f64::MIN, + 0, + 0.0, + 0.0, + 0.0, + 0.0, + )); entry.0 = entry.0.min(frame); entry.1 = entry.1.max(frame); if timestamp > 0.0 { @@ -55,20 +72,33 @@ pub async fn ingest_traces(db: &PostgresDb, file_uuid: &str) -> Result { entry.8 += h; } - let traces: Vec = trace_data.into_iter().map(|(trace_id, (first_f, last_f, first_t, last_t, count, sum_x, sum_y, sum_w, sum_h))| { - TraceAgg { - trace_id, - first_frame: first_f, - last_frame: last_f, - first_time: if first_t != f64::MAX { first_t } else { first_f as f64 / fps }, - last_time: if last_t != f64::MIN { last_t } else { last_f as f64 / fps }, - face_count: count, - avg_x: sum_x / count as f64, - avg_y: sum_y / count as f64, - avg_w: sum_w / count as f64, - avg_h: sum_h / count as f64, - } - }).collect(); + let traces: Vec = trace_data + .into_iter() + .map( + |(trace_id, (first_f, last_f, first_t, last_t, count, sum_x, sum_y, sum_w, sum_h))| { + TraceAgg { + trace_id, + first_frame: first_f, + last_frame: last_f, + first_time: if first_t != f64::MAX { + first_t + } else { + first_f as f64 / fps + }, + last_time: if last_t != f64::MIN { + last_t + } else { + last_f as f64 / fps + }, + face_count: count, + avg_x: sum_x / count as f64, + avg_y: sum_y / count as f64, + avg_w: sum_w / count as f64, + avg_h: sum_h / count as f64, + } + }, + ) + .collect(); if traces.is_empty() { info!("No traces found for {}", file_uuid); diff --git a/src/core/db/postgres_db.rs b/src/core/db/postgres_db.rs index 7ae38b3..69f427e 100644 --- a/src/core/db/postgres_db.rs +++ b/src/core/db/postgres_db.rs @@ -1325,7 +1325,8 @@ impl PostgresDb { .await?; // ── TKG Operation Log ── - sqlx::query("CREATE TABLE IF NOT EXISTS tkg_operation_log ( + sqlx::query( + "CREATE TABLE IF NOT EXISTS tkg_operation_log ( id BIGSERIAL PRIMARY KEY, file_uuid VARCHAR(32) NOT NULL, operation VARCHAR(20) NOT NULL, @@ -1342,10 +1343,15 @@ impl PostgresDb { started_at TIMESTAMPTZ DEFAULT NOW(), completed_at TIMESTAMPTZ, properties JSONB - )").execute(pool).await?; - sqlx::query("CREATE INDEX IF NOT EXISTS idx_tkg_op_file_uuid ON tkg_operation_log(file_uuid)") - .execute(pool) - .await?; + )", + ) + .execute(pool) + .await?; + sqlx::query( + "CREATE INDEX IF NOT EXISTS idx_tkg_op_file_uuid ON tkg_operation_log(file_uuid)", + ) + .execute(pool) + .await?; // ── Functions & Triggers ── sqlx::query( @@ -1902,7 +1908,7 @@ impl PostgresDb { &format!( r#" INSERT INTO {} (uuid, video_path, status, video_id, processors) - VALUES ($1, $2, 'pending', $3, ARRAY['asr','cut','ocr','face','face_cluster','pose','asrx']) + VALUES ($1, $2, 'pending', $3, ARRAY['cut','asr','face','ocr','asrx','pose','appearance']) RETURNING id, uuid, video_path, status, current_processor, progress_total, progress_current, error_count, last_error, started_at::TEXT, updated_at::TEXT, created_at::TEXT, processors, completed_processors, failed_processors, video_id "#, jobs_table @@ -2775,7 +2781,8 @@ impl PostgresDb { )], ) -> Result<()> { let table = schema::table_name("pre_chunks"); - self.delete_pre_chunks_by_processor(uuid, processor_type).await?; + self.delete_pre_chunks_by_processor(uuid, processor_type) + .await?; for (frame, ts, data, text, _) in chunks { sqlx::query(&format!( "INSERT INTO {} (file_uuid, processor_type, chunk_type, start_frame, start_time, data, text_content) \ @@ -3127,13 +3134,23 @@ impl PostgresDb { let table = schema::table_name("videos"); let key = processor.to_uppercase(); let path = format!("{{progress,{}}}", key); + let status_path = format!("{{progress,{},status}}", key); sqlx::query(&format!( - "UPDATE {} SET processing_status = jsonb_set(COALESCE(processing_status, '{{}}'::jsonb), $1::text[], $2::jsonb) WHERE file_uuid = $3", table + "UPDATE {} SET processing_status = \ + CASE \ + WHEN COALESCE(processing_status #>> $4::text[], '') = 'completed' \ + THEN processing_status \ + ELSE jsonb_set(COALESCE(processing_status, '{{}}'::jsonb), $1::text[], $2::jsonb) \ + END \ + WHERE file_uuid = $3", + table )) .bind(&path) .bind(serde_json::json!({"processed": current, "total": total, "status": status})) .bind(uuid) - .execute(&self.pool).await?; + .bind(&status_path) + .execute(&self.pool) + .await?; Ok(()) } diff --git a/src/core/db/qdrant_db.rs b/src/core/db/qdrant_db.rs index ab3db7b..e9213bf 100644 --- a/src/core/db/qdrant_db.rs +++ b/src/core/db/qdrant_db.rs @@ -876,7 +876,7 @@ impl QdrantDb { .await?; let batch_len = batch.len(); all_points.extend(batch); - if batch_len < page_size { + if batch_len < page_size || next.is_none() { break; } offset = next; diff --git a/src/core/mod.rs b/src/core/mod.rs index 45e0f53..7fbe963 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -22,6 +22,6 @@ pub mod storage; pub mod text; pub mod thumbnail; pub mod time; -pub mod tmdb; pub mod tkg; +pub mod tmdb; pub mod vision; diff --git a/src/core/pipeline/mod.rs b/src/core/pipeline/mod.rs index 36d5384..ef7a525 100644 --- a/src/core/pipeline/mod.rs +++ b/src/core/pipeline/mod.rs @@ -33,7 +33,13 @@ pub async fn store_asrx_chunks(db: &PostgresDb, uuid: &str) -> Result<()> { "start_frame": segment.start_frame, "end_frame": segment.end_frame, }); - pre_chunks.push((segment.start_frame as i64, Some(segment.start_time), data, None, None)); + pre_chunks.push(( + segment.start_frame as i64, + Some(segment.start_time), + data, + None, + None, + )); speaker_detections.push(( segment.speaker_id.clone().unwrap_or_default(), segment.start_time, diff --git a/src/core/processor/heuristic_scene.rs b/src/core/processor/heuristic_scene.rs index 11ae2c9..79a89a9 100644 --- a/src/core/processor/heuristic_scene.rs +++ b/src/core/processor/heuristic_scene.rs @@ -159,7 +159,10 @@ pub async fn build_heuristic_scene_meta( {"key": "trace_id", "match": {"value": 1}} ] }); - let points = qdrant.scroll_all_points("_faces", face_filter, 500).await.unwrap_or_default(); + let points = qdrant + .scroll_all_points("_faces", face_filter, 500) + .await + .unwrap_or_default(); let mut frame_face_counts: HashMap = HashMap::new(); for point in &points { diff --git a/src/core/progress.rs b/src/core/progress.rs index dad7bf5..cc2fbe8 100644 --- a/src/core/progress.rs +++ b/src/core/progress.rs @@ -18,9 +18,9 @@ use serde::{Deserialize, Serialize}; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct PipelineStage { pub name: String, - pub weight: f64, // Weight in overall progress (0.0-1.0) - pub progress: f64, // Stage progress (0.0-1.0) - pub status: String, // "pending", "running", "completed", "failed" + pub weight: f64, // Weight in overall progress (0.0-1.0) + pub progress: f64, // Stage progress (0.0-1.0) + pub status: String, // "pending", "running", "completed", "failed" pub detail: Option, } @@ -28,7 +28,7 @@ pub struct PipelineStage { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct PipelineProgress { pub file_uuid: String, - pub overall_progress: f64, // 0.0-1.0 weighted sum of all stages + pub overall_progress: f64, // 0.0-1.0 weighted sum of all stages pub stages: Vec, pub updated_at: String, } @@ -40,23 +40,71 @@ impl PipelineProgress { overall_progress: 0.0, stages: vec![ // Processors (30% total) - PipelineStage { name: "processors".into(), weight: 0.30, progress: 0.0, status: "pending".into(), detail: None }, + PipelineStage { + name: "processors".into(), + weight: 0.30, + progress: 0.0, + status: "pending".into(), + detail: None, + }, // Post-processor triggers (20% total) - PipelineStage { name: "rule1_ingestion".into(), weight: 0.05, progress: 0.0, status: "pending".into(), detail: None }, - PipelineStage { name: "face_tracing".into(), weight: 0.05, progress: 0.0, status: "pending".into(), detail: None }, - PipelineStage { name: "identity_agent".into(), weight: 0.10, progress: 0.0, status: "pending".into(), detail: None }, + PipelineStage { + name: "rule1_ingestion".into(), + weight: 0.05, + progress: 0.0, + status: "pending".into(), + detail: None, + }, + PipelineStage { + name: "face_tracing".into(), + weight: 0.05, + progress: 0.0, + status: "pending".into(), + detail: None, + }, + PipelineStage { + name: "identity_agent".into(), + weight: 0.10, + progress: 0.0, + status: "pending".into(), + detail: None, + }, // TKG Build (35% total) - PipelineStage { name: "tkg_nodes".into(), weight: 0.20, progress: 0.0, status: "pending".into(), detail: None }, - PipelineStage { name: "tkg_edges".into(), weight: 0.15, progress: 0.0, status: "pending".into(), detail: None }, + PipelineStage { + name: "tkg_nodes".into(), + weight: 0.20, + progress: 0.0, + status: "pending".into(), + detail: None, + }, + PipelineStage { + name: "tkg_edges".into(), + weight: 0.15, + progress: 0.0, + status: "pending".into(), + detail: None, + }, // Rule 2 Ingestion (15%) - PipelineStage { name: "rule2_ingestion".into(), weight: 0.15, progress: 0.0, status: "pending".into(), detail: None }, + PipelineStage { + name: "rule2_ingestion".into(), + weight: 0.15, + progress: 0.0, + status: "pending".into(), + detail: None, + }, ], updated_at: chrono::Utc::now().to_rfc3339(), } } /// Update a stage's progress and recalculate overall progress - pub fn update_stage(&mut self, stage_name: &str, progress: f64, status: &str, detail: Option) { + pub fn update_stage( + &mut self, + stage_name: &str, + progress: f64, + status: &str, + detail: Option, + ) { if let Some(stage) = self.stages.iter_mut().find(|s| s.name == stage_name) { stage.progress = progress.clamp(0.0, 1.0); stage.status = status.to_string(); @@ -67,7 +115,9 @@ impl PipelineProgress { /// Recalculate overall progress as weighted sum fn recalculate_overall(&mut self) { - self.overall_progress = self.stages.iter() + self.overall_progress = self + .stages + .iter() .map(|s| s.weight * s.progress) .sum::() .clamp(0.0, 1.0); @@ -90,24 +140,24 @@ impl PipelineProgress { #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum TkgPhase { - FaceTracing = 0, // Phase 0: Populate trace_id from face.json - FaceTrackNodes = 1, // Build face_track nodes - GazeTrackNodes = 2, // Build gaze_track nodes - LipTrackNodes = 3, // Build lip_track nodes - TextRegionNodes = 4, // Build text_region nodes - AppearanceNodes = 5, // Build appearance_trace nodes - AccessoryNodes = 6, // Build accessory nodes - ObjectNodes = 7, // Build yolo_object nodes - HandNodes = 8, // Build hand nodes - SpeakerNodes = 9, // Build speaker nodes - CoOccurrenceEdges = 10, // Build co_occurrence edges - SpeakerFaceEdges = 11, // Build speaker_face edges - FaceFaceEdges = 12, // Build face_face edges - MutualGazeEdges = 13, // Build mutual_gaze edges - LipSyncEdges = 14, // Build lip_sync edges - HasAppearanceEdges = 15,// Build has_appearance edges - WearsEdges = 16, // Build wears edges - HandObjectEdges = 17, // Build hand_object edges + FaceTracing = 0, // Phase 0: Populate trace_id from face.json + FaceTrackNodes = 1, // Build face_track nodes + GazeTrackNodes = 2, // Build gaze_track nodes + LipTrackNodes = 3, // Build lip_track nodes + TextRegionNodes = 4, // Build text_region nodes + AppearanceNodes = 5, // Build appearance_trace nodes + AccessoryNodes = 6, // Build accessory nodes + ObjectNodes = 7, // Build yolo_object nodes + HandNodes = 8, // Build hand nodes + SpeakerNodes = 9, // Build speaker nodes + CoOccurrenceEdges = 10, // Build co_occurrence edges + SpeakerFaceEdges = 11, // Build speaker_face edges + FaceFaceEdges = 12, // Build face_face edges + MutualGazeEdges = 13, // Build mutual_gaze edges + LipSyncEdges = 14, // Build lip_sync edges + HasAppearanceEdges = 15, // Build has_appearance edges + WearsEdges = 16, // Build wears edges + HandObjectEdges = 17, // Build hand_object edges Completed = 18, Failed = 19, } @@ -282,12 +332,7 @@ impl TkgProgress { } } - pub fn update_phase( - &mut self, - phase: TkgPhase, - phase_progress: f64, - message: &str, - ) { + pub fn update_phase(&mut self, phase: TkgPhase, phase_progress: f64, message: &str) { self.phase = phase.name().to_string(); self.phase_index = phase as usize; self.phase_progress = phase_progress.clamp(0.0, 1.0); @@ -339,12 +384,7 @@ impl AgentProgress { } } - pub fn update_phase( - &mut self, - phase: AgentPhase, - phase_progress: f64, - message: &str, - ) { + pub fn update_phase(&mut self, phase: AgentPhase, phase_progress: f64, message: &str) { self.phase = phase.name().to_string(); self.phase_index = phase as usize; self.phase_progress = phase_progress.clamp(0.0, 1.0); @@ -357,13 +397,21 @@ impl AgentProgress { } pub fn mark_completed(&mut self) { - self.update_phase(AgentPhase::Completed, 1.0, "Identity Agent processing completed"); + self.update_phase( + AgentPhase::Completed, + 1.0, + "Identity Agent processing completed", + ); self.overall_progress = 1.0; self.phase_progress = 1.0; } pub fn mark_failed(&mut self, error: &str) { - self.update_phase(AgentPhase::Failed, 0.0, &format!("Identity Agent failed: {}", error)); + self.update_phase( + AgentPhase::Failed, + 0.0, + &format!("Identity Agent failed: {}", error), + ); } } @@ -482,10 +530,7 @@ pub async fn publish_agent_progress( } } -pub async fn get_progress( - redis: &Arc, - file_uuid: &str, -) -> Option { +pub async fn get_progress(redis: &Arc, file_uuid: &str) -> Option { let tkg_key = format!( "{}progress:{}:tkg", crate::core::config::REDIS_KEY_PREFIX.as_str(), @@ -542,7 +587,11 @@ pub async fn publish_pipeline_progress( // Merge: update stages from new progress onto existing for new_stage in &progress.stages { if new_stage.status == "completed" || new_stage.progress > 0.0 { - if let Some(existing_stage) = existing.stages.iter_mut().find(|s| s.name == new_stage.name) { + if let Some(existing_stage) = existing + .stages + .iter_mut() + .find(|s| s.name == new_stage.name) + { existing_stage.status = new_stage.status.clone(); existing_stage.progress = new_stage.progress; existing_stage.detail = new_stage.detail.clone(); diff --git a/src/core/tkg/log.rs b/src/core/tkg/log.rs index 9fa53cc..8a6d586 100644 --- a/src/core/tkg/log.rs +++ b/src/core/tkg/log.rs @@ -23,7 +23,11 @@ impl TkgLogger { .bind(operation) .fetch_one(pool) .await?; - tracing::info!("[TKG-Log] Started operation {} for {}", operation, file_uuid); + tracing::info!( + "[TKG-Log] Started operation {} for {}", + operation, + file_uuid + ); Ok(id) } @@ -55,7 +59,11 @@ impl TkgLogger { pub async fn complete_operation(&self, error: Option<&str>) -> Result<()> { if let Some(log_id) = self.log_id { let table = schema::table_name("tkg_operation_log"); - let status = if error.is_some() { "failed" } else { "completed" }; + let status = if error.is_some() { + "failed" + } else { + "completed" + }; sqlx::query(&format!( "UPDATE {table} \ SET status = $1, \ @@ -68,14 +76,41 @@ impl TkgLogger { .bind(log_id) .execute(&self.pool) .await?; - tracing::info!("[TKG-Log] Operation {} completed with status: {}", log_id, status); + tracing::info!( + "[TKG-Log] Operation {} completed with status: {}", + log_id, + status + ); } Ok(()) } - pub async fn get_operations(pool: &PgPool, file_uuid: &str) -> Result> { + pub async fn get_operations( + pool: &PgPool, + file_uuid: &str, + ) -> Result> { let table = schema::table_name("tkg_operation_log"); - let rows = sqlx::query_as::<_, (i64, String, String, Option, Option, i32, i32, i32, i32, i32, i32, String, Option, String, Option, Option)>(&format!( + let rows = sqlx::query_as::< + _, + ( + i64, + String, + String, + Option, + Option, + i32, + i32, + i32, + i32, + i32, + i32, + String, + Option, + String, + Option, + Option, + ), + >(&format!( "SELECT id, file_uuid, operation, node_type, edge_type, \ nodes_created, nodes_updated, nodes_deleted, \ edges_created, edges_updated, edges_deleted, \ @@ -86,34 +121,35 @@ impl TkgLogger { .fetch_all(pool) .await?; - Ok(rows.into_iter().map(|r| crate::core::tkg::models::TkgOperationLog { - id: r.0, - file_uuid: r.1, - operation: r.2, - node_type: r.3, - edge_type: r.4, - nodes_created: r.5, - nodes_updated: r.6, - nodes_deleted: r.7, - edges_created: r.8, - edges_updated: r.9, - edges_deleted: r.10, - status: r.11, - error_message: r.12, - started_at: r.13, - completed_at: r.14, - properties: r.15, - }).collect()) + Ok(rows + .into_iter() + .map(|r| crate::core::tkg::models::TkgOperationLog { + id: r.0, + file_uuid: r.1, + operation: r.2, + node_type: r.3, + edge_type: r.4, + nodes_created: r.5, + nodes_updated: r.6, + nodes_deleted: r.7, + edges_created: r.8, + edges_updated: r.9, + edges_deleted: r.10, + status: r.11, + error_message: r.12, + started_at: r.13, + completed_at: r.14, + properties: r.15, + }) + .collect()) } pub async fn delete_operations(pool: &PgPool, file_uuid: &str) -> Result { let table = schema::table_name("tkg_operation_log"); - let result = sqlx::query(&format!( - "DELETE FROM {table} WHERE file_uuid = $1" - )) - .bind(file_uuid) - .execute(pool) - .await?; + let result = sqlx::query(&format!("DELETE FROM {table} WHERE file_uuid = $1")) + .bind(file_uuid) + .execute(pool) + .await?; Ok(result.rows_affected() as i64) } } diff --git a/src/core/tkg/mod.rs b/src/core/tkg/mod.rs index 2ac0b81..6d31bd9 100644 --- a/src/core/tkg/mod.rs +++ b/src/core/tkg/mod.rs @@ -1,7 +1,7 @@ -pub mod service; pub mod log; pub mod models; +pub mod service; -pub use service::TkgService; pub use log::TkgLogger; pub use models::*; +pub use service::TkgService; diff --git a/src/core/tkg/models.rs b/src/core/tkg/models.rs index cb1ce0b..0908772 100644 --- a/src/core/tkg/models.rs +++ b/src/core/tkg/models.rs @@ -57,15 +57,26 @@ pub struct TkgBuildStats { impl TkgBuildStats { pub fn total_nodes(&self) -> i32 { - self.face_track_nodes + self.gaze_track_nodes + self.lip_track_nodes - + self.text_region_nodes + self.appearance_trace_nodes + self.accessory_nodes - + self.object_nodes + self.hand_nodes + self.speaker_nodes + self.face_track_nodes + + self.gaze_track_nodes + + self.lip_track_nodes + + self.text_region_nodes + + self.appearance_trace_nodes + + self.accessory_nodes + + self.object_nodes + + self.hand_nodes + + self.speaker_nodes } pub fn total_edges(&self) -> i32 { - self.co_occurrence_edges + self.speaker_face_edges + self.face_face_edges - + self.mutual_gaze_edges + self.lip_sync_edges + self.has_appearance_edges - + self.wears_edges + self.hand_object_edges + self.co_occurrence_edges + + self.speaker_face_edges + + self.face_face_edges + + self.mutual_gaze_edges + + self.lip_sync_edges + + self.has_appearance_edges + + self.wears_edges + + self.hand_object_edges } pub fn to_json(&self) -> serde_json::Value { diff --git a/src/core/tkg/service.rs b/src/core/tkg/service.rs index 0b3e9e9..4db9e54 100644 --- a/src/core/tkg/service.rs +++ b/src/core/tkg/service.rs @@ -1,5 +1,5 @@ -use crate::core::db::PostgresDb; use crate::core::db::schema; +use crate::core::db::PostgresDb; use crate::core::tkg::log::TkgLogger; use crate::core::tkg::models::{TkgBuildStats, TkgOperationLog}; use anyhow::Result; @@ -20,8 +20,11 @@ impl TkgService { let mut logger = TkgLogger::new(self.db.pool().clone()); logger.log_id = Some(log_id); - let redis = crate::core::db::RedisClient::new().ok().map(|r| std::sync::Arc::new(r)); - let result = crate::core::processor::tkg::build_tkg(&self.db, file_uuid, output_dir, redis).await; + let redis = crate::core::db::RedisClient::new() + .ok() + .map(|r| std::sync::Arc::new(r)); + let result = + crate::core::processor::tkg::build_tkg(&self.db, file_uuid, output_dir, redis).await; match result { Ok(r) => { @@ -45,8 +48,12 @@ impl TkgService { hand_object_edges: r.hand_object_edges as i32, }; logger.complete_operation(None).await?; - tracing::info!("[TKG-Service] Build completed for {}: {} nodes, {} edges", - file_uuid, stats.total_nodes(), stats.total_edges()); + tracing::info!( + "[TKG-Service] Build completed for {}: {} nodes, {} edges", + file_uuid, + stats.total_nodes(), + stats.total_edges() + ); Ok(stats) } Err(e) => { @@ -57,7 +64,12 @@ impl TkgService { } } - pub async fn rebuild(&self, file_uuid: &str, output_dir: &str, force: bool) -> Result { + pub async fn rebuild( + &self, + file_uuid: &str, + output_dir: &str, + force: bool, + ) -> Result { let operation = if force { "rebuild_force" } else { "rebuild" }; let log_id = TkgLogger::start_operation(self.db.pool(), file_uuid, operation).await?; @@ -70,7 +82,11 @@ impl TkgService { // Update the original log entry if let Some(id) = Some(log_id) { let table = schema::table_name("tkg_operation_log"); - let status = if result.is_ok() { "completed" } else { "failed" }; + let status = if result.is_ok() { + "completed" + } else { + "failed" + }; let error = result.as_ref().err().map(|e| e.to_string()); sqlx::query(&format!( "UPDATE {table} SET status = $1, error_message = $2, completed_at = NOW() WHERE id = $3" @@ -92,11 +108,15 @@ impl TkgService { match result { Ok(_) => { - TkgLogger::new(self.db.pool().clone()).complete_operation(None).await?; + TkgLogger::new(self.db.pool().clone()) + .complete_operation(None) + .await?; Ok(()) } Err(e) => { - TkgLogger::new(self.db.pool().clone()).complete_operation(Some(&e.to_string())).await?; + TkgLogger::new(self.db.pool().clone()) + .complete_operation(Some(&e.to_string())) + .await?; Err(e) } } @@ -107,23 +127,23 @@ impl TkgService { let edges_table = schema::table_name("tkg_edges"); // Delete edges first (foreign key constraint) - let edges_deleted = sqlx::query(&format!( - "DELETE FROM {edges_table} WHERE file_uuid = $1" - )) - .bind(file_uuid) - .execute(self.db.pool()) - .await?; + let edges_deleted = sqlx::query(&format!("DELETE FROM {edges_table} WHERE file_uuid = $1")) + .bind(file_uuid) + .execute(self.db.pool()) + .await?; // Delete nodes - let nodes_deleted = sqlx::query(&format!( - "DELETE FROM {nodes_table} WHERE file_uuid = $1" - )) - .bind(file_uuid) - .execute(self.db.pool()) - .await?; + let nodes_deleted = sqlx::query(&format!("DELETE FROM {nodes_table} WHERE file_uuid = $1")) + .bind(file_uuid) + .execute(self.db.pool()) + .await?; - tracing::info!("[TKG-Service] Deleted {} nodes and {} edges for {}", - nodes_deleted.rows_affected(), edges_deleted.rows_affected(), file_uuid); + tracing::info!( + "[TKG-Service] Deleted {} nodes and {} edges for {}", + nodes_deleted.rows_affected(), + edges_deleted.rows_affected(), + file_uuid + ); Ok(()) } diff --git a/src/core/tmdb/face_agent.rs b/src/core/tmdb/face_agent.rs index 30195d7..5921f72 100644 --- a/src/core/tmdb/face_agent.rs +++ b/src/core/tmdb/face_agent.rs @@ -69,7 +69,10 @@ pub async fn match_faces_against_tmdb(db: &PostgresDb, file_uuid: &str) -> Resul info!("[TKG-MATCH] No TMDb identity seeds in _seeds collection"); return Ok(0); } - info!("[TKG-MATCH] {} TMDb seeds loaded from _seeds", tmdb_rows.len()); + info!( + "[TKG-MATCH] {} TMDb seeds loaded from _seeds", + tmdb_rows.len() + ); // Step 2: Load face embeddings from Qdrant _faces, grouped by trace_id let face_filter = serde_json::json!({ @@ -81,7 +84,10 @@ pub async fn match_faces_against_tmdb(db: &PostgresDb, file_uuid: &str) -> Resul let face_points = match qdrant.scroll_all_points("_faces", face_filter, 1000).await { Ok(pts) => pts, Err(e) => { - warn!("[TKG-MATCH] Failed to scroll _faces for {}: {}", file_uuid, e); + warn!( + "[TKG-MATCH] Failed to scroll _faces for {}: {}", + file_uuid, e + ); return Ok(0); } }; @@ -118,7 +124,11 @@ pub async fn match_faces_against_tmdb(db: &PostgresDb, file_uuid: &str) -> Resul } let total = trace_faces.len(); - info!("[TKG-MATCH] {} traces with {} faces", total, face_points.len()); + info!( + "[TKG-MATCH] {} traces with {} faces", + total, + face_points.len() + ); // Step 3: Iterative matching const TH: f32 = 0.50; @@ -341,7 +351,9 @@ async fn quality_check_temporal_collisions_qdrant( ] }); let payload = serde_json::json!({"identity_id": serde_json::Value::Null}); - let _ = qdrant.update_payload_by_filter("_faces", filter, payload).await; + let _ = qdrant + .update_payload_by_filter("_faces", filter, payload) + .await; unbound += 1; warn!("[TKG-QC] Collision identity={}: trace {} vs trace {} ({} overlap frames). Unbound trace {} ({} points)", diff --git a/src/main.rs b/src/main.rs index b8e6a1a..c2380ff 100644 --- a/src/main.rs +++ b/src/main.rs @@ -13,6 +13,7 @@ fn init_tracing() { tracing_subscriber::fmt() .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) .with_target(true) + .with_writer(|| std::io::LineWriter::new(std::io::stdout())) .init(); } diff --git a/src/playground.rs b/src/playground.rs index 3d9367f..87fe1c6 100644 --- a/src/playground.rs +++ b/src/playground.rs @@ -2335,7 +2335,8 @@ async fn main() -> Result<()> { // Build TKG println!("\nBuilding TKG..."); let tkg_result = - momentry_core::core::processor::tkg::build_tkg(&db, &uuid, &output_dir, None).await?; + momentry_core::core::processor::tkg::build_tkg(&db, &uuid, &output_dir, None) + .await?; println!( "✓ TKG built: {} nodes, {} edges", tkg_result.face_track_nodes + tkg_result.hand_nodes + tkg_result.object_nodes, diff --git a/src/worker/job_worker.rs b/src/worker/job_worker.rs index e9b6ec7..5524d97 100644 --- a/src/worker/job_worker.rs +++ b/src/worker/job_worker.rs @@ -9,7 +9,6 @@ use tracing::{debug, error, info, warn}; use crate::api::identity_agent_api::run_identity_agent; use crate::core::chunk::rule1_ingest; use crate::core::config::OUTPUT_DIR; -use crate::core::progress::{publish_pipeline_progress, PipelineProgress}; use crate::core::db::qdrant_db::QdrantDb; use crate::core::db::{ schema, MonitorJobStatus, PostgresDb, ProcessorJobStatus, RedisClient, VectorPayload, @@ -17,6 +16,7 @@ use crate::core::db::{ }; use crate::core::embedding::Embedder; use crate::core::processor::heuristic_scene::generate_scene_meta; +use crate::core::progress::{publish_pipeline_progress, PipelineProgress}; use crate::worker::config::WorkerConfig; use crate::worker::processor::{ProcessorPool, ProcessorTask}; use crate::worker::resources::SystemResources; @@ -206,7 +206,10 @@ impl JobWorker { let should_retry = self .check_and_complete_job(job.id, &job.uuid, &job.processors, expected_count) .await - .unwrap_or(false); + .unwrap_or_else(|e| { + error!("check_and_complete_job failed for {}: {}", job.uuid, e); + false + }); if should_retry && self.processor_pool.can_start().await { if let Err(e) = self.process_job(job.clone()).await { error!("Failed to reprocess job {}: {}", job.uuid, e); @@ -281,7 +284,10 @@ impl JobWorker { // Check if job still exists in database (may have been deleted by unregister) let current_job = self.db.get_monitor_job_by_uuid(&job.uuid).await?; if current_job.is_none() { - info!("Job {} no longer exists in database (possibly unregistered), skipping", job.uuid); + info!( + "Job {} no longer exists in database (possibly unregistered), skipping", + job.uuid + ); return Ok(()); } @@ -313,9 +319,17 @@ impl JobWorker { .await?; // Clear any stale PipelineProgress from previous jobs - let progress_key = format!("{}progress:{}:pipeline", crate::core::config::REDIS_KEY_PREFIX.as_str(), job.uuid); + let progress_key = format!( + "{}progress:{}:pipeline", + crate::core::config::REDIS_KEY_PREFIX.as_str(), + job.uuid + ); if let Ok(mut conn) = self.redis.get_conn().await { - let _: Option = redis::cmd("DEL").arg(&progress_key).query_async(&mut conn).await.ok(); + let _: Option = redis::cmd("DEL") + .arg(&progress_key) + .query_async(&mut conn) + .await + .ok(); } self.db @@ -384,14 +398,18 @@ impl JobWorker { if let Ok(meta) = std::fs::metadata(&tmp_path) { // 條件 1: 檔案 > 1KB let has_content = meta.len() > 1024; - + // 條件 2: 檔案超過 120 秒未修改（確定沒人還在寫） let is_stale = if let Ok(modified) = meta.modified() { if let Ok(elapsed) = modified.elapsed() { elapsed.as_secs() > 120 - } else { false } - } else { false }; - + } else { + false + } + } else { + false + }; + // 條件 3: 檢查程序是否還在跑 let proc_name = processor_type.as_str(); let process_running = std::process::Command::new("ps") @@ -400,11 +418,11 @@ impl JobWorker { .ok() .and_then(|out| String::from_utf8(out.stdout).ok()) .map(|out| { - out.contains(&format!("{}_processor", proc_name)) || out.contains(&format!("{}_processor", proc_name)) + || out.contains(&format!("{}_processor", proc_name)) }) .unwrap_or(false); - + if has_content && is_stale && !process_running { info!( "Found stale .tmp file ({} bytes, {}s old, process={}), renaming to .json for {}", @@ -1272,7 +1290,8 @@ impl JobWorker { // TKG may create 0 nodes/edges for videos with minimal content let has_asr_or_asrx_for_tkg = job_processors.is_empty() || job_processors.iter().any(|p| p == "asrx" || p == "asr"); - let has_face_for_tkg = job_processors.is_empty() || job_processors.iter().any(|p| p == "face"); + let has_face_for_tkg = + job_processors.is_empty() || job_processors.iter().any(|p| p == "face"); let tkg_done: bool = if has_asr_or_asrx_for_tkg && has_face_for_tkg { // TKG is done if face traces are complete (TKG runs after face tracing) @@ -1360,9 +1379,11 @@ impl JobWorker { }); // Check for missing processors (in job_processors but not in results) - let missing_processors: Vec = job_processors.iter().filter(|p| { - !results.iter().any(|r| r.processor_type.as_str() == *p) - }).cloned().collect(); + let missing_processors: Vec = job_processors + .iter() + .filter(|p| !results.iter().any(|r| r.processor_type.as_str() == *p)) + .cloned() + .collect(); if !missing_processors.is_empty() { info!( @@ -1602,8 +1623,18 @@ impl JobWorker { } } let mut pp = PipelineProgress::new(&uuid_clone); - pp.update_stage("rule1_ingestion", 1.0, "completed", Some(format!("{} chunks", count))); - publish_pipeline_progress(redis_clone.as_ref(), &uuid_clone, &pp).await; + pp.update_stage( + "rule1_ingestion", + 1.0, + "completed", + Some(format!("{} chunks", count)), + ); + publish_pipeline_progress( + redis_clone.as_ref(), + &uuid_clone, + &pp, + ) + .await; info!("📦 Phase 1 release packaging..."); let executor = match crate::core::processor::PythonExecutor::new() { @@ -1653,87 +1684,113 @@ impl JobWorker { // 🚀 P2 Trigger: Face Trace + DB Store (after Face) // Runs face_tracker.py (IoU+embedding tracking), stores trace_id + position in DB if has_face { - info!("📝 Face completed, triggering face trace + DB store..."); - let db_clone = self.db.clone(); - let redis_clone = self.redis.clone(); - let uuid_clone = uuid.to_string(); - tokio::spawn(async move { - let executor = match crate::core::processor::PythonExecutor::new() { - Ok(ex) => ex, - Err(e) => { - error!("Failed to create PythonExecutor for face trace: {}", e); - return; - } - }; - match executor - .run( - "store_traced_faces.py", - &["--file-uuid", &uuid_clone], - Some(&uuid_clone), - "TRACE_STORE", - Some(std::time::Duration::from_secs(600)), - ) - .await - { - Ok(()) => { - info!("✅ Face trace + DB store completed for {}", uuid_clone); + let traced_path = format!( + "{}{}.face_traced.json", + crate::core::config::OUTPUT_DIR + .as_str() + .trim_end_matches('/'), + uuid + ); + if std::path::Path::new(&traced_path).exists() { + info!("✅ Face trace already done for {}, skipping spawn", uuid); + } else { + info!("📝 Face completed, triggering face trace + DB store..."); + let db_clone = self.db.clone(); + let redis_clone = self.redis.clone(); + let uuid_clone = uuid.to_string(); + tokio::spawn(async move { + let executor = match crate::core::processor::PythonExecutor::new() { + Ok(ex) => ex, + Err(e) => { + error!("Failed to create PythonExecutor for face trace: {}", e); + return; + } + }; + match executor + .run( + "store_traced_faces.py", + &["--file-uuid", &uuid_clone], + Some(&uuid_clone), + "TRACE_STORE", + Some(std::time::Duration::from_secs(600)), + ) + .await + { + Ok(()) => { + info!("✅ Face trace + DB store completed for {}", uuid_clone); - // Query trace count and distribution - let trace_count = match db_clone - .get_trace_count_by_file(&uuid_clone) - .await - { - Ok(c) => c, - Err(e) => { - error!("Failed to get trace count for {}: {}", uuid_clone, e); - 0 - } - }; + // Query trace count and distribution + let trace_count = + match db_clone.get_trace_count_by_file(&uuid_clone).await { + Ok(c) => c, + Err(e) => { + error!( + "Failed to get trace count for {}: {}", + uuid_clone, e + ); + 0 + } + }; - let (single_frame, multi_frame) = match db_clone - .get_trace_frame_count_distribution(&uuid_clone) - .await - { - Ok(dist) => dist, - Err(e) => { - error!( - "Failed to get trace distribution for {}: {}", - uuid_clone, e + let (single_frame, multi_frame) = match db_clone + .get_trace_frame_count_distribution(&uuid_clone) + .await + { + Ok(dist) => dist, + Err(e) => { + error!( + "Failed to get trace distribution for {}: {}", + uuid_clone, e + ); + (0, 0) + } + }; + + let trace_status = + crate::core::processor::TraceStatus::from_trace_count( + trace_count, ); - (0, 0) - } - }; - - let trace_status = - crate::core::processor::TraceStatus::from_trace_count(trace_count); - info!( + info!( "📊 Trace status: {} (total={}, single_frame={}, multi_frame={}) for {}", trace_status, trace_count, single_frame, multi_frame, uuid_clone ); - // Update processor_results trace_status for Face - if let Err(e) = db_clone - .update_trace_status_for_face( - &uuid_clone, - &trace_status, - trace_count, - single_frame, - multi_frame, - ) - .await - { - error!("Failed to update trace_status for {}: {}", uuid_clone, e); - } + // Update processor_results trace_status for Face + if let Err(e) = db_clone + .update_trace_status_for_face( + &uuid_clone, + &trace_status, + trace_count, + single_frame, + multi_frame, + ) + .await + { + error!( + "Failed to update trace_status for {}: {}", + uuid_clone, e + ); + } - let mut pp = PipelineProgress::new(&uuid_clone); - pp.update_stage("face_tracing", 1.0, "completed", Some(format!("{} traces ({} single, {} multi)", trace_count, single_frame, multi_frame))); - publish_pipeline_progress(redis_clone.as_ref(), &uuid_clone, &pp).await; + let mut pp = PipelineProgress::new(&uuid_clone); + pp.update_stage( + "face_tracing", + 1.0, + "completed", + Some(format!( + "{} traces ({} single, {} multi)", + trace_count, single_frame, multi_frame + )), + ); + publish_pipeline_progress(redis_clone.as_ref(), &uuid_clone, &pp) + .await; + } + Err(e) => { + error!("❌ Face trace + DB store failed for {}: {}", uuid_clone, e) + } } - Err(e) => { - error!("❌ Face trace + DB store failed for {}: {}", uuid_clone, e) - } - } - }); + }); + } } // 🚀 P2.5 Trigger: TMDb Face Matching (after Face, if TMDb data exists) @@ -1841,7 +1898,8 @@ impl JobWorker { let has_seeds = { use crate::core::db::qdrant_db::QdrantDb; let qdrant = QdrantDb::new(); - let schema = std::env::var("DATABASE_SCHEMA").unwrap_or_else(|_| "dev".to_string()); + let schema = + std::env::var("DATABASE_SCHEMA").unwrap_or_else(|_| "dev".to_string()); let seeds_collection = if schema == "public" { "momentry_public_seeds" } else { @@ -1850,7 +1908,7 @@ impl JobWorker { let filter = serde_json::json!({ "must": [{"key": "file_uuid", "match": {"value": uuid}}] }); - match qdrant.scroll_all_points("_seeds", filter, 1).await { + match qdrant.scroll_all_points("_seeds", filter, 100).await { Ok(points) => !points.is_empty(), Err(e) => { warn!("Failed to check _seeds for {}: {}", uuid, e); @@ -1860,62 +1918,140 @@ impl JobWorker { }; if has_seeds { - info!("📝 Prerequisites met for Identity Agent (has seeds). Starting analysis..."); + info!( + "📝 Prerequisites met for Identity Agent (has seeds). Starting analysis..." + ); let db_clone = self.db.clone(); let redis_clone = self.redis.clone(); let uuid_clone = uuid.to_string(); tokio::spawn(async move { - match run_identity_agent(&db_clone, &uuid_clone, Some(redis_clone.clone())).await { + match run_identity_agent(&db_clone, &uuid_clone, Some(redis_clone.clone())) + .await + { Ok(()) => { info!("✅ Identity Agent completed for {}", uuid_clone); let mut pp = PipelineProgress::new(&uuid_clone); pp.update_stage("identity_agent", 1.0, "completed", None); - publish_pipeline_progress(redis_clone.as_ref(), &uuid_clone, &pp).await; + publish_pipeline_progress(redis_clone.as_ref(), &uuid_clone, &pp) + .await; } Err(e) => error!("❌ Identity Agent failed for {}: {}", uuid_clone, e), } }); } else { - info!("📝 Skipping Identity Agent for {} (no seed identities)", uuid); + info!( + "📝 Skipping Identity Agent for {} (no seed identities)", + uuid + ); } } // 🚀 P4 Trigger: TKG Build (Face + ASRX) → then Rule2 ingestion - // Note: build_tkg uses ON CONFLICT, so it's safe to call multiple times if has_face && has_asrx { - info!("📝 Prerequisites met for TKG Build. Starting graph construction..."); - let db_clone = self.db.clone(); - let redis_clone = self.redis.clone(); - let uuid_clone = uuid.to_string(); - let output_dir_clone = crate::core::config::OUTPUT_DIR.clone(); - tokio::spawn(async move { - match crate::core::processor::tkg::build_tkg(&db_clone, &uuid_clone, &output_dir_clone, Some(redis_clone.clone())).await { - Ok(r) => { - let total_nodes = r.face_track_nodes + r.gaze_track_nodes + r.lip_track_nodes + r.text_region_nodes + r.appearance_trace_nodes + r.accessory_nodes + r.object_nodes + r.hand_nodes + r.speaker_nodes; - let total_edges = r.co_occurrence_edges + r.speaker_face_edges + r.face_face_edges + r.mutual_gaze_edges + r.lip_sync_edges + r.has_appearance_edges + r.wears_edges + r.hand_object_edges; - info!("✅ TKG build completed for {}: {} nodes, {} edges", uuid_clone, total_nodes, total_edges); + let tkg_table = crate::core::db::schema::table_name("tkg_edges"); + let tkg_done: bool = sqlx::query_scalar::<_, i32>(&format!( + "SELECT 1 FROM {tkg_table} WHERE file_uuid = $1 LIMIT 1" + )) + .bind(uuid) + .fetch_optional(self.db.pool()) + .await + .unwrap_or(None) + .unwrap_or(0) + > 0; - let mut pp = PipelineProgress::new(&uuid_clone); - pp.update_stage("tkg_nodes", 1.0, "completed", Some(format!("{} nodes", total_nodes))); - pp.update_stage("tkg_edges", 1.0, "completed", Some(format!("{} edges", total_edges))); - publish_pipeline_progress(redis_clone.as_ref(), &uuid_clone, &pp).await; + if tkg_done { + info!("✅ TKG already built for {}, skipping spawn", uuid); + } else { + info!("📝 Prerequisites met for TKG Build. Starting graph construction..."); + let db_clone = self.db.clone(); + let redis_clone = self.redis.clone(); + let uuid_clone = uuid.to_string(); + let output_dir_clone = crate::core::config::OUTPUT_DIR.clone(); + tokio::spawn(async move { + match crate::core::processor::tkg::build_tkg( + &db_clone, + &uuid_clone, + &output_dir_clone, + Some(redis_clone.clone()), + ) + .await + { + Ok(r) => { + let total_nodes = r.face_track_nodes + + r.gaze_track_nodes + + r.lip_track_nodes + + r.text_region_nodes + + r.appearance_trace_nodes + + r.accessory_nodes + + r.object_nodes + + r.hand_nodes + + r.speaker_nodes; + let total_edges = r.co_occurrence_edges + + r.speaker_face_edges + + r.face_face_edges + + r.mutual_gaze_edges + + r.lip_sync_edges + + r.has_appearance_edges + + r.wears_edges + + r.hand_object_edges; + info!( + "✅ TKG build completed for {}: {} nodes, {} edges", + uuid_clone, total_nodes, total_edges + ); - // Trigger Rule 2 ingestion after TKG complete - if total_edges > 0 { - match crate::core::chunk::rule2_ingest::ingest_rule2(db_clone.pool(), &uuid_clone, None, None).await { - Ok(rule2_count) => { - info!("✅ Rule 2 ingestion completed for {}: {} relationship chunks", uuid_clone, rule2_count); - let mut pp = PipelineProgress::new(&uuid_clone); - pp.update_stage("rule2_ingestion", 1.0, "completed", Some(format!("{} chunks", rule2_count))); - publish_pipeline_progress(redis_clone.as_ref(), &uuid_clone, &pp).await; + let mut pp = PipelineProgress::new(&uuid_clone); + pp.update_stage( + "tkg_nodes", + 1.0, + "completed", + Some(format!("{} nodes", total_nodes)), + ); + pp.update_stage( + "tkg_edges", + 1.0, + "completed", + Some(format!("{} edges", total_edges)), + ); + publish_pipeline_progress(redis_clone.as_ref(), &uuid_clone, &pp) + .await; + + // Trigger Rule 2 ingestion after TKG complete + if total_edges > 0 { + match crate::core::chunk::rule2_ingest::ingest_rule2( + db_clone.pool(), + &uuid_clone, + None, + None, + ) + .await + { + Ok(rule2_count) => { + info!("✅ Rule 2 ingestion completed for {}: {} relationship chunks", uuid_clone, rule2_count); + let mut pp = PipelineProgress::new(&uuid_clone); + pp.update_stage( + "rule2_ingestion", + 1.0, + "completed", + Some(format!("{} chunks", rule2_count)), + ); + publish_pipeline_progress( + redis_clone.as_ref(), + &uuid_clone, + &pp, + ) + .await; + } + Err(e) => error!( + "❌ Rule 2 ingestion failed for {}: {}", + uuid_clone, e + ), } - Err(e) => error!("❌ Rule 2 ingestion failed for {}: {}", uuid_clone, e), } } + Err(e) => error!("❌ TKG build failed for {}: {}", uuid_clone, e), } - Err(e) => error!("❌ TKG build failed for {}: {}", uuid_clone, e), - } - }); + }); + } } if !Self::ingestion_complete(self.db.pool(), uuid, job_processors).await { diff --git a/src/worker/processor.rs b/src/worker/processor.rs index 322c90a..302d905 100644 --- a/src/worker/processor.rs +++ b/src/worker/processor.rs @@ -1505,7 +1505,13 @@ impl ProcessorPool { "end_frame": segment.end_frame, }); - pre_chunks_to_store.push((segment.start_frame as i64, Some(segment.start_time), data, None, None)); + pre_chunks_to_store.push(( + segment.start_frame as i64, + Some(segment.start_time), + data, + None, + None, + )); speaker_detections.push(( segment.speaker_id.clone().unwrap_or_default(),