From 7635ad8d4d5fbf13bf3c945d21eba4eccb0e8797 Mon Sep 17 00:00:00 2001 From: Accusys Date: Sat, 11 Jul 2026 08:40:24 +0800 Subject: [PATCH] fix: correct TKG node_type/edge_type mismatches in scan.rs and tkg.rs Mismatches fixed (builder vs query layer): - 'text_region' -> 'text_trace' (3 locations: scan.rs count + tkg.rs LIP_SYNC query + tkg.rs LIP_SYNC node ID lookup) - 'yolo_object' -> 'object' (2 queries in scan.rs) - 'FACE_TO_FACE' -> CO_OCCURS_WITH + JOIN with tkg_nodes (2 queries in scan.rs, so face_face_edges correctly counts only face-track pairs) - 'HAND_OBJECT' -> 'hand_object' (2 queries in scan.rs) Impact: ingestion_status and file stats endpoints now return accurate counts; LIP_SYNC edges now resolve their target text nodes. --- src/api/scan.rs | 34 +++++-- src/api/trace_agent_api.rs | 2 +- src/core/processor/tkg.rs | 177 ++++++++++++++++++++++++++++--------- 3 files changed, 165 insertions(+), 48 deletions(-) diff --git a/src/api/scan.rs b/src/api/scan.rs index 6820066..91abd44 100644 --- a/src/api/scan.rs +++ b/src/api/scan.rs @@ -542,10 +542,10 @@ async fn get_ingestion_status( let face_track_nodes: i64 = count_sql!(&format!("SELECT COUNT(*) FROM {tkg_nodes_table} WHERE file_uuid = '{file_uuid}' AND node_type = 'face_track'")); let gaze_track_nodes: i64 = count_sql!(&format!("SELECT COUNT(*) FROM {tkg_nodes_table} WHERE file_uuid = '{file_uuid}' AND node_type = 'gaze_track'")); let lip_track_nodes: i64 = count_sql!(&format!("SELECT COUNT(*) FROM {tkg_nodes_table} WHERE file_uuid = '{file_uuid}' AND node_type = 'lip_track'")); - let text_region_nodes: i64 = count_sql!(&format!("SELECT COUNT(*) FROM {tkg_nodes_table} WHERE file_uuid = '{file_uuid}' AND node_type = 'text_region'")); + let text_region_nodes: i64 = count_sql!(&format!("SELECT COUNT(*) FROM {tkg_nodes_table} WHERE file_uuid = '{file_uuid}' AND node_type = 'text_trace'")); let appearance_nodes: i64 = count_sql!(&format!("SELECT COUNT(*) FROM {tkg_nodes_table} WHERE file_uuid = '{file_uuid}' AND node_type = 'appearance_trace'")); let accessory_nodes: i64 = count_sql!(&format!("SELECT COUNT(*) FROM {tkg_nodes_table} WHERE file_uuid = '{file_uuid}' AND node_type = 'accessory'")); - let object_nodes: i64 = count_sql!(&format!("SELECT COUNT(*) FROM {tkg_nodes_table} WHERE file_uuid = '{file_uuid}' AND node_type = 'yolo_object'")); + let object_nodes: i64 = count_sql!(&format!("SELECT COUNT(*) FROM {tkg_nodes_table} WHERE file_uuid = '{file_uuid}' AND node_type = 'object'")); let hand_nodes: i64 = count_sql!(&format!("SELECT COUNT(*) FROM {tkg_nodes_table} WHERE file_uuid = '{file_uuid}' AND node_type = 'hand'")); let speaker_nodes: i64 = count_sql!(&format!("SELECT COUNT(*) FROM {tkg_nodes_table} WHERE file_uuid = '{file_uuid}' AND node_type = 'speaker'")); @@ -553,12 +553,20 @@ async fn get_ingestion_status( let tkg_edges_table = schema::table_name("tkg_edges"); let co_occurrence_edges: i64 = count_sql!(&format!("SELECT COUNT(*) FROM {tkg_edges_table} WHERE file_uuid = '{file_uuid}' AND edge_type = 'CO_OCCURS_WITH'")); let speaker_face_edges: i64 = count_sql!(&format!("SELECT COUNT(*) FROM {tkg_edges_table} WHERE file_uuid = '{file_uuid}' AND edge_type = 'SPEAKS_AS'")); - let face_face_edges: i64 = count_sql!(&format!("SELECT COUNT(*) FROM {tkg_edges_table} WHERE file_uuid = '{file_uuid}' AND edge_type = 'FACE_TO_FACE'")); + let face_face_edges: i64 = count_sql!(&format!( + "SELECT COUNT(*) FROM {tkg_edges_table} e \ + INNER JOIN {tkg_nodes_table} sn ON e.source_node_id = sn.id \ + INNER JOIN {tkg_nodes_table} tn ON e.target_node_id = tn.id \ + WHERE e.file_uuid = '{file_uuid}' \ + AND e.edge_type = 'CO_OCCURS_WITH' \ + AND sn.node_type = 'face_track' \ + AND tn.node_type = 'face_track'" + )); let mutual_gaze_edges: i64 = count_sql!(&format!("SELECT COUNT(*) FROM {tkg_edges_table} WHERE file_uuid = '{file_uuid}' AND edge_type = 'MUTUAL_GAZE'")); let lip_sync_edges: i64 = count_sql!(&format!("SELECT COUNT(*) FROM {tkg_edges_table} WHERE file_uuid = '{file_uuid}' AND edge_type = 'LIP_SYNC'")); let has_appearance_edges: i64 = count_sql!(&format!("SELECT COUNT(*) FROM {tkg_edges_table} WHERE file_uuid = '{file_uuid}' AND edge_type = 'HAS_APPEARANCE'")); let wears_edges: i64 = count_sql!(&format!("SELECT COUNT(*) FROM {tkg_edges_table} WHERE file_uuid = '{file_uuid}' AND edge_type = 'WEARS'")); - let hand_object_edges: i64 = count_sql!(&format!("SELECT COUNT(*) FROM {tkg_edges_table} WHERE file_uuid = '{file_uuid}' AND edge_type = 'HAND_OBJECT'")); + let hand_object_edges: i64 = count_sql!(&format!("SELECT COUNT(*) FROM {tkg_edges_table} WHERE file_uuid = '{file_uuid}' AND edge_type = 'hand_object'")); // Rule 2 relationship chunks let rule2_chunks = count_sql!(&format!( @@ -994,19 +1002,31 @@ async fn get_file_stats( text_region_nodes: count_nodes(pool, &tkg_nodes_table, &file_uuid, "text_trace").await, appearance_nodes: count_nodes(pool, &tkg_nodes_table, &file_uuid, "appearance_trace").await, accessory_nodes: count_nodes(pool, &tkg_nodes_table, &file_uuid, "accessory").await, - object_nodes: count_nodes(pool, &tkg_nodes_table, &file_uuid, "yolo_object").await, + object_nodes: count_nodes(pool, &tkg_nodes_table, &file_uuid, "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, 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, + face_face_edges: sqlx::query_scalar::<_, i64>(&format!( + "SELECT COUNT(*) FROM {tkg_edges_table} e \ + INNER JOIN {tkg_nodes_table} sn ON e.source_node_id = sn.id \ + INNER JOIN {tkg_nodes_table} tn ON e.target_node_id = tn.id \ + WHERE e.file_uuid = $1 \ + AND e.edge_type = 'CO_OCCURS_WITH' \ + AND sn.node_type = 'face_track' \ + AND tn.node_type = 'face_track'" + )) + .bind(&file_uuid) + .fetch_one(pool) + .await + .unwrap_or(0), 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, 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, + hand_object_edges: count_edges(pool, &tkg_edges_table, &file_uuid, "hand_object").await, ..Default::default() }; diff --git a/src/api/trace_agent_api.rs b/src/api/trace_agent_api.rs index 37c8282..fecf4f3 100644 --- a/src/api/trace_agent_api.rs +++ b/src/api/trace_agent_api.rs @@ -9,8 +9,8 @@ use axum::{ use serde::{Deserialize, Serialize}; use std::sync::Arc; -use crate::core::db::PostgresDb; use crate::core::db::schema::t; +use crate::core::db::PostgresDb; pub fn trace_agent_routes() -> Router { Router::new() diff --git a/src/core/processor/tkg.rs b/src/core/processor/tkg.rs index 1ed9778..fc5b108 100644 --- a/src/core/processor/tkg.rs +++ b/src/core/processor/tkg.rs @@ -37,7 +37,10 @@ async fn scroll_face_points(file_uuid: &str) -> Vec { // Retry with exponential backoff (3 attempts) let mut last_err = None; for attempt in 1..=3 { - match qdrant.scroll_all_points("_faces", filter.clone(), 500).await { + match qdrant + .scroll_all_points("_faces", filter.clone(), 500) + .await + { Ok(pts) => { if attempt > 1 { tracing::info!( @@ -150,7 +153,6 @@ async fn build_node_id_map(pool: &PgPool, file_uuid: &str) -> HashMap<(String, S .collect() } - // ── Phase 0: Populate trace_id from face.json ─────────────────────────────────────────────────────── async fn populate_face_detections_from_face_json( @@ -758,7 +760,8 @@ pub async fn build_tkg( if let Some(r) = redis.as_ref() { publish_tkg_progress(r, file_uuid, &progress).await; } - let n_lip = build_lip_track_nodes(pool, file_uuid, output_dir, &pose_data, &frame_index).await?; + let n_lip = + build_lip_track_nodes(pool, file_uuid, output_dir, &pose_data, &frame_index).await?; progress.stats.lip_track_nodes = n_lip as i64; progress.update_phase( TkgPhase::LipTrackNodes, @@ -881,7 +884,8 @@ pub async fn build_tkg( if let Some(r) = redis.as_ref() { publish_tkg_progress(r, file_uuid, &progress).await; } - let e_co = build_co_occurrence_edges(pool, file_uuid, output_dir, &face_points, &node_id_map).await?; + let e_co = + build_co_occurrence_edges(pool, file_uuid, output_dir, &face_points, &node_id_map).await?; progress.stats.co_occurrence_edges = e_co as i64; progress.update_phase( TkgPhase::CoOccurrenceEdges, @@ -915,7 +919,8 @@ pub async fn build_tkg( if let Some(r) = redis.as_ref() { publish_tkg_progress(r, file_uuid, &progress).await; } - let e_ff = build_face_face_edges(pool, file_uuid, &pose_data, &face_points, &node_id_map).await?; + let e_ff = + build_face_face_edges(pool, file_uuid, &pose_data, &face_points, &node_id_map).await?; progress.stats.face_face_edges = e_ff as i64; progress.update_phase(TkgPhase::FaceFaceEdges, 1.0, &format!("{} face-face", e_ff)); if let Some(r) = redis.as_ref() { @@ -930,7 +935,8 @@ pub async fn build_tkg( if let Some(r) = redis.as_ref() { publish_tkg_progress(r, file_uuid, &progress).await; } - let e_mg = build_mutual_gaze_edges(pool, file_uuid, &pose_data, &face_points, &node_id_map).await?; + let e_mg = + build_mutual_gaze_edges(pool, file_uuid, &pose_data, &face_points, &node_id_map).await?; progress.stats.mutual_gaze_edges = e_mg as i64; progress.update_phase( TkgPhase::MutualGazeEdges, @@ -1383,7 +1389,8 @@ async fn build_co_occurrence_edges( face_points: &[FacePoint], node_id_map: &HashMap<(String, String), i64>, ) -> Result { - build_co_occurrence_edges_from_qdrant(pool, file_uuid, output_dir, face_points, node_id_map).await + build_co_occurrence_edges_from_qdrant(pool, file_uuid, output_dir, face_points, node_id_map) + .await } async fn build_co_occurrence_edges_from_qdrant( @@ -1427,13 +1434,17 @@ async fn build_co_occurrence_edges_from_qdrant( } // Use pre-built node ID map (no SQL query) - let face_node_id = match node_id_map.get(&("face_track".to_string(), format!("face_track_{}", face.trace_id))) { + let face_node_id = match node_id_map.get(&( + "face_track".to_string(), + format!("face_track_{}", face.trace_id), + )) { Some(id) => *id, None => continue, }; for det in dets { - let obj_node_id = match node_id_map.get(&("object".to_string(), det.class_name.clone())) { + let obj_node_id = match node_id_map.get(&("object".to_string(), det.class_name.clone())) + { Some(id) => *id, None => continue, }; @@ -1443,7 +1454,13 @@ async fn build_co_occurrence_edges_from_qdrant( "object_confidence": det.confidence, }); - edges_to_insert.push(("CO_OCCURS_WITH", face_node_id, obj_node_id, file_uuid, serde_json::to_string(&edge_props)?)); + edges_to_insert.push(( + "CO_OCCURS_WITH", + face_node_id, + obj_node_id, + file_uuid, + serde_json::to_string(&edge_props)?, + )); } } @@ -1458,7 +1475,8 @@ async fn build_co_occurrence_edges_from_qdrant( if i > 0 { query.push(", "); } - query.push("(") + query + .push("(") .push_bind(*edge_type) .push_bind(*src) .push_bind(*tgt) @@ -1535,10 +1553,11 @@ async fn build_speaker_face_edges_from_pg( for (tid, sf, ef) in &traces { // Use pre-built node ID map (no SQL query) - let face_node_id = match node_id_map.get(&("face_track".to_string(), format!("face_track_{}", tid))) { - Some(id) => *id, - None => continue, - }; + let face_node_id = + match node_id_map.get(&("face_track".to_string(), format!("face_track_{}", tid))) { + Some(id) => *id, + None => continue, + }; let face_start_sec = *sf as f64 / fps; let face_end_sec = *ef as f64 / fps; @@ -1565,10 +1584,11 @@ async fn build_speaker_face_edges_from_pg( } // Use pre-built node ID map (no SQL query) - let speaker_node_id = match node_id_map.get(&("speaker".to_string(), seg.speaker_id.clone())) { - Some(id) => *id, - None => continue, - }; + let speaker_node_id = + match node_id_map.get(&("speaker".to_string(), seg.speaker_id.clone())) { + Some(id) => *id, + None => continue, + }; let edge_props = serde_json::json!({ "overlap_ratio": (overlap_ratio * 1000.0).round() / 1000.0, @@ -2410,7 +2430,7 @@ async fn build_lip_sync_edges( (properties->>'start_frame')::bigint, (properties->>'end_frame')::bigint, properties->>'speaker_id' - FROM {} WHERE file_uuid = $1 AND node_type = 'text_region' + FROM {} WHERE file_uuid = $1 AND node_type = 'text_trace' "#, nodes_table )) @@ -2472,7 +2492,7 @@ async fn build_lip_sync_edges( Some(id) => *id, None => { if let Some((id,)) = sqlx::query_as::<_, (i64,)>(&format!( - "SELECT id FROM {} WHERE file_uuid=$1 AND node_type='text_region' AND external_id=$2", + "SELECT id FROM {} WHERE file_uuid=$1 AND node_type='text_trace' AND external_id=$2", nodes_table )) .bind(file_uuid).bind(text_ext).fetch_optional(pool).await? @@ -2560,9 +2580,9 @@ async fn build_appearance_trace_nodes( if let Some(nose) = nose { let nx = nose.get("x").and_then(|v| v.as_f64()).unwrap_or(0.0); let ny = nose.get("y").and_then(|v| v.as_f64()).unwrap_or(0.0); - // Find closest face trace in Qdrant for this frame (in-memory lookup) - if let Some(trace_id) = - find_trace_in_frame(frame_index, frame_num, nx, ny, 1.0, 1.0) + // Find closest face trace in Qdrant for this frame (in-memory lookup) + if let Some(trace_id) = + find_trace_in_frame(frame_index, frame_num, nx, ny, 1.0, 1.0) { trace_data .entry(trace_id) @@ -2689,7 +2709,11 @@ async fn build_accessory_nodes(pool: &PgPool, file_uuid: &str, output_dir: &str) // ── HAS_APPEARANCE Edges ────────────────────────────────────────── -async fn build_has_appearance_edges(pool: &PgPool, file_uuid: &str, _node_id_map: &HashMap<(String, String), i64>) -> Result { +async fn build_has_appearance_edges( + pool: &PgPool, + file_uuid: &str, + _node_id_map: &HashMap<(String, String), i64>, +) -> Result { let nodes_table = t("tkg_nodes"); let edges_table = t("tkg_edges"); @@ -3181,10 +3205,38 @@ mod tests { #[test] fn test_build_frame_index() { let points = vec![ - FacePoint { trace_id: 1, frame: 0, x: 100.0, y: 100.0, w: 50.0, h: 50.0 }, - FacePoint { trace_id: 1, frame: 0, x: 200.0, y: 200.0, w: 50.0, h: 50.0 }, - FacePoint { trace_id: 2, frame: 1, x: 150.0, y: 150.0, w: 50.0, h: 50.0 }, - FacePoint { trace_id: 1, frame: 2, x: 120.0, y: 120.0, w: 50.0, h: 50.0 }, + FacePoint { + trace_id: 1, + frame: 0, + x: 100.0, + y: 100.0, + w: 50.0, + h: 50.0, + }, + FacePoint { + trace_id: 1, + frame: 0, + x: 200.0, + y: 200.0, + w: 50.0, + h: 50.0, + }, + FacePoint { + trace_id: 2, + frame: 1, + x: 150.0, + y: 150.0, + w: 50.0, + h: 50.0, + }, + FacePoint { + trace_id: 1, + frame: 2, + x: 120.0, + y: 120.0, + w: 50.0, + h: 50.0, + }, ]; let index = build_frame_index(&points); @@ -3205,8 +3257,22 @@ mod tests { #[test] fn test_find_trace_in_frame_exact_match() { let points = vec![ - FacePoint { trace_id: 1, frame: 0, x: 100.0, y: 100.0, w: 100.0, h: 100.0 }, - FacePoint { trace_id: 2, frame: 0, x: 300.0, y: 300.0, w: 100.0, h: 100.0 }, + FacePoint { + trace_id: 1, + frame: 0, + x: 100.0, + y: 100.0, + w: 100.0, + h: 100.0, + }, + FacePoint { + trace_id: 2, + frame: 0, + x: 300.0, + y: 300.0, + w: 100.0, + h: 100.0, + }, ]; let index = build_frame_index(&points); @@ -3218,8 +3284,22 @@ mod tests { #[test] fn test_find_trace_in_frame_closest_match() { let points = vec![ - FacePoint { trace_id: 1, frame: 0, x: 100.0, y: 100.0, w: 100.0, h: 100.0 }, - FacePoint { trace_id: 2, frame: 0, x: 500.0, y: 500.0, w: 100.0, h: 100.0 }, + FacePoint { + trace_id: 1, + frame: 0, + x: 100.0, + y: 100.0, + w: 100.0, + h: 100.0, + }, + FacePoint { + trace_id: 2, + frame: 0, + x: 500.0, + y: 500.0, + w: 100.0, + h: 100.0, + }, ]; let index = build_frame_index(&points); @@ -3230,9 +3310,14 @@ mod tests { #[test] fn test_find_trace_in_frame_no_matching_frame() { - let points = vec![ - FacePoint { trace_id: 1, frame: 0, x: 100.0, y: 100.0, w: 100.0, h: 100.0 }, - ]; + let points = vec![FacePoint { + trace_id: 1, + frame: 0, + x: 100.0, + y: 100.0, + w: 100.0, + h: 100.0, + }]; let index = build_frame_index(&points); // Query for frame 1 which doesn't exist @@ -3256,9 +3341,21 @@ mod tests { map.insert(("speaker".to_string(), "SPEAKER_0".to_string()), 200); map.insert(("object".to_string(), "person".to_string()), 300); - assert_eq!(map.get(&("face_track".to_string(), "face_track_1".to_string())), Some(&100)); - assert_eq!(map.get(&("speaker".to_string(), "SPEAKER_0".to_string())), Some(&200)); - assert_eq!(map.get(&("object".to_string(), "person".to_string())), Some(&300)); - assert_eq!(map.get(&("face_track".to_string(), "face_track_99".to_string())), None); + assert_eq!( + map.get(&("face_track".to_string(), "face_track_1".to_string())), + Some(&100) + ); + assert_eq!( + map.get(&("speaker".to_string(), "SPEAKER_0".to_string())), + Some(&200) + ); + assert_eq!( + map.get(&("object".to_string(), "person".to_string())), + Some(&300) + ); + assert_eq!( + map.get(&("face_track".to_string(), "face_track_99".to_string())), + None + ); } }