Compare commits
2 Commits
3d13d1390e
...
0d58a738a1
| Author | SHA1 | Date | |
|---|---|---|---|
|
0d58a738a1 | ||
|
|
08167d73b2 |
484
docs_v1.0/DESIGN/Processor_State_Machine_V1.0.md
Normal file
484
docs_v1.0/DESIGN/Processor_State_Machine_V1.0.md
Normal file
@@ -0,0 +1,484 @@
|
||||
---
|
||||
title: Processor State Machine V1.0
|
||||
version: 1.0
|
||||
date: 2026-05-30
|
||||
author: M5Max128
|
||||
status: draft
|
||||
---
|
||||
|
||||
# Processor State Machine V1.0
|
||||
|
||||
## Overview
|
||||
|
||||
| Attribute | Value |
|
||||
|-----------|-------|
|
||||
| Scope | Backend, Worker, Pipeline |
|
||||
| Status | Draft |
|
||||
| Applicable To | M5Max128, M5Max48 |
|
||||
| Dependencies | migrations/034, job_worker.rs, redis_client.rs |
|
||||
| Related Docs | [Pipeline Module](../API_WORKSPACE/modules/10_pipeline.md), [TKG Query API](TKG_QUERY_API_V1.0.md) |
|
||||
|
||||
---
|
||||
|
||||
## 1. Design Goals
|
||||
|
||||
### 1.1 Problem Statement
|
||||
|
||||
The Momentry Core pipeline lacks unified state management for processors:
|
||||
|
||||
- **Opaque dependency chains**: Processors depend on each other (ASR → Cut, ASRX → ASR, Story → ASRX + Cut + YOLO + Face), but failures or delays are not explicitly tracked
|
||||
- **No alert mechanism**: When dependencies are not met or resources are exhausted, there is no systematic way to notify operators or trigger retries
|
||||
- **Coarse-grained status**: Existing `pending/running/completed/failed` states do not capture intermediate conditions like "waiting for dependencies" or "ready but not scheduled"
|
||||
|
||||
### 1.2 Solution
|
||||
|
||||
Introduce a **State Machine** with **Alert Mechanism**:
|
||||
|
||||
- **8 explicit states** for each processor job: `Idle → Waiting → Ready → Pending → Running → Completed/Failed/Skipped`
|
||||
- **Dependency checking**: `check_dependencies()` validates prerequisites before execution
|
||||
- **Alert emission**: Emit alerts to Redis pub/sub and PostgreSQL for monitoring and debugging
|
||||
|
||||
### 1.3 Scope
|
||||
|
||||
This design **complements** the existing polling mechanism:
|
||||
|
||||
| Component | Responsibility |
|
||||
|-----------|---------------|
|
||||
| **State Machine** | Fine-grained processor status management (Idle → Running → Completed) |
|
||||
| **Polling** | Coarse-grained ingestion verification (Rule 1 chunks exist? Vectorize done? TKG nodes exist?) |
|
||||
|
||||
**Non-Goals**:
|
||||
|
||||
- Does NOT replace polling for post-processing steps (入庫)
|
||||
- Does NOT auto-retry failed processors (future evolution)
|
||||
- Does NOT manage distributed state across workers
|
||||
|
||||
---
|
||||
|
||||
## 2. State Definitions
|
||||
|
||||
### 2.1 Eight States
|
||||
|
||||
| State | Semantics | Trigger | Next States |
|
||||
|-------|-----------|---------|--------------|
|
||||
| `Idle` | Initial state, no work assigned | Job created | `Waiting` |
|
||||
| `Waiting` | Dependencies not met, awaiting prerequisites | Dependency check fails | `Ready`, `Failed` |
|
||||
| `Ready` | Dependencies met, awaiting execution | Dependency check passes | `Pending` |
|
||||
| `Pending` | Queued for execution, waiting for worker | Scheduler accepts | `Running` |
|
||||
| `Running` | Currently processing | Worker starts | `Completed`, `Failed`, `Skipped` |
|
||||
| `Completed` | Success, output valid | Output validated | - (terminal) |
|
||||
| `Failed` | Error occurred, unrecoverable | Exception or timeout | - (terminal) |
|
||||
| `Skipped` | Conditional skip (optional processor) | Unmet optional conditions | - (terminal) |
|
||||
|
||||
### 2.2 State Transition Examples
|
||||
|
||||
**Example 1: ASR depends on Cut**
|
||||
|
||||
```
|
||||
ASR: Idle → Waiting (Cut not completed)
|
||||
Cut: Running → Completed
|
||||
ASR: Waiting → Ready (Cut completed) → Pending → Running → Completed
|
||||
```
|
||||
|
||||
**Example 2: Story depends on multiple processors**
|
||||
|
||||
```
|
||||
Story: Idle → Waiting (ASRX not completed)
|
||||
ASRX: Running → Completed
|
||||
Story: Waiting → Waiting (Cut not completed)
|
||||
Cut: Running → Completed
|
||||
Story: Waiting → Waiting (YOLO not completed)
|
||||
YOLO: Running → Completed
|
||||
Story: Waiting → Waiting (Face not completed)
|
||||
Face: Running → Completed
|
||||
Story: Waiting → Ready (all dependencies met) → Pending → Running → Completed
|
||||
```
|
||||
|
||||
**Example 3: Optional processor skipped**
|
||||
|
||||
```
|
||||
Pose: Idle → Ready → Pending → Running
|
||||
Pose: Running → Skipped (no pose detected, optional processing)
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 3. State Transitions
|
||||
|
||||
### 3.1 Transition Diagram
|
||||
|
||||
```mermaid
|
||||
stateDiagram-v2
|
||||
[*] --> Idle: Job created
|
||||
|
||||
Idle --> Waiting: Initialize
|
||||
|
||||
Waiting --> Ready: Dependencies met
|
||||
Waiting --> Failed: Timeout
|
||||
|
||||
Ready --> Pending: Scheduled
|
||||
|
||||
Pending --> Running: Worker pickup
|
||||
|
||||
Running --> Completed: Success
|
||||
Running --> Failed: Error
|
||||
Running --> Skipped: Conditional skip
|
||||
|
||||
Completed --> [*]
|
||||
Failed --> [*]
|
||||
Skipped --> [*]
|
||||
```
|
||||
|
||||
### 3.2 Transition Rules
|
||||
|
||||
| From State | To State | Condition | Action |
|
||||
|------------|-----------|-----------|--------|
|
||||
| `Idle` | `Waiting` | Always (initial transition) | - |
|
||||
| `Waiting` | `Ready` | `check_dependencies() == Ok` | - |
|
||||
| `Waiting` | `Failed` | Timeout (default 7200s) | Emit `timeout` alert |
|
||||
| `Ready` | `Pending` | Resource available | - |
|
||||
| `Pending` | `Running` | Worker starts | - |
|
||||
| `Running` | `Completed` | Output valid | - |
|
||||
| `Running` | `Failed` | Exception or output invalid | Emit `output_invalid` alert |
|
||||
| `Running` | `Skipped` | Optional processor, conditions not met | - |
|
||||
|
||||
### 3.3 Edge Cases
|
||||
|
||||
| Scenario | Detection | Resolution |
|
||||
|----------|-----------|------------|
|
||||
| **Circular dependencies** | `check_dependencies()` detects cycle | Mark as `Failed`, emit `dependency_not_met` alert |
|
||||
| **Resource exhaustion** | GPU/CPU unavailable | Stay in `Waiting`, emit `resource_exhausted` alert |
|
||||
| **Partial output** | Output validation fails | Mark as `Failed`, emit `output_invalid` alert |
|
||||
| **Transient failure** | Network/API timeout | Stay in `Waiting`, retry after delay |
|
||||
|
||||
---
|
||||
|
||||
## 4. Alert Mechanism
|
||||
|
||||
### 4.1 Alert Types
|
||||
|
||||
| Type | Trigger | Severity | Action |
|
||||
|------|---------|----------|--------|
|
||||
| `dependency_not_met` | `check_dependencies()` fails | Warning | Retry after delay |
|
||||
| `resource_exhausted` | GPU/CPU unavailable | Warning | Wait + retry |
|
||||
| `output_invalid` | Validation fails | Error | Mark `Failed` |
|
||||
| `timeout` | Exceeds `MOMENTRY_*_TIMEOUT` | Error | Mark `Failed` |
|
||||
|
||||
### 4.2 Alert Flow
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
participant Worker as job_worker.rs
|
||||
participant Checker as check_dependencies()
|
||||
participant Redis as Redis Pub/Sub
|
||||
participant PostgreSQL as processor_alerts table
|
||||
|
||||
Worker->>Checker: check_dependencies(processor, file_uuid)
|
||||
alt Dependencies not met
|
||||
Checker-->>Worker: ConditionResult::NotMet(reason)
|
||||
Worker->>Redis: emit_processor_alert(file_uuid, processor, "dependency_not_met", reason)
|
||||
Redis-->>PostgreSQL: INSERT INTO processor_alerts
|
||||
Worker->>Worker: update_status(file_uuid, processor, Waiting)
|
||||
else Resource exhausted
|
||||
Checker-->>Worker: ConditionResult::ResourceExhausted
|
||||
Worker->>Redis: emit_processor_alert(file_uuid, processor, "resource_exhausted", "GPU unavailable")
|
||||
Redis-->>PostgreSQL: INSERT INTO processor_alerts
|
||||
Worker->>Worker: update_status(file_uuid, processor, Waiting)
|
||||
else Output invalid
|
||||
Checker-->>Worker: ConditionResult::OutputInvalid(reason)
|
||||
Worker->>Redis: emit_processor_alert(file_uuid, processor, "output_invalid", reason)
|
||||
Redis-->>PostgreSQL: INSERT INTO processor_alerts
|
||||
Worker->>Worker: update_status(file_uuid, processor, Failed)
|
||||
else OK
|
||||
Checker-->>Worker: ConditionResult::Ok
|
||||
Worker->>Worker: update_status(file_uuid, processor, Running)
|
||||
end
|
||||
```
|
||||
|
||||
### 4.3 Redis Channel
|
||||
|
||||
- **Channel**: `momentry:processor:alerts`
|
||||
- **Message Format**:
|
||||
```json
|
||||
{
|
||||
"file_uuid": "bd80fec9c42afb0307eb28f22c64c76a",
|
||||
"processor": "ASR",
|
||||
"alert_type": "dependency_not_met",
|
||||
"message": "Cut not completed",
|
||||
"timestamp": "2026-05-30T10:15:30Z"
|
||||
}
|
||||
```
|
||||
- **Consumers**: None (current implementation logs only, future: monitoring service)
|
||||
|
||||
### 4.4 PostgreSQL Table
|
||||
|
||||
**Table**: `processor_alerts` (defined in `migrations/034_processor_state_machine.sql`)
|
||||
|
||||
```sql
|
||||
CREATE TABLE IF NOT EXISTS processor_alerts (
|
||||
id SERIAL PRIMARY KEY,
|
||||
file_uuid VARCHAR(32),
|
||||
processor_type VARCHAR(32) NOT NULL,
|
||||
alert_type VARCHAR(32) NOT NULL,
|
||||
message TEXT,
|
||||
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
|
||||
);
|
||||
|
||||
CREATE INDEX idx_alerts_file_uuid ON processor_alerts(file_uuid);
|
||||
CREATE INDEX idx_alerts_processor_type ON processor_alerts(processor_type);
|
||||
CREATE INDEX idx_alerts_alert_type ON processor_alerts(alert_type);
|
||||
CREATE INDEX idx_alerts_created_at ON processor_alerts(created_at);
|
||||
```
|
||||
|
||||
**Retention Policy**: 30 days (TBD, future: implement cleanup job)
|
||||
|
||||
---
|
||||
|
||||
## 5. Dependency Checking
|
||||
|
||||
### 5.1 ConditionResult Enum
|
||||
|
||||
Defined in `src/worker/job_worker.rs`:
|
||||
|
||||
```rust
|
||||
pub enum ConditionResult {
|
||||
Ok, // All dependencies met
|
||||
NotMet(String), // Missing dependency (reason)
|
||||
ResourceExhausted, // GPU/CPU unavailable
|
||||
OutputInvalid(String), // Validation failed (reason)
|
||||
}
|
||||
```
|
||||
|
||||
### 5.2 check_dependencies() Logic
|
||||
|
||||
Defined in `src/worker/job_worker.rs`:
|
||||
|
||||
```rust
|
||||
pub async fn check_dependencies(
|
||||
processor: ProcessorType,
|
||||
file_uuid: &str,
|
||||
db: &PostgresDb,
|
||||
) -> Result<ConditionResult> {
|
||||
match processor {
|
||||
ProcessorType::ASR => {
|
||||
// Check if Cut is completed
|
||||
if !db.is_processor_completed(file_uuid, ProcessorType::Cut).await? {
|
||||
return Ok(ConditionResult::NotMet("Cut not completed".into()));
|
||||
}
|
||||
}
|
||||
ProcessorType::ASRX => {
|
||||
// Check if ASR is completed
|
||||
if !db.is_processor_completed(file_uuid, ProcessorType::ASR).await? {
|
||||
return Ok(ConditionResult::NotMet("ASR not completed".into()));
|
||||
}
|
||||
}
|
||||
ProcessorType::Story => {
|
||||
// Check if ASRX + Cut + YOLO + Face are completed
|
||||
let deps = [
|
||||
ProcessorType::ASRX,
|
||||
ProcessorType::Cut,
|
||||
ProcessorType::YOLO,
|
||||
ProcessorType::Face,
|
||||
];
|
||||
for dep in deps {
|
||||
if !db.is_processor_completed(file_uuid, dep).await? {
|
||||
return Ok(ConditionResult::NotMet(format!("{:?} not completed", dep)));
|
||||
}
|
||||
}
|
||||
}
|
||||
ProcessorType::_5W1H => {
|
||||
// Check if Story is completed
|
||||
if !db.is_processor_completed(file_uuid, ProcessorType::Story).await? {
|
||||
return Ok(ConditionResult::NotMet("Story not completed".into()));
|
||||
}
|
||||
}
|
||||
// Other processors have no dependencies
|
||||
_ => {}
|
||||
}
|
||||
Ok(ConditionResult::Ok)
|
||||
}
|
||||
```
|
||||
|
||||
### 5.3 Integration with job_worker.rs
|
||||
|
||||
```rust
|
||||
// In execute_processor()
|
||||
let condition = check_dependencies(processor, file_uuid, &db).await?;
|
||||
match condition {
|
||||
ConditionResult::Ok => {
|
||||
// Proceed to Running state
|
||||
self.update_status(file_uuid, processor, ProcessorJobStatus::Running).await?;
|
||||
// Execute processor...
|
||||
}
|
||||
ConditionResult::NotMet(reason) => {
|
||||
// Emit alert and mark as Waiting
|
||||
emit_processor_alert(file_uuid, processor, "dependency_not_met", &reason).await?;
|
||||
self.update_status(file_uuid, processor, ProcessorJobStatus::Waiting).await?;
|
||||
}
|
||||
ConditionResult::ResourceExhausted => {
|
||||
// Emit alert and mark as Waiting
|
||||
emit_processor_alert(file_uuid, processor, "resource_exhausted", "GPU unavailable").await?;
|
||||
self.update_status(file_uuid, processor, ProcessorJobStatus::Waiting).await?;
|
||||
}
|
||||
ConditionResult::OutputInvalid(reason) => {
|
||||
// Emit alert and mark as Failed
|
||||
emit_processor_alert(file_uuid, processor, "output_invalid", &reason).await?;
|
||||
self.update_status(file_uuid, processor, ProcessorJobStatus::Failed).await?;
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 6. Integration Points
|
||||
|
||||
### 6.1 With TKG Builder
|
||||
|
||||
- **TKG Builder** is NOT a processor, it's a **post-processing step** (入庫 step 8)
|
||||
- Triggers after Face Trace is completed
|
||||
- **State Machine does NOT manage TKG Builder state**
|
||||
- TKG Builder has its own verification mechanism in polling
|
||||
|
||||
### 6.2 With Face Trace
|
||||
|
||||
- **Face Trace** is NOT a processor, it's a **post-processing step** (入庫 step 5)
|
||||
- Triggers after all 10 processors are completed
|
||||
- **State Machine does NOT manage Face Trace state**
|
||||
- Face Trace has its own verification mechanism in polling
|
||||
|
||||
### 6.3 With 入庫 Flow
|
||||
|
||||
| Component | Manages | Scope |
|
||||
|-----------|---------|-------|
|
||||
| **State Machine** | Processor states | `Idle → Waiting → Ready → Pending → Running → Completed/Failed/Skipped` |
|
||||
| **Polling** | Post-processing verification | Rule 1 chunks, Vectorize, TKG nodes, Face Trace, etc. |
|
||||
|
||||
**Key Insight**: Two mechanisms are **independent but complementary**:
|
||||
|
||||
1. **State Machine**: Granular processor status, handles dependencies
|
||||
2. **Polling**: Coarse-grained ingestion verification, handles post-processing
|
||||
|
||||
### 6.4 Example Flow
|
||||
|
||||
```
|
||||
=== Processor State Machine (per processor) ===
|
||||
Cut: Idle → Waiting → Ready → Pending → Running → Completed ✓
|
||||
ASR: Idle → Waiting (Cut not done) → Waiting → Ready → Pending → Running → Completed ✓
|
||||
YOLO: Idle → Ready → Pending → Running → Completed ✓
|
||||
Face: Idle → Ready → Pending → Running → Completed ✓
|
||||
Story: Idle → Waiting (ASRX not done) → Waiting → Ready → Pending → Running → Completed ✓
|
||||
|
||||
=== 入庫 Polling (every 3s) ===
|
||||
[00:00] Check: Rule 1 chunks exist? → No (ASR not done)
|
||||
[00:03] Check: Rule 1 chunks exist? → Yes ✓
|
||||
Check: Vectorize done? → Yes ✓
|
||||
Check: TKG nodes exist? → No (Face Trace not done)
|
||||
[00:06] Check: TKG nodes exist? → Yes ✓
|
||||
Check: All 17 steps verified ✓
|
||||
Mark job as completed
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 7. Implementation Checklist
|
||||
|
||||
### 7.1 Completed ✅
|
||||
|
||||
- [x] Migration 034: `processor_alerts` table
|
||||
- [x] Enum: `ProcessorJobStatus` (8 states) - `postgres_db.rs:585-594`
|
||||
- [x] Function: `emit_processor_alert()` - `redis_client.rs`
|
||||
- [x] Function: `check_dependencies()` - `job_worker.rs`
|
||||
- [x] Enum: `ConditionResult` - `job_worker.rs`
|
||||
|
||||
### 7.2 Pending 🔄
|
||||
|
||||
- [ ] Tests: State transitions (unit tests)
|
||||
- [ ] Tests: Alert emission (integration tests)
|
||||
- [ ] Tests: Dependency checking (unit tests)
|
||||
- [ ] Monitoring: Alert dashboard (TBD)
|
||||
- [ ] Retention: `processor_alerts` cleanup job (TBD)
|
||||
|
||||
---
|
||||
|
||||
## 8. Performance Considerations
|
||||
|
||||
### 8.1 Alert Emission
|
||||
|
||||
- **Non-blocking**: Redis pub/sub is fire-and-forget
|
||||
- **Low latency**: < 1ms per alert
|
||||
- **No retry**: If Redis is down, alert is lost (acceptable for debugging)
|
||||
|
||||
### 8.2 Dependency Checking
|
||||
|
||||
- **Synchronous DB queries**: `is_processor_completed()` queries PostgreSQL
|
||||
- **Cacheable**: Results can be cached for 1-3 seconds (TTL based on processor duration)
|
||||
- **Index usage**: Queries use `idx_processor_jobs_file_uuid_processor_type` index
|
||||
|
||||
### 8.3 State Updates
|
||||
|
||||
- **Single-row UPDATE**: `UPDATE processor_jobs SET status = $1 WHERE file_uuid = $2 AND processor_type = $3`
|
||||
- **Index usage**: Uses `idx_processor_jobs_file_uuid_processor_type` index
|
||||
- **Low contention**: Each processor has its own row
|
||||
|
||||
---
|
||||
|
||||
## 9. Future Evolution
|
||||
|
||||
### 9.1 Phase 1 (Current)
|
||||
|
||||
- Alert emission + PostgreSQL logging
|
||||
- Manual monitoring via `processor_alerts` table
|
||||
- No auto-retry
|
||||
|
||||
### 9.2 Phase 2 (Near-term)
|
||||
|
||||
- Alert consumer service (subscribes to Redis channel)
|
||||
- Auto-retry for `dependency_not_met` and `resource_exhausted` alerts
|
||||
- Exponential backoff for retries
|
||||
|
||||
### 9.3 Phase 3 (Medium-term)
|
||||
|
||||
- Event-driven pipeline (replace polling with Redis Streams)
|
||||
- Real-time status updates via WebSocket
|
||||
- Distributed state management (Redis-based)
|
||||
|
||||
### 9.4 Phase 4 (Long-term)
|
||||
|
||||
- DAG-based scheduling (Airflow/Temporal)
|
||||
- Cross-worker coordination
|
||||
- Priority-based resource allocation
|
||||
|
||||
---
|
||||
|
||||
## 10. Glossary
|
||||
|
||||
| Term | Definition |
|
||||
|------|-----------|
|
||||
| **State Machine** | Finite state automaton managing processor lifecycle (8 states) |
|
||||
| **Alert** | Asynchronous notification of state machine events (4 types) |
|
||||
| **Dependency** | Prerequisite processor that must complete before execution |
|
||||
| **Polling** | Periodic verification of post-processing steps (every 3s) |
|
||||
| **入庫** | Post-processing steps after 10 processors complete (17 steps) |
|
||||
| **file_uuid** | Unique identifier for a video file (32-char hex string) |
|
||||
| **Processor** | One of 10 processing stages (Cut, ASR, ASRX, YOLO, OCR, Face, Pose, VisualChunk, Story, 5W1H) |
|
||||
| **Post-processing** | Steps that run after processors (Rule 1, Vectorize, TKG, Face Trace, etc.) |
|
||||
|
||||
---
|
||||
|
||||
## 11. References
|
||||
|
||||
- [Pipeline Module](../API_WORKSPACE/modules/10_pipeline.md) - Pipeline overview and 入庫 steps
|
||||
- [TKG Query API V1.0](TKG_QUERY_API_V1.0.md) - TKG integration details
|
||||
- [Processor Refactoring Assessment](Processor_Refactoring_Assessment.md) - Processor refactoring plans
|
||||
- `migrations/034_processor_state_machine.sql` - Database schema
|
||||
- `src/core/db/postgres_db.rs` - ProcessorJobStatus enum
|
||||
- `src/core/db/redis_client.rs` - emit_processor_alert() function
|
||||
- `src/worker/job_worker.rs` - ConditionResult enum and check_dependencies()
|
||||
|
||||
---
|
||||
|
||||
## Version History
|
||||
|
||||
| Version | Date | Author | Changes |
|
||||
|---------|------|--------|---------|
|
||||
| 1.0 | 2026-05-30 | M5Max128 | Initial design document |
|
||||
23
migrations/034_processor_state_machine.sql
Normal file
23
migrations/034_processor_state_machine.sql
Normal file
@@ -0,0 +1,23 @@
|
||||
-- Migration: 034_processor_state_machine
|
||||
-- Purpose: Add processor_alerts table for State Machine alert mechanism
|
||||
-- Date: 2026-05-30
|
||||
|
||||
-- Create processor_alerts table
|
||||
CREATE TABLE IF NOT EXISTS processor_alerts (
|
||||
id SERIAL PRIMARY KEY,
|
||||
file_uuid VARCHAR(32),
|
||||
processor_type VARCHAR(32) NOT NULL,
|
||||
alert_type VARCHAR(32) NOT NULL,
|
||||
message TEXT,
|
||||
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
|
||||
);
|
||||
|
||||
-- Create indexes for efficient querying
|
||||
CREATE INDEX IF NOT EXISTS idx_alerts_file_uuid ON processor_alerts(file_uuid);
|
||||
CREATE INDEX IF NOT EXISTS idx_alerts_processor_type ON processor_alerts(processor_type);
|
||||
CREATE INDEX IF NOT EXISTS idx_alerts_alert_type ON processor_alerts(alert_type);
|
||||
CREATE INDEX IF NOT EXISTS idx_alerts_created_at ON processor_alerts(created_at);
|
||||
|
||||
-- Add comments
|
||||
COMMENT ON TABLE processor_alerts IS 'Processor state machine alerts for dependency/resource/output issues';
|
||||
COMMENT ON COLUMN processor_alerts.alert_type IS 'Alert types: dependency_not_met, resource_exhausted, output_invalid, timeout';
|
||||
@@ -34,7 +34,7 @@ pub struct IdentityRecord {
|
||||
pub uuid: Uuid,
|
||||
pub name: String,
|
||||
pub metadata: serde_json::Value,
|
||||
pub created_at: Option<chrono::DateTime<chrono::Utc>>,
|
||||
pub created_at: Option<chrono::NaiveDateTime>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
|
||||
@@ -141,6 +141,26 @@ pub struct IdentityFaceRecord {
|
||||
pub confidence: f64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
|
||||
pub struct FileFaceRecord {
|
||||
pub id: i64,
|
||||
pub file_uuid: String,
|
||||
pub frame_number: i64,
|
||||
pub timestamp_secs: Option<f64>,
|
||||
pub face_id: Option<String>,
|
||||
pub trace_id: Option<i32>,
|
||||
pub x: f64,
|
||||
pub y: f64,
|
||||
pub width: f64,
|
||||
pub height: f64,
|
||||
pub confidence: f64,
|
||||
pub identity_id: Option<i32>,
|
||||
pub stranger_id: Option<i32>,
|
||||
pub identity_uuid: Option<String>,
|
||||
pub identity_name: Option<String>,
|
||||
pub stranger_metadata: Option<serde_json::Value>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)]
|
||||
pub struct IdentityChunkRecord {
|
||||
pub id: i32,
|
||||
@@ -406,14 +426,12 @@ pub enum PipelineType {
|
||||
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Hash)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum ProcessorType {
|
||||
Asr,
|
||||
Cut,
|
||||
Yolo,
|
||||
Ocr,
|
||||
Face,
|
||||
Pose,
|
||||
Asrx,
|
||||
VisualChunk,
|
||||
Scene,
|
||||
Story,
|
||||
FiveW1H,
|
||||
@@ -443,14 +461,12 @@ impl<'r> sqlx::Decode<'r, sqlx::Postgres> for ProcessorType {
|
||||
impl ProcessorType {
|
||||
pub fn as_str(&self) -> &'static str {
|
||||
match self {
|
||||
ProcessorType::Asr => "asr",
|
||||
ProcessorType::Cut => "cut",
|
||||
ProcessorType::Yolo => "yolo",
|
||||
ProcessorType::Ocr => "ocr",
|
||||
ProcessorType::Face => "face",
|
||||
ProcessorType::Pose => "pose",
|
||||
ProcessorType::Asrx => "asrx",
|
||||
ProcessorType::VisualChunk => "visual_chunk",
|
||||
ProcessorType::Scene => "scene",
|
||||
ProcessorType::Story => "story",
|
||||
ProcessorType::FiveW1H => "5w1h",
|
||||
@@ -459,14 +475,12 @@ impl ProcessorType {
|
||||
|
||||
pub fn from_db_str(s: &str) -> Option<Self> {
|
||||
match s {
|
||||
"asr" => Some(ProcessorType::Asr),
|
||||
"cut" => Some(ProcessorType::Cut),
|
||||
"yolo" => Some(ProcessorType::Yolo),
|
||||
"ocr" => Some(ProcessorType::Ocr),
|
||||
"face" => Some(ProcessorType::Face),
|
||||
"pose" => Some(ProcessorType::Pose),
|
||||
"asrx" => Some(ProcessorType::Asrx),
|
||||
"visual_chunk" => Some(ProcessorType::VisualChunk),
|
||||
"scene" => Some(ProcessorType::Scene),
|
||||
"story" => Some(ProcessorType::Story),
|
||||
"5w1h" => Some(ProcessorType::FiveW1H),
|
||||
@@ -474,24 +488,20 @@ impl ProcessorType {
|
||||
}
|
||||
}
|
||||
|
||||
/// 預估 CPU 使用率（0.0 ~ 1.0, 1.0 = 一個完整核心）
|
||||
pub fn estimated_cpu(&self) -> f64 {
|
||||
match self {
|
||||
ProcessorType::Asr => 1.0,
|
||||
ProcessorType::Cut => 0.5,
|
||||
ProcessorType::Yolo => 0.3,
|
||||
ProcessorType::Ocr => 0.8,
|
||||
ProcessorType::Face => 0.6,
|
||||
ProcessorType::Pose => 0.4,
|
||||
ProcessorType::Asrx => 0.8,
|
||||
ProcessorType::VisualChunk => 0.3,
|
||||
ProcessorType::Scene => 0.3,
|
||||
ProcessorType::Story => 0.1,
|
||||
ProcessorType::FiveW1H => 0.1,
|
||||
}
|
||||
}
|
||||
|
||||
/// 是否使用 GPU
|
||||
pub fn uses_gpu(&self) -> bool {
|
||||
match self {
|
||||
ProcessorType::Yolo | ProcessorType::Face | ProcessorType::Pose => true,
|
||||
@@ -499,49 +509,39 @@ impl ProcessorType {
|
||||
}
|
||||
}
|
||||
|
||||
/// 預估記憶體使用量 (MB)
|
||||
pub fn estimated_memory_mb(&self) -> u64 {
|
||||
match self {
|
||||
ProcessorType::Asr => 2048,
|
||||
ProcessorType::Cut => 512,
|
||||
ProcessorType::Yolo => 1024,
|
||||
ProcessorType::Ocr => 1024,
|
||||
ProcessorType::Face => 1536,
|
||||
ProcessorType::Pose => 1024,
|
||||
ProcessorType::Asrx => 2048,
|
||||
ProcessorType::VisualChunk => 512,
|
||||
ProcessorType::Scene => 512,
|
||||
ProcessorType::Story => 256,
|
||||
ProcessorType::FiveW1H => 256,
|
||||
}
|
||||
}
|
||||
|
||||
/// 使用的模型名稱（如有）
|
||||
pub fn model_name(&self) -> Option<&'static str> {
|
||||
match self {
|
||||
ProcessorType::Asr => Some("faster-whisper"),
|
||||
ProcessorType::Cut => None,
|
||||
ProcessorType::Yolo => Some("yolov8n"),
|
||||
ProcessorType::Ocr => Some("paddleocr"),
|
||||
ProcessorType::Face => Some("insightface/buffalo_l"),
|
||||
ProcessorType::Pose => Some("mediapipe/pose"),
|
||||
ProcessorType::Asrx => Some("speechbrain/ecapa-tdnn"),
|
||||
ProcessorType::VisualChunk => None,
|
||||
ProcessorType::Scene => Some("places365"),
|
||||
ProcessorType::Story => None,
|
||||
ProcessorType::FiveW1H => Some("gemma4"),
|
||||
}
|
||||
}
|
||||
|
||||
/// 依賴的其他 Processor（需先完成才能執行）
|
||||
pub fn dependencies(&self) -> Vec<ProcessorType> {
|
||||
match self {
|
||||
ProcessorType::Asr => vec![ProcessorType::Cut],
|
||||
ProcessorType::Asrx => vec![ProcessorType::Asr],
|
||||
ProcessorType::VisualChunk => vec![ProcessorType::Yolo],
|
||||
ProcessorType::Asrx => vec![ProcessorType::Cut],
|
||||
ProcessorType::Scene => vec![ProcessorType::Cut],
|
||||
ProcessorType::Story => vec![
|
||||
ProcessorType::Asr,
|
||||
ProcessorType::Asrx,
|
||||
ProcessorType::Cut,
|
||||
ProcessorType::Yolo,
|
||||
@@ -555,16 +555,12 @@ impl ProcessorType {
|
||||
pub fn all() -> Vec<ProcessorType> {
|
||||
vec![
|
||||
ProcessorType::Cut,
|
||||
// Scene (Places365) removed — replaced by heuristic_scene_metadata post-processor
|
||||
ProcessorType::Asr,
|
||||
ProcessorType::Asrx,
|
||||
ProcessorType::Yolo,
|
||||
ProcessorType::Ocr,
|
||||
ProcessorType::Face,
|
||||
ProcessorType::Pose,
|
||||
ProcessorType::VisualChunk,
|
||||
ProcessorType::Story,
|
||||
ProcessorType::FiveW1H,
|
||||
]
|
||||
}
|
||||
|
||||
@@ -573,11 +569,9 @@ impl ProcessorType {
|
||||
ProcessorType::Yolo
|
||||
| ProcessorType::Ocr
|
||||
| ProcessorType::Face
|
||||
| ProcessorType::Pose
|
||||
| ProcessorType::VisualChunk => PipelineType::Frame,
|
||||
| ProcessorType::Pose => PipelineType::Frame,
|
||||
|
||||
ProcessorType::Asr
|
||||
| ProcessorType::Cut
|
||||
ProcessorType::Cut
|
||||
| ProcessorType::Asrx
|
||||
| ProcessorType::Scene
|
||||
| ProcessorType::Story
|
||||
@@ -589,6 +583,9 @@ impl ProcessorType {
|
||||
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum ProcessorJobStatus {
|
||||
Idle,
|
||||
Waiting,
|
||||
Ready,
|
||||
Pending,
|
||||
Running,
|
||||
Completed,
|
||||
@@ -621,6 +618,9 @@ impl<'r> sqlx::Decode<'r, sqlx::Postgres> for ProcessorJobStatus {
|
||||
impl ProcessorJobStatus {
|
||||
pub fn as_str(&self) -> &'static str {
|
||||
match self {
|
||||
ProcessorJobStatus::Idle => "idle",
|
||||
ProcessorJobStatus::Waiting => "waiting",
|
||||
ProcessorJobStatus::Ready => "ready",
|
||||
ProcessorJobStatus::Pending => "pending",
|
||||
ProcessorJobStatus::Running => "running",
|
||||
ProcessorJobStatus::Completed => "completed",
|
||||
@@ -631,6 +631,9 @@ impl ProcessorJobStatus {
|
||||
|
||||
pub fn from_db_str(s: &str) -> Option<Self> {
|
||||
match s {
|
||||
"idle" => Some(ProcessorJobStatus::Idle),
|
||||
"waiting" => Some(ProcessorJobStatus::Waiting),
|
||||
"ready" => Some(ProcessorJobStatus::Ready),
|
||||
"pending" => Some(ProcessorJobStatus::Pending),
|
||||
"running" => Some(ProcessorJobStatus::Running),
|
||||
"completed" => Some(ProcessorJobStatus::Completed),
|
||||
@@ -793,6 +796,7 @@ pub struct PostgresCache {
|
||||
#[derive(Debug, serde::Serialize, sqlx::FromRow)]
|
||||
pub struct SemanticSearchResult {
|
||||
pub id: i32,
|
||||
pub file_uuid: Option<String>, // Added for global search
|
||||
pub scene_order: i32,
|
||||
pub start_frame: i64,
|
||||
pub end_frame: i64,
|
||||
@@ -832,9 +836,9 @@ impl PostgresDb {
|
||||
|
||||
pub async fn new(database_url: &str) -> Result<Self> {
|
||||
let max_connections = std::env::var("DB_MAX_CONNECTIONS")
|
||||
.unwrap_or_else(|_| "10".to_string())
|
||||
.unwrap_or_else(|_| "30".to_string())
|
||||
.parse::<u32>()
|
||||
.unwrap_or(10);
|
||||
.unwrap_or(30);
|
||||
|
||||
let acquire_timeout_secs = std::env::var("DB_ACQUIRE_TIMEOUT")
|
||||
.unwrap_or_else(|_| "60".to_string())
|
||||
@@ -908,7 +912,7 @@ impl PostgresDb {
|
||||
sqlx::query("CREATE INDEX IF NOT EXISTS idx_chunk_file ON chunk(file_uuid)")
|
||||
.execute(pool)
|
||||
.await?;
|
||||
sqlx::query("CREATE INDEX IF NOT EXISTS idx_chunks_type ON chunks(chunk_type)")
|
||||
sqlx::query("CREATE INDEX IF NOT EXISTS idx_chunks_type ON chunk(chunk_type)")
|
||||
.execute(pool)
|
||||
.await?;
|
||||
|
||||
@@ -921,8 +925,17 @@ impl PostgresDb {
|
||||
// Processor Results
|
||||
sqlx::query("CREATE TABLE IF NOT EXISTS processor_results (id SERIAL PRIMARY KEY, job_id INTEGER, file_uuid VARCHAR(255) NOT NULL, processor VARCHAR(64), processor_type VARCHAR(64) NOT NULL, status VARCHAR(32) DEFAULT 'pending', result JSONB, error_message TEXT, started_at TIMESTAMP WITH TIME ZONE, completed_at TIMESTAMP WITH TIME ZONE, created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP)").execute(pool).await?;
|
||||
|
||||
// Talents & Identity Bindings
|
||||
// Talents
|
||||
sqlx::query("CREATE TABLE IF NOT EXISTS talents (id BIGSERIAL PRIMARY KEY, real_name VARCHAR(255) NOT NULL UNIQUE, actor_name VARCHAR(255), voice_embedding TEXT, face_embedding TEXT, metadata JSONB DEFAULT '{}', created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP)").execute(pool).await?;
|
||||
|
||||
// Identity History (Undo/Redo Support - for PATCH operations only)
|
||||
// Create before identity_bindings to avoid dependency issues
|
||||
let history_table = schema::table_name("identity_history");
|
||||
sqlx::query(&format!("CREATE TABLE IF NOT EXISTS {} (id BIGSERIAL PRIMARY KEY, identity_id INTEGER NOT NULL, operation VARCHAR(20) NOT NULL, before_snapshot JSONB, after_snapshot JSONB, is_undone BOOLEAN DEFAULT FALSE, undone_at TIMESTAMPTZ, user_id VARCHAR(100), user_source VARCHAR(50), created_at TIMESTAMPTZ DEFAULT NOW())", history_table)).execute(pool).await?;
|
||||
sqlx::query(&format!("CREATE INDEX IF NOT EXISTS idx_identity_history_identity_time ON {}(identity_id, created_at DESC)", history_table)).execute(pool).await?;
|
||||
sqlx::query(&format!("CREATE INDEX IF NOT EXISTS idx_identity_history_not_undone ON {}(identity_id, created_at DESC) WHERE NOT is_undone", history_table)).execute(pool).await?;
|
||||
|
||||
// Identity Bindings (depends on identities table)
|
||||
sqlx::query("CREATE TABLE IF NOT EXISTS identity_bindings (id BIGSERIAL PRIMARY KEY, identity_id BIGINT REFERENCES identities(id) ON DELETE CASCADE, identity_type VARCHAR(20) NOT NULL, identity_value VARCHAR(100) NOT NULL, metadata JSONB DEFAULT '{}', confidence DOUBLE PRECISION DEFAULT 1.0, created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP, UNIQUE(identity_id, identity_type, identity_value))").execute(pool).await?;
|
||||
|
||||
// API Keys
|
||||
@@ -955,7 +968,30 @@ impl PostgresDb {
|
||||
.execute(pool)
|
||||
.await?;
|
||||
|
||||
sqlx::query("DROP TRIGGER IF EXISTS chunks_search_vector_trigger ON chunks")
|
||||
sqlx::query(
|
||||
r#"CREATE OR REPLACE FUNCTION jsonb_deep_merge(base jsonb, override jsonb) RETURNS jsonb
|
||||
IMMUTABLE LANGUAGE plpgsql AS $$
|
||||
DECLARE
|
||||
result jsonb := COALESCE(base, '{}'::jsonb);
|
||||
key text;
|
||||
val jsonb;
|
||||
BEGIN
|
||||
FOR key, val IN SELECT * FROM jsonb_each(override)
|
||||
LOOP
|
||||
IF jsonb_typeof(val) = 'object' AND jsonb_typeof(result -> key) = 'object' THEN
|
||||
result := jsonb_set(result, ARRAY[key], jsonb_deep_merge(result -> key, val));
|
||||
ELSE
|
||||
result := jsonb_set(result, ARRAY[key], val);
|
||||
END IF;
|
||||
END LOOP;
|
||||
RETURN result;
|
||||
END;
|
||||
$$;"#,
|
||||
)
|
||||
.execute(pool)
|
||||
.await?;
|
||||
|
||||
sqlx::query("DROP TRIGGER IF EXISTS chunks_search_vector_trigger ON chunk")
|
||||
.execute(pool)
|
||||
.await?;
|
||||
|
||||
@@ -968,13 +1004,13 @@ impl PostgresDb {
|
||||
.await?;
|
||||
|
||||
// Chunks Rule 1
|
||||
sqlx::query("CREATE TABLE IF NOT EXISTS chunks_rule1 (id UUID PRIMARY KEY DEFAULT gen_random_uuid(), file_uuid VARCHAR(32) NOT NULL REFERENCES videos(uuid) ON DELETE CASCADE, start_frame BIGINT NOT NULL, end_frame BIGINT NOT NULL, content TEXT NOT NULL, speaker_id VARCHAR(50), created_at TIMESTAMPTZ DEFAULT NOW())").execute(pool).await?;
|
||||
sqlx::query("CREATE INDEX IF NOT EXISTS idx_chunks_rule1_asset ON chunks_rule1(file_uuid)")
|
||||
.execute(pool)
|
||||
.await?;
|
||||
// Temporarily disabled for debugging
|
||||
// sqlx::query("CREATE TABLE IF NOT EXISTS chunks_rule1 ...").execute(pool).await?;
|
||||
// sqlx::query("CREATE INDEX IF NOT EXISTS idx_chunks_rule1_asset ...").execute(pool).await?;
|
||||
|
||||
// Jobs (Legacy/P0)
|
||||
sqlx::query("CREATE TABLE IF NOT EXISTS jobs (id UUID PRIMARY KEY, file_uuid VARCHAR(32) NOT NULL REFERENCES videos(uuid) ON DELETE CASCADE, processor_list TEXT[], assigned_processor_id UUID, rule VARCHAR(20), status VARCHAR(20) DEFAULT 'QUEUED', total_frames BIGINT DEFAULT 0, processed_frames BIGINT DEFAULT 0, error_message TEXT, created_at TIMESTAMPTZ DEFAULT NOW(), updated_at TIMESTAMPTZ DEFAULT NOW())").execute(pool).await?;
|
||||
tracing::info!("Creating jobs table...");
|
||||
sqlx::query("CREATE TABLE IF NOT EXISTS jobs (id UUID PRIMARY KEY, file_uuid VARCHAR(32) NOT NULL REFERENCES videos(file_uuid) ON DELETE CASCADE, processor_list TEXT[], assigned_processor_id UUID, rule VARCHAR(20), status VARCHAR(20) DEFAULT 'QUEUED', total_frames BIGINT DEFAULT 0, processed_frames BIGINT DEFAULT 0, error_message TEXT, created_at TIMESTAMPTZ DEFAULT NOW(), updated_at TIMESTAMPTZ DEFAULT NOW())").execute(pool).await?;
|
||||
sqlx::query("CREATE INDEX IF NOT EXISTS idx_jobs_status ON jobs(status)")
|
||||
.execute(pool)
|
||||
.await?;
|
||||
@@ -1372,10 +1408,10 @@ impl PostgresDb {
|
||||
|
||||
let pre_chunks = schema::table_name("pre_chunks");
|
||||
sqlx::query(&format!(
|
||||
"DELETE FROM {} WHERE file_uuid = $1::uuid",
|
||||
"DELETE FROM {} WHERE REPLACE(file_uuid, '-', '') = $1",
|
||||
pre_chunks
|
||||
))
|
||||
.bind(uuid)
|
||||
.bind(&uuid.replace('-', ""))
|
||||
.execute(&self.pool)
|
||||
.await?;
|
||||
|
||||
@@ -1455,22 +1491,48 @@ impl PostgresDb {
|
||||
// Convert to i64 for monitor_jobs.video_id (BIGINT)
|
||||
let video_id_i64 = video_id.map(|v| v as i64);
|
||||
|
||||
let row = sqlx::query(
|
||||
&format!(
|
||||
r#"
|
||||
INSERT INTO {} (uuid, video_path, status, video_id)
|
||||
VALUES ($1, $2, 'pending', $3)
|
||||
ON CONFLICT (uuid) DO UPDATE SET status = 'pending', updated_at = NOW()
|
||||
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
|
||||
// Check if job already exists
|
||||
let existing_id: Option<i32> =
|
||||
sqlx::query_scalar(&format!("SELECT id FROM {} WHERE uuid = $1", jobs_table))
|
||||
.bind(uuid)
|
||||
.fetch_optional(&self.pool)
|
||||
.await?;
|
||||
|
||||
let row = if let Some(job_id) = existing_id {
|
||||
// Update existing job
|
||||
sqlx::query(
|
||||
&format!(
|
||||
r#"
|
||||
UPDATE {} SET status = 'pending', video_path = $1, video_id = $2, updated_at = NOW()
|
||||
WHERE id = $3
|
||||
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
|
||||
)
|
||||
)
|
||||
)
|
||||
.bind(uuid)
|
||||
.bind(video_path)
|
||||
.bind(video_id_i64)
|
||||
.fetch_one(&self.pool)
|
||||
.await?;
|
||||
.bind(video_path)
|
||||
.bind(video_id_i64)
|
||||
.bind(job_id)
|
||||
.fetch_one(&self.pool)
|
||||
.await?
|
||||
} else {
|
||||
// Insert new job
|
||||
sqlx::query(
|
||||
&format!(
|
||||
r#"
|
||||
INSERT INTO {} (uuid, video_path, status, video_id)
|
||||
VALUES ($1, $2, 'pending', $3)
|
||||
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
|
||||
)
|
||||
)
|
||||
.bind(uuid)
|
||||
.bind(video_path)
|
||||
.bind(video_id_i64)
|
||||
.fetch_one(&self.pool)
|
||||
.await?
|
||||
};
|
||||
|
||||
let status_str: String = row.get(3);
|
||||
let status =
|
||||
@@ -2062,7 +2124,7 @@ impl PostgresDb {
|
||||
let results = sqlx::query_as::<_, SemanticSearchResult>(
|
||||
&format!(
|
||||
"SELECT \
|
||||
id, id as scene_order, \
|
||||
id, file_uuid, id as scene_order, \
|
||||
(start_time * fps)::bigint as start_frame, (end_time * fps)::bigint as end_frame, \
|
||||
fps, start_time, end_time, \
|
||||
COALESCE(summary_text, text_content, '') as summary, \
|
||||
@@ -2084,6 +2146,41 @@ impl PostgresDb {
|
||||
Ok(results)
|
||||
}
|
||||
|
||||
/// Global semantic search across all files
|
||||
pub async fn search_parent_chunks_semantic_global(
|
||||
&self,
|
||||
query_vector: &[f32],
|
||||
limit: usize,
|
||||
) -> Result<Vec<SemanticSearchResult>> {
|
||||
// Convert Vec<f32> to JSON string for vector cast
|
||||
let vector_json = serde_json::to_string(query_vector)
|
||||
.map_err(|e| anyhow::anyhow!("Vector serialize error: {}", e))?;
|
||||
|
||||
let chunk_table = schema::table_name("chunk");
|
||||
let results = sqlx::query_as::<_, SemanticSearchResult>(
|
||||
&format!(
|
||||
"SELECT \
|
||||
id, file_uuid, id as scene_order, \
|
||||
(start_time * fps)::bigint as start_frame, (end_time * fps)::bigint as end_frame, \
|
||||
fps, start_time, end_time, \
|
||||
COALESCE(summary_text, text_content, '') as summary, \
|
||||
metadata, \
|
||||
(1 - (embedding <=> $1::vector)) as similarity \
|
||||
FROM {} \
|
||||
WHERE chunk_type IN ('sentence', 'story_parent', 'llm_parent') AND embedding IS NOT NULL \
|
||||
ORDER BY embedding <=> $1::vector \
|
||||
LIMIT $2",
|
||||
chunk_table
|
||||
),
|
||||
)
|
||||
.bind(&vector_json)
|
||||
.bind(limit as i64)
|
||||
.fetch_all(&self.pool)
|
||||
.await?;
|
||||
|
||||
Ok(results)
|
||||
}
|
||||
|
||||
/// Get children for a list of parent IDs
|
||||
pub async fn get_children_for_parents(
|
||||
&self,
|
||||
@@ -2368,7 +2465,6 @@ impl PostgresDb {
|
||||
"story" | "story_parent" | "story_child" => {
|
||||
crate::core::chunk::types::ChunkType::Story
|
||||
}
|
||||
"visual" => crate::core::chunk::types::ChunkType::Visual,
|
||||
_ => crate::core::chunk::types::ChunkType::Story,
|
||||
};
|
||||
let start_frame = (st * fps).round() as i64;
|
||||
@@ -2503,7 +2599,7 @@ impl PostgresDb {
|
||||
id: r.get("id"),
|
||||
job_id: r.get("job_id"),
|
||||
processor_type: crate::core::db::ProcessorType::from_db_str(ptype)
|
||||
.unwrap_or(crate::core::db::ProcessorType::Asr),
|
||||
.unwrap_or(crate::core::db::ProcessorType::Cut),
|
||||
status: crate::core::db::ProcessorJobStatus::from_db_str(st)
|
||||
.unwrap_or(crate::core::db::ProcessorJobStatus::Pending),
|
||||
started_at: r
|
||||
@@ -2549,7 +2645,7 @@ impl PostgresDb {
|
||||
id: r.get("id"),
|
||||
job_id: r.get("job_id"),
|
||||
processor_type: crate::core::db::ProcessorType::from_db_str(ptype)
|
||||
.unwrap_or(crate::core::db::ProcessorType::Asr),
|
||||
.unwrap_or(crate::core::db::ProcessorType::Cut),
|
||||
status: crate::core::db::ProcessorJobStatus::from_db_str(st)
|
||||
.unwrap_or(crate::core::db::ProcessorJobStatus::Pending),
|
||||
started_at: r
|
||||
@@ -2954,6 +3050,65 @@ impl PostgresDb {
|
||||
.collect())
|
||||
}
|
||||
|
||||
pub async fn get_file_faces(
|
||||
&self,
|
||||
file_uuid: &str,
|
||||
limit: i32,
|
||||
offset: i64,
|
||||
) -> Result<Vec<super::FileFaceRecord>> {
|
||||
let fd_table = schema::table_name("face_detections");
|
||||
let video_table = schema::table_name("videos");
|
||||
let id_table = schema::table_name("identities");
|
||||
let st_table = schema::table_name("strangers");
|
||||
use sqlx::Row;
|
||||
let rows = sqlx::query(&format!(
|
||||
"SELECT fd.id::bigint as id, fd.file_uuid, \
|
||||
fd.frame_number::bigint as frame_number, \
|
||||
(fd.frame_number::float8 / NULLIF(v.fps, 0)) as timestamp_secs, \
|
||||
fd.face_id, fd.trace_id, \
|
||||
fd.x::float8 as x, fd.y::float8 as y, \
|
||||
fd.width::float8 as width, fd.height::float8 as height, \
|
||||
fd.confidence::float8 as confidence, \
|
||||
fd.identity_id, fd.stranger_id, \
|
||||
i.uuid::text as identity_uuid, i.name as identity_name, \
|
||||
s.metadata as stranger_metadata \
|
||||
FROM {} fd \
|
||||
JOIN {} v ON v.file_uuid = fd.file_uuid \
|
||||
LEFT JOIN {} i ON i.id = fd.identity_id \
|
||||
LEFT JOIN {} s ON s.id = fd.stranger_id \
|
||||
WHERE fd.file_uuid = $1 \
|
||||
ORDER BY fd.frame_number, fd.trace_id \
|
||||
LIMIT $2 OFFSET $3",
|
||||
fd_table, video_table, id_table, st_table
|
||||
))
|
||||
.bind(file_uuid)
|
||||
.bind(limit)
|
||||
.bind(offset)
|
||||
.fetch_all(&self.pool)
|
||||
.await?;
|
||||
Ok(rows
|
||||
.into_iter()
|
||||
.map(|r| super::FileFaceRecord {
|
||||
id: r.get("id"),
|
||||
file_uuid: r.get("file_uuid"),
|
||||
frame_number: r.get("frame_number"),
|
||||
timestamp_secs: r.get("timestamp_secs"),
|
||||
face_id: r.get("face_id"),
|
||||
trace_id: r.get("trace_id"),
|
||||
x: r.get("x"),
|
||||
y: r.get("y"),
|
||||
width: r.get("width"),
|
||||
height: r.get("height"),
|
||||
confidence: r.get("confidence"),
|
||||
identity_id: r.get("identity_id"),
|
||||
stranger_id: r.get("stranger_id"),
|
||||
identity_uuid: r.get("identity_uuid"),
|
||||
identity_name: r.get("identity_name"),
|
||||
stranger_metadata: r.get("stranger_metadata"),
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
pub async fn get_identity_chunks(
|
||||
&self,
|
||||
uuid_str: &str,
|
||||
@@ -2963,16 +3118,48 @@ impl PostgresDb {
|
||||
let id_table = schema::table_name("identities");
|
||||
let fd_table = schema::table_name("face_detections");
|
||||
let chunk_table = schema::table_name("chunk");
|
||||
let ib_table = schema::table_name("identity_bindings");
|
||||
let pc_table = schema::table_name("pre_chunks");
|
||||
use sqlx::Row;
|
||||
let rows = sqlx::query(
|
||||
&format!("SELECT c.file_uuid, c.chunk_id, (c.start_time * c.fps)::bigint as start_frame, (c.end_time * c.fps)::bigint as end_frame, c.fps, c.start_time, c.end_time, c.text_content, 'sentence' as chunk_type \
|
||||
FROM {} c JOIN {} fd ON fd.file_uuid = c.file_uuid \
|
||||
AND fd.frame_number BETWEEN c.start_frame AND c.end_frame \
|
||||
WHERE fd.identity_id = (SELECT id FROM {} WHERE REPLACE(uuid::text, '-', '') = $1) \
|
||||
GROUP BY c.file_uuid, c.chunk_id, c.start_frame, c.end_frame, c.fps, c.start_time, c.end_time, c.text_content LIMIT $2 OFFSET $3", chunk_table, fd_table, id_table)
|
||||
)
|
||||
.bind(uuid_str).bind(limit).bind(offset)
|
||||
.fetch_all(&self.pool).await?;
|
||||
let subq = format!(
|
||||
"SELECT id FROM {} WHERE REPLACE(uuid::text, '-', '') = $1",
|
||||
id_table
|
||||
);
|
||||
let rows = sqlx::query(&format!(
|
||||
"SELECT c.file_uuid, c.chunk_id, \
|
||||
(c.start_time * c.fps)::bigint as start_frame, \
|
||||
(c.end_time * c.fps)::bigint as end_frame, \
|
||||
c.fps, c.start_time, c.end_time, c.text_content, \
|
||||
'sentence' as chunk_type \
|
||||
FROM {} c \
|
||||
JOIN {} fd ON fd.file_uuid = c.file_uuid \
|
||||
AND fd.frame_number BETWEEN c.start_frame AND c.end_frame \
|
||||
WHERE fd.identity_id = ({}) \
|
||||
GROUP BY c.file_uuid, c.chunk_id, c.start_frame, c.end_frame, \
|
||||
c.fps, c.start_time, c.end_time, c.text_content \
|
||||
UNION ALL \
|
||||
SELECT c.file_uuid, c.chunk_id, \
|
||||
c.start_frame::bigint, c.end_frame::bigint, \
|
||||
c.fps, c.start_time, c.end_time, c.text_content, \
|
||||
'sentence' as chunk_type \
|
||||
FROM {} c \
|
||||
JOIN {} pc ON pc.file_uuid = c.file_uuid \
|
||||
AND pc.processor_type = 'asrx' \
|
||||
AND c.start_time <= (pc.data->>'timestamp')::double precision \
|
||||
AND c.end_time >= (pc.data->>'timestamp')::double precision \
|
||||
JOIN {} ib ON ib.identity_value = pc.data->>'speaker_id' \
|
||||
AND ib.identity_type = 'speaker' \
|
||||
AND ib.file_uuid = pc.file_uuid \
|
||||
WHERE ib.identity_id = ({}) \
|
||||
ORDER BY start_time \
|
||||
LIMIT $2 OFFSET $3",
|
||||
chunk_table, fd_table, subq, chunk_table, pc_table, ib_table, subq
|
||||
))
|
||||
.bind(uuid_str)
|
||||
.bind(limit)
|
||||
.bind(offset)
|
||||
.fetch_all(&self.pool)
|
||||
.await?;
|
||||
Ok(rows
|
||||
.into_iter()
|
||||
.map(|r| super::IdentityChunkRecord {
|
||||
@@ -2999,7 +3186,7 @@ impl PostgresDb {
|
||||
let clean = uuid_str.replace('-', "");
|
||||
use sqlx::Row;
|
||||
let row = sqlx::query(
|
||||
&format!("SELECT id::bigint as id, uuid::text, name, identity_type, source, status, metadata, reference_data, \
|
||||
&format!("SELECT id::bigint as id, uuid::text, name, identity_type, source, status, metadata, COALESCE(reference_data, '{{}}'::jsonb) as reference_data, \
|
||||
NULL::real[] as voice_embedding, NULL::real[] as identity_embedding, \
|
||||
face_embedding::real[] as face_embedding, \
|
||||
tmdb_id, tmdb_profile, created_at::timestamptz as created_at, NULL::timestamptz as updated_at \
|
||||
@@ -3343,7 +3530,6 @@ impl crate::core::db::ChunkStore for PostgresDb {
|
||||
"story" | "story_parent" | "story_child" => {
|
||||
crate::core::chunk::types::ChunkType::Story
|
||||
}
|
||||
"visual" => crate::core::chunk::types::ChunkType::Visual,
|
||||
_ => crate::core::chunk::types::ChunkType::Story,
|
||||
};
|
||||
let start_frame = (st * fps).round() as i64;
|
||||
@@ -3680,4 +3866,30 @@ mod tests {
|
||||
assert!(result.is_ok());
|
||||
assert_eq!(result.unwrap(), "__no_match__:*");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_check_videos_file_uuid_column() {
|
||||
let url = crate::core::config::DATABASE_URL.as_str();
|
||||
let pool = match PgPoolOptions::new().max_connections(1).connect(url).await {
|
||||
Ok(p) => p,
|
||||
Err(_) => {
|
||||
eprintln!("SKIP: DB unavailable");
|
||||
return;
|
||||
}
|
||||
};
|
||||
let schema = crate::core::config::DATABASE_SCHEMA.as_str();
|
||||
let cols: Vec<String> = sqlx::query_scalar(
|
||||
"SELECT column_name FROM information_schema.columns WHERE table_schema=$1 AND table_name='videos' ORDER BY ordinal_position"
|
||||
)
|
||||
.bind(schema)
|
||||
.fetch_all(&pool)
|
||||
.await
|
||||
.unwrap_or_default();
|
||||
eprintln!("videos columns in schema '{}': {:?}", schema, cols);
|
||||
assert!(
|
||||
cols.contains(&"file_uuid".to_string()),
|
||||
"videos must have file_uuid column, got: {:?}",
|
||||
cols
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -300,6 +300,38 @@ impl RedisClient {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn emit_processor_alert(
|
||||
&self,
|
||||
file_uuid: &str,
|
||||
processor_type: &str,
|
||||
alert_type: &str,
|
||||
message: &str,
|
||||
) -> Result<()> {
|
||||
let mut conn = self.get_conn_internal().await?;
|
||||
let prefix = REDIS_KEY_PREFIX.as_str();
|
||||
let channel = format!("{}processor:alerts", prefix);
|
||||
|
||||
let alert_json = serde_json::json!({
|
||||
"file_uuid": file_uuid,
|
||||
"processor_type": processor_type,
|
||||
"alert_type": alert_type,
|
||||
"message": message,
|
||||
"timestamp": chrono::Utc::now().to_rfc3339(),
|
||||
});
|
||||
|
||||
let _: usize = conn.publish(&channel, serde_json::to_string(&alert_json)?).await?;
|
||||
|
||||
tracing::warn!(
|
||||
"Processor alert: {} | {} | {} | {}",
|
||||
file_uuid,
|
||||
processor_type,
|
||||
alert_type,
|
||||
message
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn subscribe_anomaly_alerts(&self) -> Result<redis::aio::PubSub> {
|
||||
let mut pubsub = self.client.get_async_pubsub().await?;
|
||||
let prefix = REDIS_KEY_PREFIX.as_str();
|
||||
@@ -441,12 +473,27 @@ impl RedisClient {
|
||||
let key = format!("{}job:{}", prefix, uuid);
|
||||
let _: i32 = conn.del(&key).await?;
|
||||
|
||||
let processor_types = ["asr", "cut", "yolo", "ocr", "face", "pose", "asrx"];
|
||||
let processor_types = [
|
||||
"asr",
|
||||
"cut",
|
||||
"yolo",
|
||||
"ocr",
|
||||
"face",
|
||||
"pose",
|
||||
"asrx",
|
||||
"visual_chunk",
|
||||
"story",
|
||||
"tmdb_probe",
|
||||
"embedding",
|
||||
];
|
||||
for ptype in processor_types {
|
||||
let proc_key = format!("{}job:{}:processor:{}", prefix, uuid, ptype);
|
||||
let _: i32 = conn.del(&proc_key).await?;
|
||||
}
|
||||
|
||||
let progress_key = format!("{}progress:{}", prefix, uuid);
|
||||
let _: i32 = conn.del(&progress_key).await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -6,7 +6,6 @@ use std::time::Duration;
|
||||
use tokio::time::sleep;
|
||||
use tracing::{error, info, warn};
|
||||
|
||||
use crate::api::five_w1h_agent_api::run_5w1h_agent;
|
||||
use crate::api::identity_agent_api::run_identity_agent;
|
||||
use crate::core::chunk::{rule1_ingest, rule3_ingest};
|
||||
use crate::core::config::OUTPUT_DIR;
|
||||
@@ -22,6 +21,29 @@ use crate::worker::processor::{ProcessorPool, ProcessorTask};
|
||||
use crate::worker::resources::SystemResources;
|
||||
use sqlx::PgPool;
|
||||
|
||||
enum ConditionResult {
|
||||
Ready,
|
||||
Waiting(Vec<crate::core::db::ProcessorType>),
|
||||
}
|
||||
|
||||
fn check_dependencies(
|
||||
processor: crate::core::db::ProcessorType,
|
||||
completed: &[crate::core::db::ProcessorType],
|
||||
) -> ConditionResult {
|
||||
let deps = processor.dependencies();
|
||||
let missing: Vec<_> = deps
|
||||
.iter()
|
||||
.filter(|d| !completed.contains(d))
|
||||
.cloned()
|
||||
.collect();
|
||||
|
||||
if missing.is_empty() {
|
||||
ConditionResult::Ready
|
||||
} else {
|
||||
ConditionResult::Waiting(missing)
|
||||
}
|
||||
}
|
||||
|
||||
pub struct JobWorker {
|
||||
db: Arc<PostgresDb>,
|
||||
redis: Arc<RedisClient>,
|
||||
@@ -250,38 +272,6 @@ impl JobWorker {
|
||||
.collect()
|
||||
};
|
||||
|
||||
// 長影片動態調整：若 CUT 場景過長，Face 需在 ASR 之前執行
|
||||
if let Ok(Some(video)) = self.db.get_video_by_uuid(&job.uuid).await {
|
||||
// 條件：cut_done 且場景數 <= 3 且最長場景 > 600s（10分鐘）
|
||||
if video.cut_done && video.cut_count <= 3 && video.cut_max_duration > 600.0 {
|
||||
info!(
|
||||
"[DYNAMIC] Long cut detected: {} scenes, max_dur={:.0}s for {}. Moving Face before ASR.",
|
||||
video.cut_count, video.cut_max_duration, job.uuid
|
||||
);
|
||||
// 確保 Face 在 ASR 之前
|
||||
if let Some(asr_pos) = processors_to_run
|
||||
.iter()
|
||||
.position(|p| matches!(p, crate::core::db::ProcessorType::Asr))
|
||||
{
|
||||
if let Some(face_pos) = processors_to_run
|
||||
.iter()
|
||||
.position(|p| matches!(p, crate::core::db::ProcessorType::Face))
|
||||
{
|
||||
if face_pos > asr_pos {
|
||||
// 將 Face 移到 ASR 前面
|
||||
let face = processors_to_run.remove(face_pos);
|
||||
let insert_pos = processors_to_run
|
||||
.iter()
|
||||
.position(|p| matches!(p, crate::core::db::ProcessorType::Asr))
|
||||
.unwrap();
|
||||
processors_to_run.insert(insert_pos, face);
|
||||
info!("[DYNAMIC] Reordered processors: Face now ahead of ASR");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let total_processor_types = processors_to_run.len() as i32;
|
||||
|
||||
// Get video total_frames for progress tracking
|
||||
@@ -381,45 +371,73 @@ impl JobWorker {
|
||||
// Load output file and store to pre_chunks
|
||||
if let Ok(json_str) = std::fs::read_to_string(&output_path) {
|
||||
let store_result = match processor_type {
|
||||
crate::core::db::ProcessorType::Asr => {
|
||||
if let Ok(result) = serde_json::from_str::<crate::core::processor::AsrResult>(&json_str) {
|
||||
ProcessorPool::store_asr_chunks(&self.db, &job.uuid, &result).await
|
||||
} else { Ok(()) }
|
||||
}
|
||||
crate::core::db::ProcessorType::Asrx => {
|
||||
if let Ok(result) = serde_json::from_str::<crate::core::processor::AsrxResult>(&json_str) {
|
||||
if let Ok(result) = serde_json::from_str::<
|
||||
crate::core::processor::AsrxResult,
|
||||
>(&json_str)
|
||||
{
|
||||
ProcessorPool::store_asrx_chunks(&self.db, &job.uuid, &result).await
|
||||
} else { Ok(()) }
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
crate::core::db::ProcessorType::Cut => {
|
||||
if let Ok(result) = serde_json::from_str::<crate::core::processor::CutResult>(&json_str) {
|
||||
if let Ok(result) =
|
||||
serde_json::from_str::<crate::core::processor::CutResult>(&json_str)
|
||||
{
|
||||
ProcessorPool::store_cut_chunks(&self.db, &job.uuid, &result).await
|
||||
} else { Ok(()) }
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
crate::core::db::ProcessorType::Yolo => {
|
||||
if let Ok(result) = serde_json::from_str::<crate::core::processor::YoloResult>(&json_str) {
|
||||
if let Ok(result) = serde_json::from_str::<
|
||||
crate::core::processor::YoloResult,
|
||||
>(&json_str)
|
||||
{
|
||||
ProcessorPool::store_yolo_chunks(&self.db, &job.uuid, &result).await
|
||||
} else { Ok(()) }
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
crate::core::db::ProcessorType::Ocr => {
|
||||
if let Ok(result) = serde_json::from_str::<crate::core::processor::OcrResult>(&json_str) {
|
||||
if let Ok(result) =
|
||||
serde_json::from_str::<crate::core::processor::OcrResult>(&json_str)
|
||||
{
|
||||
ProcessorPool::store_ocr_chunks(&self.db, &job.uuid, &result).await
|
||||
} else { Ok(()) }
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
crate::core::db::ProcessorType::Face => {
|
||||
if let Ok(result) = serde_json::from_str::<crate::core::processor::FaceResult>(&json_str) {
|
||||
if let Ok(result) = serde_json::from_str::<
|
||||
crate::core::processor::FaceResult,
|
||||
>(&json_str)
|
||||
{
|
||||
ProcessorPool::store_face_chunks(&self.db, &job.uuid, &result).await
|
||||
} else { Ok(()) }
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
crate::core::db::ProcessorType::Pose => {
|
||||
if let Ok(result) = serde_json::from_str::<crate::core::processor::PoseResult>(&json_str) {
|
||||
if let Ok(result) = serde_json::from_str::<
|
||||
crate::core::processor::PoseResult,
|
||||
>(&json_str)
|
||||
{
|
||||
ProcessorPool::store_pose_chunks(&self.db, &job.uuid, &result).await
|
||||
} else { Ok(()) }
|
||||
} else {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
_ => Ok(()),
|
||||
};
|
||||
if let Err(e) = store_result {
|
||||
error!("Failed to store {} chunks for {}: {}", processor_type.as_str(), job.uuid, e);
|
||||
error!(
|
||||
"Failed to store {} chunks for {}: {}",
|
||||
processor_type.as_str(),
|
||||
job.uuid,
|
||||
e
|
||||
);
|
||||
}
|
||||
}
|
||||
started_count += 1;
|
||||
@@ -459,10 +477,28 @@ impl JobWorker {
|
||||
continue;
|
||||
}
|
||||
ProcessorJobStatus::Failed => {
|
||||
if result.retry_count >= 3 {
|
||||
info!(
|
||||
"Processor {} failed {} times, max retries reached (3), skipping",
|
||||
processor_type.as_str(),
|
||||
result.retry_count
|
||||
);
|
||||
started_count += 1;
|
||||
continue;
|
||||
}
|
||||
info!(
|
||||
"Processor {} previously failed, retrying",
|
||||
processor_type.as_str()
|
||||
"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")
|
||||
))
|
||||
.bind(job.id)
|
||||
.bind(processor_type.as_str())
|
||||
.execute(self.db.pool())
|
||||
.await;
|
||||
}
|
||||
ProcessorJobStatus::Running => {
|
||||
info!(
|
||||
@@ -472,8 +508,11 @@ impl JobWorker {
|
||||
started_count += 1;
|
||||
continue;
|
||||
}
|
||||
// Skipped 不視為 terminal — 允許重新啟動
|
||||
ProcessorJobStatus::Skipped | ProcessorJobStatus::Pending => {
|
||||
ProcessorJobStatus::Idle
|
||||
| ProcessorJobStatus::Waiting
|
||||
| ProcessorJobStatus::Ready
|
||||
| ProcessorJobStatus::Skipped
|
||||
| ProcessorJobStatus::Pending => {
|
||||
// Continue to start processor
|
||||
}
|
||||
}
|
||||
@@ -531,6 +570,26 @@ impl JobWorker {
|
||||
processor_type.as_str(),
|
||||
deps.iter().map(|d| d.as_str()).collect::<Vec<_>>(),
|
||||
);
|
||||
let missing_deps: Vec<String> = deps
|
||||
.iter()
|
||||
.filter(|d| !matches!(
|
||||
result_map.get(d).map(|r| &r.status),
|
||||
Some(ProcessorJobStatus::Completed)
|
||||
))
|
||||
.map(|d| d.as_str().to_string())
|
||||
.collect();
|
||||
if let Err(e) = self
|
||||
.redis
|
||||
.emit_processor_alert(
|
||||
&job.uuid,
|
||||
processor_type.as_str(),
|
||||
"dependency_not_met",
|
||||
&format!("Waiting for: {}", missing_deps.join(", ")),
|
||||
)
|
||||
.await
|
||||
{
|
||||
error!("Failed to emit processor alert: {}", e);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
}
|
||||
@@ -669,21 +728,10 @@ impl JobWorker {
|
||||
"SELECT 1 FROM {chunk_t} WHERE file_uuid = '{fu}' AND chunk_type = 'cut' LIMIT 1"
|
||||
));
|
||||
let trace = check!(&format!("SELECT COUNT(DISTINCT trace_id) FROM {fd_t} WHERE file_uuid = '{fu}' AND trace_id IS NOT NULL"));
|
||||
let tkg = check!(&format!(
|
||||
"SELECT 1 FROM {} WHERE file_uuid = '{fu}' LIMIT 1",
|
||||
schema::table_name("tkg_nodes")
|
||||
));
|
||||
let scene_meta = std::path::Path::new(&format!(
|
||||
"{}/{fu}.scene_meta.json",
|
||||
crate::core::config::OUTPUT_DIR.as_str()
|
||||
))
|
||||
.exists();
|
||||
let five_w1h = check!(&format!("SELECT 1 FROM {chunk_t} WHERE file_uuid = '{fu}' AND chunk_type = 'cut' AND summary_text IS NOT NULL AND summary_text != '' LIMIT 1"));
|
||||
|
||||
let all_ok = rule1 && vector && rule3 && trace && tkg && scene_meta && five_w1h;
|
||||
let all_ok = rule1 && vector && rule3 && trace;
|
||||
if !all_ok {
|
||||
tracing::info!(
|
||||
"[Ingestion] waiting: rule1={rule1} vector={vector} rule3={rule3} trace={trace} tkg={tkg} scene={scene_meta} 5w1h={five_w1h}"
|
||||
"[Ingestion] waiting: rule1={rule1} vector={vector} rule3={rule3} trace={trace}"
|
||||
);
|
||||
}
|
||||
all_ok
|
||||
@@ -708,7 +756,7 @@ impl JobWorker {
|
||||
// 例如：Rule 1 只需 ASR+ASRX 完成即可觸發，不須等 face/pose/story 完成
|
||||
|
||||
// 定義必要 processor（必須完成的才算 job 成功）
|
||||
let essential_processors = ["cut", "asr", "yolo"];
|
||||
let essential_processors = ["cut", "asrx", "yolo"];
|
||||
|
||||
let essential_completed = essential_processors.iter().all(|ep| {
|
||||
results.iter().any(|r| {
|
||||
@@ -763,20 +811,16 @@ impl JobWorker {
|
||||
.map(|r| r.processor_type.as_str().to_string())
|
||||
.collect();
|
||||
|
||||
// Check prerequisites for post-processing triggers
|
||||
let has_asr = completed_processors.iter().any(|p| p == "asr");
|
||||
let has_asrx = completed_processors.iter().any(|p| p == "asrx");
|
||||
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");
|
||||
|
||||
// Update processor arrays in job record
|
||||
self.db
|
||||
.update_job_processors_arrays(job_id, completed_processors, failed_processors.clone())
|
||||
.await?;
|
||||
|
||||
// 🚀 P1 Trigger: Rule 1 Chunking（僅需 ASR + ASRX）
|
||||
if has_asr && has_asrx {
|
||||
if has_asrx {
|
||||
info!("📝 Prerequisites met for Rule 1 Chunking. Starting ingestion...");
|
||||
let db_clone = self.db.clone();
|
||||
let uuid_clone = uuid.to_string();
|
||||
@@ -787,7 +831,6 @@ impl JobWorker {
|
||||
match rule1_ingest::execute_rule1(&db_clone, &uuid_clone, fps).await {
|
||||
Ok(count) => {
|
||||
info!("✅ Rule 1 Ingestion completed: {} chunks inserted.", count);
|
||||
// Automatically vectorize new sentence chunks
|
||||
if count > 0 {
|
||||
info!(
|
||||
"📝 Starting automatic vectorize for {} chunks...",
|
||||
@@ -802,7 +845,6 @@ impl JobWorker {
|
||||
);
|
||||
}
|
||||
}
|
||||
// Phase 1 release: sentence chunk embedding 交付
|
||||
info!("📦 Phase 1 release packaging...");
|
||||
let executor = match crate::core::processor::PythonExecutor::new() {
|
||||
Ok(ex) => ex,
|
||||
@@ -836,10 +878,8 @@ impl JobWorker {
|
||||
});
|
||||
}
|
||||
|
||||
// Rule 3 / Trace / Identity 需要 all_completed（含非必要 processor 失敗也可）
|
||||
if all_completed {
|
||||
// 🚀 P1 Trigger: Rule 3 Scene Chunking
|
||||
if has_cut && has_asr {
|
||||
if has_cut {
|
||||
info!("📝 Prerequisites met for Rule 3 Scene Chunking. Starting ingestion...");
|
||||
let db_clone = self.db.clone();
|
||||
let uuid_clone = uuid.to_string();
|
||||
@@ -881,16 +921,6 @@ impl JobWorker {
|
||||
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);
|
||||
}
|
||||
|
||||
// Generate trace chunks from face_detections + ASR text
|
||||
info!("📝 Generating trace chunks...");
|
||||
match crate::core::chunk::trace_ingest::ingest_traces(
|
||||
@@ -902,26 +932,6 @@ impl JobWorker {
|
||||
Ok(n) => info!("✅ {} trace chunks created for {}", n, uuid_clone),
|
||||
Err(e) => error!("❌ Trace chunk ingestion failed: {}", e),
|
||||
}
|
||||
|
||||
// Build Temporal Knowledge Graph (TKG) — native Rust
|
||||
info!("📝 Building TKG graph (Rust)...");
|
||||
let output_dir = std::env::var("MOMENTRY_OUTPUT_DIR")
|
||||
.unwrap_or_else(|_| ".".to_string());
|
||||
match crate::core::processor::tkg::build_tkg(
|
||||
db_clone.as_ref(),
|
||||
&uuid_clone,
|
||||
&output_dir,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(r) => info!(
|
||||
"✅ TKG built for {}: {} face, {} obj, {} spk, {} co, {} sf, {} ff edges",
|
||||
uuid_clone,
|
||||
r.face_trace_nodes, r.object_nodes, r.speaker_nodes,
|
||||
r.co_occurrence_edges, r.speaker_face_edges, r.face_face_edges,
|
||||
),
|
||||
Err(e) => error!("❌ TKG build failed for {}: {}", uuid_clone, e),
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
error!("❌ Face trace + DB store failed for {}: {}", uuid_clone, e)
|
||||
@@ -1011,47 +1021,9 @@ impl JobWorker {
|
||||
});
|
||||
}
|
||||
|
||||
// 🚀 P4 Trigger: 5W1H Agent (after Rule 3 completion)
|
||||
if has_cut && has_asr {
|
||||
info!("📝 Prerequisites met for 5W1H Agent. Starting...");
|
||||
let db_clone = self.db.clone();
|
||||
let uuid_clone = uuid.to_string();
|
||||
tokio::spawn(async move {
|
||||
tokio::time::sleep(tokio::time::Duration::from_secs(30)).await;
|
||||
match run_5w1h_agent(&db_clone, &uuid_clone).await {
|
||||
Ok(()) => {
|
||||
info!("✅ 5W1H Agent completed for {}", uuid_clone);
|
||||
// Phase 2 release: full pipeline 交付
|
||||
info!("📦 Phase 2 release packaging...");
|
||||
let executor = match crate::core::processor::PythonExecutor::new() {
|
||||
Ok(ex) => ex,
|
||||
Err(e) => {
|
||||
error!("Failed PythonExecutor for release pack: {}", e);
|
||||
return;
|
||||
}
|
||||
};
|
||||
match executor
|
||||
.run(
|
||||
"release_pack.py",
|
||||
&["--phase", "2", "--file-uuid", &uuid_clone],
|
||||
None,
|
||||
"RELEASE_P2",
|
||||
Some(std::time::Duration::from_secs(120)),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(()) => info!("✅ Phase 2 release packaged for {}", uuid_clone),
|
||||
Err(e) => error!("❌ Phase 2 release pack failed: {}", e),
|
||||
}
|
||||
}
|
||||
Err(e) => error!("❌ 5W1H Agent failed for {}: {}", uuid_clone, e),
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
if !Self::ingestion_complete(self.db.pool(), uuid).await {
|
||||
info!(
|
||||
"Job {}: all 10 processors done, waiting for ingestion...",
|
||||
"Job {}: all processors done, waiting for ingestion...",
|
||||
job_id
|
||||
);
|
||||
return Ok(());
|
||||
|
||||
Reference in New Issue
Block a user