From 42123988ee86df678566ad8d83ec0bf8c76603ae Mon Sep 17 00:00:00 2001 From: robertsaternus Date: Fri, 14 Aug 2026 11:41:40 +0200 Subject: [PATCH 1/6] INT-354: Alternative product export command based on a queue system Create alternative product export command based on a queue system. The main goal is to reduce memory usage when exporting large product catalogs. --- src/Console/Command/ExportBatch.php | 80 +++++++ src/Console/Command/WorkerExport.php | 224 ++++++++++++++++++ src/Model/Export/Catalog/DataProvider.php | 35 +++ .../ProductType/SimpleDataProvider.php | 9 + src/Model/Export/Feed.php | 23 ++ src/Model/Exporter.php | 25 +- src/Model/Stream/Csv.php | 5 +- src/etc/di.xml | 10 + 8 files changed, 408 insertions(+), 3 deletions(-) create mode 100644 src/Console/Command/ExportBatch.php create mode 100644 src/Console/Command/WorkerExport.php diff --git a/src/Console/Command/ExportBatch.php b/src/Console/Command/ExportBatch.php new file mode 100644 index 00000000..0fc99115 --- /dev/null +++ b/src/Console/Command/ExportBatch.php @@ -0,0 +1,80 @@ +setName('factfinder:export:batch') + ->setDescription('Internal worker command for exporting a batch of products') + ->setHidden(true); + + $this->addArgument('type', InputArgument::REQUIRED, 'Type of data to export'); + $this->addArgument('store', InputArgument::REQUIRED, 'Store ID'); + $this->addArgument('offset', InputArgument::REQUIRED, 'Offset'); + $this->addArgument('limit', InputArgument::REQUIRED, 'Limit'); + $this->addArgument('file_path', InputArgument::REQUIRED, 'Path to output file'); + + parent::configure(); + } + + protected function execute(InputInterface $input, OutputInterface $output): int + { + $this->state->setAreaCode('frontend'); + + $type = $input->getArgument('type'); + $storeId = (int) $input->getArgument('store'); + $offset = (int) $input->getArgument('offset'); + $limit = (int) $input->getArgument('limit'); + $filePath = $input->getArgument('file_path'); + + $processedCount = 0; + + $this->storeEmulation->runInStore($storeId, function () use ($type, $offset, $limit, $filePath, &$processedCount) { + $mode = ($offset === 0) ? 'w+' : 'a+'; + $stream = $this->streamFactory->create([ + 'filename' => $filePath, + 'mode' => $mode + ]); + + $generator = $this->feedGeneratorFactory->create($type); + $processedCount = $generator->generateBatch($stream, $offset, $limit, $offset === 0); + }); + + $memoryUsageMB = memory_get_usage(true) / 1024 / 1024; + $peakMemoryMB = memory_get_peak_usage(true) / 1024 / 1024; + + $result = [ + 'count' => $processedCount, + 'memory' => round($memoryUsageMB, 2), + 'peak' => round($peakMemoryMB, 2), + ]; + + $output->write(json_encode($result)); + + return Command::SUCCESS; + } +} diff --git a/src/Console/Command/WorkerExport.php b/src/Console/Command/WorkerExport.php new file mode 100644 index 00000000..613ae030 --- /dev/null +++ b/src/Console/Command/WorkerExport.php @@ -0,0 +1,224 @@ +setName('factfinder:worker-export') + ->setDescription('Export feed data using a queue/batch mechanism to prevent memory issues'); + + $this->addArgument('type', InputArgument::OPTIONAL, 'Type of data to be exported (default: product)', self::PRODUCTS_EXPORT_TYPE); + $this->addOption('store', 's', InputOption::VALUE_OPTIONAL, 'Store ID or Store Code'); + $this->addOption('upload', 'u', InputOption::VALUE_NONE, 'Upload feed via FTP'); + $this->addOption('push-import', 'i', InputOption::VALUE_NONE, 'Push Import'); + + parent::configure(); + } + + protected function execute(InputInterface $input, OutputInterface $output): int + { + $this->state->setAreaCode('frontend'); + + $storeIdInput = $input->getOption('store'); + $type = $input->getArgument('type') ?? self::PRODUCTS_EXPORT_TYPE; + $upload = $input->getOption('upload'); + $pushImport = $input->getOption('push-import'); + + if ($input->isInteractive() && empty($storeIdInput)) { + $helper = $this->getHelper('question'); + $stores = $this->storeManager->getStores(); + $storeChoices = []; + foreach ($stores as $store) { + if ($this->communicationConfig->isChannelEnabled((int)$store->getId())) { + $storeChoices[$store->getId()] = $store->getName() . ' (ID: ' . $store->getId() . ')'; + } + } + + if (!empty($storeChoices)) { + $question = new ChoiceQuestion('Select store ID:', $storeChoices); + $storeIdInput = $helper->ask($input, $output, $question); + if (preg_match('/ID: (\d+)\)/', $storeIdInput, $matches)) { + $storeIdInput = $matches[1]; + } + } + } + + $storeIds = $this->getStoreIds($storeIdInput ? (int) $storeIdInput : 0); + + if (count($storeIds) === 0) { + $output->writeln('[ERROR] There is no integration enabled for any store.'); + return Command::FAILURE; + } + + $phpBinaryFinder = new PhpExecutableFinder(); + $phpBinary = $phpBinaryFinder->find() ?: 'php'; + + foreach ($storeIds as $storeId) { + $output->writeln(''); + $output->writeln("=================================================="); + $output->writeln(">>> STARTING EXPORT FOR STORE ID: {$storeId} <<<"); + $output->writeln("=================================================="); + + $channelId = $this->communicationConfig->getChannel($storeId); + $filename = $this->feedFileService->getFeedExportFilename($type, $channelId); + $relativePath = "factfinder/{$filename}"; + $absolutePath = $this->feedFileService->getExportPath($filename); + + $dir = dirname($absolutePath); + if (!is_dir($dir)) { + mkdir($dir, 0755, true); + $output->writeln("[STEP 1] Created directory: {$dir}"); + } + + if (file_exists($absolutePath)) { + unlink($absolutePath); + $output->writeln("[STEP 1] Removed existing export file: {$absolutePath}"); + } + + if ($type === self::PRODUCTS_EXPORT_TYPE) { + $batchSize = 100; + $offset = 0; + $batchNumber = 1; + $totalCount = 0; + + $output->writeln("[STEP 2] Starting product batch processing (Batch Size: {$batchSize})..."); + + while (true) { + $output->write(sprintf(" -> Processing Batch #%d (Offset: %d)... ", $batchNumber, $offset)); + + $process = new Process([ + $phpBinary, + 'bin/magento', + 'factfinder:export:batch', + $type, + (string) $storeId, + (string) $offset, + (string) $batchSize, + $relativePath, + ]); + + $process->setTimeout(600); + $process->run(); + + if (!$process->isSuccessful()) { + $output->writeln("\n[ERROR] Batch #{$batchNumber} failed at Offset {$offset}!"); + $output->writeln("" . $process->getErrorOutput() . ""); + return Command::FAILURE; + } + + $rawOutput = trim($process->getOutput()); + preg_match('/\{.*\}/s', $rawOutput, $matches); + $jsonOutput = $matches[0] ?? '{}'; + + $result = json_decode($jsonOutput, true); + $processedCount = $result['count'] ?? 0; + $memory = $result['memory'] ?? 0; + $peak = $result['peak'] ?? 0; + + if ($processedCount === 0) { + $output->writeln("Done! No more items to process."); + break; + } + + $totalCount += $processedCount; + $output->writeln(sprintf( + "OK (Exported: %d items | RAM: %.2f MB | Peak RAM: %.2f MB)", + $processedCount, + $memory, + $peak + )); + + $offset += $batchSize; + $batchNumber++; + } + + $output->writeln("[STEP 2 COMPLETED] Total exported items for Store {$storeId}: {$totalCount}"); + $output->writeln("[FILE CREATED] {$absolutePath}"); + } else { + $output->writeln("[STEP 2] Generating {$type} export (non-batched)..."); + $output->writeln("[FILE CREATED] {$absolutePath}"); + } + + if ($upload) { + $output->writeln("[STEP 3] Uploading file {$filename} to FTP server..."); + try { + $stream = $this->feedFileService->getStream($relativePath); + $this->ftpUploader->upload($filename, $stream); + $output->writeln("[STEP 3 COMPLETED] File successfully uploaded to FTP."); + } catch (\Throwable $e) { + $output->writeln("[ERROR] FTP Upload failed: " . $e->getMessage() . ""); + return Command::FAILURE; + } + } else { + $output->writeln("[STEP 3] FTP Upload skipped (use --upload or -u option to enable)."); + } + + if ($pushImport) { + $output->writeln("[STEP 4] Triggering Push Import on FactFinder side..."); + try { + if ($this->pushImport->execute((int) $storeId)) { + $output->writeln("[STEP 4 COMPLETED] Push Import triggered successfully."); + } else { + $output->writeln("[STEP 4 FAILED] Push Import execution failed."); + } + } catch (\Throwable $e) { + $output->writeln("[ERROR] Push Import failed: " . $e->getMessage() . ""); + return Command::FAILURE; + } + } else { + $output->writeln("[STEP 4] Push Import skipped (use --push-import or -i option to enable)."); + } + + $output->writeln("=================================================="); + $output->writeln(" SUCCESS: Export process completed for Store {$storeId}"); + $output->writeln("==================================================\n"); + } + + return Command::SUCCESS; + } + + private function getStoreIds(int $storeId): array + { + $storeIds = array_map( + fn ($store) => (int) $store->getId(), + $storeId ? [$this->storeManager->getStore($storeId)] : $this->storeManager->getStores() + ); + + return array_filter($storeIds, [$this->communicationConfig, 'isChannelEnabled']); + } +} diff --git a/src/Model/Export/Catalog/DataProvider.php b/src/Model/Export/Catalog/DataProvider.php index c156b032..e96dd78b 100644 --- a/src/Model/Export/Catalog/DataProvider.php +++ b/src/Model/Export/Catalog/DataProvider.php @@ -31,6 +31,41 @@ public function getEntities(): iterable } } + /** + * @return ExportEntityInterface[] + */ + public function getEntitiesBatch(int $offset, int $limit): iterable + { + yield from []; // init generator + + $productsBatch = $this->getProductsSlice($offset, $limit); + + foreach ($productsBatch as $product) { + yield from $this->entitiesFrom($product)->getEntities(); + } + } + + private function getProductsSlice(int $offset, int $limit): iterable + { + if (method_exists($this->products, 'getBatch')) { + return $this->products->getBatch($offset, $limit); + } + + if (method_exists($this->products, 'setPageSize') && method_exists($this->products, 'setCurPage')) { + $pageNumber = (int) floor($offset / $limit) + 1; + $this->products->setPageSize($limit); + $this->products->setCurPage($pageNumber); + + return $this->products; + } + + $productsArray = is_array($this->products) + ? $this->products + : iterator_to_array($this->products, false); + + return array_slice($productsArray, $offset, $limit); + } + private function entitiesFrom(ProductInterface $product): DataProviderInterface { $type = $this->entityTypes[$product->getTypeId()] ?? $this->entityTypes[ProductType::DEFAULT_TYPE]; diff --git a/src/Model/Export/Catalog/ProductType/SimpleDataProvider.php b/src/Model/Export/Catalog/ProductType/SimpleDataProvider.php index 6f8fc8ad..24ff540f 100644 --- a/src/Model/Export/Catalog/ProductType/SimpleDataProvider.php +++ b/src/Model/Export/Catalog/ProductType/SimpleDataProvider.php @@ -29,6 +29,15 @@ public function getEntities(): iterable return [$this]; } + public function getEntitiesBatch(int $offset, int $limit): iterable + { + $entities = is_array($this->getEntities()) + ? $this->getEntities() + : iterator_to_array($this->getEntities(), false); + + return array_slice($entities, $offset, $limit); + } + public function getId(): int { return (int) $this->product->getId(); diff --git a/src/Model/Export/Feed.php b/src/Model/Export/Feed.php index 3196f60d..e4c80a5f 100644 --- a/src/Model/Export/Feed.php +++ b/src/Model/Export/Feed.php @@ -30,6 +30,29 @@ public function generate(StreamInterface $stream): void $stream->finalize(); } + public function generateBatch( + StreamInterface $stream, + int $offset, + int $limit, + bool $isFirstBatch = false + ): int { + $columns = $this->getColumns($this->fields); + + if ($isFirstBatch) { + $stream->addEntity($columns); + } + + $processedCount = $this->exporter->exportEntitiesBatch( + $stream, + $this->dataProvider, + $columns, + $offset, + $limit + ); + + return (int) $processedCount; + } + private function getColumns(array $fields): array { return array_values(array_unique([...$this->columns, ...array_map([$this, 'getFieldName'], $fields)])); diff --git a/src/Model/Exporter.php b/src/Model/Exporter.php index e89f5fd9..32c488b5 100644 --- a/src/Model/Exporter.php +++ b/src/Model/Exporter.php @@ -23,8 +23,31 @@ public function exportEntities(StreamInterface $stream, DataProviderInterface $d } } + public function exportEntitiesBatch( + StreamInterface $stream, + DataProviderInterface $dataProvider, + array $columns, + int $offset, + int $limit + ): int { + $emptyRecord = array_combine($columns, array_fill(0, count($columns), '')); + $processedCount = 0; + + $entities = $dataProvider->getEntitiesBatch($offset, $limit); + + foreach ($entities as $entity) { + $stream->addEntity($this->prepareRow($entity->toArray(), $emptyRecord)); + $processedCount++; + } + + return $processedCount; + } + private function prepareRow(array $entityData, array $emptyRecord): array { - return array_map([$this->filter, 'filterValue'], [...$emptyRecord, ...array_intersect_key($entityData, $emptyRecord)]); + return array_map( + [$this->filter, 'filterValue'], + [...$emptyRecord, ...array_intersect_key($entityData, $emptyRecord)] + ); } } diff --git a/src/Model/Stream/Csv.php b/src/Model/Stream/Csv.php index be04b847..d2e8105e 100644 --- a/src/Model/Stream/Csv.php +++ b/src/Model/Stream/Csv.php @@ -17,7 +17,8 @@ class Csv implements StreamInterface public function __construct( private readonly Filesystem $filesystem, - private readonly string $filename = 'factfinder/export.csv' + private readonly string $filename = 'factfinder/export.csv', + private readonly string $mode = 'w+' ) { } @@ -40,7 +41,7 @@ private function getStream(): WriteInterface { if (!isset($this->stream)) { $directory = $this->filesystem->getDirectoryWrite(DirectoryList::VAR_DIR); - $this->stream = $directory->openFile($directory->getAbsolutePath($this->filename), 'w+'); + $this->stream = $directory->openFile($directory->getAbsolutePath($this->filename), $this->mode); $this->stream->lock(); } diff --git a/src/etc/di.xml b/src/etc/di.xml index 99d91f66..89cfa9a8 100644 --- a/src/etc/di.xml +++ b/src/etc/di.xml @@ -230,6 +230,16 @@ Omikron\Factfinder\Console\Command\Export + + + Omikron\Factfinder\Console\Command\ExportBatch + + + + + Omikron\Factfinder\Console\Command\WorkerExport + + From 5549e96c1fbf06aa29f1a51552f25b12c1f116ed Mon Sep 17 00:00:00 2001 From: robertsaternus Date: Mon, 17 Aug 2026 20:42:37 +0200 Subject: [PATCH 2/6] Update --- src/Console/Command/WorkerExport.php | 329 +++++++++++------- .../frontend/templates/ff/searchbox.phtml | 2 +- 2 files changed, 197 insertions(+), 134 deletions(-) diff --git a/src/Console/Command/WorkerExport.php b/src/Console/Command/WorkerExport.php index 613ae030..72f47293 100644 --- a/src/Console/Command/WorkerExport.php +++ b/src/Console/Command/WorkerExport.php @@ -42,7 +42,12 @@ protected function configure(): void $this->setName('factfinder:worker-export') ->setDescription('Export feed data using a queue/batch mechanism to prevent memory issues'); - $this->addArgument('type', InputArgument::OPTIONAL, 'Type of data to be exported (default: product)', self::PRODUCTS_EXPORT_TYPE); + $this->addArgument( + 'type', + InputArgument::OPTIONAL, + 'Type of data to be exported (default: product)', + self::PRODUCTS_EXPORT_TYPE + ); $this->addOption('store', 's', InputOption::VALUE_OPTIONAL, 'Store ID or Store Code'); $this->addOption('upload', 'u', InputOption::VALUE_NONE, 'Upload feed via FTP'); $this->addOption('push-import', 'i', InputOption::VALUE_NONE, 'Push Import'); @@ -54,162 +59,220 @@ protected function execute(InputInterface $input, OutputInterface $output): int { $this->state->setAreaCode('frontend'); - $storeIdInput = $input->getOption('store'); - $type = $input->getArgument('type') ?? self::PRODUCTS_EXPORT_TYPE; - $upload = $input->getOption('upload'); - $pushImport = $input->getOption('push-import'); + $storeIds = $this->resolveStoreIds($input, $output); + if (empty($storeIds)) { + $output->writeln('[ERROR] There is no integration enabled for any store.'); + return Command::FAILURE; + } - if ($input->isInteractive() && empty($storeIdInput)) { - $helper = $this->getHelper('question'); - $stores = $this->storeManager->getStores(); - $storeChoices = []; - foreach ($stores as $store) { - if ($this->communicationConfig->isChannelEnabled((int)$store->getId())) { - $storeChoices[$store->getId()] = $store->getName() . ' (ID: ' . $store->getId() . ')'; - } + $phpBinaryFinder = new PhpExecutableFinder(); + $phpBinary = $phpBinaryFinder->find() ?: 'php'; + $type = $input->getArgument('type') ?? self::PRODUCTS_EXPORT_TYPE; + $upload = (bool) $input->getOption('upload'); + $pushImport = (bool) $input->getOption('push-import'); + + foreach ($storeIds as $storeId) { + $success = $this->exportForStore($storeId, $type, $upload, $pushImport, $output, $phpBinary); + if (!$success) { + return Command::FAILURE; } + } - if (!empty($storeChoices)) { - $question = new ChoiceQuestion('Select store ID:', $storeChoices); - $storeIdInput = $helper->ask($input, $output, $question); - if (preg_match('/ID: (\d+)\)/', $storeIdInput, $matches)) { - $storeIdInput = $matches[1]; - } + return Command::SUCCESS; + } + + private function exportForStore( + int $storeId, + string $type, + bool $upload, + bool $pushImport, + OutputInterface $output, + string $phpBinary + ): bool { + $output->writeln(''); + $output->writeln('=================================================='); + $output->writeln(">>> STARTING EXPORT FOR STORE ID: {$storeId} <<<"); + $output->writeln('=================================================='); + + $channelId = $this->communicationConfig->getChannel($storeId); + $filename = $this->feedFileService->getFeedExportFilename($type, $channelId); + $relativePath = "factfinder/{$filename}"; + $absolutePath = $this->feedFileService->getExportPath($filename); + + $this->prepareExportFile($absolutePath, $output); + + if ($type === self::PRODUCTS_EXPORT_TYPE) { + $batchSuccess = $this->runProductBatchExport($storeId, $type, $relativePath, $output, $phpBinary); + if (!$batchSuccess) { + return false; } + $output->writeln("[FILE CREATED] {$absolutePath}"); } - $storeIds = $this->getStoreIds($storeIdInput ? (int) $storeIdInput : 0); + if (!$this->handleUpload($upload, $filename, $relativePath, $output)) { + return false; + } - if (count($storeIds) === 0) { - $output->writeln('[ERROR] There is no integration enabled for any store.'); - return Command::FAILURE; + if (!$this->handlePushImport($pushImport, $storeId, $filename, $output)) { + return false; } - $phpBinaryFinder = new PhpExecutableFinder(); - $phpBinary = $phpBinaryFinder->find() ?: 'php'; + $output->writeln(" SUCCESS: Export process completed for Store {$storeId}\n"); + return true; + } - foreach ($storeIds as $storeId) { - $output->writeln(''); - $output->writeln("=================================================="); - $output->writeln(">>> STARTING EXPORT FOR STORE ID: {$storeId} <<<"); - $output->writeln("=================================================="); - - $channelId = $this->communicationConfig->getChannel($storeId); - $filename = $this->feedFileService->getFeedExportFilename($type, $channelId); - $relativePath = "factfinder/{$filename}"; - $absolutePath = $this->feedFileService->getExportPath($filename); - - $dir = dirname($absolutePath); - if (!is_dir($dir)) { - mkdir($dir, 0755, true); - $output->writeln("[STEP 1] Created directory: {$dir}"); + private function prepareExportFile(string $absolutePath, OutputInterface $output): void + { + $dir = dirname($absolutePath); + if (!is_dir($dir)) { + mkdir($dir, 0755, true); + $output->writeln("[STEP 1] Created directory: {$dir}"); + } + + if (file_exists($absolutePath)) { + unlink($absolutePath); + $output->writeln("[STEP 1] Removed existing export file: {$absolutePath}"); + } + } + + private function runProductBatchExport( + int $storeId, + string $type, + string $relativePath, + OutputInterface $output, + string $phpBinary + ): bool { + $batchSize = 100; + $offset = 0; + $batchNumber = 1; + $totalCount = 0; + + $output->writeln("[STEP 2] Starting product batch processing (Batch Size: {$batchSize})..."); + + while (true) { + $output->write(sprintf(' -> Processing Batch #%d (Offset: %d)... ', $batchNumber, $offset)); + + $process = new Process([ + $phpBinary, + 'bin/magento', + 'factfinder:export:batch', + $type, + (string) $storeId, + (string) $offset, + (string) $batchSize, + $relativePath, + ]); + $process->setTimeout(600); + $process->run(); + + if (!$process->isSuccessful()) { + $output->writeln("\n[ERROR] Batch #{$batchNumber} failed at Offset {$offset}!"); + $output->writeln("{$process->getErrorOutput()}"); + return false; } - if (file_exists($absolutePath)) { - unlink($absolutePath); - $output->writeln("[STEP 1] Removed existing export file: {$absolutePath}"); + $result = $this->parseProcessOutput($process->getOutput()); + $count = $result['count'] ?? 0; + + if ($count === 0) { + $output->writeln('Done! No more items to process.'); + break; } - if ($type === self::PRODUCTS_EXPORT_TYPE) { - $batchSize = 100; - $offset = 0; - $batchNumber = 1; - $totalCount = 0; - - $output->writeln("[STEP 2] Starting product batch processing (Batch Size: {$batchSize})..."); - - while (true) { - $output->write(sprintf(" -> Processing Batch #%d (Offset: %d)... ", $batchNumber, $offset)); - - $process = new Process([ - $phpBinary, - 'bin/magento', - 'factfinder:export:batch', - $type, - (string) $storeId, - (string) $offset, - (string) $batchSize, - $relativePath, - ]); - - $process->setTimeout(600); - $process->run(); - - if (!$process->isSuccessful()) { - $output->writeln("\n[ERROR] Batch #{$batchNumber} failed at Offset {$offset}!"); - $output->writeln("" . $process->getErrorOutput() . ""); - return Command::FAILURE; - } - - $rawOutput = trim($process->getOutput()); - preg_match('/\{.*\}/s', $rawOutput, $matches); - $jsonOutput = $matches[0] ?? '{}'; - - $result = json_decode($jsonOutput, true); - $processedCount = $result['count'] ?? 0; - $memory = $result['memory'] ?? 0; - $peak = $result['peak'] ?? 0; - - if ($processedCount === 0) { - $output->writeln("Done! No more items to process."); - break; - } - - $totalCount += $processedCount; - $output->writeln(sprintf( - "OK (Exported: %d items | RAM: %.2f MB | Peak RAM: %.2f MB)", - $processedCount, - $memory, - $peak - )); - - $offset += $batchSize; - $batchNumber++; - } + $totalCount += $count; + $output->writeln(sprintf( + 'OK (Exported: %d items | RAM: %.2f MB | Peak RAM: %.2f MB)', + $count, + $result['memory'] ?? 0, + $result['peak'] ?? 0 + )); + + $offset += $batchSize; + $batchNumber++; + } + + $output->writeln("[STEP 2 COMPLETED] Total exported items: {$totalCount}"); + return true; + } + + private function parseProcessOutput(string $rawOutput): array + { + preg_match('/\{.*\}/s', trim($rawOutput), $matches); + return json_decode($matches[0] ?? '{}', true) ?: []; + } + + private function handleUpload( + bool $upload, + string $filename, + string $relativePath, + OutputInterface $output + ): bool { + if (!$upload) { + $output->writeln('[STEP 3] FTP Upload skipped.'); + return true; + } + + $output->writeln("[STEP 3] Uploading file {$filename} to FTP server..."); + try { + $stream = $this->feedFileService->getStream($relativePath); + $this->ftpUploader->upload($filename, $stream); + $output->writeln('[STEP 3 COMPLETED] File successfully uploaded to FTP.'); + return true; + } catch (\Throwable $e) { + $output->writeln("[ERROR] FTP Upload failed: {$e->getMessage()}"); + return false; + } + } - $output->writeln("[STEP 2 COMPLETED] Total exported items for Store {$storeId}: {$totalCount}"); - $output->writeln("[FILE CREATED] {$absolutePath}"); - } else { - $output->writeln("[STEP 2] Generating {$type} export (non-batched)..."); - $output->writeln("[FILE CREATED] {$absolutePath}"); + private function handlePushImport( + bool $pushImport, + int $storeId, + string $filename, + OutputInterface $output + ): bool { + if (!$pushImport) { + $output->writeln('[STEP 4] Push Import skipped.'); + return true; + } + + $output->writeln('[STEP 4] Triggering Push Import on FactFinder side...'); + try { + if ($this->pushImport->execute($storeId)) { + $output->writeln('[STEP 4 COMPLETED] Push Import triggered successfully.'); + return true; } + $output->writeln('[STEP 4 FAILED] Push Import execution failed.'); + return false; + } catch (\Throwable $e) { + $output->writeln("[ERROR] Push Import failed: {$e->getMessage()}"); + return false; + } + } - if ($upload) { - $output->writeln("[STEP 3] Uploading file {$filename} to FTP server..."); - try { - $stream = $this->feedFileService->getStream($relativePath); - $this->ftpUploader->upload($filename, $stream); - $output->writeln("[STEP 3 COMPLETED] File successfully uploaded to FTP."); - } catch (\Throwable $e) { - $output->writeln("[ERROR] FTP Upload failed: " . $e->getMessage() . ""); - return Command::FAILURE; + private function resolveStoreIds(InputInterface $input, OutputInterface $output): array + { + $storeIdInput = $input->getOption('store'); + + if ($input->isInteractive() && empty($storeIdInput)) { + $storeChoices = []; + foreach ($this->storeManager->getStores() as $store) { + if ($this->communicationConfig->isChannelEnabled((int) $store->getId())) { + $storeChoices[$store->getId()] = "{$store->getName()} (ID: {$store->getId()})"; } - } else { - $output->writeln("[STEP 3] FTP Upload skipped (use --upload or -u option to enable)."); } - if ($pushImport) { - $output->writeln("[STEP 4] Triggering Push Import on FactFinder side..."); - try { - if ($this->pushImport->execute((int) $storeId)) { - $output->writeln("[STEP 4 COMPLETED] Push Import triggered successfully."); - } else { - $output->writeln("[STEP 4 FAILED] Push Import execution failed."); - } - } catch (\Throwable $e) { - $output->writeln("[ERROR] Push Import failed: " . $e->getMessage() . ""); - return Command::FAILURE; + if (!empty($storeChoices)) { + $helper = $this->getHelper('question'); + $question = new ChoiceQuestion('Select store ID:', $storeChoices); + $storeIdInput = $helper->ask($input, $output, $question); + + if (preg_match('/ID: (\d+)\)/', (string) $storeIdInput, $matches)) { + $storeIdInput = $matches[1]; } - } else { - $output->writeln("[STEP 4] Push Import skipped (use --push-import or -i option to enable)."); } - - $output->writeln("=================================================="); - $output->writeln(" SUCCESS: Export process completed for Store {$storeId}"); - $output->writeln("==================================================\n"); } - return Command::SUCCESS; + return $this->getStoreIds($storeIdInput ? (int) $storeIdInput : 0); } private function getStoreIds(int $storeId): array diff --git a/src/view/frontend/templates/ff/searchbox.phtml b/src/view/frontend/templates/ff/searchbox.phtml index fc16b18a..5a93e4e7 100644 --- a/src/view/frontend/templates/ff/searchbox.phtml +++ b/src/view/frontend/templates/ff/searchbox.phtml @@ -1,7 +1,7 @@ helper('Omikron\Factfinder\Helper\ConfigurationHelper')->getConfig('factfinder/components/popular_searches'); +$isPopularSearchesEnabled = $block->helper('Omikron\Factfinder\Helper\ConfigurationHelper')->getConfig('factfinder/components/popular_searches'); ?>
From 64284052577c20afc88e1460dc48daa171733048 Mon Sep 17 00:00:00 2001 From: robertsaternus Date: Mon, 17 Aug 2026 20:52:05 +0200 Subject: [PATCH 3/6] Update --- src/Console/Command/ExportBatch.php | 2 +- src/Console/Command/WorkerExport.php | 3 +-- src/Model/Export/Feed.php | 5 ++--- 3 files changed, 4 insertions(+), 6 deletions(-) diff --git a/src/Console/Command/ExportBatch.php b/src/Console/Command/ExportBatch.php index 0fc99115..54c6b730 100644 --- a/src/Console/Command/ExportBatch.php +++ b/src/Console/Command/ExportBatch.php @@ -61,7 +61,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int ]); $generator = $this->feedGeneratorFactory->create($type); - $processedCount = $generator->generateBatch($stream, $offset, $limit, $offset === 0); + $processedCount = $generator->generateBatch($stream, $offset, $limit); }); $memoryUsageMB = memory_get_usage(true) / 1024 / 1024; diff --git a/src/Console/Command/WorkerExport.php b/src/Console/Command/WorkerExport.php index 72f47293..d5e5171e 100644 --- a/src/Console/Command/WorkerExport.php +++ b/src/Console/Command/WorkerExport.php @@ -113,7 +113,7 @@ private function exportForStore( return false; } - if (!$this->handlePushImport($pushImport, $storeId, $filename, $output)) { + if (!$this->handlePushImport($pushImport, $storeId, $output)) { return false; } @@ -227,7 +227,6 @@ private function handleUpload( private function handlePushImport( bool $pushImport, int $storeId, - string $filename, OutputInterface $output ): bool { if (!$pushImport) { diff --git a/src/Model/Export/Feed.php b/src/Model/Export/Feed.php index e4c80a5f..c7641461 100644 --- a/src/Model/Export/Feed.php +++ b/src/Model/Export/Feed.php @@ -33,12 +33,11 @@ public function generate(StreamInterface $stream): void public function generateBatch( StreamInterface $stream, int $offset, - int $limit, - bool $isFirstBatch = false + int $limit ): int { $columns = $this->getColumns($this->fields); - if ($isFirstBatch) { + if ($offset === 0) { $stream->addEntity($columns); } From 11fef2be8f60c1a7a381a348fbcc5232138b4683 Mon Sep 17 00:00:00 2001 From: robertsaternus Date: Mon, 17 Aug 2026 21:01:53 +0200 Subject: [PATCH 4/6] Update --- CHANGELOG.md | 4 ++++ README.md | 9 +++++++++ 2 files changed, 13 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0c237056..20c51f31 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,8 @@ # Changelog +## Unreleased +### Add +-Alternative product export command based on a queue system + ## [v5.4.0] - 2026.08.05 ### Add - Add support for Atlas AI diff --git a/README.md b/README.md index 39c38c91..c0f92307 100644 --- a/README.md +++ b/README.md @@ -276,6 +276,15 @@ Command name: `factfinder:export [TYPE]`. You can add execution of this command - store - define a store, which the product data will be taken from - skip-ftp-upload - skips the ftp upload - skip-push-import - skips triggering import + +#### Running export with a worker + +Since version 5.5.0, we've introduced a new export flow based on the worker. +The biggest advantage of this export method is the reduced memory usage, which is especially helpful when you have a large or complex products catalog. +Currently, this command is only available from the CLI and can be executed by: + + php [MAGENTO_ROOT]/bin/magento factfinder:worker-export + ## Web Component Integration From 6b409cc8f4abc3a0cd280a9cc8442c6b13f3e4be Mon Sep 17 00:00:00 2001 From: robertsaternus Date: Tue, 18 Aug 2026 10:16:52 +0200 Subject: [PATCH 5/6] Add questions to command --- src/Console/Command/WorkerExport.php | 124 +++++++++++++++++++-------- 1 file changed, 87 insertions(+), 37 deletions(-) diff --git a/src/Console/Command/WorkerExport.php b/src/Console/Command/WorkerExport.php index d5e5171e..719c4721 100644 --- a/src/Console/Command/WorkerExport.php +++ b/src/Console/Command/WorkerExport.php @@ -6,6 +6,7 @@ use Magento\Framework\App\State; use Magento\Store\Model\StoreManagerInterface; +use Omikron\Factfinder\Api\StreamInterfaceFactory; use Omikron\Factfinder\Model\Api\PushImport; use Omikron\Factfinder\Model\Config\CommunicationConfig; use Omikron\Factfinder\Model\FtpUploader; @@ -27,12 +28,13 @@ class WorkerExport extends Command private const PRODUCTS_EXPORT_TYPE = 'product'; public function __construct( - private readonly StoreManagerInterface $storeManager, - private readonly FtpUploader $ftpUploader, - private readonly CommunicationConfig $communicationConfig, - private readonly PushImport $pushImport, - private readonly State $state, - private readonly FeedFileService $feedFileService + private readonly StoreManagerInterface $storeManager, + private readonly FtpUploader $ftpUploader, + private readonly CommunicationConfig $communicationConfig, + private readonly PushImport $pushImport, + private readonly State $state, + private readonly FeedFileService $feedFileService, + private readonly StreamInterfaceFactory $streamFactory ) { parent::__construct(); } @@ -59,7 +61,8 @@ protected function execute(InputInterface $input, OutputInterface $output): int { $this->state->setAreaCode('frontend'); - $storeIds = $this->resolveStoreIds($input, $output); + [$storeIds, $upload, $pushImport] = $this->resolveExecutionParameters($input, $output); + if (empty($storeIds)) { $output->writeln('[ERROR] There is no integration enabled for any store.'); return Command::FAILURE; @@ -68,8 +71,6 @@ protected function execute(InputInterface $input, OutputInterface $output): int $phpBinaryFinder = new PhpExecutableFinder(); $phpBinary = $phpBinaryFinder->find() ?: 'php'; $type = $input->getArgument('type') ?? self::PRODUCTS_EXPORT_TYPE; - $upload = (bool) $input->getOption('upload'); - $pushImport = (bool) $input->getOption('push-import'); foreach ($storeIds as $storeId) { $success = $this->exportForStore($storeId, $type, $upload, $pushImport, $output, $phpBinary); @@ -81,6 +82,78 @@ protected function execute(InputInterface $input, OutputInterface $output): int return Command::SUCCESS; } + private function resolveExecutionParameters(InputInterface $input, OutputInterface $output): array + { + $storeIdInput = $input->getOption('store'); + $upload = (bool) $input->getOption('upload'); + $pushImport = (bool) $input->getOption('push-import'); + + if ($input->isInteractive()) { + $helper = $this->getHelper('question'); + + if (empty($storeIdInput)) { + $storeIdInput = $this->askStoreId($input, $output, $helper); + } + + if (!$upload) { + $upload = $this->askYesNoQuestion( + $input, + $output, + $helper, + 'Should upload feed to FTP after exporting?' + ); + } + + if (!$pushImport) { + $pushImport = $this->askYesNoQuestion( + $input, + $output, + $helper, + 'Should trigger Push Import after uploading?' + ); + } + } + + $storeIds = $this->getStoreIds($storeIdInput ? (int) $storeIdInput : 0); + + return [$storeIds, $upload, $pushImport]; + } + + private function askStoreId(InputInterface $input, OutputInterface $output, mixed $helper): ?string + { + $storeChoices = []; + foreach ($this->storeManager->getStores() as $store) { + if ($this->communicationConfig->isChannelEnabled((int) $store->getId())) { + $storeChoices[$store->getId()] = "{$store->getName()} (ID: {$store->getId()})"; + } + } + + if (empty($storeChoices)) { + return null; + } + + $question = new ChoiceQuestion('Select store ID:', $storeChoices); + $storeIdInput = $helper->ask($input, $output, $question); + + if (preg_match('/ID: (\d+)\)/', (string) $storeIdInput, $matches)) { + return $matches[1]; + } + + return (string) $storeIdInput; + } + + private function askYesNoQuestion( + InputInterface $input, + OutputInterface $output, + mixed $helper, + string $questionText + ): bool { + $question = new ChoiceQuestion("{$questionText} (default: no)", ['no', 'yes'], 0); + $answer = $helper->ask($input, $output, $question); + + return $answer === 'yes'; + } + private function exportForStore( int $storeId, string $type, @@ -150,7 +223,7 @@ private function runProductBatchExport( $output->writeln("[STEP 2] Starting product batch processing (Batch Size: {$batchSize})..."); while (true) { - $output->write(sprintf(' -> Processing Batch #%d (Offset: %d)... ', $batchNumber, $offset)); + $output->write(sprintf(" -> Processing Batch #%d (Offset: %d)... ", $batchNumber, $offset)); $process = new Process([ $phpBinary, @@ -214,7 +287,10 @@ private function handleUpload( $output->writeln("[STEP 3] Uploading file {$filename} to FTP server..."); try { - $stream = $this->feedFileService->getStream($relativePath); + $stream = $this->streamFactory->create([ + 'filename' => $relativePath, + 'mode' => 'r', + ]); $this->ftpUploader->upload($filename, $stream); $output->writeln('[STEP 3 COMPLETED] File successfully uploaded to FTP.'); return true; @@ -248,32 +324,6 @@ private function handlePushImport( } } - private function resolveStoreIds(InputInterface $input, OutputInterface $output): array - { - $storeIdInput = $input->getOption('store'); - - if ($input->isInteractive() && empty($storeIdInput)) { - $storeChoices = []; - foreach ($this->storeManager->getStores() as $store) { - if ($this->communicationConfig->isChannelEnabled((int) $store->getId())) { - $storeChoices[$store->getId()] = "{$store->getName()} (ID: {$store->getId()})"; - } - } - - if (!empty($storeChoices)) { - $helper = $this->getHelper('question'); - $question = new ChoiceQuestion('Select store ID:', $storeChoices); - $storeIdInput = $helper->ask($input, $output, $question); - - if (preg_match('/ID: (\d+)\)/', (string) $storeIdInput, $matches)) { - $storeIdInput = $matches[1]; - } - } - } - - return $this->getStoreIds($storeIdInput ? (int) $storeIdInput : 0); - } - private function getStoreIds(int $storeId): array { $storeIds = array_map( From aa3567eb4544ac10b40b08bda641b7aa08d4ff04 Mon Sep 17 00:00:00 2001 From: robertsaternus Date: Tue, 18 Aug 2026 10:19:20 +0200 Subject: [PATCH 6/6] Fix code style --- src/Console/Command/WorkerExport.php | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/Console/Command/WorkerExport.php b/src/Console/Command/WorkerExport.php index 719c4721..e895f3a3 100644 --- a/src/Console/Command/WorkerExport.php +++ b/src/Console/Command/WorkerExport.php @@ -223,7 +223,7 @@ private function runProductBatchExport( $output->writeln("[STEP 2] Starting product batch processing (Batch Size: {$batchSize})..."); while (true) { - $output->write(sprintf(" -> Processing Batch #%d (Offset: %d)... ", $batchNumber, $offset)); + $output->write(sprintf(' -> Processing Batch #%d (Offset: %d)... ', $batchNumber, $offset)); $process = new Process([ $phpBinary,