Files
newznab-tmux/app/Services/Runners/PostProcessRunner.php
T
2026-08-11 10:43:22 +02:00

755 lines
26 KiB
PHP

<?php
declare(strict_types=1);
namespace App\Services\Runners;
use App\Models\Category;
use App\Models\Settings;
use App\Services\AdditionalProcessing\AdditionalCandidateQuery;
use App\Services\AdditionalProcessing\AdditionalProcessingOrchestrator;
use App\Services\NfoService;
use App\Services\TempWorkspaceService;
use Illuminate\Support\Facades\Concurrency;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;
class PostProcessRunner extends BaseRunner
{
public const int ADDITIONAL_WORKER_MAX_BATCHES = 4;
private function guidBucketExpression(string $column = 'leftguid'): string
{
return DB::getDriverName() === 'sqlite'
? 'substr('.$column.', 1, 1)'
: 'LEFT('.$column.', 1)';
}
/**
* @param array<int, object{id?: mixed}> $releases
*/
private function runPostProcess(array $releases, int $maxProcesses, string $type, string $desc): void
{
if (empty($releases)) {
$this->headerNone();
return;
}
// Additional post-processing can run for a long time per bucket. Always
// stream it so tmux shows active parallel children instead of buffering
// all output until the entire pool finishes.
if ((bool) config('nntmux.stream_fork_output', false) === true || $type === 'additional') {
$commands = [];
foreach ($releases as $release) {
// id may already be a single GUID bucket char; if not, take first char defensively
$char = isset($release->id) ? substr((string) $release->id, 0, 1) : '';
// Use postprocess:guid command which accepts the GUID character
$command = PHP_BINARY.' artisan postprocess:guid '.$type.' '.$char;
if ($type === 'additional') {
$command .= ' --worker --max-batches='.self::ADDITIONAL_WORKER_MAX_BATCHES;
}
$commands[] = $command;
}
$this->runStreamingCommands($commands, $maxProcesses, $desc); // @phpstan-ignore argument.type
return;
}
$count = count($releases);
$this->headerStart('postprocess: '.$desc, $count, $maxProcesses);
if ($count <= 1 || $maxProcesses <= 1) {
foreach ($releases as $release) {
$char = isset($release->id) ? substr((string) $release->id, 0, 1) : '';
$command = PHP_BINARY.' artisan postprocess:guid '.$type.' '.$char;
echo $this->executeCommand($command);
cli()->primary('Finished task for '.$desc);
}
return;
}
$commands = [];
foreach ($releases as $idx => $release) {
$char = isset($release->id) ? substr((string) $release->id, 0, 1) : '';
$commands[$idx] = PHP_BINARY.' artisan postprocess:guid '.$type.' '.$char;
}
try {
$results = $this->runParallelCommands($commands, $maxProcesses, $this->concurrencyTimeout());
foreach ($results as $output) {
echo $output;
cli()->primary('Finished task for '.$desc);
}
} catch (\Throwable $e) {
Log::error('Postprocess batch failed: '.$e->getMessage());
cli()->error('Postprocess batch failed: '.$e->getMessage());
}
}
/**
* @param array<int, object{type: string, id: string}> $tasks
*/
private function runPostProcessMixed(array $tasks, int $maxProcesses, string $desc): void
{
if ($tasks === []) {
$this->headerNone();
return;
}
if ((bool) config('nntmux.stream_fork_output', false) === true) {
$commands = [];
foreach ($tasks as $task) {
$char = substr((string) $task->id, 0, 1);
$commands[] = PHP_BINARY.' artisan postprocess:guid '.$task->type.' '.$char;
}
$this->runStreamingCommands($commands, $maxProcesses, $desc); // @phpstan-ignore argument.type
return;
}
$count = count($tasks);
$this->headerStart('postprocess: '.$desc, $count, $maxProcesses);
if ($count <= 1 || $maxProcesses <= 1) {
foreach ($tasks as $task) {
$char = substr((string) $task->id, 0, 1);
$command = PHP_BINARY.' artisan postprocess:guid '.$task->type.' '.$char;
echo $this->executeCommand($command);
cli()->primary('Finished task for '.$desc);
}
return;
}
$batches = array_chunk($tasks, max(1, $maxProcesses));
foreach ($batches as $batchIndex => $batch) {
$runTasks = [];
foreach ($batch as $idx => $task) {
$char = substr((string) $task->id, 0, 1);
$command = PHP_BINARY.' artisan postprocess:guid '.$task->type.' '.$char;
$runTasks[$idx] = fn () => $this->executeCommand($command);
}
try {
$results = Concurrency::run($runTasks, $this->concurrencyTimeout());
foreach ($results as $output) {
echo $output;
cli()->primary('Finished task for '.$desc);
}
} catch (\Throwable $e) {
Log::error('Postprocess mixed batch failed: '.$e->getMessage());
cli()->error('Batch '.($batchIndex + 1).' failed: '.$e->getMessage());
}
}
}
/**
* @return array<int, object{id: string}>
*/
private function getBooksBuckets(): array
{
$bucketExpr = $this->guidBucketExpression();
return DB::select('
SELECT DISTINCT '.$bucketExpr.' AS id
FROM releases
WHERE (
categories_id BETWEEN '.Category::BOOKS_ROOT.' AND '.Category::BOOKS_UNKNOWN.'
OR categories_id = '.Category::MUSIC_AUDIOBOOK.'
)
AND (
bookinfo_id IS NULL
OR searchname LIKE "N:/NZB%"
OR searchname LIKE "N_NZB_%"
OR name LIKE "N:/NZB%"
OR name LIKE "N_NZB_%"
)
LIMIT 16');
}
/**
* @return array<int, object{id: string}>
*/
private function getMusicBuckets(): array
{
$bucketExpr = $this->guidBucketExpression();
return DB::select('
SELECT DISTINCT '.$bucketExpr.' AS id
FROM releases
WHERE categories_id IN ('.Category::MUSIC_MP3.', '.Category::MUSIC_LOSSLESS.', '.Category::MUSIC_OTHER.')
AND musicinfo_id IS NULL
LIMIT 16');
}
/**
* @return array<int, object{id: string}>
*/
private function getConsoleBuckets(): array
{
$bucketExpr = $this->guidBucketExpression();
$renamedFilter = (int) Settings::settingValue('lookupgames') === 2 ? 'AND isrenamed = 1' : '';
return DB::select('
SELECT DISTINCT '.$bucketExpr.' AS id
FROM releases
WHERE categories_id BETWEEN '.Category::GAME_ROOT.' AND '.Category::GAME_OTHER.'
AND consoleinfo_id IS NULL
'.$renamedFilter.'
LIMIT 16');
}
/**
* @return array<int, object{id: string}>
*/
private function getGamesBuckets(): array
{
$bucketExpr = $this->guidBucketExpression();
$renamedFilter = (int) Settings::settingValue('lookupgames') === 2 ? 'AND isrenamed = 1' : '';
return DB::select('
SELECT DISTINCT '.$bucketExpr.' AS id
FROM releases
WHERE categories_id = '.Category::PC_GAMES.'
AND gamesinfo_id = 0
'.$renamedFilter.'
LIMIT 16');
}
public function processAdditional(): void
{
// Bucket-selection predicates and size filters are owned by
// AdditionalCandidateQuery so they cannot drift away from the
// per-worker fetch in AdditionalProcessingOrchestrator::fetchReleases().
$bucketCounts = AdditionalCandidateQuery::availableBucketCounts();
$chars = array_column($bucketCounts, 'bucket');
$maxProcesses = (int) Settings::settingValue('postthreads');
$queryLimit = (int) (Settings::settingValue('maxaddprocessed') ?: 25);
// Normalize to the shape the rest of runPostProcess() expects:
// an array of objects with an `id` (first GUID char) property. If the
// backlog is concentrated in fewer buckets than configured threads,
// repeat buckets so claimBatch() can split one hot bucket across
// multiple workers.
$queue = $this->additionalQueue($bucketCounts, $maxProcesses, $queryLimit);
if (! $this->prepareAdditionalTempBuckets($chars)) {
return;
}
if ($maxProcesses <= 1) {
if ($queue === []) {
$this->headerNone();
return;
}
$this->headerStart('postprocess: additional postprocessing', count($queue), 1);
$orchestrator = app(AdditionalProcessingOrchestrator::class);
foreach ($chars as $char) {
$workerToken = bin2hex(random_bytes(16));
$excludedReleaseIds = [];
try {
for ($batch = 0; $batch < self::ADDITIONAL_WORKER_MAX_BATCHES; $batch++) {
$result = $orchestrator->start('', $char, $workerToken, $excludedReleaseIds);
if ($result->claimedCount() === 0) {
break;
}
$excludedReleaseIds = array_values(array_unique([
...$excludedReleaseIds,
...$result->claimedIds,
]));
}
} finally {
$orchestrator->finish();
}
cli()->primary('Finished task for additional postprocessing');
}
return;
}
$this->runPostProcess($queue, $maxProcesses, 'additional', 'additional postprocessing');
}
/**
* @param list<array{bucket: string, count: int}> $bucketCounts
* @return list<object{id: string}>
*/
private function additionalQueue(array $bucketCounts, int $maxProcesses, int $queryLimit): array
{
if ($bucketCounts === []) {
return [];
}
$chars = array_column($bucketCounts, 'bucket');
$queue = array_map(static fn (string $c): object => (object) ['id' => $c], $chars);
$allocations = array_fill(0, count($bucketCounts), 1);
$demand = [];
foreach ($bucketCounts as $index => $bucketCount) {
// Demand is intentionally measured in immediately claimable batches:
// postthreads remains the hard connection cap while hot buckets use
// available parallelism instead of draining all batches serially.
$demand[$index] = max(1, (int) ceil($bucketCount['count'] / max(1, $queryLimit)));
}
$target = max(count($queue), min(max(1, $maxProcesses), array_sum($demand)));
while (count($queue) < $target) {
$nextIndex = null;
$largestRemainingDemand = 0;
foreach ($chars as $index => $char) {
$remainingDemand = $demand[$index] - $allocations[$index];
if ($remainingDemand > $largestRemainingDemand) {
$largestRemainingDemand = $remainingDemand;
$nextIndex = $index;
}
}
if ($nextIndex === null) {
break;
}
$queue[] = (object) ['id' => $chars[$nextIndex]];
$allocations[$nextIndex]++;
}
return $queue;
}
/**
* @param list<string> $chars
*/
private function prepareAdditionalTempBuckets(array $chars): bool
{
if ($chars === []) {
return true;
}
$tempWorkspace = app(TempWorkspaceService::class);
$basePath = (string) config('nntmux.tmp_unrar_path');
try {
foreach ($chars as $char) {
$bucketPath = $tempWorkspace->ensureMainTempPath($basePath, $char);
$tempWorkspace->pruneStaleWorkerDirectories(
$bucketPath,
max(
3600,
(int) config('nntmux.multiprocessing_max_child_time', 1800) * 2,
),
);
}
} catch (\Throwable $e) {
cli()->warning('Additional post-processing skipped: '.$e->getMessage());
Log::error('Additional post-processing temp bucket preflight failed', [
'tmp_unrar_path' => $basePath,
'exception' => $e,
]);
return false;
}
return true;
}
public function processNfo(): void
{
if ((int) Settings::settingValue('lookupnfo') !== 1) {
$this->headerNone();
return;
}
$nfoQuery = NfoService::NfoQueryString();
$checkSql = 'SELECT r.id FROM releases r WHERE 1=1 '.$nfoQuery.' LIMIT 1';
if (count(DB::select($checkSql)) === 0) {
$this->headerNone();
return;
}
$bucketExpr = $this->guidBucketExpression('r.leftguid');
$sql = '
SELECT DISTINCT '.$bucketExpr.' AS id
FROM releases r
WHERE 1=1 '.$nfoQuery.'
LIMIT 16';
$queue = DB::select($sql);
$maxProcesses = (int) Settings::settingValue('nfothreads');
$this->runPostProcess($queue, $maxProcesses, 'nfo', 'nfo postprocessing');
}
public function processMovies(bool $renamedOnly): void
{
if ((int) Settings::settingValue('lookupimdb') <= 0) {
$this->headerNone();
return;
}
$condLookup = ((int) Settings::settingValue('lookupimdb') === 2 ? 'AND isrenamed = 1' : '');
$condRenamedOnly = ($renamedOnly ? 'AND isrenamed = 1' : '');
$checkSql = '
SELECT id
FROM releases
WHERE categories_id BETWEEN 2000 AND 2999
AND '.imdb_id_needs_lookup_sql('imdbid').'
'.$condLookup.' '.$condRenamedOnly.'
LIMIT 1';
if (count(DB::select($checkSql)) === 0) {
$this->headerNone();
return;
}
$renamedFlag = ($renamedOnly ? 2 : 1);
$bucketExpr = $this->guidBucketExpression();
$sql = '
SELECT DISTINCT '.$bucketExpr.' AS id, '.$renamedFlag.' AS renamed
FROM releases
WHERE categories_id BETWEEN 2000 AND 2999
AND '.imdb_id_needs_lookup_sql('imdbid').'
'.$condLookup.' '.$condRenamedOnly.'
LIMIT 16';
$queue = DB::select($sql);
$maxProcesses = (int) Settings::settingValue('postthreadsnon');
$this->runPostProcess($queue, $maxProcesses, 'movie', 'movies postprocessing');
}
public function processTv(bool $renamedOnly): void
{
if ((int) Settings::settingValue('lookuptv') <= 0) {
$this->headerNone();
return;
}
$condLookup = ((int) Settings::settingValue('lookuptv') === 2 ? 'AND isrenamed = 1' : '');
$condRenamedOnly = ($renamedOnly ? 'AND isrenamed = 1' : '');
$checkSql = '
SELECT id
FROM releases
WHERE categories_id BETWEEN 5000 AND 5999
AND categories_id != 5070
AND videos_id = 0
AND size > 1048576
AND tv_episodes_id BETWEEN -3 AND 0
'.$condLookup.' '.$condRenamedOnly.'
LIMIT 1';
if (count(DB::select($checkSql)) === 0) {
$this->headerNone();
return;
}
$renamedFlag = ($renamedOnly ? 2 : 1);
$bucketExpr = $this->guidBucketExpression();
$sql = '
SELECT DISTINCT '.$bucketExpr.' AS id, '.$renamedFlag.' AS renamed
FROM releases
WHERE categories_id BETWEEN 5000 AND 5999
AND categories_id != 5070
AND videos_id = 0
AND tv_episodes_id BETWEEN -3 AND 0
AND size > 1048576
'.$condLookup.' '.$condRenamedOnly.'
LIMIT 16';
$queue = DB::select($sql);
$maxProcesses = (int) Settings::settingValue('postthreadsnon');
// Use pipelined TV processing for better efficiency
$this->runPostProcessTvPipeline($queue, $maxProcesses, 'tv postprocessing (pipelined)', $renamedOnly);
}
/**
* Run pipelined TV post-processing across multiple GUID buckets in parallel.
* Each parallel process runs the full provider pipeline sequentially.
*
* @param array<string, mixed> $releases
*/
private function runPostProcessTvPipeline(array $releases, int $maxProcesses, string $desc, bool $renamedOnly): void
{
if (empty($releases)) {
$this->headerNone();
return;
}
// If streaming is enabled, run commands with real-time output
if ((bool) config('nntmux.stream_fork_output', false) === true) {
$commands = [];
foreach ($releases as $release) {
$char = isset($release->id) ? substr((string) $release->id, 0, 1) : '';
$renamed = isset($release->renamed) ? $release->renamed : '';
// Use the pipelined TV command
$commands[] = PHP_BINARY.' artisan postprocess:tv-pipeline '.$char.($renamed ? ' '.$renamed : '').' --mode=pipeline';
}
$this->runStreamingCommands($commands, $maxProcesses, $desc); // @phpstan-ignore argument.type
return;
}
$count = count($releases);
$this->headerStart('postprocess: '.$desc, $count, $maxProcesses);
// Process in batches using Laravel's native Concurrency facade
$batches = array_chunk($releases, max(1, $maxProcesses));
foreach ($batches as $batchIndex => $batch) {
$tasks = [];
foreach ($batch as $idx => $release) {
$char = isset($release->id) ? substr((string) $release->id, 0, 1) : '';
$renamed = isset($release->renamed) ? $release->renamed : '';
// Use the pipelined TV command for each GUID bucket
$command = PHP_BINARY.' artisan postprocess:tv-pipeline '.$char.($renamed ? ' '.$renamed : '').' --mode=pipeline';
$tasks[$idx] = fn () => $this->executeCommand($command);
}
try {
$results = Concurrency::run($tasks, $this->concurrencyTimeout());
foreach ($results as $taskIdx => $output) {
echo $output;
cli()->primary('Finished task for '.$desc);
}
} catch (\Throwable $e) {
Log::error('TV pipeline batch failed: '.$e->getMessage());
cli()->error('Batch '.($batchIndex + 1).' failed: '.$e->getMessage());
}
}
}
/**
* Lightweight check to determine if there is any TV work to process.
*/
public function hasTvWork(bool $renamedOnly): bool
{
if ((int) Settings::settingValue('lookuptv') <= 0) {
return false;
}
$condLookup = ((int) Settings::settingValue('lookuptv') === 2 ? 'AND isrenamed = 1' : '');
$condRenamedOnly = ($renamedOnly ? 'AND isrenamed = 1' : '');
$checkSql = '
SELECT id
FROM releases
WHERE categories_id BETWEEN 5000 AND 5999
AND categories_id != 5070
AND videos_id = 0
AND size > 1048576
AND tv_episodes_id BETWEEN -3 AND 0
'.$condLookup.' '.$condRenamedOnly.'
LIMIT 1';
return count(DB::select($checkSql)) > 0;
}
public function processAnime(): void
{
if ((int) Settings::settingValue('lookupanidb') <= 0) {
$this->headerNone();
return;
}
$checkSql = '
SELECT id
FROM releases
WHERE categories_id = 5070
AND anidbid IS NULL
LIMIT 1';
if (count(DB::select($checkSql)) === 0) {
$this->headerNone();
return;
}
$bucketExpr = $this->guidBucketExpression();
$sql = '
SELECT DISTINCT '.$bucketExpr.' AS id
FROM releases
WHERE categories_id = 5070
AND anidbid IS NULL
LIMIT 16';
$queue = DB::select($sql);
$maxProcesses = (int) Settings::settingValue('postthreadsnon');
$this->runPostProcess($queue, $maxProcesses, 'anime', 'anime postprocessing');
}
public function processBooks(): void
{
if ((int) Settings::settingValue('lookupbooks') <= 0) {
$this->headerNone();
return;
}
$checkSql = '
SELECT id
FROM releases
WHERE (
categories_id BETWEEN 7000 AND 7999
OR categories_id = 3030
)
AND (
bookinfo_id IS NULL
OR searchname LIKE "N:/NZB%"
OR searchname LIKE "N_NZB_%"
OR name LIKE "N:/NZB%"
OR name LIKE "N_NZB_%"
)
LIMIT 1';
if (count(DB::select($checkSql)) === 0) {
$this->headerNone();
return;
}
$bucketExpr = $this->guidBucketExpression();
$sql = '
SELECT DISTINCT '.$bucketExpr.' AS id
FROM releases
WHERE (
categories_id BETWEEN 7000 AND 7999
OR categories_id = 3030
)
AND (
bookinfo_id IS NULL
OR searchname LIKE "N:/NZB%"
OR searchname LIKE "N_NZB_%"
OR name LIKE "N:/NZB%"
OR name LIKE "N_NZB_%"
)
LIMIT 16';
$queue = DB::select($sql);
$maxProcesses = (int) Settings::settingValue('postthreadsamazon');
$this->runPostProcess($queue, $maxProcesses, 'books', 'books postprocessing');
}
public function processMusic(): void
{
if ((int) Settings::settingValue('lookupmusic') <= 0) {
$this->headerNone();
return;
}
$checkSql = '
SELECT id
FROM releases
WHERE categories_id IN ('.Category::MUSIC_MP3.', '.Category::MUSIC_LOSSLESS.', '.Category::MUSIC_OTHER.')
AND musicinfo_id IS NULL
LIMIT 1';
if (count(DB::select($checkSql)) === 0) {
$this->headerNone();
return;
}
$queue = $this->getMusicBuckets();
$maxProcesses = (int) Settings::settingValue('postthreadsamazon');
$this->runPostProcess($queue, $maxProcesses, 'music', 'music postprocessing');
}
public function processConsoles(): void
{
if ((int) Settings::settingValue('lookupgames') <= 0) {
$this->headerNone();
return;
}
$renamedFilter = (int) Settings::settingValue('lookupgames') === 2 ? 'AND isrenamed = 1' : '';
$checkSql = '
SELECT id
FROM releases
WHERE categories_id BETWEEN '.Category::GAME_ROOT.' AND '.Category::GAME_OTHER.'
AND consoleinfo_id IS NULL
'.$renamedFilter.'
LIMIT 1';
if (count(DB::select($checkSql)) === 0) {
$this->headerNone();
return;
}
$queue = $this->getConsoleBuckets();
$maxProcesses = (int) Settings::settingValue('postthreadsamazon');
$this->runPostProcess($queue, $maxProcesses, 'console', 'console postprocessing');
}
public function processGames(): void
{
if ((int) Settings::settingValue('lookupgames') <= 0) {
$this->headerNone();
return;
}
$renamedFilter = (int) Settings::settingValue('lookupgames') === 2 ? 'AND isrenamed = 1' : '';
$checkSql = '
SELECT id
FROM releases
WHERE categories_id = '.Category::PC_GAMES.'
AND gamesinfo_id = 0
'.$renamedFilter.'
LIMIT 1';
if (count(DB::select($checkSql)) === 0) {
$this->headerNone();
return;
}
$queue = $this->getGamesBuckets();
$maxProcesses = (int) Settings::settingValue('postthreadsamazon');
$this->runPostProcess($queue, $maxProcesses, 'games', 'games postprocessing');
}
public function processAmazon(): void
{
$maxProcesses = (int) Settings::settingValue('postthreadsamazon');
$tasks = [];
if ((int) Settings::settingValue('lookupbooks') > 0) {
foreach ($this->getBooksBuckets() as $row) {
$tasks[] = (object) ['type' => 'books', 'id' => (string) $row->id];
}
}
if ((int) Settings::settingValue('lookupmusic') > 0) {
foreach ($this->getMusicBuckets() as $row) {
$tasks[] = (object) ['type' => 'music', 'id' => (string) $row->id];
}
}
if ((int) Settings::settingValue('lookupgames') > 0) {
foreach ($this->getConsoleBuckets() as $row) {
$tasks[] = (object) ['type' => 'console', 'id' => (string) $row->id];
}
foreach ($this->getGamesBuckets() as $row) {
$tasks[] = (object) ['type' => 'games', 'id' => (string) $row->id];
}
}
$this->runPostProcessMixed($tasks, $maxProcesses, 'amazon (books+music+console+games)');
}
}