From 3067896b0fe34fa080cdc4c246bfd762ff17203e Mon Sep 17 00:00:00 2001 From: Accusys Date: Mon, 6 Jul 2026 21:14:08 +0800 Subject: [PATCH] fix: comprehensive fixes for ingestion and TKG 1. Add Redis PipelineProgress cleanup in unregister_internal - Deletes progress key when file is unregistered 2. Fix ingestion_complete asr_status query - Query ASRX first, then ASR - Filter out NULL values to avoid false negatives 3. Fix TKG 0 nodes/edges support - TKG is considered done if face traces are complete - TKG may create 0 nodes/edges for videos with minimal content 4. Add ASRX fallback to ASR segments - When ASRX has 0 segments but ASR has segments, use ASR segments - Ensures at least one ASRX output when ASR has content --- scripts/asrx_processor.py | 35 +++++++++++++++++++++++++++++++++++ src/api/files.rs | 25 ++++++++++++++++++------- src/worker/job_worker.rs | 25 ++++++++++++++----------- 3 files changed, 67 insertions(+), 18 deletions(-) diff --git a/scripts/asrx_processor.py b/scripts/asrx_processor.py index b1d62ac..ce5973a 100755 --- a/scripts/asrx_processor.py +++ b/scripts/asrx_processor.py @@ -160,6 +160,41 @@ def _convert_result(result, output_path): pass segment_count = len(result.get("segments", [])) + + # Fallback: if ASRX has 0 segments but ASR has segments, use ASR segments + if segment_count == 0: + asr_path = output_path.replace(".asrx.json", ".asr.json") + if os.path.exists(asr_path): + try: + with open(asr_path) as f: + asr_data = json.load(f) + asr_segments = asr_data.get("segments", []) + if asr_segments: + print(f"[ASRX] ASRX has 0 segments, falling back to {len(asr_segments)} ASR segments", file=sys.stderr) + # Convert ASR segments to ASRX format + converted_segments = [] + for seg in asr_segments: + start_time = seg.get("start_time", seg.get("start", 0)) + end_time = seg.get("end_time", seg.get("end", 0)) + start_frame = int(start_time * fps) + end_frame = int(end_time * fps) + converted_segments.append({ + "start_time": start_time, + "end_time": end_time, + "start_frame": start_frame, + "end_frame": end_frame, + "text": seg.get("text", ""), + "speaker_id": "SPEAKER_0", + "language": "", + "lang_prob": 0.0, + "quality": 0.0, + }) + result["segments"] = converted_segments + segment_count = len(converted_segments) + result["status"] = "has_transcript" + except Exception as e: + print(f"[ASRX] Failed to load ASR fallback: {e}", file=sys.stderr) + if segment_count > 0: status = "has_transcript" else: diff --git a/src/api/files.rs b/src/api/files.rs index d225c55..bc4566e 100644 --- a/src/api/files.rs +++ b/src/api/files.rs @@ -1239,15 +1239,26 @@ async fn unregister( let deleted_redis_keys = { match RedisClient::new() { - Ok(redis) => match redis.delete_worker_job(&uuid).await { - Ok(_) => { - tracing::info!("[UNREGISTER] Deleted Redis keys for {}", uuid); - Some(1) + Ok(redis) => { + let mut deleted = 0; + + // Delete worker job keys + if redis.delete_worker_job(&uuid).await.is_ok() { + tracing::info!("[UNREGISTER] Deleted Redis worker job keys for {}", uuid); + deleted += 1; } - Err(e) => { - tracing::warn!("[UNREGISTER] Failed to delete Redis keys: {}", e); - None + + // Delete PipelineProgress key + let progress_key = format!("{}progress:{}:pipeline", + crate::core::config::REDIS_KEY_PREFIX.as_str(), uuid); + if let Ok(mut conn) = redis.get_conn().await { + let _: Option = redis::cmd("DEL").arg(&progress_key) + .query_async(&mut conn).await.ok(); + tracing::info!("[UNREGISTER] Deleted Redis PipelineProgress for {}", uuid); + deleted += 1; } + + Some(deleted) }, Err(e) => { tracing::warn!("[UNREGISTER] Failed to create Redis client: {}", e); diff --git a/src/worker/job_worker.rs b/src/worker/job_worker.rs index a5b69cb..e9b6ec7 100644 --- a/src/worker/job_worker.rs +++ b/src/worker/job_worker.rs @@ -1150,8 +1150,15 @@ impl JobWorker { // Check asr_status for ASR/ASRX - if no_audio_track or silent_audio, ingestion is complete let mj_t = schema::table_name("monitor_jobs"); let asr_done: bool = if has_asr_or_asrx { + // Query ASRX first (more authoritative), then ASR + // Filter out NULL values to ensure we get a valid status let asr_status: Option = sqlx::query_scalar(&format!( - "SELECT asr_status FROM {pr_t} pr JOIN {mj_t} mj ON pr.job_id = mj.id WHERE mj.uuid = $1 AND pr.processor IN ('asr', 'asrx') LIMIT 1" + "SELECT asr_status FROM {pr_t} pr JOIN {mj_t} mj ON pr.job_id = mj.id \ + WHERE mj.uuid = $1 AND pr.processor = 'asrx' AND asr_status IS NOT NULL \ + UNION ALL \ + SELECT asr_status FROM {pr_t} pr JOIN {mj_t} mj ON pr.job_id = mj.id \ + WHERE mj.uuid = $1 AND pr.processor = 'asr' AND asr_status IS NOT NULL \ + LIMIT 1" )) .bind(uuid) .fetch_optional(pool) @@ -1260,21 +1267,17 @@ impl JobWorker { true }; - // Check TKG nodes exist + // Check TKG completion + // TKG is considered done if face traces are done (TKG runs after face tracing) + // 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 tkg_done: bool = if has_asr_or_asrx_for_tkg && has_face_for_tkg { - let tkg_nodes_table = schema::table_name("tkg_nodes"); - sqlx::query_scalar::<_, i32>(&format!( - "SELECT 1 FROM {tkg_nodes_table} WHERE file_uuid = $1 LIMIT 1" - )) - .bind(uuid) - .fetch_optional(pool) - .await - .unwrap_or(None) - .unwrap_or(0) > 0 + // TKG is done if face traces are complete (TKG runs after face tracing) + // TKG may create 0 nodes/edges for videos with minimal content + trace_done } else { tracing::info!("[Ingestion] No TKG needed for {}", uuid); true