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

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\Helper;
  16. use Blackbit\DataDirectorBundle\lib\Pim\Import\CallbackFunction;
  17. use Blackbit\DataDirectorBundle\lib\Pim\Item\ImporterInterface;
  18. use Blackbit\DataDirectorBundle\lib\Pim\Item\ItemMoldBuilder;
  19. use Blackbit\DataDirectorBundle\lib\Pim\RawData\Importer;
  20. use Blackbit\DataDirectorBundle\lib\Pim\Serializer;
  21. use Blackbit\DataDirectorBundle\model\Dataport;
  22. use Blackbit\DataDirectorBundle\model\DataportResource;
  23. use Blackbit\DataDirectorBundle\model\Fieldmapping;
  24. use Blackbit\DataDirectorBundle\model\ImportStatus;
  25. use Blackbit\DataDirectorBundle\model\PimcoreDbRepository;
  26. use Blackbit\DataDirectorBundle\model\Queue;
  27. use Blackbit\DataDirectorBundle\model\RawItem;
  28. use Blackbit\DataDirectorBundle\model\RawItemData;
  29. use Blackbit\DataDirectorBundle\Tools\Installer;
  30. use InvalidArgumentException;
  31. use Pimcore\Db;
  32. use Pimcore\Event\Model\AssetEvent;
  33. use Pimcore\Event\Model\ElementEventInterface;
  34. use Pimcore\Logger;
  35. use Pimcore\Model\AbstractModel;
  36. use Pimcore\Model\Asset;
  37. use Pimcore\Model\DataObject\AbstractObject;
  38. use Pimcore\Model\DataObject\Concrete;
  39. use Pimcore\Model\Element\DirtyIndicatorInterface;
  40. use Pimcore\Model\Element\ElementInterface;
  41. use Pimcore\Model\Element\Service;
  42. use Pimcore\Model\User;
  43. use Pimcore\Model\Version;
  44. use Pimcore\Tool;
  45. use Blackbit\DataDirectorBundle\lib\Pim\Cli;
  46. use Psr\Log\LoggerAwareInterface;
  47. use Psr\Log\LoggerAwareTrait;
  48. use Psr\Log\LoggerInterface;
  49. use Symfony\Component\HttpFoundation\Request;
  51. class AutomaticImportListener implements LoggerAwareInterface
  52. {
  53.     use LoggerAwareTrait;
  55.     private static $processCommandRegistered = [];
  57.     private static $dataports = null;
  59.     /** @var ItemMoldBuilder */
  60.     private $itemMoldBuilder;
  62.     public function __construct(ItemMoldBuilder $itemMoldBuilder)
  63.     {
  64.         $this->itemMoldBuilder = $itemMoldBuilder;
  65.     }
  67.     private static function getDataports() {
  68.         if(self::$dataports === null) {
  69.             self::$dataports = Dataport::getInstance()->find([], 'name');
  70.             foreach(self::$dataports as &$dataport) {
  71.                 $dataport['sourceconfig'] = unserialize($dataport['sourceconfig'], ['allowed_classes' => false]);
  72.                 $dataport['targetconfig'] = unserialize($dataport['targetconfig'], ['allowed_classes' => false]);
  73.             }
  74.             unset($dataport);
  75.         }
  77.         return self::$dataports;
  78.     }
  80.     public function startImports(ElementEventInterface $e) {
  81.         if(\method_exists($e, 'getArgument')) {
  82.             try {
  83.                 $saveVersionOnly = $e->getArgument('saveVersionOnly');
  84.                 if($saveVersionOnly) {
  85.                     return;
  86.                 }
  87.             } catch(\InvalidArgumentException $exception) {
  88.             }
  89.         }
  91.         $object = $e->getElement();
  93.         $user = Helper::getUser();
  94.         if($user instanceof User) {
  95.             PimcoreDbRepository::getInstance()->execute('UPDATE edit_lock SET date=? WHERE cid=? AND ctype=? AND userId=?', [time(), $object->getId(), Service::getElementType($object), $user->getId()]);
  96.         }
  98.         $dataports = Dataport::getInstance();
  99.         $dataportResourceRepository = DataportResource::getInstance();
  101.         $queue = Queue::getInstance();
  103.         $queueAddItems = [];
  104.         foreach(self::getDataports() as $dataport) {
  105.             $originalLocale = \Pimcore::getContainer()->get('pimcore.locale')->getLocale() ?? Tool::getDefaultLanguage();
  106.             try {
  107.                 $sourceConfig = $dataport['sourceconfig'];
  108.                 $targetConfig = $dataport['targetconfig'];
  110.                 if(!empty($sourceConfig['autoImport'])) {
  111.                     $parser = $dataports->getParser($dataport['id']);
  112.                     if(method_exists($parser, 'disableLoggingNotFoundImportResource')) {
  113.                         $parser->disableLoggingNotFoundImportResource();
  114.                     }
  116.                     if(\method_exists($parser, 'setSourceFile') && \method_exists($parser, 'getFileConditionFromObject')) {
  117.                         if (empty($targetConfig['itemClass'])) {
  118.                             // Export
  119.                             if (empty($sourceConfig['incrementalExport'])) {
  120.                                 $dataportResources = $dataportResourceRepository->find(['dataportId = ?' => $dataport['id']]);
  121.                             } else {
  122.                                 $dataportResources = [
  123.                                     $dataportResourceRepository->create(
  124.                                         [
  125.                                             'dataportId' => $dataport['id'],
  126.                                             'resource' => \json_encode([], JSON_UNESCAPED_SLASHES)
  127.                                         ]
  128.                                     )
  129.                                 ];
  130.                             }
  132.                             $dataportResourceIdsStillContainingObject = [];
  133.                             foreach ($dataportResources as $dataportResource) {
  134.                                 try {
  135.                                     $resourceSettings = \json_decode($dataportResource['resource'], true);
  136.                                     $parser->setSourceFile($resourceSettings['file'] ?? null);
  137.                                     if (!empty($resourceSettings['locale'])) {
  138.                                         \Pimcore::getContainer()->get('pimcore.locale')->setLocale($resourceSettings['locale']);
  139.                                     } else {
  140.                                         $resourceSettings['locale'] = \Pimcore::getContainer()->get('pimcore.locale')->getLocale();
  141.                                         if ($resourceSettings['locale'] === null) {
  142.                                             $resourceSettings['locale'] = Tool::getDefaultLanguage();
  143.                                         }
  144.                                     }
  146.                                     $source = $parser->getFileConditionFromObject($object);
  147.                                     if ($source !== null) {
  148.                                         $dataportResourceIdsStillContainingObject[$source][$resourceSettings['locale']][] = $dataportResource['id'];
  149.                                     } elseif (empty($sourceConfig['incrementalExport'])) {
  150.                                         $source = $parser->getFileConditionFromObject($object, false);
  151.                                         if ($source !== null) {
  152.                                             $tmpSourceConfig = $sourceConfig;
  153.                                             if (\method_exists($parser, 'setConfig')) {
  154.                                                 if (\method_exists($parser, 'getConfig')) {
  155.                                                     $tmpSourceConfig = $parser->getConfig();
  156.                                                 }
  157.                                                 $tmpSourceConfig['file'] = '';
  159.                                                 $tmpSourceConfig['dataportId'] = $dataport['id'];
  160.                                                 $parser->setConfig($tmpSourceConfig);
  161.                                             }
  163.                                             $parser->setSourceFile($source);
  165.                                             $keyFields = [];
  166.                                             foreach ($tmpSourceConfig['fields'] as $fieldIndex => $field) {
  167.                                                 if (!empty($field['exportKey'])) {
  168.                                                     $keyFields[$fieldIndex] = $field;
  169.                                                 }
  170.                                             }
  172.                                             $hashs = [];
  173.                                             if ($dataport['sourcetype'] === 'pimcore') {
  174.                                                 $locales = Tool::getValidLanguages();
  175.                                                 if ($dataportResource !== null) {
  176.                                                     $resource = \json_decode($dataportResource['resource'], true);
  177.                                                     if ($resource['locale']) {
  178.                                                         $locales = [$resource['locale']];
  179.                                                     }
  180.                                                 }
  181.                                                 foreach ($locales as $language) {
  182.                                                     \Pimcore::getContainer()->get('pimcore.locale')->setLocale($language);
  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.                                                 }
  200.                                             } else {
  201.                                                 foreach ($parser as $rawItemData) {
  202.                                                     if ($rawItemData === null) {
  203.                                                         continue;
  204.                                                     }
  206.                                                     $rawItemData = array_filter(
  207.                                                         $rawItemData,
  208.                                                         static function ($fieldId) use ($keyFields) {
  209.                                                             return isset($keyFields[$fieldId]);
  210.                                                         },
  211.                                                         ARRAY_FILTER_USE_KEY
  212.                                                     );
  214.                                                     $hashs[] = Importer::getHash($rawItemData);
  215.                                                 }
  216.                                             }
  218.                                             if(count($hashs) > 0) {
  219.                                                 $countDeletedRawItems = Db::get()->executeUpdate('DELETE FROM '.Installer::TABLE_RAWITEM.' WHERE dataport_resource_id = ? AND hash IN ("'.implode('","', $hashs).'")', [$dataportResource['id']]);
  220.                                                 if ($countDeletedRawItems) {
  221.                                                     $this->queueProcessRawData($dataportResource['id'], $dataport['id']);
  222.                                                 }
  223.                                             }
  224.                                         }
  225.                                     }
  226.                                 } catch (\Throwable $e) {
  227.                                     if(!$e instanceof SkipTriggerAutomaticImportException) {
  228.                                         $error = 'Check for automatic start for dataport #'.$dataport['id'].' failed: '.(string)$e;
  229.                                         if($this->logger) {
  230.                                             $this->logger->error($error);
  231.                                         } else {
  232.                                             Logger::error($error);
  233.                                         }
  234.                                     }
  235.                                 }
  236.                             }
  238.                             foreach($dataportResourceIdsStillContainingObject as $dataportResourceQuery => $dataportResourcesWithSameLocale) {
  239.                                 foreach($dataportResourcesWithSameLocale as $locale => $dataportResourceIds) {
  240.                                     if (empty($sourceConfig['incrementalExport'])) {
  241.                                         if ($dataport['sourcetype'] === 'pimcore' && $this->extractIds($dataportResourceQuery, $match)) {
  242.                                             $ids = $match[1];
  243.                                             sort($ids, SORT_NUMERIC);
  244.                                             foreach ($ids as $id) {
  245.                                                 $queueAddItems[] = [
  246.                                                     'command' => 'data-director:extract '.$dataport['id'].' "'.str_replace([$match[0], '"', '$', '`'], ['='.$id, '\\"', '\\$', '\\`'], $dataportResourceQuery).'"'.(!empty($locale) ? ' --locale='.$locale : '').' --dataport-resource-id='.implode(',', $dataportResourceIds),
  247.                                                     'triggered_by' => 'Element '.$object->getFullPath().' (#'.$object->getId().') got saved'.(($user instanceof User) ? ' by '.$user->getUsername() : ''),
  248.                                                     'worker_id' => $dataport['id']
  249.                                                 ];
  250.                                             }
  251.                                         } else {
  252.                                             $queueAddItems[] = [
  253.                                                 'command' => 'data-director:extract '.$dataport['id'].' "'.str_replace(['"', '$', '`'], ['\\"', '\\$', '\\`'], $dataportResourceQuery).'"'.(!empty($locale) ? ' --locale='.$locale : '').' --dataport-resource-id='.implode(',', $dataportResourceIds),
  254.                                                 'triggered_by' => 'Element '.$object->getFullPath().' (#'.$object->getId().') got saved'.(($user instanceof User) ? ' by '.$user->getUsername() : ''),
  255.                                                 'worker_id' => $dataport['id']
  256.                                             ];
  257.                                         }
  258.                                     } elseif ($dataport['sourcetype'] === 'pimcore' && ($ids = $this->extractIds($dataportResourceQuery, $match))) {
  259.                                         $ids = $match[1];
  260.                                         sort($ids, SORT_NUMERIC);
  261.                                         foreach ($ids as $id) {
  262.                                             $queueAddItems[] = [
  263.                                                 'command' => 'data-director:complete '.$dataport['id'].' "'.str_replace([$match[0], '"', '$', '`'], ['='.$id, '\\"', '\\$', '\\`'], $dataportResourceQuery).'"',
  264.                                                 'triggered_by' => 'Element '.$object->getFullPath().' (#'.$object->getId().') got saved'.(($user instanceof User) ? ' by '.$user->getUsername() : ''),
  265.                                                 'worker_id' => $dataport['id']
  266.                                             ];
  267.                                         }
  268.                                     } else {
  269.                                         $queueAddItems[] = [
  270.                                             'command' => 'data-director:complete '.$dataport['id'].' "'.str_replace(['"', '$', '`'], ['\\"', '\\$', '\\`'], $dataportResourceQuery).'"',
  271.                                             'triggered_by' => 'Element '.$object->getFullPath().' (#'.$object->getId().') got saved'.(($user instanceof User) ? ' by '.$user->getUsername() : ''),
  272.                                             'worker_id' => $dataport['id']
  273.                                         ];
  274.                                     }
  275.                                 }
  276.                             }
  277.                         } else {
  278.                             try {
  279.                                 // Import
  280.                                 if (\method_exists($parser, 'getConfig')) {
  281.                                     $parser->setSourceFile($parser->getConfig()['file'] ?? null);
  282.                                 }
  284.                                 $source = $parser->getFileConditionFromObject($object);
  286.                                 if ($source !== null) {
  287.                                     if ($dataport['sourcetype'] === 'pimcore' && preg_match('/ IN \(((\d+,?)+)\)/', $source, $match)) {
  288.                                         $ids = explode(',', $match[1]);
  289.                                         sort($ids, SORT_NUMERIC);
  290.                                         foreach ($ids as $id) {
  291.                                             $queueAddItems[] = [
  292.                                                 'command' => 'data-director:complete '.$dataport['id'].' "'.str_replace([$match[0], '"', '$', '`'], ['='.$id, '\\"', '\\$', '\\`'], $source).'"'.(($user instanceof User) ? ' --user='.$user->getId() : ''),
  293.                                                 'triggered_by' => 'Element '.$object->getFullPath().' (#'.$object->getId().') got saved'.(($user instanceof User) ? ' by '.$user->getUsername() : ''),
  294.                                                 'worker_id' => $dataport['id']
  295.                                             ];
  296.                                         }
  297.                                     } else {
  298.                                         $queueAddItems[] = [
  299.                                             'command' => 'data-director:complete '.$dataport['id'].' "'.str_replace(['"', '$', '`'], ['\\"', '\\$', '\\`'], $source).'"',
  300.                                             'triggered_by' => 'Element '.$object->getFullPath().' (#'.$object->getId().') got saved'.(($user instanceof User) ? ' by '.$user->getUsername() : ''),
  301.                                             'worker_id' => $dataport['id']
  302.                                         ];
  303.                                     }
  304.                                 }
  305.                             } catch(SkipTriggerAutomaticImportException $e) {
  306.                             }
  307.                         }
  308.                     }
  309.                 }
  310.             } catch (\Exception $e) {
  311.                 if($this->logger) {
  312.                     $this->logger->error('Automatic processing of dataport #'.$dataport['id'].' "'.$dataport['name'].'" failed:'. (string)$e);
  313.                 } else {
  314.                     Logger::error('Automatic processing of dataport #'.$dataport['id'].' "'.$dataport['name'].'" failed:'.(string)$e);
  315.                 }
  316.             } finally {
  317.                 \Pimcore::getContainer()->get('pimcore.locale')->setLocale($originalLocale);
  318.             }
  319.         }
  321.         if (count($queueAddItems) > 0) {
  322.             $queue->create($queueAddItems);
  323.         }
  324.     }
  326.     public function deleteRawdata(ElementEventInterface $e)
  327.     {
  328.         if (\method_exists($e, 'getArgument')) {
  329.             try {
  330.                 $saveVersionOnly = $e->getArgument('saveVersionOnly');
  331.                 if ($saveVersionOnly) {
  332.                     return;
  333.                 }
  334.             } catch (\InvalidArgumentException $exception) {
  335.             }
  336.         }
  338.         $object = $e->getElement();
  339.         $dataports = Dataport::getInstance();
  340.         $objectType = Service::getElementType($object);
  341.         $commandPrefix = '"'.Cli::getPhpCli().'" '.realpath(PIMCORE_PROJECT_ROOT.DIRECTORY_SEPARATOR.'bin'.DIRECTORY_SEPARATOR.'console');
  343.         foreach (self::getDataports() as $dataport) {
  344.             $sourceConfig = $dataport['sourceconfig'];
  345.             $targetConfig = $dataport['targetconfig'];
  347.             if (!empty($sourceConfig['autoImport']) && empty($targetConfig['itemClass'])) {
  348.                 $itemMold = $this->itemMoldBuilder->getItemMoldByClassId($sourceConfig['sourceClass'] ?? null);
  349.                 if ($object instanceof $itemMold) {
  350.                     Cli::exec($commandPrefix.' data-director:delete-rawdata --dataport='.$dataport['id'].' --object-id='.$object->getId().' --object-type='.$objectType);
  351.                 }
  352.             }
  353.         }
  354.     }
  356.     private function queueProcessRawData($dataportResourceId, $dataportId) {
  357.         if (!isset(self::$processCommandRegistered[$dataportResourceId])) {
  358.             self::$processCommandRegistered[$dataportResourceId] = true;
  360.             $queue = Queue::getInstance();
  361.             $cmd = 'data-director:process ' . $dataportId . ' --dataport-resource-id=' . $dataportResourceId;
  362.             register_shutdown_function(
  363.                 static function () use ($queue, $cmd, $dataportId) {
  364.                     $queue->create([
  365.                         'command' => $cmd,
  366.                         'triggered_by' => 'An item got deleted from raw data -> new result document has to be generated',
  367.                         'worker_id' => $dataportId
  368.                     ]);
  369.                 }
  370.             );
  371.         }
  372.     }
  374.     private function extractIds($query, &$match) {
  375.         $startTerm = ' IN (';
  376.         $endTerm = ')';
  377.         $start = strpos($query, $startTerm);
  378.         if($start === false) {
  379.             return null;
  380.         }
  382.         $end = strpos($query, $endTerm, $start);
  384.         if($end === false) {
  385.             return null;
  386.         }
  388.         $match = [
  389.             substr($query, $start, $end - $start+strlen($endTerm)),
  390.             explode(',', substr($query, $start + strlen($startTerm), $end - $start - strlen($startTerm)))
  391.         ];
  393.         return true;
  394.     }
  395. }