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

refactor: cargo fmt across 27 files + behavioral fixes

Browse Source 
Behavioral changes:
- postgres_db: reorder default processors (cut first), prevent overwriting completed status
- qdrant_db: fix scroll pagination exit condition (next.is_none())
- job_worker: add idempotency guards for face trace / TKG build; better error logging
- main.rs: add LineWriter for stdout buffering

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