VideoDetect/REFACTOR.md

REFACTOR.md — Worker-to-API Refactor Plan

Context

VideoDetect workers currently run as long-lived processes (main.py → WorkerPool) that query the database directly via DBConnector. The re-architecture moves to independent, externally-invoked task processes that communicate through a REST API provided by the Perl api/app.pl module. This enables finer-grained scaling, resilience to individual process crashes, and separation of concerns between scanning (discovery) and processing (AI inference).

---

Target Architecture

┌───────────┐     /api/v1/nexttask/AISCAN     ┌─────────────┐
│  Scanner   │ ──────────────────────────────> │  Perl API    │
│  (long-    │ <────────────────────────────── │  (Dancer2)   │
│   lived)   │  task reservation + assign_key  ├─────────────┤
└───────────┘                                  │   DB         │
                                                 │             │
     ┌───────────────────────────────────────────┼──────────┐
     │                                           ▼          │
     │                                    ┌──────────────┐  │
     │       POST /api/v1/task/:id/complete              │  │
     │      ◄─────────────────────────────────────────►  │  │
     │                                          │         │
     │          ┌───────────────────────────────┘        │
     ▼          ▼                                        ▼
┌────────┐  ┌────────┐  ┌────────┐              ┌──────────────┐
│worker-1│  │worker-2│  │worker-N│              │   MariaDB     │
│(short- │  │(short- │  │(short- │              │              │
│ lived) │  │ lived) │  │ lived) │              └──────────────┘
└────────┘  └────────┘  └────────┘

Each worker process:

1. Starts → accepts --tasks N (default 1) via CLI argument
2. Loops up to N times: claim task → fetch video data → process → submit results
3. Exits cleanly after completing the requested count

---

API Reference (from api/app.pl)

GET /api/v1/nexttask/:type

Claims the next pending task of the given type and reserves it.

Field	Type	Example
task	object	Task record from DB
assign_key	string	"worker_338" (server-generated)

On success: HTTP 200 with JSON body:

{
    "task": {
        "id": 1,
        "video_id": 42,
        "task_type": "AISCAN",
        "status": "PENDING",
        "created_at": "2026-09-09T13:05:56",
        "updated_at": "2026-09-09T14:53:57",
        "assign_key": null,
        "results": null,
        "assigned_at": "2026-09-09T14:43:31"
    },
    "assign_key": "worker_338"
}

On no tasks available: HTTP 404 with {"message":"No task"}.

NOTE: The Perl API marks the task IN_PROGRESS and sets assign_key + assigned_at atomically (UPDATE ... WHERE id=? AND status='PENDING'). This is safe for multiple workers racing to claim the same task.

---

GET /api/v1/video/:id

Returns full video metadata by ID.

Field	Type	Example
id	integer	1
file_path	string	"/data/The Fappening/Sextape - Alyson Hannigan (American actress - American pie).wmv"
file_size	integer	19358016
file_hash	string	"1ffd178e9ee23039aebffd79ddcbc88e983633edcb75062e6bd4c269f4d7bf94"
resolution_w	integer	640
resolution_h	integer	480
codec	string	"wmv1"
duration	float	102.499
last_scan_time	datetime	"2026-09-09T13:05:56"
created_at	datetime	"2026-09-09T13:05:56"
updated_at	datetime	"2026-09-09T13:05:56"

On not found: HTTP 404 with {"message":"No video"}.

---

POST /api/v1/task/:task/complete

Submits processing results for a claimed task. The API verifies the task is IN_PROGRESS and assigned to the given assign_key.

Required parameters (form-encoded or JSON body):

Field	Type	Description
assign_key	string	Must match the key returned by /nexttask
results	object	JSON-serializable dict to store in DB results column

On success: HTTP 200 with {"message":"Task completed successfully"}.

On mismatch or task not IN_PROGRESS: HTTP 403 with error message.

NOTE: The Perl API currently uses body_parameters->get("results"), which reads form-encoded data. For JSON body submission, the Dancer2 config may need serializer: JSON (which is already set in api/config.yml). However, body_parameters only parses form-encoded fields — JSON body content goes to $app->request->body. This needs verification before implementation.

---

Refactor Steps

Phase 1 — Create the Task Worker Module

Goal: A single entry-point script that can run as a standalone process.

File: src/task_worker.py (new)

Acceptance Criteria:

• Starts with python3 src/task_worker.py --tasks N (default 1 task)
• Prints startup log message showing number of tasks requested and model version
• Exits cleanly after processing the requested count

# Default: process 1 task and exit
python3 -m src.task_worker

# Explicit: process 5 tasks and exit
python3 -m src.task_worker --tasks 5

# Also supported via direct script invocation
python3 src/task_worker.py --tasks 10

CLI Implementation:

import argparse

parser = argparse.ArgumentParser(description="VideoDetect task worker")
parser.add_argument("--tasks", type=int, default=1,
                    help="Number of tasks to process before exiting (default: 1)")
args = parser.parse_args()

---

Phase 2 — API Client Module

Goal: A thin HTTP client layer for communicating with the Perl API.

File: src/api_client.py (new)

Acceptance Criteria:

• Encapsulates all API calls in one class: ApiClient(base_url)
• Handles JSON serialization/deserialization automatically
• Raises a custom ApiError on HTTP errors (with status code and response body)
• Returns None for 404 responses (no task remaining) from /nexttask

API Error handling:

class ApiError(Exception):
    def __init__(self, status_code: int, message: str):
        self.status_code = status_code
        self.message = message
        super().__init__(f"API error {status_code}: {message}")

# Usage patterns:
try:
    response = client.get_next_task("AISCAN")
except ApiError as e:
    if e.status_code == 404:
        logger.info("No more tasks available")
        break  # exit processing loop
    raise

result = client.submit_results(task_id, assign_key, results_dict)

Methods to implement:

Method	Endpoint	Returns	Special handling
get_next_task(task_type)	GET /api/v1/nexttask/{type}	{"task": {...}, "assign_key": str} | None	Returns None on 404
get_video(video_id)	GET /api/v1/video/{id}	dict with video metadata	Raises ApiError on 404/5xx
submit_results(task_id, assign_key, results)	POST /api/v1/task/{task}/complete	dict with response	Passes assign_key + serialized results

Notes for implementation:

• Use Python's standard library only (urllib.request) if possible to avoid adding dependencies. If JSON body submission is needed for the Perl API, check whether body_parameters->get('results') in Dancer2 handles raw JSON (it typically does not — it expects form-encoded data).
• If form-encoding is required: requests.post(url, data={'assign_key': key, 'results': json.dumps(results)})
• If JSON body works: requests.post(url, json={'assign_key': key, 'results': results})
• Recommendation: Use the requests library (already in worker requirements.txt likely). Document both approaches and implement the one that matches your Dancer2 config.

---

Phase 3 — Implement the AI Processing Pipeline

Goal: Extract the core AI processing logic from orchestrator.py into a reusable function callable by the task worker.

File: New or updated module in src/ (tentatively src/task_processor.py)

Acceptance Criteria:

• Takes video metadata dict + file path as input
• Returns a results dict matching what will be stored in the DB results column
• Handles all error cases gracefully (returns errors instead of crashing)
• Logs all significant decisions at INFO level

Input:

video = {
    "id": 1,
    "file_path": "/data/input/...",
    "codec": "wmv1",
    "duration": 102.499,
    "resolution_w": 640,
    "resolution_h": 480,
    "file_size": 19358016,
    "file_hash": "abc...",
}

Output (results dict stored in DB):

{
    "status": "COMPLETED",           # or "FAILED" on error
    "confidence": 0.87,              # video-level confidence score
    "routing_decision": "MATCH",     # MATCH | REVIEW | SKIP
    "face_count": 142,               # total faces detected across all frames
    "frame_count": 23,               # frames extracted and processed
    "model_version": "v0.0.0-placeholder",
    "processing_time_seconds": 12.4,
    "error": None,                   # error message if FAILED
}

Processing flow (extracted from orchestrator.py):

def process_video(video: dict) -> dict:
    """Run the full AI scan pipeline on a single video.
    
    Returns results dict for submission via API.
    """
    start_time = time.time()
    file_path = video["file_path"]
    video_id = video["id"]
    
    try:
        # 1. Probe (if metadata is incomplete)
        prober = VideoProber(timeout=10)
        metadata = prober.probe(file_path)
        
        # 2. Extract frames to scratch
        scratch = ScratchManager(base_path="/scratch", video_id=str(video_id))
        frame_dir = scratch.ensure_frame_dir()
        sampler = FrameSampler(interval_seconds=30, quality=2)
        frame_paths = sampler.extract_frames(
            video_path=file_path, output_dir=str(frame_dir),
            duration=metadata.duration, resolution=(metadata.resolution_w, metadata.resolution_h)
        )
        
        # 3. Detect faces
        detector = FaceDetector(engine_path="/models/face_detector/face_detector.trt")
        detections_per_frame = detector.detect_faces(frame_paths)
        total_faces = sum(len(dets) for dets in detections_per_frame)
        
        # 4. Crop and classify (aggregate confidences from face crops)
        classifier = FaceClassifier(engine_path="/models/classifier/classifier.trt")
        # ... crop paths → classify → get confidence list
        
        # 5. Aggregate
        confidence = aggregate(confidences, strategy="max")
        
        # 6. Route
        routing = router.route(confidence)
        
        processing_time = time.time() - start_time
        
        return {
            "status": "COMPLETED",
            "confidence": round(confidence, 4),
            "routing_decision": routing,
            "face_count": total_faces,
            "frame_count": len(frame_paths),
            "model_version": get_model_version(),
            "processing_time_seconds": round(processing_time, 2),
            "error": None,
        }
        
    except Exception as exc:
        logger.error("Processing failed for video %d: %s", video_id, exc)
        return {
            "status": "FAILED",
            "confidence": 0.0,
            "routing_decision": "REVIEW",
            "face_count": 0,
            "frame_count": 0,
            "model_version": get_model_version(),
            "processing_time_seconds": round(time.time() - start_time, 2),
            "error": str(exc),
        }

Notes for implementation:

• Reuse existing classes from orchestrator.py (FaceDetector, FaceClassifier, FrameSampler, ScratchManager, VideoProber, aggregate, router) without refactoring them — just import and use them.
• GPU setup (torch CUDA, GPUMemoryManager) can be done once at module level or lazily inside process_video() to avoid overhead per task invocation.
• Model loading is expensive (~seconds). Consider lazy initialization or process-level caching via a singleton pattern.

---

Phase 4 — Wire It Together in task_worker.py

Goal: Combine the API client, processing pipeline, and CLI entry point into a working script.

Acceptance Criteria:

• Full end-to-end: claim → fetch → process → submit → repeat N times
• Exits with code 0 on success, non-zero on unrecoverable errors
• Logs startup/shutdown counts (tasks attempted, tasks succeeded, tasks failed)
• Graceful shutdown on SIGTERM/SIGINT (finish current task, then exit)

Expected lifecycle log output:

{"timestamp": "...", "level": "INFO", "message": "Task worker starting. Will process 3 task(s)."}
{"timestamp": "...", "level": "INFO", "message": "API client configured: base_url=http://localhost:8890"}
{"timestamp": "...", "level": "INFO", "message": "Claiming task 1 (video_id=42) for AISCAN"}
{"timestamp": "...", "level": "INFO", "message": "Processing video 42: /data/input/..."}
{"timestamp": "...", "level": "INFO", "message": "Video 42 complete: C=0.87 routing=MATCH faces=142 frames=23 time=12.4s"}
{"timestamp": "...", "level": "INFO", "message": "Submitting results for task 1 via API"}
{"timestamp": "...", "level": "INFO", "message": "Claiming task 2 (video_id=99) for AISCAN"}
{"timestamp": "...", "level": "INFO", "message": "Processing video 99: /data/input/..."}
{"timestamp": "...", "level": "INFO", "message": "Video 99 complete: C=0.31 routing=SKIP faces=0 frames=18 time=8.7s"}
{"timestamp": "...", "level": "INFO", "message": "Submitting results for task 2 via API"}
{"timestamp": "...", "level": "INFO", "message": "Claiming task 3 (video_id=157) for AISCAN"}
{"timestamp": "...", "level": "WARNING", "message": "Processing video 157 failed: ffprobe error: cannot decode stream"}
{"timestamp": "...", "level": "INFO", "message": "Submitting FAILED results for task 3 via API"}
{"timestamp": "...", "level": "INFO", "message": "No more tasks available (API returned 404)"}
{"timestamp": "...", "level": "INFO", "message": "Task worker finished: 3 attempted, 2 succeeded, 1 failed"}

Core loop:

for i in range(1, args.tasks + 1):
    logger.info("=== Processing task %d/%d ===", i, args.tasks)
    
    # Claim a task
    response = api.get_next_task("AISCAN")
    if response is None:
        logger.info("No more tasks available. Done.")
        break
    
    task = response["task"]
    assign_key = response["assign_key"]
    video_id = task["video_id"]
    
    # Fetch video metadata
    video = api.get_video(video_id)
    
    # Process
    results = process_video(video)
    
    # Submit
    api.submit_results(task["id"], assign_key, results)

logger.info("Finished: %d attempted, %d succeeded, %d failed", total_attempted, total_succeeded, total_failed)

---

Phase 5 — Update docker-compose.yml

Goal: Change the worker service from a long-lived WorkerPool to a short-lived task process.

Changes to docker-compose.yml worker section:

1. Update command to invoke the task worker instead of main.py:

   command: python3 -m src.task_worker --tasks 50
   

2. Add restart policy considerations — since workers are now short-lived, you have two options:

   • Option A (simpler): Run a single worker container with a large --tasks count (e.g., 100) so it processes many videos before exiting, then rely on Kubernetes/Cron/external scheduler to restart.
   • Option B (more flexible): Make --tasks configurable via environment variable:

     environment:
       - TASK_COUNT=${TASK_COUNT:-50}
     command: >
       python3 -m src.task_worker --tasks ${TASK_COUNT}
     

3. Remove the WorkerPool initialization from main.py — it becomes unused (or is removed entirely in a later cleanup).

---

Phase 6 — Verify API Compatibility

Goal: Confirm the Dancer2 Perl API correctly handles JSON body submission from Python's requests library on /api/v1/task/:task/complete.

Acceptance Criteria:

• POST with Content-Type: application/json body containing assign_key + results (as a dict, not a string) returns 200
• POST with form-encoded data also works as fallback
• The Perl API correctly stores the results JSON in the database

Test procedure:

# Step 1: Create a test task manually (via MySQL or via scanner)
mysql videodetect -e "INSERT INTO tasks (task_type, video_id, status) VALUES ('AISCAN', 1, 'PENDING');"

# Step 2: Claim the task via API
curl -s http://localhost:8890/api/v1/nexttask/AISCAN | python3 -m json.tool

# Step 3: Submit results (adjust video_id to a real one)
curl -X POST http://localhost:8890/api/v1/task/1/complete \
  -H "Content-Type: application/json" \
  -d '{"assign_key":"worker_999","results":{"confidence":0.87,"routing_decision":"MATCH"}}' | python3 -m json.tool

# Step 4: Verify results stored in DB
mysql videodetect -e "SELECT id, status, results FROM tasks WHERE id=1;"

If JSON body does NOT work, update the Dancer2 API to handle JSON explicitly:

post '/api/v1/task/:task/complete' => sub {
    my $task = route_parameters->get("task");
    
    # Handle both JSON body and form-encoded data
    my $body;
    if (request_content_type eq 'application/json') {
        use Dancer2::Core::Request::Entity;
        $body = decode_json(request_body);
    } else {
        $body = body_parameters->to_hash;
    }
    
    my $assign_key = $body->{assign_key};
    my $results    = $body->{results};
    
    # ... rest of the handler unchanged
};

---

Files Modified / Created Summary

File	Action	Description
src/task_worker.py	NEW	CLI entry point: --tasks N, loop claim→process→submit
src/api_client.py	NEW	HTTP client for the Perl API (ApiClient)
src/task_processor.py	NEW	process_video(video_dict) — core AI pipeline callable standalone
docker-compose.yml	MODIFY	Change worker command to invoke task_worker; add TASK_COUNT env var
api/app.pl	MAYBE MODIFY	Update /task/:task/complete to handle JSON body if Dancer2 doesn't support it natively
src/main.py	MODIFY (cleanup later)	Remove or deprecate WorkerPool usage (long-lived process no longer needed)
REFACTOR.md	CURRENT FILE	This plan document

---

Risks & Considerations

1. GPU model loading per invocation: Each new Python process loads the ONNX/TensorRT engine into GPU memory. With --tasks 1, this is wasteful. Mitigation: use larger --tasks values (e.g., 20-100) so amortization is favorable, or implement a local cache server pattern later.

2. Permanence of scratch space: The ScratchManager creates per-video temp files under /scratch. Since workers are now short-lived, ensure cleanup=True is always set (it is by default in the existing code).

3. API JSON body compatibility: Dancer2's body_parameters may not parse JSON bodies — it expects form-encoded data. This is the highest-risk item. Test Phase 6 early.

4. Task idempotency: If a worker crashes between processing and submitting results, the task remains IN_PROGRESS. The Perl API prevents re-claiming (the WHERE clause checks status='PENDING'). A manual SQL update or a /api/v1/task/:task/reset endpoint may be needed for recovery. This is outside the scope of this refactor but worth noting.

---

Suggested Order of Execution

1. Phase 6 first — verify API JSON compatibility (takes 5 minutes, unblocks everything)
2. Phase 2 — create api_client.py with minimal methods
3. Phase 3 — create task_processor.py by extracting from orchestrator.py
4. Phase 1 — create task_worker.py with CLI + loop
5. Phase 4 — wire together and test locally (python3 -m src.task_worker --tasks 3)
6. Phase 5 — update docker-compose.yml and deploy