fix(cli/file-import): improvements and config(s) hashes or files

EMS/common-bundle/src/Common/CoreApi/Client.php

@@ -12,7 +12,6 @@
 use Psr\Http\Message\StreamInterface;
 use Psr\Log\LoggerAwareInterface;
 use Psr\Log\LoggerInterface;
-use Symfony\Component\HttpClient\CurlHttpClient;
 use Symfony\Component\HttpClient\Response\StreamableInterface;
 use Symfony\Component\HttpFoundation\Request;
 use Symfony\Component\HttpFoundation\Response;
@@ -24,19 +23,13 @@ class Client
 {
     /** @var array<string, string> */
     private array $headers = [];
-    private readonly HttpClientInterface $client;
     private LoggerInterface $logger;
 
-    public function __construct(private readonly string $baseUrl, LoggerInterface $logger, bool $verify, int $timeout)
-    {
-        $this->client = new CurlHttpClient([
-            'base_uri' => $baseUrl,
-            'headers' => ['Content-Type' => 'application/json'],
-            'verify_host' => $verify,
-            'verify_peer' => $verify,
-            'timeout' => $timeout,
-        ]);
-
+    public function __construct(
+        private readonly HttpClientInterface $httpClient,
+        private readonly string $baseUrl,
+        LoggerInterface $logger,
+    ) {
         $this->setLogger($logger);
     }
 
@@ -64,6 +57,17 @@ public function getHeader(string $name): string
         return $this->headers[$name];
     }
 
+    /**
+     * @param array<mixed> $options
+     */
+    public function asyncRequest(string $method, string $resource, array $options = []): ResponseInterface
+    {
+        return $this->httpClient->request($method, $resource, [
+            ...['headers' => $this->headers],
+            ...$options,
+        ]);
+    }
+
     /**
      * @param array<mixed> $query
      */
@@ -145,7 +149,7 @@ public function delete(string $resource): Result
 
     public function head(string $resource): bool
     {
-        $response = $this->client->request(Request::METHOD_HEAD, $resource, [
+        $response = $this->httpClient->request(Request::METHOD_HEAD, $resource, [
             'headers' => $this->headers,
         ]);
 
@@ -164,8 +168,8 @@ public function setLogger(LoggerInterface $logger): void
     {
         $this->logger = $logger;
 
-        if ($this->client instanceof LoggerAwareInterface) {
-            $this->client->setLogger($logger);
+        if ($this->httpClient instanceof LoggerAwareInterface) {
+            $this->httpClient->setLogger($logger);
         }
     }
 
@@ -178,7 +182,7 @@ private function getResponse(string $method, string $resource, array $options):
             throw new BaseUrlNotDefinedException();
         }
 
-        $response = $this->client->request($method, $resource, $options);
+        $response = $this->httpClient->request($method, $resource, $options);
 
         if (Response::HTTP_UNAUTHORIZED === $response->getStatusCode()) {
             throw new NotAuthenticatedException($response);

EMS/common-bundle/src/Common/CoreApi/CoreApi.php

@@ -50,6 +50,11 @@ public function authenticate(string $username, string $password): CoreApiInterfa
         return $this;
     }
 
+    public function queue(int $flushSize): ResponseQueue
+    {
+        return new ResponseQueue($flushSize);
+    }
+
     public function data(string $contentType): DataInterface
     {
         $versions = $this->admin()->getVersions();

EMS/common-bundle/src/Common/CoreApi/CoreApiFactory.php

@@ -8,21 +8,38 @@
 use EMS\CommonBundle\Contracts\CoreApi\CoreApiInterface;
 use EMS\CommonBundle\Storage\StorageManager;
 use Psr\Log\LoggerInterface;
+use Symfony\Component\HttpClient\CurlHttpClient;
 
 final class CoreApiFactory implements CoreApiFactoryInterface
 {
+    /**
+     * @param array{ headers: array<string, string>, max_connections: int, verify: bool, timeout: int } $options
+     */
     public function __construct(
         private readonly LoggerInterface $logger,
         private readonly StorageManager $storageManager,
-        private readonly bool $verify = true,
-        private readonly int $timeout = 30,
+        private readonly array $options,
     ) {
     }
 
     public function create(string $baseUrl): CoreApiInterface
     {
-        $client = new Client($baseUrl, $this->logger, $this->verify, $this->timeout);
+        $httpClient = new CurlHttpClient(
+            defaultOptions: [
+                'base_uri' => $baseUrl,
+                'headers' => [
+                    ...$this->options['headers'],
+                    ...['Content-Type' => 'application/json'],
+                ],
+                'verify_host' => $this->options['verify'],
+                'verify_peer' => $this->options['verify'],
+                'timeout' => $this->options['timeout'],
+            ],
+            maxHostConnections: $this->options['max_connections']
+        );
 
-        return new CoreApi($client, $this->storageManager);
+        $coreApiClient = new Client($httpClient, $baseUrl, $this->logger);
+
+        return new CoreApi($coreApiClient, $this->storageManager);
     }
 }

EMS/common-bundle/src/Common/CoreApi/Endpoint/Data/Data.php

@@ -9,6 +9,8 @@
 use EMS\CommonBundle\Contracts\CoreApi\Endpoint\Data\DataInterface;
 use EMS\CommonBundle\Contracts\CoreApi\Endpoint\Data\DraftInterface;
 use EMS\CommonBundle\Contracts\CoreApi\Endpoint\Data\RevisionInterface;
+use Symfony\Component\HttpFoundation\Request;
+use Symfony\Contracts\HttpClient\ResponseInterface;
 
 final class Data implements DataInterface
 {
@@ -132,6 +134,17 @@ public function index(?string $ouuid, array $rawData, bool $merge = false, bool
         return new Index($this->client->post($resource, $rawData));
     }
 
+    public function indexAsync(?string $ouuid, array $rawData, bool $merge = false, bool $refresh = false): ResponseInterface
+    {
+        $resource = $this->makeResource($merge && $ouuid ? 'update' : 'index', $ouuid);
+
+        if ($refresh) {
+            $resource .= '?refresh=true';
+        }
+
+        return $this->client->asyncRequest(Request::METHOD_POST, $resource, ['json' => $rawData]);
+    }
+
     private function makeResource(?string ...$path): string
     {
         return \implode('/', \array_merge($this->endPoint, \array_filter($path)));

EMS/common-bundle/src/Common/CoreApi/ResponseQueue.php

@@ -0,0 +1,56 @@
+<?php
+
+declare(strict_types=1);
+
+namespace EMS\CommonBundle\Common\CoreApi;
+
+use Symfony\Contracts\HttpClient\ResponseInterface;
+
+class ResponseQueue implements \Countable
+{
+    private int $count = 0;
+    /** @var ResponseInterface[] */
+    private array $responses = [];
+    /** @var callable */
+    private $flushCallback;
+
+    public function __construct(private readonly ?int $flushSize = 10)
+    {
+    }
+
+    public function count(): int
+    {
+        return $this->count;
+    }
+
+    public function addFlushCallback(callable $callback): self
+    {
+        $this->flushCallback = $callback;
+
+        return $this;
+    }
+
+    public function add(ResponseInterface $response): self
+    {
+        $this->responses[] = $response;
+        ++$this->count;
+
+        if ($this->flushSize && \count($this->responses) === $this->flushSize) {
+            $this->flush();
+        }
+
+        return $this;
+    }
+
+    public function flush(): self
+    {
+        foreach ($this->responses as $i => $response) {
+            if ($this->flushCallback) {
+                ($this->flushCallback)($response);
+            }
+            unset($this->responses[$i]);
+        }
+
+        return $this;
+    }
+}

EMS/common-bundle/src/Common/File/FileReader.php

@@ -5,6 +5,7 @@
 namespace EMS\CommonBundle\Common\File;
 
 use EMS\CommonBundle\Contracts\File\FileReaderInterface;
+use EMS\Helpers\File\CsvFile;
 use PhpOffice\PhpSpreadsheet\IOFactory;
 use PhpOffice\PhpSpreadsheet\Reader\Csv;
 use PhpOffice\PhpSpreadsheet\Reader\Html;
@@ -15,19 +16,73 @@ final class FileReader implements FileReaderInterface
     /**
      * {@inheritDoc}
      */
-    public function getData(string $filename, bool $skipFirstRow = false, string $encoding = null): array
+    public function getData(string $filename, array $options = []): array
     {
         $reader = IOFactory::createReaderForFile($filename);
+
+        $encoding = $options['encoding'] ?? null;
         if (($reader instanceof Csv || $reader instanceof Html || $reader instanceof Slk) && null !== $encoding) {
             $reader->setInputEncoding($encoding);
         }
 
-        $data = $reader->load($filename)->getActiveSheet()->toArray();
+        if ($reader instanceof Csv && isset($options['delimiter'])) {
+            $reader->setDelimiter($options['delimiter']);
+        }
+
+        return $reader->load($filename)->getActiveSheet()->toArray();
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    public function readCells(string $filename, array $options = []): \Generator
+    {
+        $isCsv = 0 === \strcasecmp(\pathinfo($filename, PATHINFO_EXTENSION), 'csv');
+
+        if ($isCsv) {
+            $csv = new CsvFile(
+                filename: $filename,
+                delimiter: ($options['delimiter'] ?? null),
+                encoding: ($options['encoding'] ?? null)
+            );
+            $total = \count($csv);
+            $data = $csv;
+        } else {
+            $data = $this->getData($filename, $options);
+            $total = \count($data);
+        }
+
+        $excludeRows = ($options['exclude_rows'] ?? []);
+        $excludeIndexes = \array_map(static fn (int $i) => $i < 0 ? $total + $i : $i, $excludeRows);
+        $headings = false;
+        $invalid = [];
+        $limit = $options['limit'] ?? false;
+
+        foreach ($data as $index => $row) {
+            if (\in_array($index, $excludeIndexes, true)) {
+                continue;
+            }
+
+            if (!$headings) {
+                $headings = \array_map('trim', $row);
+                continue;
+            }
+
+            if (\count($headings) !== \count($row)) {
+                $invalid[] = $row;
+                continue;
+            }
+
+            $rowData = \array_filter(\array_combine($headings, $row));
+            if (\count($rowData) > 0) {
+                yield $rowData;
+            }
 
-        if ($skipFirstRow) {
-            unset($data[0]);
+            if ($limit && 0 === --$limit) {
+                break;
+            }
         }
 
-        return $data;
+        return $invalid;
     }
 }

EMS/common-bundle/src/Contracts/CoreApi/CoreApiInterface.php

@@ -6,6 +6,7 @@
 
 use EMS\CommonBundle\Common\CoreApi\Endpoint\File\DataExtract;
 use EMS\CommonBundle\Common\CoreApi\Endpoint\Search\Search;
+use EMS\CommonBundle\Common\CoreApi\ResponseQueue;
 use EMS\CommonBundle\Contracts\CoreApi\Endpoint\Admin\AdminInterface;
 use EMS\CommonBundle\Contracts\CoreApi\Endpoint\Admin\MetaInterface;
 use EMS\CommonBundle\Contracts\CoreApi\Endpoint\Data\DataInterface;
@@ -26,6 +27,8 @@ interface CoreApiInterface
      */
     public function authenticate(string $username, string $password): CoreApiInterface;
 
+    public function queue(int $flushSize): ResponseQueue;
+
     public function data(string $contentType): DataInterface;
 
     public function file(): FileInterface;

EMS/common-bundle/src/Contracts/CoreApi/Endpoint/Data/DataInterface.php

@@ -6,6 +6,7 @@
 
 use EMS\CommonBundle\Common\CoreApi\Endpoint\Data\Index;
 use EMS\CommonBundle\Contracts\CoreApi\CoreApiExceptionInterface;
+use Symfony\Contracts\HttpClient\ResponseInterface;
 
 interface DataInterface
 {
@@ -41,6 +42,11 @@ public function finalize(int $revisionId): string;
      */
     public function index(?string $ouuid, array $rawData, bool $merge = false, bool $refresh = false): Index;
 
+    /**
+     * @param array<string, mixed> $rawData
+     */
+    public function indexAsync(?string $ouuid, array $rawData, bool $merge = false, bool $refresh = false): ResponseInterface;
+
     /**
      * @throws CoreApiExceptionInterface
      */

EMS/common-bundle/src/Contracts/File/FileReaderInterface.php

@@ -7,7 +7,24 @@
 interface FileReaderInterface
 {
     /**
-     * @return array<mixed>
+     * @param array{
+     *     delimiter?: ?string,
+     *     encoding?: ?string,
+     * } $options
+     *
+     * @return array<int, array<mixed>>
      */
-    public function getData(string $filename, bool $skipFirstRow = false, string $encoding = null): array;
+    public function getData(string $filename, array $options = []): array;
+
+    /**
+     * @param array{
+     *      delimiter?: ?string,
+     *      encoding?: ?string,
+     *      exclude_rows?: int[],
+     *      limit?: ?int,
+     *  } $options
+     *
+     * @return \Generator<mixed>
+     */
+    public function readCells(string $filename, array $options = []): \Generator;
 }