Add interpretation-stage events: PreInterpretFileEvent and PreQueueRowEvent

doc/06_Extending/02_Events.md

@@ -12,6 +12,8 @@ Listening for events customizes import behaviour without replacing any component
 
 | Event | Fired |
 |---|---|
+| `PreInterpretFileEvent` | Before the interpreter starts reading the source file. |
+| `PreQueueRowEvent` | For every extracted row, before it is added to the processing queue. |
 | `DataObject\PreSaveEvent` | Before an imported data object is saved. |
 | `DataObject\PostSaveEvent` | After an imported data object is saved. |
 | `DataObject\ProcessElementExceptionEvent` | When processing a record throws an exception. |
@@ -27,6 +29,87 @@ that failed, when the failure can be attributed to one.
 
 `PostPreparationEvent` exposes the configuration name, the execution type, and whether the source file was interpreted.
 
+## Interpretation-Stage Events
+
+The two interpretation-stage events customize how the source file turns into import rows without writing a custom
+interpreter. Both expose the configuration name and the execution type, and both are also dispatched (with
+`isPreview()` returning `true`) when Pimcore Studio renders the source preview and the available mapping columns, so
+the configuration UI shows exactly the data an actual import would produce. The preview dispatches `PreQueueRowEvent`
+only for the record being displayed - preceding records are not replayed. Listeners that carry state across rows (for
+example a group marker taken from an earlier row) should check `isPreview()` and fall back to stateless behavior, as
+the carried state is not available in preview mode.
+
+### PreInterpretFileEvent
+
+Dispatched before the interpreter validates and reads the source file. `setPath()` replaces the file that gets
+interpreted - use it to normalize a file (transcode it, strip a report preamble, rewrite delimiters) while keeping the
+standard interpreter. It also marks the start of an interpretation run, which stateful `PreQueueRowEvent` listeners can
+use as a reset signal.
+
+### PreQueueRowEvent
+
+Dispatched for every row the interpreter extracted, right before the row is added to the processing queue. The listener
+receives the row exactly as the interpreter produced it - before the delta check, the resolver's identifier extraction,
+and the mapping pipeline - so changed values (including the ID column) affect which element a row resolves to.
+
+| Method | Purpose |
+|---|---|
+| `getOriginalRow()` | The row as extracted, unaffected by other listeners. |
+| `getRows()` / `setRows(array $rows)` | The rows that will be queued. Set one row to modify it, an empty array to skip it, or multiple rows to fan the source row out into multiple elements. |
+| `skipRow(bool $keepInCleanupIdentifierCache = false)` | Skip the row. See the cleanup warning below. |
+
+```php
+namespace App\EventListener;
+
+use Pimcore\Bundle\DataImporterBundle\Event\PreQueueRowEvent;
+use Symfony\Component\EventDispatcher\Attribute\AsEventListener;
+
+#[AsEventListener]
+final class ProductRowListener
+{
+    public function __invoke(PreQueueRowEvent $event): void
+    {
+        if ($event->getConfigName() !== 'my-product-import') {
+            return;
+        }
+
+        $row = $event->getOriginalRow();
+
+        // skip discontinued products, but keep their existing objects
+        if (($row['status'] ?? '') === 'discontinued') {
+            $event->skipRow(keepInCleanupIdentifierCache: true);
+
+            return;
+        }
+
+        // add a computed column that mapping, resolver and location strategies can use
+        $row['path'] = '/products/' . $row['category'] . '/' . $row['sku'];
+
+        // fan out: one source row per configured sales channel becomes one element each
+        $rows = [];
+        foreach (explode(',', $row['channels']) as $channel) {
+            $rows[] = ['channel' => $channel] + $row;
+        }
+
+        $event->setRows($rows);
+    }
+}
+```
+
+:::warning
+
+When the import uses an active cleanup strategy, every element whose identifier is not seen during interpretation is
+deleted or unpublished. A skipped row's element counts as "not seen". Skip rows with
+`skipRow(keepInCleanupIdentifierCache: true)` when their existing elements must survive the cleanup.
+
+:::
+
+Custom interpreters that extend `AbstractInterpreter` get both events automatically: the interpreter compiler pass
+wires the event dispatcher into every tagged interpreter service that provides a `setEventDispatcher()` method, and
+the base class dispatches the events in `interpretFile()` and `processImportRow()`. An interpreter that implements
+`InterpreterInterface` directly still receives the dispatcher through that setter, but must dispatch
+`PreInterpretFileEvent` and `PreQueueRowEvent` itself from its own reading loop.
+
 ## Example
 
 Adjust a data object right before it is saved:

src/DataSource/Interpreter/AbstractInterpreter.php

@@ -15,6 +15,8 @@
 use Pimcore\Bundle\ApplicationLoggerBundle\ApplicationLogger;
 use Pimcore\Bundle\ApplicationLoggerBundle\FileObject;
 use Pimcore\Bundle\DataImporterBundle\DataSource\Interpreter\DeltaChecker\DeltaChecker;
+use Pimcore\Bundle\DataImporterBundle\Event\PreInterpretFileEvent;
+use Pimcore\Bundle\DataImporterBundle\Event\PreQueueRowEvent;
 use Pimcore\Bundle\DataImporterBundle\Exception\InvalidInputException;
 use Pimcore\Bundle\DataImporterBundle\PimcoreDataImporterBundle;
 use Pimcore\Bundle\DataImporterBundle\Processing\ImportProcessingService;
@@ -23,6 +25,7 @@
 use Pimcore\Model\Tool\TmpStore;
 use Pimcore\Tool\Admin;
 use Psr\Log\LoggerAwareTrait;
+use Symfony\Contracts\EventDispatcher\EventDispatcherInterface;
 
 /**
  * @internal
@@ -45,6 +48,8 @@ abstract class AbstractInterpreter implements InterpreterInterface
 
     protected Resolver $resolver;
 
+    protected ?EventDispatcherInterface $eventDispatcher = null;
+
     /**
      * @var string[]
      */
@@ -128,11 +133,22 @@ public function setResolver(Resolver $resolver): void
         $this->resolver = $resolver;
     }
 
+    public function setEventDispatcher(?EventDispatcherInterface $eventDispatcher): void
+    {
+        $this->eventDispatcher = $eventDispatcher;
+    }
+
     public function interpretFile(string $path): bool
     {
         $success = false;
         $this->resetIdentifierCache();
 
+        if ($this->eventDispatcher !== null) {
+            $event = new PreInterpretFileEvent($this->configName, $this->executionType, $path);
+            $this->eventDispatcher->dispatch($event);
+            $path = $event->getPath();
+        }
+
         if ($this->fileValid($path)) {
             $archiveLogMessage = 'Interpreted source file and created queue items.';
             $this->doInterpretFileAndCallProcessRow($path);
@@ -161,6 +177,37 @@ public function interpretFile(string $path): bool
     abstract protected function doInterpretFileAndCallProcessRow(string $path): void;
 
     protected function processImportRow(array $data)
+    {
+        foreach ($this->applyPreQueueRowEvent($data) as $row) {
+            $this->addRowToQueue($row);
+        }
+    }
+
+    /**
+     * Lets PreQueueRowEvent listeners modify, skip, or fan out the row the interpreter
+     * extracted. Without a dispatcher or listeners the row passes through unchanged.
+     *
+     * @return array<int, array> the rows to queue
+     */
+    private function applyPreQueueRowEvent(array $data): array
+    {
+        if ($this->eventDispatcher === null) {
+            return [$data];
+        }
+
+        $event = new PreQueueRowEvent($this->configName, $this->executionType, $data);
+        $this->eventDispatcher->dispatch($event);
+
+        if ($event->isRowSkipped() && $event->shouldKeepSkippedRowInIdentifierCache()) {
+            // Register the skipped row's identifier so an active cleanup strategy does not
+            // treat its existing element as removed from the source.
+            $this->addToIdentifierCache($event->getOriginalRow());
+        }
+
+        return $event->getRows();
+    }
+
+    private function addRowToQueue(array $data): void
     {
         $this->assertValidRowEncoding($data);
 

src/DependencyInjection/CompilerPass/InterpreterConfigurationFactoryPass.php

@@ -13,8 +13,10 @@
 namespace Pimcore\Bundle\DataImporterBundle\DependencyInjection\CompilerPass;
 
 use Pimcore\Bundle\DataImporterBundle\DataSource\Interpreter\InterpreterFactory;
+use Symfony\Component\DependencyInjection\ChildDefinition;
 use Symfony\Component\DependencyInjection\Compiler\CompilerPassInterface;
 use Symfony\Component\DependencyInjection\ContainerBuilder;
+use Symfony\Component\DependencyInjection\Definition;
 use Symfony\Component\DependencyInjection\Reference;
 
 /**
@@ -28,15 +30,70 @@ public function process(ContainerBuilder $container): void
     {
         $taggedServices = $container->findTaggedServiceIds(self::INTERPRETER_TAG);
         $interpreters = [];
-        if (sizeof($taggedServices)) {
-            foreach ($taggedServices as $id => $tags) {
-                foreach ($tags as $attributes) {
-                    $interpreters[$attributes['type']] = new Reference($id);
+        foreach ($taggedServices as $id => $tags) {
+            foreach ($tags as $attributes) {
+                $interpreters[$attributes['type']] = new Reference($id);
+            }
+        }
+
+        // Wire the event dispatcher into every interpreter that supports it (built-in
+        // and custom alike), so PreInterpretFileEvent/PreQueueRowEvent are dispatched.
+        // This pass runs before child definitions and class parameters are resolved, so
+        // both are resolved here. Parents go first: a child inherits their method calls.
+        $chains = [];
+        foreach (array_keys($taggedServices) as $id) {
+            $chains[$id] = $this->getDefinitionChain($container, $id);
+        }
+        uasort($chains, static fn (array $a, array $b): int => count($a) <=> count($b));
+
+        foreach ($chains as $id => $chain) {
+            $class = $this->resolveClass($container, $chain, $id);
+            if ($class === null || !method_exists($class, 'setEventDispatcher')) {
+                continue;
+            }
+
+            foreach ($chain as $definition) {
+                if ($definition->hasMethodCall('setEventDispatcher')) {
+                    continue 2;
                 }
             }
+
+            $chain[0]->addMethodCall('setEventDispatcher', [new Reference('event_dispatcher')]);
         }
 
         $serviceLocator = $container->getDefinition(InterpreterFactory::class);
         $serviceLocator->setArgument('$interpreterBluePrints', $interpreters);
     }
+
+    /**
+     * @return list<Definition> the service definition followed by its parent definitions
+     */
+    private function getDefinitionChain(ContainerBuilder $container, string $id): array
+    {
+        $chain = [$container->getDefinition($id)];
+        while (end($chain) instanceof ChildDefinition && $container->hasDefinition(end($chain)->getParent())) {
+            $chain[] = $container->getDefinition(end($chain)->getParent());
+        }
+
+        return $chain;
+    }
+
+    /**
+     * @param list<Definition> $chain
+     */
+    private function resolveClass(ContainerBuilder $container, array $chain, string $id): ?string
+    {
+        $class = $id;
+        foreach ($chain as $definition) {
+            if ($definition->getClass() !== null) {
+                $class = $definition->getClass();
+
+                break;
+            }
+        }
+
+        $class = $container->getParameterBag()->resolveValue($class);
+
+        return is_string($class) ? $class : null;
+    }
 }

src/Event/PreInterpretFileEvent.php

@@ -0,0 +1,63 @@
+<?php
+declare(strict_types=1);
+
+/**
+ * This source file is available under the terms of the
+ * Pimcore Open Core License (POCL)
+ * Full copyright and license information is available in
+ * LICENSE.md which is distributed with this source code.
+ *
+ *  @copyright  Copyright (c) Pimcore GmbH (https://www.pimcore.com)
+ *  @license    Pimcore Open Core License (POCL)
+ */
+
+namespace Pimcore\Bundle\DataImporterBundle\Event;
+
+use Symfony\Contracts\EventDispatcher\Event;
+
+/**
+ * Dispatched before an interpreter starts reading the source file, both for real imports
+ * and for the Studio preview. Listeners may replace the file path, e.g. to normalize the
+ * file (transcode, strip a report preamble, rewrite delimiters) without replacing the
+ * interpreter. Stateful PreQueueRowEvent listeners can also use it as a per-run reset signal.
+ */
+final class PreInterpretFileEvent extends Event
+{
+    public function __construct(
+        private readonly string $configName,
+        private readonly string $executionType,
+        private string $path,
+        private readonly bool $preview = false,
+    ) {
+    }
+
+    public function getConfigName(): string
+    {
+        return $this->configName;
+    }
+
+    public function getExecutionType(): string
+    {
+        return $this->executionType;
+    }
+
+    public function getPath(): string
+    {
+        return $this->path;
+    }
+
+    public function setPath(string $path): self
+    {
+        $this->path = $path;
+
+        return $this;
+    }
+
+    /**
+     * True when the file is being read for the Studio preview instead of an actual import.
+     */
+    public function isPreview(): bool
+    {
+        return $this->preview;
+    }
+}