vendor/blackbit/data-director/EventListener/AutomaticImportListener.php line 307

Open in your IDE?
  1. <?php
  2. /**
  3.  * Copyright Blackbit digital Commerce GmbH <info@blackbit.de>
  4.  *
  5.  * This program is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation; either version 3 of the License, or (at your option) any later version.
  6.  *
  7.  * This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
  8.  *
  9.  * You should have received a copy of the GNU General Public License along with this program; if not, write to the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  10.  */
  12. namespace Blackbit\DataDirectorBundle\EventListener;
  14. use Blackbit\DataDirectorBundle\Controller\ImportController;
  15. use Blackbit\DataDirectorBundle\lib\Pim\Import\CallbackFunction;
  16. use Blackbit\DataDirectorBundle\lib\Pim\Item\ImporterInterface;
  17. use Blackbit\DataDirectorBundle\lib\Pim\Item\ItemMoldBuilder;
  18. use Blackbit\DataDirectorBundle\lib\Pim\RawData\Importer;
  19. use Blackbit\DataDirectorBundle\lib\Pim\Serializer;
  20. use Blackbit\DataDirectorBundle\model\Dataport;
  21. use Blackbit\DataDirectorBundle\model\DataportResource;
  22. use Blackbit\DataDirectorBundle\model\Fieldmapping;
  23. use Blackbit\DataDirectorBundle\model\ImportStatus;
  24. use Blackbit\DataDirectorBundle\model\PimcoreDbRepository;
  25. use Blackbit\DataDirectorBundle\model\Queue;
  26. use Blackbit\DataDirectorBundle\model\RawItem;
  27. use Blackbit\DataDirectorBundle\model\RawItemData;
  28. use Blackbit\DataDirectorBundle\Tools\Installer;
  29. use InvalidArgumentException;
  30. use Pimcore\Db;
  31. use Pimcore\Event\Model\AssetEvent;
  32. use Pimcore\Event\Model\ElementEventInterface;
  33. use Pimcore\Logger;
  34. use Pimcore\Model\AbstractModel;
  35. use Pimcore\Model\Asset;
  36. use Pimcore\Model\DataObject\AbstractObject;
  37. use Pimcore\Model\DataObject\Concrete;
  38. use Pimcore\Model\Element\DirtyIndicatorInterface;
  39. use Pimcore\Model\Element\ElementInterface;
  40. use Pimcore\Model\Element\Service;
  41. use Pimcore\Model\User;
  42. use Pimcore\Model\Version;
  43. use Pimcore\Tool;
  44. use Blackbit\DataDirectorBundle\lib\Pim\Cli;
  45. use Psr\Log\LoggerAwareInterface;
  46. use Psr\Log\LoggerAwareTrait;
  47. use Psr\Log\LoggerInterface;
  48. use Symfony\Component\HttpFoundation\Request;
  50. class AutomaticImportListener implements LoggerAwareInterface
  51. {
  52.     use LoggerAwareTrait;
  54.     private static $processCommandRegistered = [];
  56.     /** @var ItemMoldBuilder */
  57.     private $itemMoldBuilder;
  59.     public function __construct(ItemMoldBuilder $itemMoldBuilder)
  60.     {
  61.         $this->itemMoldBuilder = $itemMoldBuilder;
  62.     }
  64.     public function startImports(ElementEventInterface $e) {
  65.         if(\method_exists($e, 'getArgument')) {
  66.             try {
  67.                 $saveVersionOnly = $e->getArgument('saveVersionOnly');
  68.                 if($saveVersionOnly) {
  69.                     return;
  70.                 }
  71.             } catch(\InvalidArgumentException $exception) {
  72.             }
  73.         }
  75.         $object = $e->getElement();
  77.         $user = Tool\Admin::getCurrentUser();
  78.         if($user instanceof User) {
  79.             PimcoreDbRepository::getInstance()->execute('UPDATE edit_lock SET date=? WHERE cid=? AND ctype=? AND userId=?', [time(), $object->getId(), Service::getElementType($object), $user->getId()]);
  80.         }
  82.         $dataports = Dataport::getInstance();
  83.         $dataportResourceRepository = DataportResource::getInstance();
  85.         $queue = Queue::getInstance();
  87.         foreach($dataports->find([], 'name') as $dataport) {
  88.             $originalLocale = \Pimcore::getContainer()->get('pimcore.locale')->getLocale();
  89.             try {
  90.                 $sourceConfig = unserialize($dataport['sourceconfig'], ['allowed_classes' => false]);
  91.                 $targetConfig = unserialize($dataport['targetconfig'], ['allowed_classes' => false]);
  93.                 if(!empty($sourceConfig['autoImport'])) {
  94.                     $parser = $dataports->getParser($dataport['id']);
  95.                     if(method_exists($parser, 'disableLoggingNotFoundImportResource')) {
  96.                         $parser->disableLoggingNotFoundImportResource();
  97.                     }
  99.                     if(\method_exists($parser, 'setSourceFile') && \method_exists($parser, 'getFileConditionFromObject')) {
  100.                         if (empty($targetConfig['itemClass'])) {
  101.                             // Export
  102.                             if (empty($sourceConfig['incrementalExport'])) {
  103.                                 $dataportResources = $dataportResourceRepository->find(['dataportId = ?' => $dataport['id']]);
  104.                             } else {
  105.                                 $dataportResources = [
  106.                                     $dataportResourceRepository->create(
  107.                                         [
  108.                                             'dataportId' => $dataport['id'],
  109.                                             'resource' => \json_encode([], JSON_UNESCAPED_SLASHES)
  110.                                         ]
  111.                                     )
  112.                                 ];
  113.                             }
  115.                             $dataportResourceIdsStillContainingObject = [];
  116.                             foreach ($dataportResources as $dataportResource) {
  117.                                 try {
  118.                                     $resourceSettings = \json_decode($dataportResource['resource'], true);
  119.                                     $parser->setSourceFile($resourceSettings['file'] ?? null);
  120.                                     if (!empty($resourceSettings['locale'])) {
  121.                                         \Pimcore::getContainer()->get('pimcore.locale')->setLocale($resourceSettings['locale']);
  122.                                     } else {
  123.                                         $resourceSettings['locale'] = \Pimcore::getContainer()->get('pimcore.locale')->getLocale();
  124.                                         if ($resourceSettings['locale'] === null) {
  125.                                             $resourceSettings['locale'] = Tool::getDefaultLanguage();
  126.                                         }
  127.                                     }
  129.                                     $source = $parser->getFileConditionFromObject($object);
  130.                                     if ($source !== null) {
  131.                                         $dataportResourceIdsStillContainingObject[$source][$resourceSettings['locale']][] = $dataportResource['id'];
  132.                                     } elseif (empty($sourceConfig['incrementalExport'])) {
  133.                                         $source = $parser->getFileConditionFromObject($object, false);
  134.                                         if ($source !== null) {
  135.                                             $tmpSourceConfig = $sourceConfig;
  136.                                             if (\method_exists($parser, 'setConfig')) {
  137.                                                 if (\method_exists($parser, 'getConfig')) {
  138.                                                     $tmpSourceConfig = $parser->getConfig();
  139.                                                 }
  140.                                                 $tmpSourceConfig['file'] = '';
  142.                                                 $tmpSourceConfig['dataportId'] = $dataport['id'];
  143.                                                 $parser->setConfig($tmpSourceConfig);
  144.                                             }
  146.                                             $parser->setSourceFile($source);
  148.                                             $keyFields = [];
  149.                                             foreach ($tmpSourceConfig['fields'] as $fieldIndex => $field) {
  150.                                                 if (!empty($field['exportKey'])) {
  151.                                                     $keyFields[$fieldIndex] = $field;
  152.                                                 }
  153.                                             }
  155.                                             $hashs = [];
  156.                                             if ($dataport['sourcetype'] === 'pimcore') {
  157.                                                 $locales = Tool::getValidLanguages();
  158.                                                 if ($dataportResource !== null) {
  159.                                                     $resource = \json_decode($dataportResource['resource'], true);
  160.                                                     if ($resource['locale']) {
  161.                                                         $locales = [$resource['locale']];
  162.                                                     }
  163.                                                 }
  164.                                                 foreach ($locales as $language) {
  165.                                                     \Pimcore::getContainer()->get('pimcore.locale')->setLocale($language);
  167.                                                     foreach ($parser as $rawItemData) {
  168.                                                         if ($rawItemData === null) {
  169.                                                             continue;
  170.                                                         }
  172.                                                         $rawItemData = array_filter(
  173.                                                             $rawItemData,
  174.                                                             static function ($fieldId) use ($keyFields) {
  175.                                                                 return isset($keyFields[$fieldId]);
  176.                                                             },
  177.                                                             ARRAY_FILTER_USE_KEY
  178.                                                         );
  180.                                                         $hashs[] = Importer::getHash($rawItemData);
  181.                                                     }
  182.                                                 }
  183.                                             } else {
  184.                                                 foreach ($parser as $rawItemData) {
  185.                                                     if ($rawItemData === null) {
  186.                                                         continue;
  187.                                                     }
  189.                                                     $rawItemData = array_filter(
  190.                                                         $rawItemData,
  191.                                                         static function ($fieldId) use ($keyFields) {
  192.                                                             return isset($keyFields[$fieldId]);
  193.                                                         },
  194.                                                         ARRAY_FILTER_USE_KEY
  195.                                                     );
  197.                                                     $hashs[] = Importer::getHash($rawItemData);
  198.                                                 }
  199.                                             }
  201.                                             if(count($hashs) > 0) {
  202.                                                 $countDeletedRawItems = Db::get()->executeUpdate('DELETE FROM '.Installer::TABLE_RAWITEM.' WHERE dataport_resource_id = ? AND hash IN ("'.implode('","', $hashs).'")', [$dataportResource['id']]);
  203.                                                 if ($countDeletedRawItems) {
  204.                                                     $this->queueProcessRawData($dataportResource['id'], $dataport['id']);
  205.                                                 }
  206.                                             }
  207.                                         }
  208.                                     }
  209.                                 } catch (\Throwable $e) {
  210.                                     if(!$e instanceof SkipTriggerAutomaticImportException) {
  211.                                         $error = 'Check for automatic start for dataport #'.$dataport['id'].' failed: '.(string)$e;
  212.                                         if($this->logger) {
  213.                                             $this->logger->error($error);
  214.                                         } else {
  215.                                             Logger::error($error);
  216.                                         }
  217.                                     }
  218.                                 }
  219.                             }
  221.                             foreach($dataportResourceIdsStillContainingObject as $dataportResourceQuery => $dataportResourcesWithSameLocale) {
  222.                                 foreach($dataportResourcesWithSameLocale as $locale => $dataportResourceIds) {
  223.                                     if (empty($sourceConfig['incrementalExport'])) {
  224.                                         if ($dataport['sourcetype'] === 'pimcore' && preg_match('/ IN \(((\d+,?)+)\)/', $dataportResourceQuery, $match)) {
  225.                                             $ids = explode(',', $match[1]);
  226.                                             sort($ids, SORT_NUMERIC);
  227.                                             foreach ($ids as $id) {
  228.                                                 $queue->create(
  229.                                                     [
  230.                                                         'command' => 'data-director:extract '.$dataport['id'].' "'.str_replace([$match[0], '"', '$', '`'], ['='.$id, '\\"', '\\$', '\\`'], $dataportResourceQuery).'"'.(!empty($locale) ? ' --locale='.$locale : '').' --dataport-resource-id='.implode(',', $dataportResourceIds),
  231.                                                         'triggered_by' => 'Element '.$object->getFullPath().' (#'.$object->getId().') got saved'.(($user instanceof User) ? ' by '.$user->getUsername() : ''),
  232.                                                         'worker_id' => $dataport['id']
  233.                                                     ]
  234.                                                 );
  235.                                             }
  236.                                         } else {
  237.                                             $queue->create(
  238.                                                 [
  239.                                                     'command' => 'data-director:extract '.$dataport['id'].' "'.str_replace(['"', '$', '`'], ['\\"', '\\$', '\\`'], $dataportResourceQuery).'"'.(!empty($locale) ? ' --locale='.$locale : '').' --dataport-resource-id='.implode(',', $dataportResourceIds),
  240.                                                     'triggered_by' => 'Element '.$object->getFullPath().' (#'.$object->getId().') got saved'.(($user instanceof User) ? ' by '.$user->getUsername() : ''),
  241.                                                     'worker_id' => $dataport['id']
  242.                                                 ]
  243.                                             );
  244.                                         }
  245.                                     } elseif ($dataport['sourcetype'] === 'pimcore' && preg_match('/ IN \(((\d+,?)+)\)/', $dataportResourceQuery, $match)) {
  246.                                         $ids = explode(',', $match[1]);
  247.                                         sort($ids, SORT_NUMERIC);
  248.                                         foreach ($ids as $id) {
  249.                                             $queue->create(
  250.                                                 [
  251.                                                     'command' => 'data-director:complete '.$dataport['id'].' "'.str_replace([$match[0], '"', '$', '`'], ['='.$id, '\\"', '\\$', '\\`'], $dataportResourceQuery).'"',
  252.                                                     'triggered_by' => 'Element '.$object->getFullPath().' (#'.$object->getId().') got saved'.(($user instanceof User) ? ' by '.$user->getUsername() : ''),
  253.                                                     'worker_id' => $dataport['id']
  254.                                                 ]
  255.                                             );
  256.                                         }
  257.                                     } else {
  258.                                         $queue->create([
  259.                                             'command' => 'data-director:complete '.$dataport['id'].' "'.str_replace(['"', '$', '`'], ['\\"', '\\$', '\\`'], $dataportResourceQuery).'"',
  260.                                             'triggered_by' => 'Element '.$object->getFullPath().' (#'.$object->getId().') got saved'.(($user instanceof User) ? ' by '.$user->getUsername() : ''),
  261.                                             'worker_id' => $dataport['id']
  262.                                         ]);
  263.                                     }
  264.                                 }
  265.                             }
  266.                         } else {
  267.                             try {
  268.                                 // Import
  269.                                 $source = $parser->getFileConditionFromObject($object);
  271.                                 if ($source !== null) {
  272.                                     if ($dataport['sourcetype'] === 'pimcore' && preg_match('/ IN \(((\d+,?)+)\)/', $source, $match)) {
  273.                                         $ids = explode(',', $match[1]);
  274.                                         sort($ids, SORT_NUMERIC);
  275.                                         foreach ($ids as $id) {
  276.                                             $queue->create([
  277.                                                 'command' => 'data-director:complete '.$dataport['id'].' "'.str_replace([$match[0], '"', '$', '`'], ['='.$id, '\\"', '\\$', '\\`'], $source).'"',
  278.                                                 'triggered_by' => 'Element '.$object->getFullPath().' (#'.$object->getId().') got saved'.(($user instanceof User) ? ' by '.$user->getUsername() : ''),
  279.                                                 'worker_id' => $dataport['id']
  280.                                             ]);
  281.                                         }
  282.                                     } else {
  283.                                         $queue->create([
  284.                                             'command' => 'data-director:complete '.$dataport['id'].' "'.str_replace(['"', '$', '`'], ['\\"', '\\$', '\\`'], $source).'"',
  285.                                             'triggered_by' => 'Element '.$object->getFullPath().' (#'.$object->getId().') got saved'.(($user instanceof User) ? ' by '.$user->getUsername() : ''),
  286.                                             'worker_id' => $dataport['id']
  287.                                         ]);
  288.                                     }
  289.                                 }
  290.                             } catch(SkipTriggerAutomaticImportException $e) {
  291.                             }
  292.                         }
  293.                     }
  294.                 }
  295.             } catch (\Exception $e) {
  296.                 if($this->logger) {
  297.                     $this->logger->error((string)$e);
  298.                 } else {
  299.                     Logger::error((string)$e);
  300.                 }
  301.             } finally {
  302.                 \Pimcore::getContainer()->get('pimcore.locale')->setLocale($originalLocale);
  303.             }
  304.         }
  305.     }
  307.     public function deleteRawdata(ElementEventInterface $e)
  308.     {
  309.         if (\method_exists($e, 'getArgument')) {
  310.             try {
  311.                 $saveVersionOnly = $e->getArgument('saveVersionOnly');
  312.                 if ($saveVersionOnly) {
  313.                     return;
  314.                 }
  315.             } catch (\InvalidArgumentException $exception) {
  316.             }
  317.         }
  319.         $object = $e->getElement();
  320.         $dataports = Dataport::getInstance();
  321.         $objectType = Service::getElementType($object);
  322.         $commandPrefix = '"'.Cli::getPhpCli().'" '.realpath(PIMCORE_PROJECT_ROOT.DIRECTORY_SEPARATOR.'bin'.DIRECTORY_SEPARATOR.'console');
  324.         foreach ($dataports->find([], 'name') as $dataport) {
  325.             $sourceConfig = unserialize($dataport['sourceconfig'], ['allowed_classes' => false]);
  326.             $targetConfig = unserialize($dataport['targetconfig'], ['allowed_classes' => false]);
  328.             if (!empty($sourceConfig['autoImport']) && empty($targetConfig['itemClass'])) {
  329.                 $itemMold = $this->itemMoldBuilder->getItemMoldByClassId($sourceConfig['sourceClass']);
  330.                 if ($object instanceof $itemMold) {
  331.                     Cli::exec($commandPrefix.' data-director:delete-rawdata --dataport='.$dataport['id'].' --object-id='.$object->getId().' --object-type='.$objectType);
  332.                 }
  333.             }
  334.         }
  335.     }
  337.     private function queueProcessRawData($dataportResourceId, $dataportId) {
  338.         if (!isset(self::$processCommandRegistered[$dataportResourceId])) {
  339.             self::$processCommandRegistered[$dataportResourceId] = true;
  341.             $queue = Queue::getInstance();
  342.             $cmd = 'data-director:process ' . $dataportId . ' --dataport-resource-id=' . $dataportResourceId;
  343.             register_shutdown_function(
  344.                 static function () use ($queue, $cmd, $dataportId) {
  345.                     $queue->create([
  346.                         'command' => $cmd,
  347.                         'triggered_by' => 'An item got deleted from raw data -> new result document has to be generated',
  348.                         'worker_id' => $dataportId
  349.                     ]);
  350.                 }
  351.             );
  352.         }
  353.     }
  354. }