245 lines
7.6 KiB
Perl
245 lines
7.6 KiB
Perl
#!/usr/bin/perl
|
|
use strict;
|
|
use Dancer2;
|
|
use Dancer2::Plugin::Database;
|
|
use Data::Dumper;
|
|
use SQL::SplitStatement;
|
|
use HTTP::Lite;
|
|
|
|
$SIG{'INT'} = sub { exit; };
|
|
|
|
|
|
hook before => sub {
|
|
my $origin = request_header('Origin') || 'http://localhost:8890';
|
|
my $method = request_header('Access-Control-Request-Method') || '';
|
|
my $headers = request_header('Access-Control-Request-Headers') || '';
|
|
|
|
response_header 'Access-Control-Allow-Origin' => $origin;
|
|
response_header 'Access-Control-Allow-Methods' => $method;
|
|
response_header 'Access-Control-Allow-Headers' => $headers;
|
|
};
|
|
|
|
|
|
options qr{/api/v1/.+} => sub {
|
|
return {};
|
|
};
|
|
|
|
get '/api/v1/hello' => sub {
|
|
my $user = session('user') || 'world';
|
|
return {'hello' => $user };
|
|
};
|
|
|
|
get '/api/v1/videos' => sub {
|
|
my $sth = database->prepare("SELECT id, file_path FROM videos ORDER BY id");
|
|
$sth->execute();
|
|
my $ref = $sth->fetchall_hashref('id');
|
|
$sth->finish();
|
|
|
|
return $ref;
|
|
};
|
|
|
|
get '/api/v1/video' => sub {
|
|
my $path = query_parameters->get("path");
|
|
my $sth = database->prepare("SELECT * FROM videos WHERE file_path=?");
|
|
$sth->execute($path);
|
|
my $ref = $sth->fetchrow_hashref();
|
|
$sth->finish();
|
|
|
|
unless(ref $ref) {
|
|
send_error("No video", 404);
|
|
}
|
|
|
|
return $ref;
|
|
};
|
|
|
|
get '/api/v1/video/:video' => sub {
|
|
my $video = route_parameters->get("video");
|
|
my $sth = database->prepare("SELECT * FROM videos WHERE id=?");
|
|
$sth->execute($video);
|
|
my $ref = $sth->fetchrow_hashref();
|
|
$sth->finish();
|
|
|
|
unless(ref $ref) {
|
|
send_error("No video", 404);
|
|
}
|
|
|
|
return $ref;
|
|
};
|
|
|
|
post '/api/v1/video' => sub {
|
|
my $file_path = body_parameters->get("file_path");
|
|
my $file_hash = body_parameters->get("file_hash") || undef;
|
|
my $resolution_w = body_parameters->get("resolution_w");
|
|
my $resolution_h = body_parameters->get("resolution_h");
|
|
my $codec = body_parameters->get("codec");
|
|
|
|
my $sth = database->prepare("INSERT INTO videos (file_path, file_hash, resolution_w, resolution_h, codec) VALUES (?, ?, ?, ?, ?)");
|
|
$sth->execute($file_path, $file_hash, $resolution_w, $resolution_h, $codec);
|
|
my $video_id = database->last_insert_id(undef, undef, 'videos', undef);
|
|
$sth->finish();
|
|
|
|
return { id => $video_id };
|
|
};
|
|
|
|
get '/api/v1/tasks' => sub {
|
|
my $status = query_parameters->get("status") || 'PENDING';
|
|
my $type = query_parameters->get("type");
|
|
|
|
unless ($type) {
|
|
send_error("Missing 'type' query parameter", 400);
|
|
}
|
|
|
|
# Optional pagination
|
|
my $page = query_parameters->get("page") || 1;
|
|
my $per_page = query_parameters->get("per_page") || 20;
|
|
my $offset = ($page - 1) * $per_page;
|
|
|
|
my $sth = database->prepare(
|
|
"SELECT id, video_id, task_type AS type, status FROM tasks WHERE status=? AND task_type=? ORDER BY id LIMIT ? OFFSET ?"
|
|
);
|
|
$sth->execute($status, $type, $per_page, $offset);
|
|
my $ref = $sth->fetchall_hashref('id');
|
|
$sth->finish();
|
|
|
|
# Total count for pagination metadata
|
|
my $count_sth = database->prepare(
|
|
"SELECT COUNT(*) AS cnt FROM tasks WHERE status=? AND task_type=?"
|
|
);
|
|
$count_sth->execute($status, $type);
|
|
my $cnt = ($count_sth->fetchrow_hashref() || { cnt => 0 })->{cnt};
|
|
$count_sth->finish();
|
|
|
|
return {
|
|
tasks => $ref,
|
|
page => $page,
|
|
per_page => $per_page,
|
|
total => $cnt,
|
|
total_pages => max(1, int(($cnt + $per_page - 1) / $per_page)),
|
|
};
|
|
};
|
|
|
|
get '/api/v1/task/:task' => sub {
|
|
my $task = route_parameters->get("task");
|
|
my $sth = database->prepare("SELECT * FROM tasks WHERE id=?");
|
|
$sth->execute($task);
|
|
my $ref = $sth->fetchrow_hashref();
|
|
$sth->finish();
|
|
|
|
unless(ref $ref) {
|
|
send_error("No task", 404);
|
|
}
|
|
|
|
return $ref;
|
|
};
|
|
|
|
get '/api/v1/nexttask/:type' => sub {
|
|
my $type = route_parameters->get("type");
|
|
my $sth = database->prepare("SELECT * FROM tasks WHERE task_type=? AND status='PENDING' ORDER BY updated_at ASC LIMIT 1");
|
|
$sth->execute($type);
|
|
my $ref = $sth->fetchrow_hashref();
|
|
$sth->finish();
|
|
|
|
unless(ref $ref) {
|
|
send_error("No task", 404);
|
|
}
|
|
|
|
# Mark the task as in progress and assign it to a worker
|
|
my $assign_key = "worker_" . int(rand(1000)); # Example assign key, you can customize this
|
|
my $update_sth = database->prepare("UPDATE tasks SET status='IN_PROGRESS', assign_key=?, assigned_at=NOW() WHERE id=? AND status='PENDING'");
|
|
$update_sth->execute($assign_key, $ref->{id});
|
|
if($update_sth->rows == 0) {
|
|
send_error("Failed to claim task", 409);
|
|
}
|
|
$update_sth->finish();
|
|
|
|
return { task => $ref, assign_key => $assign_key };
|
|
};
|
|
|
|
post '/api/v1/task/:task/complete' => sub {
|
|
my $task_id = route_parameters->get("task");
|
|
|
|
# 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");
|
|
}
|
|
|
|
# Optional worker verification: if assign_key is provided, require IN_PROGRESS + matching key.
|
|
# When omitted (e.g., UI-driven workflows), skip worker checks and just complete the task.
|
|
if ($assign_key) {
|
|
my $sth = database->prepare("SELECT * FROM tasks WHERE id=? AND assign_key=? AND status='IN_PROGRESS'");
|
|
$sth->execute($task_id, $assign_key);
|
|
my $ref = $sth->fetchrow_hashref();
|
|
$sth->finish();
|
|
|
|
unless(ref $ref) {
|
|
send_error("Task not assigned to this worker or not in progress", 403);
|
|
}
|
|
} else {
|
|
# No assign_key provided — caller may be an unauthenticated UI, etc.
|
|
my $sth = database->prepare("SELECT id FROM tasks WHERE id=?");
|
|
$sth->execute($task_id);
|
|
unless ($sth->fetchrow_hashref()) {
|
|
send_error("Task not found", 404);
|
|
}
|
|
$sth->finish();
|
|
}
|
|
|
|
# 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_json, $task_id);
|
|
if($update_sth->rows == 0) {
|
|
send_error("Failed to complete task", 500);
|
|
}
|
|
$update_sth->finish();
|
|
|
|
return { message => "Task completed successfully" };
|
|
};
|
|
|
|
post '/api/v1/task' => sub {
|
|
my $video_id = body_parameters->get("video_id");
|
|
my $task_type = body_parameters->get("task_type");
|
|
|
|
database->do("DELETE FROM tasks WHERE video_id=? AND task_type=?", undef, $video_id, $task_type);
|
|
|
|
my $sth = database->prepare("INSERT INTO tasks (video_id, task_type, status) VALUES (?, ?, 'PENDING')");
|
|
$sth->execute($video_id, $task_type);
|
|
my $task_id = database->last_insert_id(undef, undef, 'tasks', undef);
|
|
$sth->finish();
|
|
|
|
return { id => $task_id };
|
|
};
|
|
|
|
post '/api/v1/task/:task/unassign' => sub {
|
|
my $task_id = route_parameters->get("task");
|
|
|
|
my $update_sth = database->prepare(
|
|
"UPDATE tasks SET status='PENDING', assign_key=NULL, assigned_at=NULL, updated_at=NOW() WHERE id=?"
|
|
);
|
|
$update_sth->execute($task_id);
|
|
|
|
if ($update_sth->rows == 0) {
|
|
send_error("Task not found or already in PENDING state", 404);
|
|
}
|
|
|
|
|
|
return { message => 'Task unassigned' };
|
|
};
|
|
|
|
start();
|
|
|
|
|
|
sub max {
|
|
return (sort { $b <=> $a } @_)[0];
|
|
}
|
|
|
|
sub min {
|
|
return (sort { $a <=> $b } @_)[0];
|
|
} |