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

feat: media API (video/bbox/thumbnail), UUID unification, dot matrix text, portal fixes, API dictionary V1.3

Browse Source
This commit is contained in:
Warren
2026-05-06 13:34:49 +08:00
parent e75c4d6f07
commit 74b6182eba
197 changed files with 17511 additions and 8759 deletions
Download Patch File Download Diff File Expand all files Collapse all files

79
src/worker/job_worker.rs
View File

@@ -55,6 +55,7 @@ impl JobWorker {
// 檢查系統資源並寫入 Redis
let resources = SystemResources::check();
let dynamic_max = resources.safe_max_concurrent(self.config.max_concurrent);
self.processor_pool.sweep_stale().await;
let health_key = format!("{}health", crate::core::config::REDIS_KEY_PREFIX.as_str());
let now = chrono::Utc::now().to_rfc3339();
@@ -264,14 +265,29 @@ impl JobWorker {
.update_worker_job_status(&job.uuid, job.id, "running", None, 0, total_processor_types)
.await?;
// Get existing processor results for this job
// Get existing processor results for this job AND completed results from previous jobs
let existing_results = self.db.get_processor_results_by_job(job.id).await?;
let mut result_map = HashMap::new();
for result in existing_results {
result_map.insert(result.processor_type, result);
}
// 補入跨 job 的 completed 狀態（避免 dependency 找不到之前完成的 processor）
let cross_job_results = self
.db
.get_latest_processor_results_by_file_uuid(&job.uuid)
.await
.unwrap_or_default();
for result in cross_job_results {
if !result_map.contains_key(&result.processor_type)
&& matches!(result.status, ProcessorJobStatus::Completed)
{
result_map.insert(result.processor_type, result);
}
}
let mut started_count = 0i32;
for processor_type in &processors_to_run {
// Update processor status to running
self.db
@@ -296,9 +312,10 @@ impl JobWorker {
continue;
}
ProcessorJobStatus::Failed => {
info!("Processor {} failed, skipping", processor_type.as_str());
started_count += 1;
continue;
info!(
"Processor {} previously failed, retrying",
processor_type.as_str()
);
}
ProcessorJobStatus::Running => {
info!(
@@ -427,6 +444,7 @@ impl JobWorker {
job: job.clone(),
processor_type: *processor_type,
processor_result_id,
frame_dir: None,
};
self.processor_pool.start_processor(task).await?;
@@ -555,31 +573,66 @@ impl JobWorker {
});
}
// 🚀 P2 Trigger: Trace Face Aggregation (after Face)
// 🚀 P2 Trigger: Face Trace + DB Store (after Face)
// Runs face_tracker.py (IoU+embedding tracking), stores trace_id + position in DB
if has_face {
info!("📝 Face completed, triggering trace_face aggregation...");
let db_clone = self.db.clone();
info!("📝 Face completed, triggering face trace + DB store...");
let uuid_clone = uuid.to_string();
tokio::spawn(async move {
let executor = match crate::core::processor::PythonExecutor::new() {
Ok(ex) => ex,
Err(e) => {
error!("Failed to create PythonExecutor for trace_face: {}", e);
error!("Failed to create PythonExecutor for face trace: {}", e);
return;
}
};
match executor
.run(
"trace_face_aggregator.py",
"store_traced_faces.py",
&["--file-uuid", &uuid_clone],
Some(&uuid_clone),
"TRACE_FACE",
Some(std::time::Duration::from_secs(300)),
"TRACE_STORE",
Some(std::time::Duration::from_secs(600)),
)
.await
{
Ok(()) => info!("✅ Trace Face aggregation completed for {}", uuid_clone),
Err(e) => error!("❌ Trace Face aggregation failed: {}", e),
Ok(()) => {
info!("✅ Face trace + DB store completed for {}", uuid_clone);
// Sync face embeddings to Qdrant for ANN search
info!("📝 Syncing face embeddings to Qdrant...");
if let Err(e) =
crate::core::db::qdrant_db::sync_face_embeddings(&uuid_clone).await
{
error!("❌ Qdrant face sync failed for {}: {}", uuid_clone, e);
} else {
info!("✅ Qdrant face sync completed for {}", uuid_clone);
}
}
Err(e) => {
error!("❌ Face trace + DB store failed for {}: {}", uuid_clone, e)
}
}
});
}
// 🚀 P2.5 Trigger: TMDb Face Matching (after Face, if TMDb data exists)
if has_face && *crate::core::config::tmdb::PROBE_ENABLED {
info!("📝 Face completed, triggering TMDb face matching...");
let db_clone = self.db.clone();
let uuid_clone = uuid.to_string();
tokio::spawn(async move {
match crate::core::tmdb::face_agent::match_faces_against_tmdb(
&db_clone,
&uuid_clone,
)
.await
{
Ok(count) => info!(
"✅ TMDb face matching: {} bindings created for {}",
count, uuid_clone
),
Err(e) => error!("❌ TMDb face matching failed for {}: {}", uuid_clone, e),
}
});
}

74
src/worker/processor.rs
View File

@@ -3,7 +3,7 @@ use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use tokio::sync::{mpsc, RwLock};
use tracing::{error, info};
use tracing::{error, info, warn};
use crate::core::config::{OUTPUT_DIR, PYTHON_PATH, SCRIPTS_DIR};
use crate::core::db::{
@@ -36,6 +36,7 @@ pub struct ProcessorTask {
pub job: MonitorJob,
pub processor_type: ProcessorType,
pub processor_result_id: i32,
pub frame_dir: Option<String>,
}
pub struct ProcessorPool {
@@ -50,6 +51,7 @@ struct ProcessorHandle {
#[allow(dead_code)]
processor_type: ProcessorType,
cancel_tx: mpsc::Sender<()>,
child_pid: Arc<RwLock<Option<i32>>>,
}
impl ProcessorPool {
@@ -75,6 +77,30 @@ impl ProcessorPool {
count < max
}
/// 清理 stale running state：若系統中實際運行的 processor 比記錄少，修正 count
pub async fn sweep_stale(&self) {
let handle_count = self.running.read().await.len();
let count = *self.running_count.read().await;
if handle_count != count {
warn!(
"[ProcessorPool] Stale count detected: handles={}, count={}, fixing",
handle_count, count
);
let mut c = self.running_count.write().await;
*c = handle_count;
}
if handle_count == 0 && count == 0 {
if let Err(e) = self
.db
.reset_stale_processor_results(ProcessorJobStatus::Failed, "Worker restarted")
.await
{
error!("Failed to reset stale processor results: {}", e);
}
}
}
pub async fn start_processor(&self, task: ProcessorTask) -> Result<()> {
let (cancel_tx, cancel_rx) = mpsc::channel(1);
let job_id = task.job.id;
@@ -94,11 +120,13 @@ impl ProcessorPool {
let running = self.running.clone();
let running_count = self.running_count.clone();
let child_pid: Arc<RwLock<Option<i32>>> = Arc::new(RwLock::new(None));
running.write().await.insert(
job_id,
ProcessorHandle {
processor_type,
cancel_tx,
child_pid: child_pid.clone(),
},
);
@@ -108,6 +136,13 @@ impl ProcessorPool {
let processor_result_id = task.processor_result_id;
let processor_name = processor_type.as_str().to_string();
// 設置共享 frame 目錄環境變數（若有）
if let Some(ref fd) = task.frame_dir {
std::env::set_var("MOMENTRY_FRAME_DIR", fd);
} else {
std::env::remove_var("MOMENTRY_FRAME_DIR");
}
tokio::spawn(async move {
info!("Starting processor {} for job {}", processor_name, job.uuid);
@@ -136,6 +171,12 @@ impl ProcessorPool {
let result = Self::run_processor(&db, &redis, &job, processor_type, cancel_rx).await;
// Store child PID for stability
{
let mut pid_lock = child_pid.write().await;
*pid_lock = Some(0);
}
{
let mut running_guard = running.write().await;
running_guard.remove(&job_id);
@@ -489,6 +530,37 @@ impl ProcessorPool {
pid: 0,
})
}
ProcessorType::Story => {
let executor = crate::core::processor::PythonExecutor::new()?;
let _ = executor
.run(
"parent_chunk_5w1h.py",
&["--file-uuid", &job.uuid, "--max-scenes", "300"],
uuid,
"STORY",
Some(std::time::Duration::from_secs(300)),
)
.await;
let narratives_path = output_dir.join(format!("{}.narratives.json", job.uuid));
let chunks_produced = if narratives_path.exists() {
let content = std::fs::read_to_string(&narratives_path).unwrap_or_default();
let count: i32 = serde_json::from_str::<Vec<String>>(&content)
.map(|v| v.len() as i32)
.unwrap_or(0);
tracing::info!("Story generated {} narratives for {}", count, job.uuid);
count
} else {
0
};
Ok(ProcessorOutput {
data: serde_json::Value::Null,
chunks_produced,
frames_processed: total_frames,
total_frames,
retry_count: 0,
pid: 0,
})
}
}
}