admin/momentry_core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

release: v1.3.0 - TKG node type renaming

Browse Source 
Changes:
- Rust: face_trace → face_track (45 occurrences in 8 files)
- Rust: gaze_trace → gaze_track, lip_trace → lip_track
- Python: tkg_builder.py unified + pipeline_checklist.py fixed
- Swift: swift_hand.swift hand state detection (empty vs holding)

Node type changes:
  face_trace    → face_track
  person_trace  → body_track
  gaze_trace    → gaze_track
  lip_trace     → lip_track
  hand_trace    → hand_track
  speaker       → speaker_segment
  object        → detected_object
  text_trace    → text_region

Migration:
  PUBLIC schema: 12970 + 892 + 305 rows updated
This commit is contained in:
Accusys
2026-06-22 07:18:21 +08:00
parent bce9435823
commit 7e548f8b08
35 changed files with 2789 additions and 481 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/api/agent_search.rs
View File

@@ -247,10 +247,10 @@ fn make_tools(pool: &sqlx::PgPool) -> Vec<ToolDef> {
),
function_calling::make_tool(
"tkg_nodes_query",
"查詢 TKG 知識圖譜的節點列表。可依照節點類型篩選（face_trace, gaze_trace, lip_trace, text_trace, appearance_trace, skin_tone_trace, object, speaker）。適合查詢影片中有多少人物軌跡、文字片段等。",
"查詢 TKG 知識圖譜的節點列表。可依照節點類型篩選（face_track, gaze_track, lip_track, text_region, appearance_trace, skin_tone_trace, object, speaker）。適合查詢影片中有多少人物軌跡、文字片段等。",
serde_json::json!({
"file_uuid": {"type": "string", "description": "影片 UUID"},
"node_type": {"type": "string", "description": "節點類型（可選）: face_trace, gaze_trace, lip_trace, text_trace, appearance_trace, skin_tone_trace, object, speaker"},
"node_type": {"type": "string", "description": "節點類型（可選）: face_track, gaze_track, lip_track, text_region, appearance_trace, skin_tone_trace, object, speaker"},
"page": {"type": "integer", "default": 1},
"page_size": {"type": "integer", "default": 20}
}),

129
src/api/identity_agent_api.rs
View File

@@ -200,7 +200,7 @@ async fn match_from_photo(
// 4. Find best matching trace (highest similarity, no threshold)
let fd_table = schema::table_name("face_detections");
let best_match: Option<(i32, i32, f64)> = sqlx::query_as(&format!(
r#"SELECT id, trace_id,
r#"SELECT id, face_track_id,
1 - (embedding::vector <=> $1::vector) as similarity
FROM {}
WHERE file_uuid = $2 AND embedding IS NOT NULL
@@ -242,7 +242,7 @@ async fn match_from_photo(
matches: 1,
traces_matched,
message: format!(
"Best trace: trace_id={}, similarity={:.4}",
"Best trace: face_track_id={}, similarity={:.4}",
fb_trace, fb_sim
),
}))
@@ -276,7 +276,7 @@ async fn match_from_trace(
let fd_table = schema::table_name("face_detections");
let all_faces: Vec<(Vec<f32>, i64)> = sqlx::query_as::<_, (Vec<f32>, i64)>(&format!(
"SELECT embedding, frame_number FROM {} \
WHERE file_uuid = $1 AND trace_id = $2 AND embedding IS NOT NULL \
WHERE file_uuid = $1 AND face_track_id = $2 AND embedding IS NOT NULL \
ORDER BY frame_number ASC",
fd_table
))
@@ -313,7 +313,7 @@ async fn match_from_trace(
// Get width*height info if available (not all pipelines store it)
let face_sizes: Vec<(i64, i32)> = sqlx::query_as::<_, (i64, i32)>(&format!(
"SELECT frame_number, COALESCE(width, 0) * COALESCE(height, 0) AS area \
FROM {} WHERE file_uuid = $1 AND trace_id = $2 AND embedding IS NOT NULL \
FROM {} WHERE file_uuid = $1 AND face_track_id = $2 AND embedding IS NOT NULL \
ORDER BY frame_number ASC",
fd_table
))
@@ -352,7 +352,7 @@ async fn match_from_trace(
for qemb in &query_embeddings {
let top = sqlx::query_as::<_, (i32, i32, f64)>(&format!(
r#"SELECT id, trace_id,
r#"SELECT id, face_track_id,
1 - (embedding::vector <=> $1::vector) as similarity
FROM {}
WHERE file_uuid = $2
@@ -374,9 +374,9 @@ async fn match_from_trace(
)
})?;
if let Some((cface_id, c_trace_id, c_sim)) = top {
if seen_trace_ids.insert(c_trace_id) {
validated.push((cface_id, c_trace_id, c_sim));
if let Some((cface_id, c_face_track_id, c_sim)) = top {
if seen_trace_ids.insert(c_face_track_id) {
validated.push((cface_id, c_face_track_id, c_sim));
}
}
}
@@ -411,7 +411,7 @@ async fn match_from_trace(
// 4. Update matched face_detections
let mut traces_matched: Vec<i32> = Vec::new();
for (id, trace_id, _similarity) in &validated {
for (id, face_track_id, _similarity) in &validated {
if let Err(e) = sqlx::query(&format!(
"UPDATE {} SET identity_id = $1 WHERE id = $2",
fd_table
@@ -427,15 +427,15 @@ async fn match_from_trace(
e
);
} else {
if !traces_matched.contains(trace_id) {
traces_matched.push(*trace_id);
if !traces_matched.contains(face_track_id) {
traces_matched.push(*face_track_id);
}
}
}
// 5. Also bind the source trace itself
let _ = sqlx::query(&format!(
"UPDATE {} SET identity_id = $1 WHERE file_uuid = $2 AND trace_id = $3",
"UPDATE {} SET identity_id = $1 WHERE file_uuid = $2 AND face_track_id = $3",
fd_table
))
.bind(identity_id)
@@ -452,7 +452,7 @@ async fn match_from_trace(
let _ = crate::core::identity::storage::save_identity_file(&*state.db, &uuid_clean).await;
let match_count = validated.len() + 1;
let trace_count = traces_matched.len();
let face_track_count = traces_matched.len();
Ok(Json(MatchFromPhotoResponse {
success: true,
identity_uuid: uuid_clean,
@@ -461,7 +461,7 @@ async fn match_from_trace(
traces_matched,
message: format!(
"Matched {} faces ({} unique traces)",
match_count, trace_count
match_count, face_track_count
),
}))
}
@@ -647,22 +647,25 @@ async fn match_faces_iterative(pool: &sqlx::PgPool, file_uuid: &str) -> anyhow::
let qdrant_embeddings = face_db.get_all_embeddings_for_file(file_uuid).await?;
if qdrant_embeddings.is_empty() {
tracing::warn!("[FaceMatch-Qdrant] No face embeddings in Qdrant for {}", file_uuid);
tracing::warn!(
"[FaceMatch-Qdrant] No face embeddings in Qdrant for {}",
file_uuid
);
return match_faces_iterative_pg(pool, file_uuid).await; // Fallback to PG
}
// Group: trace_id → Vec<(frame, embedding)>
let mut trace_faces_raw: HashMap<i32, Vec<(i64, Vec<f32>)>> = HashMap::new();
let mut face_track_faces_raw: HashMap<i32, Vec<(i64, Vec<f32>)>> = HashMap::new();
for (_, emb, payload) in &qdrant_embeddings {
trace_faces_raw
face_track_faces_raw
.entry(payload.trace_id)
.or_default()
.push((payload.frame, emb.clone()));
}
// Sample 3 embeddings per trace (front, mid, back)
let mut trace_samples: HashMap<i32, Vec<Vec<f32>>> = HashMap::new();
for (tid, mut faces) in trace_faces_raw {
let mut face_track_samples: HashMap<i32, Vec<Vec<f32>>> = HashMap::new();
for (tid, mut faces) in face_track_faces_raw {
faces.sort_by_key(|(frame, _)| *frame);
let n = faces.len();
let indices = if n <= 3 {
@@ -671,11 +674,11 @@ async fn match_faces_iterative(pool: &sqlx::PgPool, file_uuid: &str) -> anyhow::
vec![0, n / 2, n - 1]
};
let samples: Vec<Vec<f32>> = indices.iter().map(|&i| faces[i].1.clone()).collect();
trace_samples.insert(tid, samples);
face_track_samples.insert(tid, samples);
}
let total_traces = trace_samples.len();
let sample_count: usize = trace_samples.values().map(|v| v.len()).sum();
let total_traces = face_track_samples.len();
let sample_count: usize = face_track_samples.values().map(|v| v.len()).sum();
tracing::info!(
"[FaceMatch-Qdrant] Loaded {} traces, sampled {} embeddings",
total_traces,
@@ -687,7 +690,7 @@ async fn match_faces_iterative(pool: &sqlx::PgPool, file_uuid: &str) -> anyhow::
let tmdb_seeds: Vec<(i32, String, Vec<f32>)> = tmdb_rows;
let mut matched: HashMap<i32, String> = HashMap::new();
for (&tid, samples) in &trace_samples {
for (&tid, samples) in &face_track_samples {
let mut best_name = String::new();
let mut best_sim = 0.0f32;
for (_, ref name, ref tmdb_emb) in &tmdb_seeds {
@@ -711,19 +714,19 @@ async fn match_faces_iterative(pool: &sqlx::PgPool, file_uuid: &str) -> anyhow::
// Round 2+: Propagate
let mut round = 2;
while matched.len() < trace_samples.len() {
while matched.len() < face_track_samples.len() {
let prev_count = matched.len();
// Collect new matches in separate HashMap
let mut new_matches: HashMap<i32, String> = HashMap::new();
for (&tid, samples) in &trace_samples {
for (&tid, samples) in &face_track_samples {
if matched.contains_key(&tid) {
continue;
}
for (matched_tid, matched_name) in &matched {
if let Some(matched_embs) = trace_samples.get(matched_tid) {
if let Some(matched_embs) = face_track_samples.get(matched_tid) {
for face_emb in samples {
for ref_emb in matched_embs {
let s = cosine_similarity(face_emb, ref_emb);
@@ -776,7 +779,7 @@ async fn match_faces_iterative(pool: &sqlx::PgPool, file_uuid: &str) -> anyhow::
let identity_id = identities_map.get(name);
if let Some(id) = identity_id {
let rows = sqlx::query(&format!(
"UPDATE {} SET identity_id = $1 WHERE file_uuid = $2 AND trace_id = $3",
"UPDATE {} SET identity_id = $1 WHERE file_uuid = $2 AND face_track_id = $3",
fd_table
))
.bind(*id)
@@ -788,13 +791,13 @@ async fn match_faces_iterative(pool: &sqlx::PgPool, file_uuid: &str) -> anyhow::
updated += rows as usize;
// Phase 3: Also update TKG node
let external_id = format!("trace_{}", tid);
let external_id = format!("face_track_{}", tid);
let identity_name = identity_names.get(id);
let _ = sqlx::query(&format!(
"UPDATE {} SET properties = jsonb_set(\
jsonb_set(properties, '{{identity_id}}', $1::jsonb, false),\
'{{identity_name}}', $2::jsonb, false)\
WHERE file_uuid = $3 AND node_type = 'face_trace' AND external_id = $4",
WHERE file_uuid = $3 AND node_type = 'face_track' AND external_id = $4",
nodes_table
))
.bind(*id)
@@ -828,12 +831,12 @@ async fn match_faces_iterative_pg(pool: &sqlx::PgPool, file_uuid: &str) -> anyho
tmdb_rows.len()
);
// Step 2: 載入所有 face_detections（含 frame_number），按 trace_id 分組
// Step 2: 載入所有 face_detections（含 frame_number），按 face_track_id 分組
let fd_table = schema::table_name("face_detections");
let fd_rows = sqlx::query_as::<_, (i32, i64, Vec<f32>)>(&format!(
"SELECT trace_id, frame_number, embedding FROM {} \
WHERE file_uuid=$1 AND trace_id IS NOT NULL AND embedding IS NOT NULL \
ORDER BY trace_id, frame_number",
"SELECT face_track_id, frame_number, embedding FROM {} \
WHERE file_uuid=$1 AND face_track_id IS NOT NULL AND embedding IS NOT NULL \
ORDER BY face_track_id, frame_number",
fd_table
))
.bind(file_uuid)
@@ -845,19 +848,19 @@ async fn match_faces_iterative_pg(pool: &sqlx::PgPool, file_uuid: &str) -> anyho
return Ok(0);
}
// 分組：trace_id → (frame_number, embedding)
// 分組：face_track_id → (frame_number, embedding)
use std::collections::HashMap;
let mut trace_faces_raw: HashMap<i32, Vec<(i64, Vec<f32>)>> = HashMap::new();
let mut face_track_faces_raw: HashMap<i32, Vec<(i64, Vec<f32>)>> = HashMap::new();
for (tid, frame, emb) in &fd_rows {
trace_faces_raw
face_track_faces_raw
.entry(*tid)
.or_insert_with(Vec::new)
.push((*frame, emb.clone()));
}
// 從每個 trace 選取不同角度的 3 個 face embedding
let mut trace_samples: HashMap<i32, Vec<Vec<f32>>> = HashMap::new();
for (tid, mut faces) in trace_faces_raw {
let mut face_track_samples: HashMap<i32, Vec<Vec<f32>>> = HashMap::new();
for (tid, mut faces) in face_track_faces_raw {
faces.sort_by_key(|(frame, _)| *frame);
let n = faces.len();
let indices = if n <= 3 {
@@ -867,11 +870,11 @@ async fn match_faces_iterative_pg(pool: &sqlx::PgPool, file_uuid: &str) -> anyho
vec![0, mid, n - 1]
};
let samples: Vec<Vec<f32>> = indices.iter().map(|&i| faces[i].1.clone()).collect();
trace_samples.insert(tid, samples);
face_track_samples.insert(tid, samples);
}
let total_traces = trace_samples.len();
let sample_count: usize = trace_samples.values().map(|v| v.len()).sum();
let total_traces = face_track_samples.len();
let sample_count: usize = face_track_samples.values().map(|v| v.len()).sum();
tracing::info!(
"[FaceMatch-PG] Loaded {} traces, sampled {} embeddings (3-angle)",
total_traces,
@@ -883,10 +886,10 @@ async fn match_faces_iterative_pg(pool: &sqlx::PgPool, file_uuid: &str) -> anyho
// Step 4: 迭代匹配
const TH: f32 = 0.50;
let mut matched: HashMap<i32, String> = HashMap::new(); // trace_id → identity_name
let mut matched: HashMap<i32, String> = HashMap::new(); // face_track_id → identity_name
// Round 1: 用 3-angle samples 比對 TMDb
for (&tid, samples) in &trace_samples {
for (&tid, samples) in &face_track_samples {
let mut best_name = String::new();
let mut best_sim = 0.0f32;
for (_, ref name, ref tmdb_emb) in &tmdb_seeds {
@@ -924,7 +927,7 @@ async fn match_faces_iterative_pg(pool: &sqlx::PgPool, file_uuid: &str) -> anyho
.await?;
if let Some(identity_id) = id_opt {
let _ = sqlx::query(&format!(
"UPDATE {} SET identity_id=$1 WHERE file_uuid=$2 AND trace_id=$3",
"UPDATE {} SET identity_id=$1 WHERE file_uuid=$2 AND face_track_id=$3",
fd_table
))
.bind(identity_id)
@@ -934,12 +937,12 @@ async fn match_faces_iterative_pg(pool: &sqlx::PgPool, file_uuid: &str) -> anyho
.await;
// Phase 3: Also update TKG node
let external_id = format!("trace_{}", tid);
let external_id = format!("face_track_{}", tid);
let _ = sqlx::query(&format!(
"UPDATE {} SET properties = jsonb_set(\
jsonb_set(properties, '{{identity_id}}', $1::jsonb, false),\
'{{identity_name}}', $2::jsonb, false)\
WHERE file_uuid = $3 AND node_type = 'face_trace' AND external_id = $4",
WHERE file_uuid = $3 AND node_type = 'face_track' AND external_id = $4",
nodes_table
))
.bind(identity_id)
@@ -961,7 +964,7 @@ async fn match_faces_iterative_pg(pool: &sqlx::PgPool, file_uuid: &str) -> anyho
// 建立 seed pool: name → Vec<embedding>
let mut seed_pool: HashMap<String, Vec<&Vec<f32>>> = HashMap::new();
for (&tid, name) in &matched {
if let Some(samples) = trace_samples.get(&tid) {
if let Some(samples) = face_track_samples.get(&tid) {
seed_pool
.entry(name.clone())
.or_default()
@@ -970,7 +973,7 @@ async fn match_faces_iterative_pg(pool: &sqlx::PgPool, file_uuid: &str) -> anyho
}
let mut new_matches: Vec<(i32, String)> = Vec::new();
for (&tid, samples) in &trace_samples {
for (&tid, samples) in &face_track_samples {
if matched.contains_key(&tid) {
continue;
}
@@ -1014,11 +1017,11 @@ async fn match_faces_iterative_pg(pool: &sqlx::PgPool, file_uuid: &str) -> anyho
// Step 6: 未匹配的 trace 設 stranger_id = strangers.id (FK)
// First: ensure strangers records exist
let _ = sqlx::query(&format!(
"INSERT INTO {} (file_uuid, trace_id) \
SELECT $1, fd.trace_id FROM {} fd \
WHERE fd.file_uuid = $1 AND fd.trace_id IS NOT NULL \
"INSERT INTO {} (file_uuid, face_track_id) \
SELECT $1, fd.face_track_id FROM {} fd \
WHERE fd.file_uuid = $1 AND fd.face_track_id IS NOT NULL \
AND fd.identity_id IS NULL \
ON CONFLICT (file_uuid, trace_id) DO NOTHING",
ON CONFLICT (file_uuid, face_track_id) DO NOTHING",
strangers_table, fd_table
))
.bind(file_uuid)
@@ -1029,9 +1032,9 @@ async fn match_faces_iterative_pg(pool: &sqlx::PgPool, file_uuid: &str) -> anyho
let stranger_update = sqlx::query(&format!(
"UPDATE {} fd SET stranger_id = s.id \
FROM {} s \
WHERE s.file_uuid = fd.file_uuid AND s.trace_id = fd.trace_id \
WHERE s.file_uuid = fd.file_uuid AND s.face_track_id = fd.face_track_id \
AND fd.file_uuid = $1 AND fd.identity_id IS NULL \
AND fd.trace_id IS NOT NULL AND fd.stranger_id IS NULL",
AND fd.face_track_id IS NOT NULL AND fd.stranger_id IS NULL",
fd_table, strangers_table
))
.bind(file_uuid)
@@ -1069,16 +1072,16 @@ async fn match_faces_iterative_pg(pool: &sqlx::PgPool, file_uuid: &str) -> anyho
}
/// Bind ASRX speakers to face traces based on temporal overlap.
/// Reads face_detections (trace_id, identity_id, frame_number) and ASRX
/// Reads face_detections (face_track_id, identity_id, frame_number) and ASRX
/// segments (speaker_id, start_time, end_time), computes overlap,
/// and stores bindings in identity_bindings table.
pub async fn bind_speakers(pool: &sqlx::PgPool, file_uuid: &str) -> anyhow::Result<usize> {
// Load face traces with identity_id and frame numbers
let fd_table = schema::table_name("face_detections");
let traces = sqlx::query_as::<_, (i32, Vec<i32>)>(&format!(
"SELECT trace_id, array_agg(frame_number ORDER BY frame_number) \
FROM {} WHERE file_uuid=$1 AND trace_id IS NOT NULL AND identity_id IS NOT NULL \
GROUP BY trace_id",
"SELECT face_track_id, array_agg(frame_number ORDER BY frame_number) \
FROM {} WHERE file_uuid=$1 AND face_track_id IS NOT NULL AND identity_id IS NOT NULL \
GROUP BY face_track_id",
fd_table
))
.bind(file_uuid)
@@ -1141,7 +1144,7 @@ pub async fn bind_speakers(pool: &sqlx::PgPool, file_uuid: &str) -> anyhow::Resu
// For each trace, compute overlap with each speaker
let mut bindings = 0usize;
for (trace_id, frames) in &traces {
for (face_track_id, frames) in &traces {
if frames.is_empty() {
continue;
}
@@ -1149,9 +1152,9 @@ pub async fn bind_speakers(pool: &sqlx::PgPool, file_uuid: &str) -> anyhow::Resu
// Get identity_id for this trace
let fd_table = schema::table_name("face_detections");
let identity_id: Option<i32> = sqlx::query_scalar(
&format!("SELECT identity_id FROM {} WHERE file_uuid=$1 AND trace_id=$2 AND identity_id IS NOT NULL LIMIT 1", fd_table)
&format!("SELECT identity_id FROM {} WHERE file_uuid=$1 AND face_track_id=$2 AND identity_id IS NOT NULL LIMIT 1", fd_table)
)
.bind(file_uuid).bind(trace_id)
.bind(file_uuid).bind(face_track_id)
.fetch_optional(pool).await?.flatten();
if identity_id.is_none() {
@@ -1184,7 +1187,7 @@ pub async fn bind_speakers(pool: &sqlx::PgPool, file_uuid: &str) -> anyhow::Resu
let overlap_ratio = best_overlap as f64 / frames.len() as f64;
if overlap_ratio > 0.3 && !best_speaker.is_empty() {
let metadata = serde_json::json!({
"trace_id": trace_id,
"trace_id": face_track_id,
"overlap_frames": best_overlap,
"total_frames": frames.len(),
"overlap_ratio": overlap_ratio,
@@ -1278,7 +1281,7 @@ pub async fn run_identity_agent(db: &PostgresDb, file_uuid: &str) -> anyhow::Res
"reasoning": identities[0].reasoning,
});
let _ = sqlx::query(&format!(
"INSERT INTO {} (file_uuid, trace_id, metadata) \
"INSERT INTO {} (file_uuid, face_track_id, metadata) \
VALUES ($1, NULL, $2::jsonb) ON CONFLICT DO NOTHING",
schema::table_name("strangers")
))

42
src/api/identity_binding.rs
View File

@@ -225,7 +225,7 @@ pub async fn unbind_identity(
)
})?;
// Phase 2.3: Also update TKG node (find trace_id first)
// Phase 2.3: Also update TKG node (find face_track_id first)
let trace_id_opt: Option<i32> = sqlx::query_scalar(&format!(
"SELECT trace_id FROM {} WHERE file_uuid = $1 AND face_id = $2",
table
@@ -239,10 +239,10 @@ pub async fn unbind_identity(
if let Some(trace_id) = trace_id_opt {
let nodes_table = crate::core::db::schema::table_name("tkg_nodes");
let external_id = format!("trace_{}", trace_id);
let external_id = format!("face_track_{}", trace_id);
let _ = sqlx::query(&format!(
"UPDATE {} SET properties = properties - 'identity_id' - 'identity_name' \
WHERE file_uuid = $1 AND node_type = 'face_trace' AND external_id = $2",
WHERE file_uuid = $1 AND node_type = 'face_track' AND external_id = $2",
nodes_table
))
.bind(&req.file_uuid)
@@ -789,7 +789,7 @@ pub async fn bind_identity_trace(
// Capture old identity_id before bind trace (use first face in trace as reference)
let old_identity_id: Option<i32> = sqlx::query_scalar(&format!(
"SELECT identity_id FROM {} WHERE file_uuid = $1 AND trace_id = $2 LIMIT 1",
"SELECT identity_id FROM {} WHERE trace_id = $2 LIMIT 1",
fd_table
))
.bind(&req.file_uuid)
@@ -805,7 +805,7 @@ pub async fn bind_identity_trace(
.flatten();
let result = sqlx::query(&format!(
"UPDATE {} SET identity_id = $1 WHERE file_uuid = $2 AND trace_id = $3",
"UPDATE {} SET identity_id = $1 WHERE trace_id = $3",
fd_table
))
.bind(identity_id)
@@ -820,24 +820,22 @@ pub async fn bind_identity_trace(
)
})?;
// Phase 2.3: Also update TKG node properties
// Phase 2.3: Also update TKG node properties
let nodes_table = crate::core::db::schema::table_name("tkg_nodes");
let external_id = format!("trace_{}", req.trace_id);
let identity_name: Option<String> = sqlx::query_scalar(&format!(
"SELECT name FROM {} WHERE id = $1",
id_table
))
.bind(identity_id)
.fetch_optional(state.db.pool())
.await
.ok()
.flatten();
let external_id = format!("face_track_{}", req.trace_id);
let identity_name: Option<String> =
sqlx::query_scalar(&format!("SELECT name FROM {} WHERE id = $1", id_table))
.bind(identity_id)
.fetch_optional(state.db.pool())
.await
.ok()
.flatten();
let _ = sqlx::query(&format!(
"UPDATE {} SET properties = jsonb_set(\
jsonb_set(properties, '{{identity_id}}', $1::jsonb, false),\
'{{identity_name}}', $2::jsonb, false)\
WHERE file_uuid = $3 AND node_type = 'face_trace' AND external_id = $4",
WHERE file_uuid = $3 AND node_type = 'face_track' AND external_id = $4",
nodes_table
))
.bind(identity_id)
@@ -941,8 +939,8 @@ pub async fn get_identity_traces(
FROM {} fd
LEFT JOIN dev.videos v ON fd.file_uuid = v.file_uuid
WHERE fd.identity_id = $1
GROUP BY fd.file_uuid, fd.trace_id, v.fps
ORDER BY fd.file_uuid, fd.trace_id
GROUP BY trace_id, v.fps
ORDER BY trace_id
LIMIT $2 OFFSET $3"#,
fd_table
))
@@ -955,7 +953,7 @@ pub async fn get_identity_traces(
// Get total count for pagination
let total: (i64,) = sqlx::query_as(&format!(
"SELECT COUNT(*) FROM (SELECT 1 FROM {} fd WHERE fd.identity_id = $1 GROUP BY fd.file_uuid, fd.trace_id) sub",
"SELECT COUNT(*) FROM (SELECT 1 FROM {} fd WHERE trace_id) sub",
fd_table
))
.bind(identity_id)
@@ -1563,7 +1561,7 @@ async fn apply_bind_snapshot(
Ok(rows.rows_affected() as i64)
} else if let Some(trace_id) = snapshot.get("trace_id").and_then(|v| v.as_i64()) {
let rows = sqlx::query(&format!(
"UPDATE {} SET identity_id = $1 WHERE file_uuid = $2 AND trace_id = $3",
"UPDATE {} SET identity_id = $1 WHERE trace_id = $3",
face_table
))
.bind(id_val)
@@ -1581,7 +1579,7 @@ async fn apply_bind_snapshot(
} else {
Err((
StatusCode::INTERNAL_SERVER_ERROR,
Json(serde_json::json!({"error": "Snapshot has neither face_id nor trace_id"})),
Json(serde_json::json!({"error": "Snapshot has neither face_id nor face_track_id"})),
))
}
}

2
src/api/scan.rs
View File

@@ -469,7 +469,7 @@ async fn get_ingestion_status(
Some(format!("{scene_count} scene chunks"))
),
step!(
"face_trace",
"face_track",
trace_count > 0,
Some(format!("{trace_count} traces / {face_total} detections"))
),

30
src/api/trace_agent_api.rs
View File

@@ -983,7 +983,10 @@ async fn rebuild_tkg(
+ r.wears_edges;
if total_edges > 0 {
info!("[TKG] {} relationship edges found, triggering Rule 2 ingestion...", total_edges);
info!(
"[TKG] {} relationship edges found, triggering Rule 2 ingestion...",
total_edges
);
match ingest_rule2(state.db.pool(), &file_uuid).await {
Ok(count) => info!("[TKG] Rule 2 created {} relationship chunks", count),
Err(e) => info!("[TKG] Rule 2 ingestion failed: {}", e),
@@ -994,10 +997,10 @@ async fn rebuild_tkg(
success: true,
file_uuid,
result: Some(serde_json::json!({
"face_trace_nodes": r.face_trace_nodes,
"gaze_trace_nodes": r.gaze_trace_nodes,
"lip_trace_nodes": r.lip_trace_nodes,
"text_trace_nodes": r.text_trace_nodes,
"face_track_nodes": r.face_track_nodes,
"gaze_track_nodes": r.gaze_track_nodes,
"lip_track_nodes": r.lip_track_nodes,
"text_region_nodes": r.text_region_nodes,
"appearance_trace_nodes": r.appearance_trace_nodes,
"skin_tone_trace_nodes": r.skin_tone_trace_nodes,
"accessory_nodes": r.accessory_nodes,
@@ -1517,9 +1520,9 @@ async fn ingest_rule2(
Path(file_uuid): Path<String>,
) -> Result<Json<IngestRule2Response>, (StatusCode, Json<serde_json::Value>)> {
use crate::core::chunk::rule2_ingest::ingest_rule2;
use crate::core::embedding::Embedder;
use crate::core::db::schema;
use crate::core::db::qdrant_db::{QdrantDb, VectorPayload};
use crate::core::db::schema;
use crate::core::embedding::Embedder;
use tracing::info;
let result = ingest_rule2(state.db.pool(), &file_uuid).await;
@@ -1559,7 +1562,12 @@ async fn ingest_rule2(
continue;
}
if let Ok(vector) = embedder.embed_document(&text).await {
if state.db.store_vector(&chunk_id, &vector, &file_uuid).await.is_ok() {
if state
.db
.store_vector(&chunk_id, &vector, &file_uuid)
.await
.is_ok()
{
let payload = VectorPayload {
file_uuid: file_uuid.clone(),
chunk_id: chunk_id.clone(),
@@ -1570,7 +1578,11 @@ async fn ingest_rule2(
end_time: *end_time,
text: Some(text.clone()),
};
if qdrant.upsert_vector(&chunk_id, &vector, payload).await.is_ok() {
if qdrant
.upsert_vector(&chunk_id, &vector, payload)
.await
.is_ok()
{
vectorized += 1;
}
}

14
src/cli/agent.rs
View File

@@ -7,8 +7,8 @@ pub async fn handle_agent(tool: &str, args_str: &str) -> Result<()> {
let db = PostgresDb::init()
.await
.context("Failed to initialize database")?;
let args: serde_json::Value = serde_json::from_str(args_str)
.context("Failed to parse JSON arguments")?;
let args: serde_json::Value =
serde_json::from_str(args_str).context("Failed to parse JSON arguments")?;
let pool = db.pool();
let result = match tool {
@@ -35,12 +35,10 @@ pub async fn handle_agent(tool: &str, args_str: &str) -> Result<()> {
};
match result {
Ok(json_str) => {
match serde_json::from_str::<serde_json::Value>(&json_str) {
Ok(value) => println!("{}", serde_json::to_string_pretty(&value)?),
Err(_) => println!("{}", json_str),
}
}
Ok(json_str) => match serde_json::from_str::<serde_json::Value>(&json_str) {
Ok(value) => println!("{}", serde_json::to_string_pretty(&value)?),
Err(_) => println!("{}", json_str),
},
Err(e) => {
eprintln!("Error: {}", e);
std::process::exit(1);

79
src/core/agent/tools.rs
View File

@@ -14,7 +14,10 @@ fn t(name: &str) -> String {
}
}
pub async fn exec_find_file(pool: &sqlx::PgPool, args: &serde_json::Value) -> Result<String, String> {
pub async fn exec_find_file(
pool: &sqlx::PgPool,
args: &serde_json::Value,
) -> Result<String, String> {
let query = args.get("query").and_then(|v| v.as_str()).unwrap_or("");
let videos = schema::table_name("videos");
let fd_table = schema::table_name("face_detections");
@@ -41,7 +44,10 @@ pub async fn exec_find_file(pool: &sqlx::PgPool, args: &serde_json::Value) -> Re
Ok(serde_json::json!({"found": true, "files": files}).to_string())
}
pub async fn exec_list_files(pool: &sqlx::PgPool, args: &serde_json::Value) -> Result<String, String> {
pub async fn exec_list_files(
pool: &sqlx::PgPool,
args: &serde_json::Value,
) -> Result<String, String> {
let limit = args.get("limit").and_then(|v| v.as_i64()).unwrap_or(10);
let videos = schema::table_name("videos");
let fd_table = schema::table_name("face_detections");
@@ -63,7 +69,10 @@ pub async fn exec_list_files(pool: &sqlx::PgPool, args: &serde_json::Value) -> R
Ok(serde_json::json!({"files": files}).to_string())
}
pub async fn exec_tkg_query(pool: &sqlx::PgPool, args: &serde_json::Value) -> Result<String, String> {
pub async fn exec_tkg_query(
pool: &sqlx::PgPool,
args: &serde_json::Value,
) -> Result<String, String> {
let file_uuid = args.get("file_uuid").and_then(|v| v.as_str()).unwrap_or("");
let query_type = args
.get("query_type")
@@ -137,8 +146,8 @@ pub async fn exec_tkg_query(pool: &sqlx::PgPool, args: &serde_json::Value) -> Re
FROM {} e \
JOIN {} a ON a.id = e.source_node_id \
JOIN {} b ON b.id = e.target_node_id \
JOIN {} fd_a ON fd_a.file_uuid = $1 AND fd_a.trace_id = REPLACE(a.external_id, 'trace_', '')::int \
JOIN {} fd_b ON fd_b.file_uuid = $1 AND fd_b.trace_id = REPLACE(b.external_id, 'trace_', '')::int \
JOIN {} fd_a ON fd_a.file_uuid = $1 AND fd_a.face_track_id = REPLACE(a.external_id, 'face_track_', '')::int \
JOIN {} fd_b ON fd_b.file_uuid = $1 AND fd_b.face_track_id = REPLACE(b.external_id, 'face_track_', '')::int \
JOIN {} ia ON ia.id = fd_a.identity_id \
JOIN {} ib ON ib.id = fd_b.identity_id \
WHERE e.file_uuid = $1 AND ia.name ILIKE $2 AND ib.name ILIKE $3 \
@@ -156,8 +165,8 @@ pub async fn exec_tkg_query(pool: &sqlx::PgPool, args: &serde_json::Value) -> Re
FROM {} e \
JOIN {} a ON a.id = e.source_node_id \
JOIN {} b ON b.id = e.target_node_id \
JOIN {} fd_a ON fd_a.trace_id = REPLACE(a.external_id, 'trace_', '')::int AND fd_a.file_uuid = $1 \
JOIN {} fd_b ON fd_b.trace_id = REPLACE(b.external_id, 'trace_', '')::int AND fd_b.file_uuid = $1 \
JOIN {} fd_a ON fd_a.face_track_id = REPLACE(a.external_id, 'face_track_', '')::int AND fd_a.file_uuid = $1 \
JOIN {} fd_b ON fd_b.face_track_id = REPLACE(b.external_id, 'face_track_', '')::int AND fd_b.file_uuid = $1 \
JOIN {} ia ON ia.id = fd_a.identity_id \
JOIN {} ib ON ib.id = fd_b.identity_id \
WHERE e.file_uuid = $1 AND e.edge_type = 'CO_OCCURS_WITH' \
@@ -174,10 +183,10 @@ pub async fn exec_tkg_query(pool: &sqlx::PgPool, args: &serde_json::Value) -> Re
"identity_traces" => {
let name = identity_name.unwrap_or("");
let rows: Vec<(i32, i64, i64, i64)> = sqlx::query_as(&format!(
"SELECT fd.trace_id, COUNT(*)::bigint, MIN(fd.frame_number)::bigint, MAX(fd.frame_number)::bigint \
"SELECT fd.face_track_id, COUNT(*)::bigint, MIN(fd.frame_number)::bigint, MAX(fd.frame_number)::bigint \
FROM {} fd JOIN {} i ON i.id = fd.identity_id \
WHERE fd.file_uuid = $1 AND i.name ILIKE $2 \
GROUP BY fd.trace_id ORDER BY COUNT(*) DESC LIMIT $3",
GROUP BY fd.face_track_id ORDER BY COUNT(*) DESC LIMIT $3",
fd_table, id_table
))
.bind(file_uuid).bind(name).bind(limit)
@@ -203,8 +212,8 @@ pub async fn exec_tkg_query(pool: &sqlx::PgPool, args: &serde_json::Value) -> Re
FROM {} i \
JOIN {} fd ON fd.identity_id = i.id AND ($2::text IS NULL OR fd.file_uuid = $2) \
JOIN {} fn ON fn.file_uuid = fd.file_uuid \
AND fn.node_type = 'face_trace' \
AND fn.external_id = CONCAT('trace_', fd.trace_id) \
AND fn.node_type = 'face_track' \
AND fn.external_id = CONCAT('face_track_', fd.face_track_id) \
JOIN {} e ON e.source_node_id = fn.id \
AND e.edge_type = 'SPEAKS_AS' \
AND ($2::text IS NULL OR e.file_uuid = $2) \
@@ -242,8 +251,8 @@ pub async fn exec_tkg_query(pool: &sqlx::PgPool, args: &serde_json::Value) -> Re
FROM {} i \
JOIN {} fd ON fd.identity_id = i.id AND ($3::text IS NULL OR fd.file_uuid = $3) \
JOIN {} fn ON fn.file_uuid = fd.file_uuid \
AND fn.node_type = 'face_trace' \
AND fn.external_id = CONCAT('trace_', fd.trace_id) \
AND fn.node_type = 'face_track' \
AND fn.external_id = CONCAT('face_track_', fd.face_track_id) \
JOIN {} e ON e.source_node_id = fn.id \
AND e.edge_type = 'SPEAKS_AS' \
AND ($3::text IS NULL OR e.file_uuid = $3) \
@@ -371,7 +380,7 @@ pub async fn exec_identity_text(
let sql = format!(
"SELECT c.chunk_id, c.start_time, c.end_time, c.text_content, \
i.name AS identity_name, fd.trace_id, i.source AS identity_source \
i.name AS identity_name, fd.face_track_id, i.source AS identity_source \
FROM {} c \
JOIN {} fd ON fd.file_uuid = c.file_uuid \
AND fd.frame_number BETWEEN c.start_frame AND c.end_frame \
@@ -408,7 +417,7 @@ pub async fn exec_identity_text(
"end_time": et,
"text": txt,
"identity_name": name,
"trace_id": tid,
"face_track_id": tid,
"source": src
})
} ).collect::<Vec<_>>()})
@@ -435,7 +444,7 @@ pub async fn exec_identities_search(
let sql = format!(
"SELECT DISTINCT ON (i.name, c.chunk_id) \
i.name, c.chunk_id, c.start_time, c.end_time, c.text_content, fd.trace_id \
i.name, c.chunk_id, c.start_time, c.end_time, c.text_content, fd.face_track_id \
FROM {} i \
JOIN {} fd ON fd.identity_id = i.id \
JOIN {} c ON c.file_uuid = fd.file_uuid \
@@ -465,7 +474,7 @@ pub async fn exec_identities_search(
"start_time": st,
"end_time": et,
"text": txt,
"trace_id": tid,
"face_track_id": tid,
})
}).collect::<Vec<_>>()})
.to_string(),
@@ -549,29 +558,25 @@ pub async fn exec_analyze_frame(
let frame_number = match args.get("frame_number").and_then(|v| v.as_i64()) {
Some(f) => f,
None => {
match query_auto_representative_frame(pool, file_uuid)
None => match query_auto_representative_frame(pool, file_uuid).await {
Ok(r) => r.frame_number,
Err(_) => {
let duration: f64 = sqlx::query_scalar(&format!(
"SELECT COALESCE(duration, 0) FROM {} WHERE file_uuid = $1",
videos
))
.bind(file_uuid)
.fetch_optional(pool)
.await
{
Ok(r) => r.frame_number,
Err(_) => {
let duration: f64 = sqlx::query_scalar(&format!(
"SELECT COALESCE(duration, 0) FROM {} WHERE file_uuid = $1",
videos
))
.bind(file_uuid)
.fetch_optional(pool)
.await
.map_err(|e| e.to_string())?
.unwrap_or(0.0);
if duration > 0.0 {
((duration / 2.0) * fps) as i64
} else {
0
}
.map_err(|e| e.to_string())?
.unwrap_or(0.0);
if duration > 0.0 {
((duration / 2.0) * fps) as i64
} else {
0
}
}
}
},
};
let timestamp_secs = frame_number as f64 / fps;

53
src/core/chunk/rule2_ingest.rs
View File

@@ -99,8 +99,11 @@ pub async fn ingest_rule2(pool: &PgPool, file_uuid: &str) -> Result<usize> {
let (src_type, src_ext_id, src_label, _src_props) = source_node.unwrap();
let (tgt_type, tgt_ext_id, tgt_label, tgt_props) = target_node.unwrap();
// Resolve identity names for face_trace/gaze_trace/lip_trace nodes (Phase 2.7)
let src_identity: Option<String> = if src_type == "face_trace" || src_type == "gaze_trace" || src_type == "lip_trace" {
// Resolve identity names for face_track/gaze_track/lip_track nodes (Phase 2.7)
let src_identity: Option<String> = if src_type == "face_track"
|| src_type == "gaze_track"
|| src_type == "lip_track"
{
sqlx::query_scalar(&format!(
"SELECT i.name FROM {} n \
JOIN {} i ON i.id = (n.properties->>'identity_id')::bigint \
@@ -116,7 +119,10 @@ pub async fn ingest_rule2(pool: &PgPool, file_uuid: &str) -> Result<usize> {
None
};
let tgt_identity: Option<String> = if tgt_type == "face_trace" || tgt_type == "gaze_trace" || tgt_type == "lip_trace" {
let tgt_identity: Option<String> = if tgt_type == "face_track"
|| tgt_type == "gaze_track"
|| tgt_type == "lip_track"
{
sqlx::query_scalar(&format!(
"SELECT i.name FROM {} n \
JOIN {} i ON i.id = (n.properties->>'identity_id')::bigint \
@@ -246,19 +252,37 @@ pub async fn ingest_rule2(pool: &PgPool, file_uuid: &str) -> Result<usize> {
/// Generate natural language description for a relationship (template-based).
fn generate_description(context: &Value) -> String {
let edge_type = context.get("edge_type").and_then(|v| v.as_str()).unwrap_or("");
let edge_type = context
.get("edge_type")
.and_then(|v| v.as_str())
.unwrap_or("");
let src = context.get("source_node").unwrap();
let tgt = context.get("target_node").unwrap();
let props = context.get("properties").unwrap();
let src_identity = src.get("identity_name").and_then(|v| v.as_str());
let tgt_identity = tgt.get("identity_name").and_then(|v| v.as_str());
let src_ext_id = src.get("external_id").and_then(|v| v.as_str()).unwrap_or("");
let tgt_ext_id = tgt.get("external_id").and_then(|v| v.as_str()).unwrap_or("");
let src_ext_id = src
.get("external_id")
.and_then(|v| v.as_str())
.unwrap_or("");
let tgt_ext_id = tgt
.get("external_id")
.and_then(|v| v.as_str())
.unwrap_or("");
let first_frame = props.get("first_frame").and_then(|v| v.as_i64()).unwrap_or(0);
let last_frame = props.get("last_frame").and_then(|v| v.as_i64()).unwrap_or(first_frame);
let frame_count = props.get("frame_count").and_then(|v| v.as_i64()).unwrap_or(0);
let first_frame = props
.get("first_frame")
.and_then(|v| v.as_i64())
.unwrap_or(0);
let last_frame = props
.get("last_frame")
.and_then(|v| v.as_i64())
.unwrap_or(first_frame);
let frame_count = props
.get("frame_count")
.and_then(|v| v.as_i64())
.unwrap_or(0);
let src_display = src_identity.unwrap_or(src_ext_id);
let tgt_display = tgt_identity.unwrap_or(tgt_ext_id);
@@ -277,19 +301,16 @@ fn generate_description(context: &Value) -> String {
)
}
"CO_OCCURS_WITH" => {
// Check if both nodes are face_trace (face-face co-occurrence)
// Check if both nodes are face_track (face-face co-occurrence)
let src_type = src.get("node_type").and_then(|v| v.as_str()).unwrap_or("");
let tgt_type = tgt.get("node_type").and_then(|v| v.as_str()).unwrap_or("");
if src_type == "face_trace" && tgt_type == "face_trace" {
if src_type == "face_track" && tgt_type == "face_track" {
format!(
"{} 和 {} 同框 {} 幀，從 frame {} 到 frame {}",
src_display, tgt_display, frame_count, first_frame, last_frame
)
} else {
format!(
"{} 和 {} 在同一畫面出現",
src_display, tgt_display
)
format!("{} 和 {} 在同一畫面出現", src_display, tgt_display)
}
}
"HAS_APPEARANCE" => {
@@ -324,4 +345,4 @@ fn generate_description(context: &Value) -> String {
)
}
}
}
}

6
src/core/config.rs
View File

@@ -168,6 +168,12 @@ pub mod processor {
.parse()
.unwrap_or(7200)
});
pub static FORCE_RETRY: Lazy<bool> = Lazy::new(|| {
env::var("MOMENTRY_FORCE_RETRY")
.map(|v| v == "true" || v == "1")
.unwrap_or(false)
});
}
pub mod cache {

164
src/core/db/face_embedding_db.rs
View File

@@ -62,7 +62,10 @@ impl FaceEmbeddingDb {
.await?;
if response.status().is_success() {
tracing::info!("[FaceEmbedding] Collection {} already exists", self.collection_name);
tracing::info!(
"[FaceEmbedding] Collection {} already exists",
self.collection_name
);
return Ok(());
}
@@ -83,7 +86,10 @@ impl FaceEmbeddingDb {
.await
.context("Failed to create face embeddings collection")?;
tracing::info!("[FaceEmbedding] Created collection {} (dim=512)", self.collection_name);
tracing::info!(
"[FaceEmbedding] Created collection {} (dim=512)",
self.collection_name
);
Ok(())
}
@@ -226,8 +232,8 @@ impl FaceEmbeddingDb {
payload: HashMap<String, serde_json::Value>,
}
let parsed: SearchResult = serde_json::from_str(&text)
.context("Failed to parse Qdrant search response")?;
let parsed: SearchResult =
serde_json::from_str(&text).context("Failed to parse Qdrant search response")?;
let results: Vec<FaceEmbeddingPoint> = parsed
.result
@@ -240,28 +246,54 @@ impl FaceEmbeddingDb {
_ => "unknown".to_string(),
};
let payload = FaceEmbeddingPayload {
file_uuid: r.payload.get("file_uuid")
.and_then(|v| v.as_str()).unwrap_or("").to_string(),
trace_id: r.payload.get("trace_id")
.and_then(|v| v.as_i64()).unwrap_or(0) as i32,
frame: r.payload.get("frame")
.and_then(|v| v.as_i64()).unwrap_or(0),
bbox_x: r.payload.get("bbox_x")
.and_then(|v| v.as_f64()).unwrap_or(0.0),
bbox_y: r.payload.get("bbox_y")
.and_then(|v| v.as_f64()).unwrap_or(0.0),
bbox_w: r.payload.get("bbox_w")
.and_then(|v| v.as_f64()).unwrap_or(0.0),
bbox_h: r.payload.get("bbox_h")
.and_then(|v| v.as_f64()).unwrap_or(0.0),
confidence: r.payload.get("confidence")
.and_then(|v| v.as_f64()).unwrap_or(0.0),
yaw: r.payload.get("yaw")
.and_then(|v| v.as_f64()).unwrap_or(0.0),
pitch: r.payload.get("pitch")
.and_then(|v| v.as_f64()).unwrap_or(0.0),
roll: r.payload.get("roll")
.and_then(|v| v.as_f64()).unwrap_or(0.0),
file_uuid: r
.payload
.get("file_uuid")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string(),
trace_id: r
.payload
.get("trace_id")
.and_then(|v| v.as_i64())
.unwrap_or(0) as i32,
frame: r.payload.get("frame").and_then(|v| v.as_i64()).unwrap_or(0),
bbox_x: r
.payload
.get("bbox_x")
.and_then(|v| v.as_f64())
.unwrap_or(0.0),
bbox_y: r
.payload
.get("bbox_y")
.and_then(|v| v.as_f64())
.unwrap_or(0.0),
bbox_w: r
.payload
.get("bbox_w")
.and_then(|v| v.as_f64())
.unwrap_or(0.0),
bbox_h: r
.payload
.get("bbox_h")
.and_then(|v| v.as_f64())
.unwrap_or(0.0),
confidence: r
.payload
.get("confidence")
.and_then(|v| v.as_f64())
.unwrap_or(0.0),
yaw: r.payload.get("yaw").and_then(|v| v.as_f64()).unwrap_or(0.0),
pitch: r
.payload
.get("pitch")
.and_then(|v| v.as_f64())
.unwrap_or(0.0),
roll: r
.payload
.get("roll")
.and_then(|v| v.as_f64())
.unwrap_or(0.0),
};
FaceEmbeddingPoint {
id,
@@ -330,8 +362,8 @@ impl FaceEmbeddingDb {
vector: Vec<f32>,
}
let parsed: ScrollResult = serde_json::from_str(&text)
.context("Failed to parse Qdrant scroll response")?;
let parsed: ScrollResult =
serde_json::from_str(&text).context("Failed to parse Qdrant scroll response")?;
let results: Vec<(String, Vec<f32>)> = parsed
.result
@@ -404,8 +436,8 @@ impl FaceEmbeddingDb {
payload: HashMap<String, serde_json::Value>,
}
let parsed: ScrollResult = serde_json::from_str(&text)
.context("Failed to parse Qdrant scroll response")?;
let parsed: ScrollResult =
serde_json::from_str(&text).context("Failed to parse Qdrant scroll response")?;
let results: Vec<(String, Vec<f32>, FaceEmbeddingPayload)> = parsed
.result
@@ -418,28 +450,54 @@ impl FaceEmbeddingDb {
_ => "unknown".to_string(),
};
let payload = FaceEmbeddingPayload {
file_uuid: r.payload.get("file_uuid")
.and_then(|v| v.as_str()).unwrap_or("").to_string(),
trace_id: r.payload.get("trace_id")
.and_then(|v| v.as_i64()).unwrap_or(0) as i32,
frame: r.payload.get("frame")
.and_then(|v| v.as_i64()).unwrap_or(0),
bbox_x: r.payload.get("bbox_x")
.and_then(|v| v.as_f64()).unwrap_or(0.0),
bbox_y: r.payload.get("bbox_y")
.and_then(|v| v.as_f64()).unwrap_or(0.0),
bbox_w: r.payload.get("bbox_w")
.and_then(|v| v.as_f64()).unwrap_or(0.0),
bbox_h: r.payload.get("bbox_h")
.and_then(|v| v.as_f64()).unwrap_or(0.0),
confidence: r.payload.get("confidence")
.and_then(|v| v.as_f64()).unwrap_or(0.0),
yaw: r.payload.get("yaw")
.and_then(|v| v.as_f64()).unwrap_or(0.0),
pitch: r.payload.get("pitch")
.and_then(|v| v.as_f64()).unwrap_or(0.0),
roll: r.payload.get("roll")
.and_then(|v| v.as_f64()).unwrap_or(0.0),
file_uuid: r
.payload
.get("file_uuid")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string(),
trace_id: r
.payload
.get("trace_id")
.and_then(|v| v.as_i64())
.unwrap_or(0) as i32,
frame: r.payload.get("frame").and_then(|v| v.as_i64()).unwrap_or(0),
bbox_x: r
.payload
.get("bbox_x")
.and_then(|v| v.as_f64())
.unwrap_or(0.0),
bbox_y: r
.payload
.get("bbox_y")
.and_then(|v| v.as_f64())
.unwrap_or(0.0),
bbox_w: r
.payload
.get("bbox_w")
.and_then(|v| v.as_f64())
.unwrap_or(0.0),
bbox_h: r
.payload
.get("bbox_h")
.and_then(|v| v.as_f64())
.unwrap_or(0.0),
confidence: r
.payload
.get("confidence")
.and_then(|v| v.as_f64())
.unwrap_or(0.0),
yaw: r.payload.get("yaw").and_then(|v| v.as_f64()).unwrap_or(0.0),
pitch: r
.payload
.get("pitch")
.and_then(|v| v.as_f64())
.unwrap_or(0.0),
roll: r
.payload
.get("roll")
.and_then(|v| v.as_f64())
.unwrap_or(0.0),
};
(id, r.vector, payload)
})
@@ -485,4 +543,4 @@ impl Default for FaceEmbeddingDb {
fn default() -> Self {
Self::new()
}
}
}

97
src/core/processor/executor.rs
View File

@@ -8,6 +8,8 @@ use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::Command;
use tokio::time::{sleep, timeout};
use crate::core::config::{DATABASE_SCHEMA, OUTPUT_DIR, REDIS_KEY_PREFIX};
#[derive(Debug, Clone)]
pub struct RetryConfig {
pub max_attempts: u32,
@@ -292,6 +294,10 @@ impl PythonExecutor {
}
let mut cmd = Command::new(&self.python_path);
cmd.env("MOMENTRY_OUTPUT_DIR", &*OUTPUT_DIR);
cmd.env("DATABASE_SCHEMA", &*DATABASE_SCHEMA);
cmd.env("MOMENTRY_DB_SCHEMA", &*DATABASE_SCHEMA);
cmd.env("MOMENTRY_REDIS_PREFIX", &*REDIS_KEY_PREFIX);
cmd.arg(&script_path);
for arg in args {
@@ -302,11 +308,18 @@ impl PythonExecutor {
cmd.arg("--uuid").arg(u);
}
// Pass frame list for 8Hz sampling
// Pass frame list for 8Hz sampling (only if non-empty)
if let Some(frames) = frames {
let frames_str = Self::format_frames_arg(frames);
cmd.arg("--frames").arg(&frames_str);
tracing::info!("[{}] 8Hz sampling: {} frames", log_prefix, frames.len());
if !frames.is_empty() {
let frames_str = Self::format_frames_arg(frames);
cmd.arg("--frames").arg(&frames_str);
tracing::info!("[{}] 8Hz sampling: {} frames", log_prefix, frames.len());
} else {
tracing::info!(
"[{}] 8Hz sampling: 0 frames (skipping --frames arg)",
log_prefix
);
}
}
cmd.stdout(Stdio::piped());
@@ -419,6 +432,10 @@ impl PythonExecutor {
}
let mut cmd = Command::new(&self.python_path);
cmd.env("MOMENTRY_OUTPUT_DIR", &*OUTPUT_DIR);
cmd.env("DATABASE_SCHEMA", &*DATABASE_SCHEMA);
cmd.env("MOMENTRY_DB_SCHEMA", &*DATABASE_SCHEMA);
cmd.env("MOMENTRY_REDIS_PREFIX", &*REDIS_KEY_PREFIX);
cmd.arg(&script_path);
for arg in args {
@@ -593,6 +610,10 @@ impl PythonExecutor {
}
let mut cmd = Command::new(&self.python_path);
cmd.env("MOMENTRY_OUTPUT_DIR", &*OUTPUT_DIR);
cmd.env("DATABASE_SCHEMA", &*DATABASE_SCHEMA);
cmd.env("MOMENTRY_DB_SCHEMA", &*DATABASE_SCHEMA);
cmd.env("MOMENTRY_REDIS_PREFIX", &*REDIS_KEY_PREFIX);
cmd.arg(&script_path);
for arg in args {
@@ -603,11 +624,18 @@ impl PythonExecutor {
cmd.arg("--uuid").arg(u);
}
// Pass frame list for 8Hz sampling
// Pass frame list for 8Hz sampling (only if non-empty)
if let Some(frames) = frames {
let frames_str = Self::format_frames_arg(frames);
cmd.arg("--frames").arg(&frames_str);
tracing::info!("[{}] 8Hz sampling: {} frames", log_prefix, frames.len());
if !frames.is_empty() {
let frames_str = Self::format_frames_arg(frames);
cmd.arg("--frames").arg(&frames_str);
tracing::info!("[{}] 8Hz sampling: {} frames", log_prefix, frames.len());
} else {
tracing::info!(
"[{}] 8Hz sampling: 0 frames (skipping --frames arg)",
log_prefix
);
}
}
cmd.stdout(Stdio::piped());
@@ -826,6 +854,59 @@ impl Default for PythonExecutor {
#[cfg(test)]
mod tests {
use super::*;
use std::process::Stdio;
#[tokio::test]
async fn test_executor_passes_env_vars() {
let executor = PythonExecutor::new().unwrap();
let mut cmd = Command::new(&executor.python_path);
cmd.env("MOMENTRY_OUTPUT_DIR", &*OUTPUT_DIR);
cmd.env("DATABASE_SCHEMA", &*DATABASE_SCHEMA);
cmd.env("MOMENTRY_DB_SCHEMA", &*DATABASE_SCHEMA);
cmd.env("MOMENTRY_REDIS_PREFIX", &*REDIS_KEY_PREFIX);
cmd.args([
"-c",
"import os; print(f'ENV_DATABASE_SCHEMA={os.environ.get(\"DATABASE_SCHEMA\",\"\")}'); print(f'ENV_MOMENTRY_DB_SCHEMA={os.environ.get(\"MOMENTRY_DB_SCHEMA\",\"\")}'); print(f'ENV_MOMENTRY_OUTPUT_DIR={os.environ.get(\"MOMENTRY_OUTPUT_DIR\",\"\")}'); print(f'ENV_MOMENTRY_REDIS_PREFIX={os.environ.get(\"MOMENTRY_REDIS_PREFIX\",\"\")}');",
]);
cmd.stdout(Stdio::piped());
cmd.stderr(Stdio::piped());
let output = cmd.output().await.expect("Failed to run inline Python");
let stdout = String::from_utf8_lossy(&output.stdout);
let stderr = String::from_utf8_lossy(&output.stderr);
println!("stdout: {}", stdout);
if !stderr.is_empty() {
println!("stderr: {}", stderr);
}
assert!(
output.status.success(),
"Python inline script failed: {}",
stderr
);
assert!(
stdout.contains(&format!("ENV_DATABASE_SCHEMA={}", *DATABASE_SCHEMA)),
"DATABASE_SCHEMA mismatch:\n{}",
stdout
);
assert!(
stdout.contains(&format!("ENV_MOMENTRY_DB_SCHEMA={}", *DATABASE_SCHEMA)),
"MOMENTRY_DB_SCHEMA mismatch:\n{}",
stdout
);
assert!(
stdout.contains(&format!("ENV_MOMENTRY_OUTPUT_DIR={}", *OUTPUT_DIR)),
"MOMENTRY_OUTPUT_DIR mismatch:\n{}",
stdout
);
assert!(
stdout.contains(&format!("ENV_MOMENTRY_REDIS_PREFIX={}", *REDIS_KEY_PREFIX)),
"MOMENTRY_REDIS_PREFIX mismatch:\n{}",
stdout
);
}
#[test]
fn test_python_executor_new() {

354
src/core/processor/tkg.rs
View File

@@ -26,7 +26,7 @@ async fn populate_face_detections_from_face_json(
use tracing::info;
let fd_table = t("face_detections");
// Check if trace_id is already populated
let traced_count: i64 = sqlx::query_scalar(&format!(
"SELECT COUNT(*) FROM {} WHERE file_uuid = $1 AND trace_id IS NOT NULL",
@@ -37,7 +37,10 @@ async fn populate_face_detections_from_face_json(
.await?;
if traced_count > 0 {
info!("[TKG-Phase0] face_detections already traced for {} ({} rows with trace_id)", file_uuid, traced_count);
info!(
"[TKG-Phase0] face_detections already traced for {} ({} rows with trace_id)",
file_uuid, traced_count
);
return Ok(());
}
@@ -50,11 +53,17 @@ async fn populate_face_detections_from_face_json(
.await?;
if total_count == 0 {
info!("[TKG-Phase0] No face_detections for {}, need face processor first", file_uuid);
info!(
"[TKG-Phase0] No face_detections for {}, need face processor first",
file_uuid
);
return Ok(());
}
info!("[TKG-Phase0] {} faces exist but trace_id=NULL, calling store_traced_faces.py...", total_count);
info!(
"[TKG-Phase0] {} faces exist but trace_id=NULL, calling store_traced_faces.py...",
total_count
);
let executor = PythonExecutor::new()?;
@@ -77,11 +86,17 @@ async fn populate_face_detections_from_face_json(
.bind(file_uuid)
.fetch_one(pool)
.await?;
info!("[TKG-Phase0] Traced {} face_detections for {}", new_traced_count, file_uuid);
info!(
"[TKG-Phase0] Traced {} face_detections for {}",
new_traced_count, file_uuid
);
Ok(())
}
Err(e) => {
info!("[TKG-Phase0] Failed to trace face_detections: {} (continuing with TKG build)", e);
info!(
"[TKG-Phase0] Failed to trace face_detections: {} (continuing with TKG build)",
e
);
Ok(())
}
}
@@ -103,7 +118,11 @@ async fn populate_face_embeddings_to_qdrant(
// Check if embeddings already exist
let existing = face_db.get_all_embeddings_for_file(file_uuid).await?;
if !existing.is_empty() {
info!("[TKG-Phase1] {} embeddings already in Qdrant for {}", existing.len(), file_uuid);
info!(
"[TKG-Phase1] {} embeddings already in Qdrant for {}",
existing.len(),
file_uuid
);
return Ok(existing.len());
}
@@ -129,8 +148,8 @@ async fn populate_face_embeddings_to_qdrant(
let mut points: Vec<(String, Vec<f32>, FaceEmbeddingPayload)> = Vec::new();
for (trace_id, frame, x, y, w, h, confidence, embedding) in &rows {
if let Some(emb) = embedding {
let (yaw, pitch, roll) = get_pose_for_face(*frame, *x, *y, *w, *h, &pose_data)
.unwrap_or((0.0, 0.0, 0.0));
let (yaw, pitch, roll) =
get_pose_for_face(*frame, *x, *y, *w, *h, &pose_data).unwrap_or((0.0, 0.0, 0.0));
// Generate unique numeric point ID (trace_id * 100000 + frame)
let point_id = format!("{}", (*trace_id as u64) * 100000 + (*frame as u64));
@@ -152,7 +171,10 @@ async fn populate_face_embeddings_to_qdrant(
}
let count = face_db.batch_upsert(points).await?;
info!("[TKG-Phase1] Stored {} face embeddings in Qdrant for {}", count, file_uuid);
info!(
"[TKG-Phase1] Stored {} face embeddings in Qdrant for {}",
count, file_uuid
);
Ok(count)
}
@@ -461,10 +483,10 @@ struct FaceDetectionRow {
// ── Public API ────────────────────────────────────────────────────
pub struct TkgResult {
pub face_trace_nodes: usize,
pub gaze_trace_nodes: usize,
pub lip_trace_nodes: usize,
pub text_trace_nodes: usize,
pub face_track_nodes: usize,
pub gaze_track_nodes: usize,
pub lip_track_nodes: usize,
pub text_region_nodes: usize,
pub appearance_trace_nodes: usize,
pub skin_tone_trace_nodes: usize,
pub accessory_nodes: usize,
@@ -486,28 +508,36 @@ pub async fn build_tkg(db: &PostgresDb, file_uuid: &str, output_dir: &str) -> Re
// Phase 0: Populate face_detections from face.json (if not exists)
if let Err(e) = populate_face_detections_from_face_json(pool, output_dir, file_uuid).await {
tracing::warn!("[TKG-Phase0] populate_face_detections failed: {} (continuing)", e);
tracing::warn!(
"[TKG-Phase0] populate_face_detections failed: {} (continuing)",
e
);
}
// Phase 1: Populate face embeddings to Qdrant (for TKG-only migration)
if let Err(e) = populate_face_embeddings_to_qdrant(pool, output_dir, file_uuid).await {
tracing::warn!("[TKG-Phase1] populate_face_embeddings failed: {} (continuing)", e);
tracing::warn!(
"[TKG-Phase1] populate_face_embeddings failed: {} (continuing)",
e
);
}
let pose_data = load_face_pose_data(output_dir, file_uuid).map_err(|e| {
tracing::error!("[TKG] Failed to load face pose data: {}", e);
e
}).unwrap_or_default();
let pose_data = load_face_pose_data(output_dir, file_uuid)
.map_err(|e| {
tracing::error!("[TKG] Failed to load face pose data: {}", e);
e
})
.unwrap_or_default();
tracing::info!(
"[TKG] Loaded {} pose entries from face.json (output_dir={})",
pose_data.len(),
output_dir
);
let n_face = build_face_trace_nodes(pool, file_uuid, &pose_data).await?;
let n_gaze = build_gaze_trace_nodes(pool, file_uuid, &pose_data).await?;
let n_lip = build_lip_trace_nodes(pool, file_uuid, output_dir, &pose_data).await?;
let n_text = build_text_trace_nodes(pool, file_uuid).await?;
let n_face = build_face_track_nodes(pool, file_uuid, &pose_data).await?;
let n_gaze = build_gaze_track_nodes(pool, file_uuid, &pose_data).await?;
let n_lip = build_lip_track_nodes(pool, file_uuid, output_dir, &pose_data).await?;
let n_text = build_text_region_nodes(pool, file_uuid).await?;
let n_appearance =
build_appearance_trace_nodes(pool, file_uuid, output_dir, &pose_data).await?;
let n_skin = build_skin_tone_trace_nodes(pool, file_uuid, output_dir, &pose_data).await?;
@@ -524,10 +554,10 @@ pub async fn build_tkg(db: &PostgresDb, file_uuid: &str, output_dir: &str) -> Re
let e_w = build_wears_edges(pool, file_uuid).await?;
Ok(TkgResult {
face_trace_nodes: n_face,
gaze_trace_nodes: n_gaze,
lip_trace_nodes: n_lip,
text_trace_nodes: n_text,
face_track_nodes: n_face,
gaze_track_nodes: n_gaze,
lip_track_nodes: n_lip,
text_region_nodes: n_text,
appearance_trace_nodes: n_appearance,
skin_tone_trace_nodes: n_skin,
accessory_nodes: n_accessories,
@@ -545,7 +575,7 @@ pub async fn build_tkg(db: &PostgresDb, file_uuid: &str, output_dir: &str) -> Re
// ── Node builders ─────────────────────────────────────────────────
async fn build_face_trace_nodes(
async fn build_face_track_nodes(
pool: &PgPool,
file_uuid: &str,
pose_data: &[FacePose],
@@ -557,20 +587,28 @@ async fn build_face_trace_nodes(
let qdrant_embeddings = face_db.get_all_embeddings_for_file(file_uuid).await?;
if !qdrant_embeddings.is_empty() {
tracing::info!("[TKG-Phase2] Building face_trace nodes from Qdrant ({} embeddings)", qdrant_embeddings.len());
return build_face_trace_nodes_from_qdrant(pool, file_uuid, pose_data, qdrant_embeddings).await;
tracing::info!(
"[TKG-Phase2] Building face_track nodes from Qdrant ({} embeddings)",
qdrant_embeddings.len()
);
return build_face_track_nodes_from_qdrant(pool, file_uuid, pose_data, qdrant_embeddings)
.await;
}
// Fallback to PostgreSQL
tracing::info!("[TKG-Phase2] No Qdrant embeddings, falling back to PostgreSQL");
build_face_trace_nodes_from_pg(pool, file_uuid, pose_data).await
build_face_track_nodes_from_pg(pool, file_uuid, pose_data).await
}
async fn build_face_trace_nodes_from_qdrant(
async fn build_face_track_nodes_from_qdrant(
pool: &PgPool,
file_uuid: &str,
pose_data: &[FacePose],
qdrant_embeddings: Vec<(String, Vec<f32>, crate::core::db::face_embedding_db::FaceEmbeddingPayload)>,
qdrant_embeddings: Vec<(
String,
Vec<f32>,
crate::core::db::face_embedding_db::FaceEmbeddingPayload,
)>,
) -> Result<usize> {
use crate::core::db::face_embedding_db::FaceEmbeddingPayload;
let nodes_table = t("tkg_nodes");
@@ -598,7 +636,7 @@ async fn build_face_trace_nodes_from_qdrant(
// Build aggregates
let mut count = 0;
for (tid, frames) in &trace_frames {
let external_id = format!("trace_{}", tid);
let external_id = format!("face_track_{}", tid);
let label = format!("Face Trace {}", tid);
let frame_count = frames.len() as i64;
@@ -625,7 +663,11 @@ async fn build_face_trace_nodes_from_qdrant(
}
let (avg_yaw, avg_pitch, avg_roll) = if pose_count > 0 {
(yaw_sum / pose_count as f64, pitch_sum / pose_count as f64, roll_sum / pose_count as f64)
(
yaw_sum / pose_count as f64,
pitch_sum / pose_count as f64,
roll_sum / pose_count as f64,
)
} else {
(0.0, 0.0, 0.0)
};
@@ -653,7 +695,7 @@ async fn build_face_trace_nodes_from_qdrant(
nodes_table
))
.bind(file_uuid)
.bind("face_trace")
.bind("face_track")
.bind(&external_id)
.bind(&label)
.bind(serde_json::to_string(&props)?)
@@ -663,11 +705,11 @@ async fn build_face_trace_nodes_from_qdrant(
count += 1;
}
tracing::info!("[TKG-Phase2] Built {} face_trace nodes from Qdrant", count);
tracing::info!("[TKG-Phase2] Built {} face_track nodes from Qdrant", count);
Ok(count)
}
async fn build_face_trace_nodes_from_pg(
async fn build_face_track_nodes_from_pg(
pool: &PgPool,
file_uuid: &str,
pose_data: &[FacePose],
@@ -720,7 +762,7 @@ async fn build_face_trace_nodes_from_pg(
let mut count = 0;
for row in &rows {
let tid = row.trace_id;
let external_id = format!("trace_{}", tid);
let external_id = format!("face_track_{}", tid);
let label = format!("Face Trace {}", tid);
// Compute average pose for this trace
@@ -779,7 +821,7 @@ async fn build_face_trace_nodes_from_pg(
"#,
nodes_table
))
.bind("face_trace")
.bind("face_track")
.bind(&external_id)
.bind(file_uuid)
.bind(&label)
@@ -944,7 +986,13 @@ async fn build_co_occurrence_edges(
"[TKG-Phase2.6.1] Building co_occurrence edges from Qdrant ({} embeddings)",
qdrant_embeddings.len()
);
return build_co_occurrence_edges_from_qdrant(pool, file_uuid, output_dir, qdrant_embeddings).await;
return build_co_occurrence_edges_from_qdrant(
pool,
file_uuid,
output_dir,
qdrant_embeddings,
)
.await;
}
tracing::info!("[TKG-Phase2.6.1] No Qdrant embeddings, falling back to PostgreSQL");
@@ -955,7 +1003,11 @@ async fn build_co_occurrence_edges_from_qdrant(
pool: &PgPool,
file_uuid: &str,
output_dir: &str,
qdrant_embeddings: Vec<(String, Vec<f32>, crate::core::db::face_embedding_db::FaceEmbeddingPayload)>,
qdrant_embeddings: Vec<(
String,
Vec<f32>,
crate::core::db::face_embedding_db::FaceEmbeddingPayload,
)>,
) -> Result<usize> {
use crate::core::db::face_embedding_db::FaceEmbeddingPayload;
@@ -974,10 +1026,13 @@ async fn build_co_occurrence_edges_from_qdrant(
for (_, _, payload) in &qdrant_embeddings {
let frame = payload.frame;
let trace_id = payload.trace_id as i64;
frame_faces
.entry(frame)
.or_default()
.push((trace_id, payload.bbox_x, payload.bbox_y, payload.bbox_w, payload.bbox_h));
frame_faces.entry(frame).or_default().push((
trace_id,
payload.bbox_x,
payload.bbox_y,
payload.bbox_w,
payload.bbox_h,
));
}
let mut edge_count = 0;
@@ -999,9 +1054,9 @@ async fn build_co_occurrence_edges_from_qdrant(
}
for (trace_id, _, _, _, _) in faces {
let external_id = format!("trace_{}", trace_id);
let external_id = format!("face_track_{}", trace_id);
let face_node: Option<(i64,)> = sqlx::query_as(&format!(
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='face_trace' AND external_id=$2",
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='face_track' AND external_id=$2",
nodes_table
))
.bind(file_uuid)
@@ -1113,9 +1168,9 @@ async fn build_co_occurrence_edges_from_pg(
continue;
}
let external_id = format!("trace_{}", face.trace_id);
let external_id = format!("face_track_{}", face.trace_id);
let face_node: Option<(i64,)> = sqlx::query_as(&format!(
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='face_trace' AND external_id=$2",
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='face_track' AND external_id=$2",
nodes_table
))
.bind(file_uuid)
@@ -1196,7 +1251,13 @@ async fn build_speaker_face_edges(
"[TKG-Phase2.6.3] Building speaker_face edges from Qdrant ({} embeddings)",
qdrant_embeddings.len()
);
return build_speaker_face_edges_from_qdrant(pool, file_uuid, output_dir, qdrant_embeddings).await;
return build_speaker_face_edges_from_qdrant(
pool,
file_uuid,
output_dir,
qdrant_embeddings,
)
.await;
}
tracing::info!("[TKG-Phase2.6.3] No Qdrant embeddings, falling back to PostgreSQL");
@@ -1207,7 +1268,11 @@ async fn build_speaker_face_edges_from_qdrant(
pool: &PgPool,
file_uuid: &str,
output_dir: &str,
qdrant_embeddings: Vec<(String, Vec<f32>, crate::core::db::face_embedding_db::FaceEmbeddingPayload)>,
qdrant_embeddings: Vec<(
String,
Vec<f32>,
crate::core::db::face_embedding_db::FaceEmbeddingPayload,
)>,
) -> Result<usize> {
use crate::core::db::face_embedding_db::FaceEmbeddingPayload;
@@ -1245,9 +1310,9 @@ async fn build_speaker_face_edges_from_qdrant(
let mut edge_count = 0;
for (tid, (sf, ef)) in &trace_ranges {
let face_ext_id = format!("trace_{}", tid);
let face_ext_id = format!("face_track_{}", tid);
let face_node: Option<(i64,)> = sqlx::query_as(&format!(
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='face_trace' AND external_id=$2",
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='face_track' AND external_id=$2",
nodes_table
))
.bind(file_uuid)
@@ -1370,9 +1435,9 @@ async fn build_speaker_face_edges_from_pg(
let mut edge_count = 0;
for (tid, sf, ef) in &traces {
let face_ext_id = format!("trace_{}", tid);
let face_ext_id = format!("face_track_{}", tid);
let face_node: Option<(i64,)> = sqlx::query_as(&format!(
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='face_trace' AND external_id=$2",
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='face_track' AND external_id=$2",
nodes_table
))
.bind(file_uuid)
@@ -1469,7 +1534,8 @@ async fn build_face_face_edges(
"[TKG-Phase2.6.2] Building face_face edges from Qdrant ({} embeddings)",
qdrant_embeddings.len()
);
return build_face_face_edges_from_qdrant(pool, file_uuid, pose_data, qdrant_embeddings).await;
return build_face_face_edges_from_qdrant(pool, file_uuid, pose_data, qdrant_embeddings)
.await;
}
tracing::info!("[TKG-Phase2.6.2] No Qdrant embeddings, falling back to PostgreSQL");
@@ -1480,7 +1546,11 @@ async fn build_face_face_edges_from_qdrant(
pool: &PgPool,
file_uuid: &str,
pose_data: &[FacePose],
qdrant_embeddings: Vec<(String, Vec<f32>, crate::core::db::face_embedding_db::FaceEmbeddingPayload)>,
qdrant_embeddings: Vec<(
String,
Vec<f32>,
crate::core::db::face_embedding_db::FaceEmbeddingPayload,
)>,
) -> Result<usize> {
use crate::core::db::face_embedding_db::FaceEmbeddingPayload;
@@ -1489,20 +1559,31 @@ async fn build_face_face_edges_from_qdrant(
let mut frame_faces: HashMap<i64, Vec<FaceEmbeddingPayload>> = HashMap::new();
for (_, _, payload) in &qdrant_embeddings {
frame_faces.entry(payload.frame).or_default().push(payload.clone());
frame_faces
.entry(payload.frame)
.or_default()
.push(payload.clone());
}
let mut frame_map: HashMap<(i64, i64), (f64, f64, f64, f64)> = HashMap::new();
for (_, _, payload) in &qdrant_embeddings {
let trace_id = payload.trace_id as i64;
let frame = payload.frame;
frame_map.insert((trace_id, frame), (payload.bbox_x, payload.bbox_y, payload.bbox_w, payload.bbox_h));
frame_map.insert(
(trace_id, frame),
(
payload.bbox_x,
payload.bbox_y,
payload.bbox_w,
payload.bbox_h,
),
);
}
let mut rows: Vec<(i64, i64, i64)> = Vec::new();
for (frame, faces) in frame_faces.iter() {
for i in 0..faces.len() {
for j in (i+1)..faces.len() {
for j in (i + 1)..faces.len() {
let tid_a = faces[i].trace_id as i64;
let tid_b = faces[j].trace_id as i64;
let min_tid = tid_a.min(tid_b);
@@ -1536,14 +1617,14 @@ async fn build_face_face_edges_from_qdrant(
let mut edge_count = 0;
let mut node_id_cache: HashMap<i64, i64> = HashMap::new();
for ((tid_a, tid_b), frame_data) in &pair_frames {
let ext_a = format!("trace_{}", tid_a);
let ext_b = format!("trace_{}", tid_b);
let ext_a = format!("face_track_{}", tid_a);
let ext_b = format!("face_track_{}", tid_b);
let n_a_id = match node_id_cache.get(tid_a) {
Some(id) => *id,
None => {
if let Some((id,)) = sqlx::query_as::<_, (i64,)>(&format!(
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='face_trace' AND external_id=$2",
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='face_track' AND external_id=$2",
nodes_table
))
.bind(file_uuid).bind(&ext_a).fetch_optional(pool).await?
@@ -1558,7 +1639,7 @@ async fn build_face_face_edges_from_qdrant(
Some(id) => *id,
None => {
if let Some((id,)) = sqlx::query_as::<_, (i64,)>(&format!(
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='face_trace' AND external_id=$2",
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='face_track' AND external_id=$2",
nodes_table
))
.bind(file_uuid).bind(&ext_b).fetch_optional(pool).await?
@@ -1711,14 +1792,14 @@ async fn build_face_face_edges_from_pg(
let mut edge_count = 0;
let mut node_id_cache: HashMap<i64, i64> = HashMap::new();
for ((tid_a, tid_b), frame_data) in &pair_frames {
let ext_a = format!("trace_{}", tid_a);
let ext_b = format!("trace_{}", tid_b);
let ext_a = format!("face_track_{}", tid_a);
let ext_b = format!("face_track_{}", tid_b);
let n_a_id = match node_id_cache.get(tid_a) {
Some(id) => *id,
None => {
if let Some((id,)) = sqlx::query_as::<_, (i64,)>(&format!(
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='face_trace' AND external_id=$2",
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='face_track' AND external_id=$2",
nodes_table
))
.bind(file_uuid).bind(&ext_a).fetch_optional(pool).await?
@@ -1733,7 +1814,7 @@ async fn build_face_face_edges_from_pg(
Some(id) => *id,
None => {
if let Some((id,)) = sqlx::query_as::<_, (i64,)>(&format!(
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='face_trace' AND external_id=$2",
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='face_track' AND external_id=$2",
nodes_table
))
.bind(file_uuid).bind(&ext_b).fetch_optional(pool).await?
@@ -1820,7 +1901,7 @@ async fn build_face_face_edges_from_pg(
// ── Gaze Trace Nodes ──────────────────────────────────────────────
async fn build_gaze_trace_nodes(
async fn build_gaze_track_nodes(
pool: &PgPool,
file_uuid: &str,
pose_data: &[FacePose],
@@ -1832,19 +1913,27 @@ async fn build_gaze_trace_nodes(
let qdrant_embeddings = face_db.get_all_embeddings_for_file(file_uuid).await?;
if !qdrant_embeddings.is_empty() {
tracing::info!("[TKG-Phase2.5] Building gaze_trace nodes from Qdrant ({} embeddings)", qdrant_embeddings.len());
return build_gaze_trace_nodes_from_qdrant(pool, file_uuid, pose_data, qdrant_embeddings).await;
tracing::info!(
"[TKG-Phase2.5] Building gaze_track nodes from Qdrant ({} embeddings)",
qdrant_embeddings.len()
);
return build_gaze_track_nodes_from_qdrant(pool, file_uuid, pose_data, qdrant_embeddings)
.await;
}
tracing::info!("[TKG-Phase2.5] No Qdrant embeddings, falling back to PostgreSQL");
build_gaze_trace_nodes_from_pg(pool, file_uuid, pose_data).await
build_gaze_track_nodes_from_pg(pool, file_uuid, pose_data).await
}
async fn build_gaze_trace_nodes_from_qdrant(
async fn build_gaze_track_nodes_from_qdrant(
pool: &PgPool,
file_uuid: &str,
pose_data: &[FacePose],
qdrant_embeddings: Vec<(String, Vec<f32>, crate::core::db::face_embedding_db::FaceEmbeddingPayload)>,
qdrant_embeddings: Vec<(
String,
Vec<f32>,
crate::core::db::face_embedding_db::FaceEmbeddingPayload,
)>,
) -> Result<usize> {
use crate::core::db::face_embedding_db::FaceEmbeddingPayload;
let nodes_table = t("tkg_nodes");
@@ -1873,11 +1962,11 @@ async fn build_gaze_trace_nodes_from_qdrant(
for (tid, frames) in &trace_frames {
let external_id = format!("gaze_{}", tid);
// Phase 2.7: Query face_trace identity_id
let face_ext_id = format!("trace_{}", tid);
// Phase 2.7: Query face_track identity_id
let face_ext_id = format!("face_track_{}", tid);
let face_identity_id: Option<i64> = sqlx::query_scalar(&format!(
"SELECT (properties->>'identity_id')::bigint FROM {}
WHERE file_uuid=$1 AND node_type='face_trace' AND external_id=$2",
WHERE file_uuid=$1 AND node_type='face_track' AND external_id=$2",
nodes_table
))
.bind(file_uuid)
@@ -1969,7 +2058,7 @@ async fn build_gaze_trace_nodes_from_qdrant(
"#,
nodes_table
))
.bind("gaze_trace")
.bind("gaze_track")
.bind(&external_id)
.bind(file_uuid)
.bind(&external_id)
@@ -1980,11 +2069,14 @@ async fn build_gaze_trace_nodes_from_qdrant(
count += 1;
}
tracing::info!("[TKG-Phase2.5] Built {} gaze_trace nodes from Qdrant", count);
tracing::info!(
"[TKG-Phase2.5] Built {} gaze_track nodes from Qdrant",
count
);
Ok(count)
}
async fn build_gaze_trace_nodes_from_pg(
async fn build_gaze_track_nodes_from_pg(
pool: &PgPool,
file_uuid: &str,
pose_data: &[FacePose],
@@ -2103,7 +2195,7 @@ async fn build_gaze_trace_nodes_from_pg(
"#,
nodes_table
))
.bind("gaze_trace")
.bind("gaze_track")
.bind(&external_id)
.bind(file_uuid)
.bind(&format!("Gaze Trace {}", tid))
@@ -2203,15 +2295,15 @@ async fn build_mutual_gaze_edges(
let mut node_id_cache: HashMap<i64, i64> = HashMap::new();
for ((tid_a, tid_b), frames) in &pair_gaze_frames {
let ext_a = format!("trace_{}", tid_a);
let ext_b = format!("trace_{}", tid_b);
let ext_a = format!("face_track_{}", tid_a);
let ext_b = format!("face_track_{}", tid_b);
// Get node IDs
let n_a_id = match node_id_cache.get(tid_a) {
Some(id) => *id,
None => {
if let Some((id,)) = sqlx::query_as::<_, (i64,)>(&format!(
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='face_trace' AND external_id=$2",
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='face_track' AND external_id=$2",
nodes_table
))
.bind(file_uuid).bind(&ext_a).fetch_optional(pool).await?
@@ -2226,7 +2318,7 @@ async fn build_mutual_gaze_edges(
Some(id) => *id,
None => {
if let Some((id,)) = sqlx::query_as::<_, (i64,)>(&format!(
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='face_trace' AND external_id=$2",
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='face_track' AND external_id=$2",
nodes_table
))
.bind(file_uuid).bind(&ext_b).fetch_optional(pool).await?
@@ -2284,7 +2376,7 @@ async fn build_mutual_gaze_edges(
// ── Lip Trace Nodes ───────────────────────────────────────────────
async fn build_lip_trace_nodes(
async fn build_lip_track_nodes(
pool: &PgPool,
file_uuid: &str,
output_dir: &str,
@@ -2297,20 +2389,31 @@ async fn build_lip_trace_nodes(
let qdrant_embeddings = face_db.get_all_embeddings_for_file(file_uuid).await?;
if !qdrant_embeddings.is_empty() {
tracing::info!("[TKG-Phase2.5] Building lip_trace nodes from Qdrant + face.json");
return build_lip_trace_nodes_from_qdrant(pool, file_uuid, output_dir, pose_data, qdrant_embeddings).await;
tracing::info!("[TKG-Phase2.5] Building lip_track nodes from Qdrant + face.json");
return build_lip_track_nodes_from_qdrant(
pool,
file_uuid,
output_dir,
pose_data,
qdrant_embeddings,
)
.await;
}
tracing::info!("[TKG-Phase2.5] No Qdrant embeddings, falling back to PostgreSQL");
build_lip_trace_nodes_from_pg(pool, file_uuid, output_dir, pose_data).await
build_lip_track_nodes_from_pg(pool, file_uuid, output_dir, pose_data).await
}
async fn build_lip_trace_nodes_from_qdrant(
async fn build_lip_track_nodes_from_qdrant(
pool: &PgPool,
file_uuid: &str,
output_dir: &str,
pose_data: &[FacePose],
qdrant_embeddings: Vec<(String, Vec<f32>, crate::core::db::face_embedding_db::FaceEmbeddingPayload)>,
qdrant_embeddings: Vec<(
String,
Vec<f32>,
crate::core::db::face_embedding_db::FaceEmbeddingPayload,
)>,
) -> Result<usize> {
use crate::core::db::face_embedding_db::FaceEmbeddingPayload;
let nodes_table = t("tkg_nodes");
@@ -2328,16 +2431,13 @@ async fn build_lip_trace_nodes_from_qdrant(
// Build trace_id mapping from Qdrant: frame → Vec<(trace_id, bbox)>
let mut frame_trace_map: HashMap<i64, Vec<(i64, f64, f64, f64, f64)>> = HashMap::new();
for (_, _, payload) in &qdrant_embeddings {
frame_trace_map
.entry(payload.frame)
.or_default()
.push((
payload.trace_id as i64,
payload.bbox_x,
payload.bbox_y,
payload.bbox_w,
payload.bbox_h,
));
frame_trace_map.entry(payload.frame).or_default().push((
payload.trace_id as i64,
payload.bbox_x,
payload.bbox_y,
payload.bbox_w,
payload.bbox_h,
));
}
// Helper function to match trace_id by bbox distance
@@ -2411,11 +2511,11 @@ async fn build_lip_trace_nodes_from_qdrant(
for (tid, frames) in &lip_data {
let external_id = format!("lip_{}", tid);
// Phase 2.7: Query face_trace identity_id
let face_ext_id = format!("trace_{}", tid);
// Phase 2.7: Query face_track identity_id
let face_ext_id = format!("face_track_{}", tid);
let face_identity_id: Option<i64> = sqlx::query_scalar(&format!(
"SELECT (properties->>'identity_id')::bigint FROM {}
WHERE file_uuid=$1 AND node_type='face_trace' AND external_id=$2",
WHERE file_uuid=$1 AND node_type='face_track' AND external_id=$2",
nodes_table
))
.bind(file_uuid)
@@ -2500,7 +2600,7 @@ async fn build_lip_trace_nodes_from_qdrant(
"#,
nodes_table
))
.bind("lip_trace")
.bind("lip_track")
.bind(&external_id)
.bind(file_uuid)
.bind(&format!("Lip Trace {}", tid))
@@ -2511,11 +2611,11 @@ async fn build_lip_trace_nodes_from_qdrant(
count += 1;
}
tracing::info!("[TKG-Phase2.5] Built {} lip_trace nodes from Qdrant", count);
tracing::info!("[TKG-Phase2.5] Built {} lip_track nodes from Qdrant", count);
Ok(count)
}
async fn build_lip_trace_nodes_from_pg(
async fn build_lip_track_nodes_from_pg(
pool: &PgPool,
file_uuid: &str,
output_dir: &str,
@@ -2658,7 +2758,7 @@ async fn build_lip_trace_nodes_from_pg(
"#,
nodes_table
))
.bind("lip_trace")
.bind("lip_track")
.bind(&external_id)
.bind(file_uuid)
.bind(&format!("Lip Trace {}", tid))
@@ -2750,7 +2850,7 @@ async fn get_trace_for_face(
// ── Text/Sentence Trace Nodes ─────────────────────────────────────
async fn build_text_trace_nodes(pool: &PgPool, file_uuid: &str) -> Result<usize> {
async fn build_text_region_nodes(pool: &PgPool, file_uuid: &str) -> Result<usize> {
let chunk_table = t("chunk");
let nodes_table = t("tkg_nodes");
@@ -2827,14 +2927,14 @@ async fn build_lip_sync_edges(
let edges_table = t("tkg_edges");
// Get lip traces
let lip_traces: Vec<(i64, String, i64, i64, i64, f64)> = sqlx::query_as(&format!(
let lip_tracks: Vec<(i64, String, i64, i64, i64, f64)> = sqlx::query_as(&format!(
r#"
SELECT id::bigint, external_id,
(properties->>'start_frame')::bigint,
(properties->>'end_frame')::bigint,
(properties->>'speaking_frames')::bigint,
(properties->>'avg_openness')::float8
FROM {} WHERE file_uuid = $1 AND node_type = 'lip_trace'
FROM {} WHERE file_uuid = $1 AND node_type = 'lip_track'
"#,
nodes_table
))
@@ -2843,13 +2943,13 @@ async fn build_lip_sync_edges(
.await?;
// Get text traces
let text_traces: Vec<(i64, String, i64, i64, Option<String>)> = sqlx::query_as(&format!(
let text_regions: Vec<(i64, String, i64, i64, Option<String>)> = sqlx::query_as(&format!(
r#"
SELECT id::bigint, external_id,
(properties->>'start_frame')::bigint,
(properties->>'end_frame')::bigint,
properties->>'speaker_id'
FROM {} WHERE file_uuid = $1 AND node_type = 'text_trace'
FROM {} WHERE file_uuid = $1 AND node_type = 'text_region'
"#,
nodes_table
))
@@ -2860,8 +2960,8 @@ async fn build_lip_sync_edges(
let mut edge_count = 0;
let mut node_id_cache: HashMap<String, i64> = HashMap::new();
for (lip_id, lip_ext, lip_start, lip_end, lip_speaking, lip_openness) in &lip_traces {
for (text_id, text_ext, text_start, text_end, speaker_id) in &text_traces {
for (lip_id, lip_ext, lip_start, lip_end, lip_speaking, lip_openness) in &lip_tracks {
for (text_id, text_ext, text_start, text_end, speaker_id) in &text_regions {
// Check time overlap
let overlap_start = lip_start.max(text_start);
let overlap_end = lip_end.min(text_end);
@@ -2887,7 +2987,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='lip_trace' AND external_id=$2",
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='lip_track' AND external_id=$2",
nodes_table
))
.bind(file_uuid).bind(lip_ext).fetch_optional(pool).await?
@@ -2902,7 +3002,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_trace' AND external_id=$2",
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='text_region' AND external_id=$2",
nodes_table
))
.bind(file_uuid).bind(text_ext).fetch_optional(pool).await?
@@ -3245,7 +3345,7 @@ async fn build_has_appearance_edges(pool: &PgPool, file_uuid: &str) -> Result<us
let nodes_table = t("tkg_nodes");
let edges_table = t("tkg_edges");
// Match appearance_trace to face_trace via trace_id
// Match appearance_trace to face_track via trace_id
let appearance_traces: Vec<(i64, String, Option<i64>)> = sqlx::query_as(&format!(
r#"
SELECT id::bigint, external_id,
@@ -3263,14 +3363,14 @@ async fn build_has_appearance_edges(pool: &PgPool, file_uuid: &str) -> Result<us
for (app_id, app_ext, trace_id) in &appearance_traces {
if let Some(tid) = trace_id {
let face_ext = format!("trace_{}", tid);
let face_ext = format!("face_track_{}", tid);
// Get face trace node ID
let face_node_id = match node_id_cache.get(&face_ext) {
Some(id) => *id,
None => {
if let Some((id,)) = sqlx::query_as::<_, (i64,)>(&format!(
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='face_trace' AND external_id=$2",
"SELECT id FROM {} WHERE file_uuid=$1 AND node_type='face_track' AND external_id=$2",
nodes_table
))
.bind(file_uuid).bind(&face_ext).fetch_optional(pool).await?
@@ -3636,10 +3736,10 @@ mod tests {
#[test]
fn test_tkg_result() {
let r = TkgResult {
face_trace_nodes: 5,
gaze_trace_nodes: 5,
lip_trace_nodes: 4,
text_trace_nodes: 20,
face_track_nodes: 5,
gaze_track_nodes: 5,
lip_track_nodes: 4,
text_region_nodes: 20,
appearance_trace_nodes: 3,
skin_tone_trace_nodes: 5,
accessory_nodes: 0,
@@ -3653,7 +3753,7 @@ mod tests {
has_appearance_edges: 3,
wears_edges: 0,
};
assert_eq!(r.face_trace_nodes, 5);
assert_eq!(r.face_track_nodes, 5);
assert_eq!(r.object_nodes, 10);
assert_eq!(r.speaker_nodes, 3);
}

2
src/core/tmdb/face_agent.rs
View File

@@ -226,7 +226,7 @@ pub async fn match_faces_against_tmdb(db: &PostgresDb, file_uuid: &str) -> Resul
async fn quality_check_temporal_collisions(pool: &sqlx::PgPool, file_uuid: &str) -> Result<usize> {
let fd_table = schema::table_name("face_detections");
// Find all collision pairs: same identity, same frame, different trace
let collisions = sqlx::query_as::<_, (i32, i32, i32, i32)>(&format!(
let collisions = sqlx::query_as::<_, (i32, i32, i32, i64)>(&format!(
"SELECT a.identity_id, a.trace_id, b.trace_id, a.frame_number \
FROM {} a \
JOIN {} b \

1
src/processing/handlers.rs
View File

@@ -390,7 +390,6 @@ pub async fn handle_gitea(
Ok(())
}
/// Handle store-asrx command
pub async fn handle_store_asrx(uuid: &str) -> Result<()> {
let db = momentry_core::core::db::postgres_db::PostgresDb::new(

115
src/worker/job_worker.rs
View File

@@ -743,7 +743,9 @@ impl JobWorker {
continue;
}
ProcessorJobStatus::Failed => {
if result.retry_count >= 3 {
if result.retry_count >= 3
&& !crate::core::config::processor::FORCE_RETRY.clone()
{
info!(
"Processor {} failed {} times, max retries reached (3), skipping",
processor_type.as_str(),
@@ -752,11 +754,19 @@ impl JobWorker {
started_count += 1;
continue;
}
info!(
"Processor {} previously failed (retry {}/3), retrying",
if crate::core::config::processor::FORCE_RETRY.clone() {
info!(
"Processor {} previously failed (retry {}), FORCE_RETRY enabled, retrying",
processor_type.as_str(),
result.retry_count + 1
result.retry_count
);
} else {
info!(
"Processor {} previously failed (retry {}/3), retrying",
processor_type.as_str(),
result.retry_count + 1
);
}
let _ = sqlx::query(&format!(
"UPDATE {} SET retry_count = retry_count + 1 WHERE job_id = $1 AND processor = $2",
schema::table_name("processor_results")
@@ -988,17 +998,6 @@ impl JobWorker {
let chunk_t = schema::table_name("chunk");
let fd_t = schema::table_name("face_detections");
macro_rules! check {
($sql:expr) => {
sqlx::query_scalar::<_, i32>($sql)
.fetch_one(pool)
.await
.unwrap_or(0)
> 0
};
}
let fu = uuid;
// Only check conditions relevant to the job's processors
let has_asr_or_asrx =
job_processors.is_empty() || job_processors.iter().any(|p| p == "asrx" || p == "asr");
@@ -1006,21 +1005,57 @@ impl JobWorker {
let has_face = job_processors.is_empty() || job_processors.iter().any(|p| p == "face");
let rule1 = !has_asr_or_asrx
|| check!(&format!(
"SELECT 1 FROM {chunk_t} WHERE file_uuid = '{fu}' AND chunk_type = 'sentence' LIMIT 1"
));
|| sqlx::query_scalar::<_, i32>(&format!(
"SELECT 1 FROM {chunk_t} WHERE file_uuid = $1 AND chunk_type = 'sentence' LIMIT 1"
))
.bind(uuid)
.fetch_optional(pool)
.await
.unwrap_or(None)
.unwrap_or(0)
> 0;
let vector = !has_asr_or_asrx
|| check!(&format!("SELECT 1 FROM {chunk_t} WHERE file_uuid = '{fu}' AND chunk_type = 'sentence' AND embedding IS NOT NULL LIMIT 1"));
|| sqlx::query_scalar::<_, i32>(&format!(
"SELECT 1 FROM {chunk_t} WHERE file_uuid = $1 AND chunk_type = 'sentence' AND embedding IS NOT NULL LIMIT 1"
))
.bind(uuid)
.fetch_optional(pool)
.await
.unwrap_or(None)
.unwrap_or(0)
> 0;
let rule3 = !has_cut
|| check!(&format!(
"SELECT 1 FROM {chunk_t} WHERE file_uuid = '{fu}' AND chunk_type = 'cut' LIMIT 1"
));
|| sqlx::query_scalar::<_, i32>(&format!(
"SELECT 1 FROM {chunk_t} WHERE file_uuid = $1 AND chunk_type = 'cut' LIMIT 1"
))
.bind(uuid)
.fetch_optional(pool)
.await
.unwrap_or(None)
.unwrap_or(0)
> 0;
let trace = !has_face
|| check!(&format!("SELECT COUNT(DISTINCT trace_id) FROM {fd_t} WHERE file_uuid = '{fu}' AND trace_id IS NOT NULL"));
|| sqlx::query_scalar::<_, i64>(&format!(
"SELECT COUNT(DISTINCT trace_id) FROM {fd_t} WHERE file_uuid = $1 AND trace_id IS NOT NULL"
))
.bind(uuid)
.fetch_one(pool)
.await
.unwrap_or(0)
> 0;
let all_ok = rule1 && vector && rule3 && trace;
if !all_ok {
tracing::info!(
"[Ingestion] waiting (uuid={fu}): rule1={rule1} vector={vector} rule3={rule3} trace={trace}"
"[Ingestion] waiting (uuid={}): rule1={} vector={} rule3={} trace={}",
uuid,
rule1,
vector,
rule3,
trace
);
}
all_ok
@@ -1057,18 +1092,22 @@ impl JobWorker {
let all_completed = results
.iter()
.filter(|r| job_processors.contains(&r.processor_type.as_str().to_string()))
.all(|r| matches!(r.status, crate::core::db::ProcessorJobStatus::Completed));
let any_failed = results
.iter()
.filter(|r| job_processors.contains(&r.processor_type.as_str().to_string()))
.any(|r| matches!(r.status, crate::core::db::ProcessorJobStatus::Failed));
let any_pending = results
.iter()
.filter(|r| job_processors.contains(&r.processor_type.as_str().to_string()))
.any(|r| matches!(r.status, crate::core::db::ProcessorJobStatus::Pending));
let any_skipped = results
.iter()
.filter(|r| job_processors.contains(&r.processor_type.as_str().to_string()))
.any(|r| matches!(r.status, crate::core::db::ProcessorJobStatus::Skipped));
let completed_count = results
@@ -1101,7 +1140,9 @@ impl JobWorker {
.map(|r| r.processor_type.as_str().to_string())
.collect();
let has_asrx = completed_processors.iter().any(|p| p == "asrx");
let has_asr_or_asrx = completed_processors
.iter()
.any(|p| p == "asrx" || p == "asr");
let has_cut = completed_processors.iter().any(|p| p == "cut");
let has_face = completed_processors.iter().any(|p| p == "face");
let has_yolo = completed_processors.iter().any(|p| p == "yolo");
@@ -1110,7 +1151,7 @@ impl JobWorker {
.update_job_processors_arrays(job_id, completed_processors, failed_processors.clone())
.await?;
if has_asrx {
if has_asr_or_asrx {
// Guard: only spawn Rule 1 if sentence chunks don't exist yet
let chunk_t = schema::table_name("chunk");
let already_spawned: bool = sqlx::query_scalar::<_, i32>(&format!(
@@ -1321,7 +1362,7 @@ impl JobWorker {
}
// 🚀 P3 Trigger: Identity Agent (Face + ASRX)
if has_face && has_asrx {
if has_face && has_asr_or_asrx {
info!("📝 Prerequisites met for Identity Agent. Starting analysis...");
let db_clone = self.db.clone();
let uuid_clone = uuid.to_string();
@@ -1513,21 +1554,22 @@ impl JobWorker {
let pool = db.pool();
let chunk_table = schema::table_name("chunk");
let rows = sqlx::query_as::<_, (String, String, i64, i64, f64, f64)>(
&format!(
"SELECT chunk_id, text_content, start_frame, end_frame, start_time, end_time \
let rows = sqlx::query_as::<_, (String, String, i64, i64, f64, f64)>(&format!(
"SELECT chunk_id, text_content, start_frame, end_frame, start_time, end_time \
FROM {} WHERE file_uuid = $1 AND chunk_type = 'relationship' \
AND embedding IS NULL AND (text_content IS NOT NULL AND text_content != '') \
ORDER BY id",
chunk_table
),
)
chunk_table
))
.bind(uuid)
.fetch_all(pool)
.await?;
if rows.is_empty() {
info!("[Vectorize-R2] No relationship chunks to vectorize for {}", uuid);
info!(
"[Vectorize-R2] No relationship chunks to vectorize for {}",
uuid
);
return Ok(());
}
@@ -1560,7 +1602,10 @@ impl JobWorker {
text: Some(text.clone()),
};
if let Err(e) = qdrant.upsert_vector(&chunk_id, &vector, payload).await {
error!("[Vectorize-R2] Qdrant upsert failed for {}: {}", chunk_id, e);
error!(
"[Vectorize-R2] Qdrant upsert failed for {}: {}",
chunk_id, e
);
continue;
}
stored += 1;