diff --git a/REFACTOR.md b/REFACTOR.md new file mode 100644 index 0000000..3221531 --- /dev/null +++ b/REFACTOR.md @@ -0,0 +1,486 @@ +# 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: +```json +{ + "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 + +```bash +# 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:** + +```python +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:** + +```python +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:** +```python +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):** +```python +{ + "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):** + +```python +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:** + +```json +{"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:** + +```python +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`: + ```yaml + 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: + ```yaml + 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:** + +```bash +# 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: + +```perl +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 diff --git a/api/app.pl b/api/app.pl index 86ed8ab..e469589 100644 --- a/api/app.pl +++ b/api/app.pl @@ -130,8 +130,22 @@ get '/api/v1/nexttask/:type' => sub { post '/api/v1/task/:task/complete' => sub { my $task = route_parameters->get("task"); - my $assign_key = body_parameters->get("assign_key"); - my $results = body_parameters->get("results"); + + # Accept both JSON body and form-encoded data for assign_key/results. + my ($assign_key, $results); + my $ct = request_header('Content-Type') || ''; + if ($ct eq 'application/json') { + my $body = decode_json(request->body()); + $assign_key = $body->{assign_key}; + $results = $body->{results}; + } else { + $assign_key = body_parameters->get("assign_key"); + $results = body_parameters->get("results"); + } + + unless ($assign_key) { + send_error("Missing assign_key", 400); + } # Verify the task is assigned to the worker my $sth = database->prepare("SELECT * FROM tasks WHERE id=? AND assign_key=? AND status='IN_PROGRESS'"); @@ -143,9 +157,10 @@ post '/api/v1/task/:task/complete' => sub { send_error("Task not assigned to this worker or not in progress", 403); } - # Update the task as completed + # Update the task as completed (store results as JSON text for MariaDB) + my $results_json = defined $results ? encode_json($results) : 'null'; my $update_sth = database->prepare("UPDATE tasks SET status='COMPLETED', results=?, updated_at=NOW() WHERE id=?"); - $update_sth->execute($results, $task); + $update_sth->execute($results_json, $task); if($update_sth->rows == 0) { send_error("Failed to complete task", 500); } diff --git a/docker-compose.yml b/docker-compose.yml index 6e03503..6230b8e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -7,7 +7,7 @@ services: MYSQL_ROOT_PASSWORD: ${DB_ROOT_PASSWORD:-rootpass} MYSQL_DATABASE: ${DB_NAME:-videodetect} MYSQL_USER: ${DB_USER:-videodetect} - MYSQL_PASSWORD: ${DB_PASSWORD:-videodetect123} + MYSQL_PASSWORD: ${DB_PASSWORD:-changeme_videodetect} ports: - "3306:3306" volumes: @@ -38,7 +38,7 @@ services: - DB_PORT=3306 - DB_NAME=videodetect - DB_USER=videodetect - - DB_PASSWORD=${DB_PASSWORD:-videodetect123} + - DB_PASSWORD=${DB_PASSWORD:-changeme_videodetect} - FLASK_ENV=production volumes: - ./ui:/app @@ -67,7 +67,7 @@ services: - DB_PORT=3306 - DB_NAME=videodetect - DB_USER=videodetect - - DB_PASSWORD=${DB_PASSWORD:-videodetect123} + - DB_PASSWORD=${DB_PASSWORD:-changeme_videodetect} - DANCER_ENVIRONMENT=production volumes: - ./api:/app