starred/newznab-tmux
1
0
Fork 0
mirror of https://github.com/NNTmux/newznab-tmux.git synced 2026-08-29 01:08:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use proper multiprocessing

Browse Source
This commit is contained in:
DariusIII
2026-05-15 14:33:41 +02:00
parent efa7b748cd
commit f62336305a
13 changed files with 356 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/Console/Commands/PostProcessGuid.php
+5 -2
View File
@@ -20,7 +20,7 @@ class PostProcessGuid extends Command
* @var string
*/
protected $signature = 'postprocess:guid
{type : Type: additional, nfo, movie, tv, anime, or books}
{type : Type: additional, nfo, movie, tv, anime, books, music, console, or games}
{guid : First character of release guid (a-f, 0-9)}
{renamed? : For movie/tv: process renamed only (optional)}';
@@ -61,8 +61,11 @@ class PostProcessGuid extends Command
'tv' => $this->postProcessService->processTv('', $guid, $renamed),
'anime' => $this->postProcessService->processAnime('', $guid),
'books' => $this->postProcessService->processBooks('', $guid),
'music' => $this->postProcessService->processMusic('', $guid),
'console' => $this->postProcessService->processConsoles('', $guid),
'games' => $this->postProcessService->processGames('', $guid),
default => throw new \InvalidArgumentException(
'Invalid type. Must be: additional, nfo, movie, tv, anime, or books.'
'Invalid type. Must be: additional, nfo, movie, tv, anime, books, music, console, or games.'
),
};
app/Console/Commands/ProcessPostProcess.php
+13 -8
View File
@@ -15,7 +15,7 @@ class ProcessPostProcess extends Command
* @var string
*/
protected $signature = 'multiprocessing:postprocess
{type : Type: ama, add, ani, mov, nfo or tv}
{type : Type: ama, add, ani, mov, nfo, tv, boo, mus, con, gam}
{renamed=false : For mov/tv: only post-process renamed releases (true/false)}';
/**
@@ -30,16 +30,17 @@ class ProcessPostProcess extends Command
*/
public function handle(): int
{
$this->warn('⚠️ WARNING: This command is to be used from cli.');
$this->line('');
$type = $this->argument('type');
$renamed = $this->argument('renamed');
if (! \in_array($type, ['ama', 'add', 'ani', 'mov', 'nfo', 'tv'], true)) {
$this->error('Type must be one of: ama, add, ani, mov, nfo, sha, tv');
if (! \in_array($type, ['ama', 'add', 'ani', 'mov', 'nfo', 'tv', 'boo', 'mus', 'con', 'gam'], true)) {
$this->error('Type must be one of: ama, add, ani, mov, nfo, tv, boo, mus, con, gam');
$this->line('');
$this->line('ama => Do amazon/books processing');
$this->line('ama => Do amazon (books+music+console+games) processing in parallel');
$this->line('boo => Do books processing');
$this->line('mus => Do music processing');
$this->line('con => Do console processing');
$this->line('gam => Do games processing');
$this->line('add => Do additional (rar|zip) processing');
$this->line('ani => Do anime processing');
$this->line('mov => Do movie processing');
@@ -54,12 +55,16 @@ class ProcessPostProcess extends Command
$service = new ForkingService;
match ($type) {
'ama' => $service->processBooks(),
'ama' => $service->processAmazon(),
'boo' => $service->processBooks(),
'add' => $service->processAdditional(),
'ani' => $service->processAnime(),
'mov' => $service->processMovies($renamedOnly),
'nfo' => $service->processNfo(),
'tv' => $service->processTv($renamedOnly),
'mus' => $service->processMusic(),
'con' => $service->processConsoles(),
'gam' => $service->processGames(),
};
return self::SUCCESS;
app/Services/ConsoleService.php
+9 -1
View File
@@ -491,13 +491,21 @@ class ConsoleService
*
* @throws \Exception
*/
public function processConsoleReleases(): void
public function processConsoleReleases(string $groupID = '', string $guidChar = ''): void
{
$query = Release::query()
->select(['searchname', 'id'])
->whereBetween('categories_id', [Category::GAME_ROOT, Category::GAME_OTHER])
->whereNull('consoleinfo_id');
if ($guidChar !== '') {
$query->where('leftguid', 'like', $guidChar.'%');
}
if ($groupID !== '') {
$query->where('groups_id', $groupID);
}
if ($this->renamed === true) {
$query->where('isrenamed', '=', 1);
}
app/Services/ConsolesProcessor.php
+2 -2
View File
@@ -16,10 +16,10 @@ class ConsolesProcessor
$this->echooutput = $echooutput;
}
public function process(): void
public function process(string $groupID = '', string $guidChar = ''): void
{
if ((int) Settings::settingValue('lookupgames') !== 0) {
(new ConsoleService)->processConsoleReleases();
(new ConsoleService)->processConsoleReleases($groupID, $guidChar);
}
}
}
app/Services/ForkingService.php
+38 -2
View File
@@ -172,7 +172,39 @@ class ForkingService
*/
public function processBooks(): void
{
$this->runWithTiming('postProcess_ama', fn () => $this->postProcessRunner->processBooks());
$this->runWithTiming('postProcess_boo', fn () => $this->postProcessRunner->processBooks());
}
/**
* Process music metadata.
*/
public function processMusic(): void
{
$this->runWithTiming('postProcess_mus', fn () => $this->postProcessRunner->processMusic());
}
/**
* Process console game metadata.
*/
public function processConsoles(): void
{
$this->runWithTiming('postProcess_con', fn () => $this->postProcessRunner->processConsoles());
}
/**
* Process PC game metadata.
*/
public function processGames(): void
{
$this->runWithTiming('postProcess_gam', fn () => $this->postProcessRunner->processGames());
}
/**
* Process books, music, console, and games metadata in parallel.
*/
public function processAmazon(): void
{
$this->runWithTiming('postProcess_ama', fn () => $this->postProcessRunner->processAmazon());
}
/**
@@ -192,8 +224,12 @@ class ForkingService
'releases' => $this->releases(),
'postProcess_add' => $this->processAdditional(),
'postProcess_ani' => $this->processAnime(),
'postProcess_ama' => $this->processBooks(),
'postProcess_ama' => $this->processAmazon(),
'postProcess_boo' => $this->processBooks(),
'postProcess_con' => $this->processConsoles(),
'postProcess_gam' => $this->processGames(),
'postProcess_mov' => $this->processMovies($options[0] ?? false),
'postProcess_mus' => $this->processMusic(),
'postProcess_nfo' => $this->processNfo(),
'postProcess_tv' => $this->processTv($options[0] ?? false),
'safe_backfill' => $this->safeBackfill(),
app/Services/GamesProcessor.php
+2 -2
View File
@@ -19,10 +19,10 @@ class GamesProcessor
$this->gamesService = $gamesService ?? new GamesService;
}
public function process(): void
public function process(string $groupID = '', string $guidChar = ''): void
{
if ((int) Settings::settingValue('lookupgames') !== 0) {
$this->gamesService->processGamesReleases();
$this->gamesService->processGamesReleases($groupID, $guidChar);
}
}
}
app/Services/GamesService.php
+9 -1
View File
@@ -730,7 +730,7 @@ class GamesService
*
* @throws \Exception
*/
public function processGamesReleases(): void
public function processGamesReleases(string $groupID = '', string $guidChar = ''): void
{
// Reset stats
$this->processedCount = 0;
@@ -744,6 +744,14 @@ class GamesService
->where('gamesinfo_id', '=', 0)
->where('categories_id', '=', Category::PC_GAMES);
if ($guidChar !== '') {
$query->where('leftguid', 'like', $guidChar.'%');
}
if ($groupID !== '') {
$query->where('groups_id', $groupID);
}
if ((int) Settings::settingValue('lookupgames') === 2) {
$query->where('isrenamed', '=', 1);
}
app/Services/MusicProcessor.php
+2 -2
View File
@@ -16,10 +16,10 @@ class MusicProcessor
$this->echooutput = $echooutput;
}
public function process(): void
public function process(string $groupID = '', string $guidChar = ''): void
{
if ((int) Settings::settingValue('lookupmusic') !== 0) {
(new MusicService)->processMusicReleases();
(new MusicService)->processMusicReleases(false, $groupID, $guidChar);
}
}
}
app/Services/MusicService.php
+8 -1
View File
@@ -460,8 +460,11 @@ class MusicService
*
* @throws \Exception
*/
public function processMusicReleases(bool $local = false): void
public function processMusicReleases(bool $local = false, string $groupID = '', string $guidChar = ''): void
{
$guidFilter = $guidChar !== '' ? 'AND leftguid LIKE '.DB::getPdo()->quote($guidChar.'%') : '';
$groupFilter = $groupID !== '' ? 'AND groups_id = '.(int) $groupID : '';
$res = DB::select(
sprintf(
'
@@ -469,10 +472,14 @@ class MusicService
FROM releases
WHERE musicinfo_id IS NULL
%s
%s
%s
AND categories_id IN (%s, %s, %s)
ORDER BY postdate DESC
LIMIT %d',
$this->renamed,
$guidFilter,
$groupFilter,
Category::MUSIC_MP3,
Category::MUSIC_LOSSLESS,
Category::MUSIC_OTHER,
app/Services/PostProcessService.php
+6 -6
View File
@@ -140,9 +140,9 @@ final class PostProcessService
*
* @throws \Exception
*/
public function processConsoles(): void
public function processConsoles(string $groupID = '', string $guidChar = ''): void
{
$this->consolesProcessor->process();
$this->consolesProcessor->process($groupID, $guidChar);
}
/**
@@ -150,9 +150,9 @@ final class PostProcessService
*
* @throws \Exception
*/
public function processGames(): void
public function processGames(string $groupID = '', string $guidChar = ''): void
{
$this->gamesProcessor->process();
$this->gamesProcessor->process($groupID, $guidChar);
}
/**
@@ -177,9 +177,9 @@ final class PostProcessService
*
* @throws \Exception
*/
public function processMusic(): void
public function processMusic(string $groupID = '', string $guidChar = ''): void
{
$this->musicProcessor->process();
$this->musicProcessor->process($groupID, $guidChar);
}
/**
app/Services/Runners/PostProcessRunner.php
+245 -2
View File
@@ -4,6 +4,7 @@ declare(strict_types=1);
namespace App\Services\Runners;
use App\Models\Category;
use App\Models\Settings;
use App\Services\NfoService;
use Illuminate\Support\Facades\Concurrency;
@@ -20,7 +21,7 @@ class PostProcessRunner extends BaseRunner
}
/**
* @param array<string, mixed> $releases
* @param array<int, object{id?: mixed}> $releases
*/
private function runPostProcess(array $releases, int $maxProcesses, string $type, string $desc): void
{
@@ -84,6 +85,139 @@ class PostProcessRunner extends BaseRunner
}
}
/**
* @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
{
$bucketExpr = $this->guidBucketExpression('r.leftguid');
@@ -386,7 +520,116 @@ class PostProcessRunner extends BaseRunner
LIMIT 16';
$queue = DB::select($sql);
$maxProcesses = (int) Settings::settingValue('postthreadsnon');
$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)');
}
}
app/Services/Tmux/TmuxTaskRunner.php
+10 -44
View File
@@ -565,20 +565,20 @@ class TmuxTaskRunner
if ($postSetting === 1) {
// Post = 1: Additional processing only
if ($hasWork) {
$commands[] = "nice -n{$niceness} ".PHP_BINARY." artisan update:postprocess additional true 2>&1 | tee -a {$log}";
$commands[] = "nice -n{$niceness} ".PHP_BINARY." artisan multiprocessing:postprocess add 2>&1 | tee -a {$log}";
}
} elseif ($postSetting === 2) {
// Post = 2: NFO processing only
if ($hasNfo) {
$commands[] = "nice -n{$niceness} ".PHP_BINARY." artisan update:postprocess nfo true 2>&1 | tee -a {$log}";
$commands[] = "nice -n{$niceness} ".PHP_BINARY." artisan multiprocessing:postprocess nfo 2>&1 | tee -a {$log}";
}
} elseif ($postSetting === 3) {
// Post = 3: Both additional and NFO
if ($hasWork) {
$commands[] = "nice -n{$niceness} ".PHP_BINARY." artisan update:postprocess additional true 2>&1 | tee -a {$log}";
$commands[] = "nice -n{$niceness} ".PHP_BINARY." artisan multiprocessing:postprocess add 2>&1 | tee -a {$log}";
}
if ($hasNfo) {
$commands[] = "nice -n{$niceness} ".PHP_BINARY." artisan update:postprocess nfo true 2>&1 | tee -a {$log}";
$commands[] = "nice -n{$niceness} ".PHP_BINARY." artisan multiprocessing:postprocess nfo 2>&1 | tee -a {$log}";
}
}
@@ -625,14 +625,14 @@ class TmuxTaskRunner
$processTv = (int) ($runVar['settings']['processtvrage'] ?? 0);
$hasTvWork = (int) ($runVar['counts']['now']['processtv'] ?? 0) > 0;
if ($processTv > 0 && $hasTvWork) {
$commands[] = "nice -n{$niceness} {$artisan} update:postprocess tv 2>&1 | tee -a {$log}";
$commands[] = "nice -n{$niceness} {$artisan} multiprocessing:postprocess tv 2>&1 | tee -a {$log}";
}
// Anime processing
$processAnime = (int) ($runVar['settings']['processanime'] ?? 0);
$hasAnimeWork = (int) ($runVar['counts']['now']['processanime'] ?? 0) > 0;
if ($processAnime > 0 && $hasAnimeWork) {
$commands[] = "nice -n{$niceness} {$artisan} update:postprocess anime true 2>&1 | tee -a {$log}";
$commands[] = "nice -n{$niceness} {$artisan} multiprocessing:postprocess ani 2>&1 | tee -a {$log}";
}
// If no work available for any enabled type, disable the pane
@@ -693,7 +693,7 @@ class TmuxTaskRunner
}
$sleep = (int) ($runVar['settings']['post_timer_non'] ?? 300);
$command = "nice -n{$niceness} {$artisan} update:postprocess movies true 2>&1 | tee -a {$log}";
$command = "nice -n{$niceness} {$artisan} multiprocessing:postprocess mov false 2>&1 | tee -a {$log}";
$sleepCommand = $this->buildSleepCommand($sleep);
$fullCommand = "{$command}; date +'%Y-%m-%d %T'; {$sleepCommand}";
@@ -738,45 +738,11 @@ class TmuxTaskRunner
$niceness = Settings::settingValue('niceness') ?? 2;
$log = $this->getLogFile('post_amazon');
$artisan = PHP_BINARY.' artisan';
$commands = [];
// Books processing
$processBooks = (int) ($runVar['settings']['processbooks'] ?? 0);
$hasBooksWork = (int) ($runVar['counts']['now']['processbooks'] ?? 0) > 0;
if ($processBooks > 0 && $hasBooksWork) {
$commands[] = "nice -n{$niceness} {$artisan} update:postprocess book 2>&1 | tee -a {$log}";
}
// Music processing
$processMusic = (int) ($runVar['settings']['processmusic'] ?? 0);
$hasMusicWork = (int) ($runVar['counts']['now']['processmusic'] ?? 0) > 0;
if ($processMusic > 0 && $hasMusicWork) {
$commands[] = "nice -n{$niceness} {$artisan} update:postprocess music 2>&1 | tee -a {$log}";
}
// Console processing (uses same lookupgames setting as games)
$processConsole = (int) ($runVar['settings']['processgames'] ?? 0);
$hasConsoleWork = (int) ($runVar['counts']['now']['processconsole'] ?? 0) > 0;
if ($processConsole > 0 && $hasConsoleWork) {
$commands[] = "nice -n{$niceness} {$artisan} update:postprocess console 2>&1 | tee -a {$log}";
}
// Games processing
$processGames = (int) ($runVar['settings']['processgames'] ?? 0);
$hasGamesWork = (int) ($runVar['counts']['now']['processgames'] ?? 0) > 0;
if ($processGames > 0 && $hasGamesWork) {
$commands[] = "nice -n{$niceness} {$artisan} update:postprocess games 2>&1 | tee -a {$log}";
}
// If no commands were added (no work available), disable the pane
if (empty($commands)) {
return $this->disablePane($pane, 'Post-process Metadata', 'no work available for any enabled type');
}
$sleep = (int) ($runVar['settings']['post_timer_amazon'] ?? 300);
$allCommands = implode('; ', $commands);
$command = "nice -n{$niceness} {$artisan} multiprocessing:postprocess ama 2>&1 | tee -a {$log}";
$sleepCommand = $this->buildSleepCommand($sleep);
$fullCommand = "{$allCommands}; date +'%Y-%m-%d %T'; {$sleepCommand}";
$fullCommand = "{$command}; date +'%Y-%m-%d %T'; {$sleepCommand}";
return $this->paneManager->respawnPane($pane, $fullCommand);
}