From 179cbe0ea4e3258c06fdfe4d0b9cb198a237f418 Mon Sep 17 00:00:00 2001
From: Norbert Orzechowicz <1921950+norberttech@users.noreply.github.com>
Date: Thu, 18 Jan 2024 13:22:41 +0100
Subject: [PATCH] CSV options detection (#918)

* Checkpoint

* Added automated detection of CSV separator and enclousure
---
 .../src/Flow/ETL/Adapter/CSV/CSVDetector.php  | 80 ++++++++++++++++
 .../src/Flow/ETL/Adapter/CSV/CSVExtractor.php | 20 ++--
 .../Flow/ETL/Adapter/CSV/Detector/Option.php  | 94 ++++++++++++++++++
 .../Flow/ETL/Adapter/CSV/Detector/Options.php | 85 ++++++++++++++++
 .../CSV/Exception/CantDetectCSVOptions.php    | 11 +++
 .../src/Flow/ETL/Adapter/CSV/functions.php    | 19 +++-
 .../CSV/Tests/Integration/CSVDetectorTest.php | 96 +++++++++++++++++++
 .../CSV/Tests/Unit/Detector/OptionTest.php    | 37 +++++++
 8 files changed, 432 insertions(+), 10 deletions(-)
 create mode 100644 src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/CSVDetector.php
 create mode 100644 src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/Detector/Option.php
 create mode 100644 src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/Detector/Options.php
 create mode 100644 src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/Exception/CantDetectCSVOptions.php
 create mode 100644 src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Integration/CSVDetectorTest.php
 create mode 100644 src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Unit/Detector/OptionTest.php

diff --git a/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/CSVDetector.php b/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/CSVDetector.php
new file mode 100644
index 000000000..8da757b54
--- /dev/null
+++ b/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/CSVDetector.php
@@ -0,0 +1,80 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Adapter\CSV;
+
+use Flow\ETL\Adapter\CSV\Detector\Option;
+use Flow\ETL\Adapter\CSV\Detector\Options;
+use Flow\ETL\Adapter\CSV\Exception\CantDetectCSVOptions;
+use Flow\ETL\Exception\InvalidArgumentException;
+
+final class CSVDetector
+{
+    private ?Option $fallback;
+
+    private Options $options;
+
+    /**
+     * @var resource
+     */
+    private $resource;
+
+    private int $startingPosition;
+
+    /**
+     * @param resource $resource
+     */
+    public function __construct($resource, ?Option $fallback = new Option(',', '"', '\\'), ?Options $options = null)
+    {
+        /** @psalm-suppress DocblockTypeContradiction */
+        if (!\is_resource($resource)) {
+            throw new InvalidArgumentException('Argument must be a valid resource');
+        }
+
+        $this->resource = $resource;
+        /** @phpstan-ignore-next-line */
+        $this->startingPosition = \ftell($resource);
+        $this->options = $options ?? Options::all();
+        $this->fallback = $fallback;
+    }
+
+    public function __destruct()
+    {
+        \fseek($this->resource, $this->startingPosition);
+    }
+
+    /**
+     * @throws CantDetectCSVOptions|InvalidArgumentException
+     */
+    public function detect(int $lines = 5) : Option
+    {
+        if ($lines < 1) {
+            throw new InvalidArgumentException('Lines must be greater than 0');
+        }
+
+        $readLines = 1;
+
+        while ($line = \fgets($this->resource)) {
+            $this->options->parse($line);
+
+            if ($readLines++ >= $lines) {
+                break;
+            }
+        }
+
+        try {
+            $bestOption = $this->options->onlyValid()->best();
+        } catch (CantDetectCSVOptions $e) {
+            if ($this->fallback) {
+                return $this->fallback;
+            }
+
+            throw $e;
+        }
+
+        $this->options = $this->options->reset();
+
+        return $bestOption;
+    }
+}
diff --git a/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/CSVExtractor.php b/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/CSVExtractor.php
index 2eddee69a..acbdd5903 100644
--- a/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/CSVExtractor.php
+++ b/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/CSVExtractor.php
@@ -28,9 +28,9 @@ public function __construct(
         private readonly Path $path,
         private readonly bool $withHeader = true,
         private readonly bool $emptyToNull = true,
-        private readonly string $separator = ',',
-        private readonly string $enclosure = '"',
-        private readonly string $escape = '\\',
+        private readonly string|null $separator = null,
+        private readonly string|null $enclosure = null,
+        private readonly string|null $escape = null,
         private readonly int $charactersReadInLine = 1000
     ) {
         $this->resetLimit();
@@ -43,15 +43,21 @@ public function extract(FlowContext $context) : \Generator
         foreach ($context->streams()->fs()->scan($this->path, $this->partitionFilter()) as $path) {
             $stream = $context->streams()->fs()->open($path, Mode::READ);
 
+            $option = \Flow\ETL\Adapter\CSV\csv_detect_separator($stream->resource());
+
+            $separator = $this->separator ?? $option->separator;
+            $enclosure = $this->enclosure ?? $option->enclosure;
+            $escape = $this->escape ?? $option->escape;
+
             $headers = [];
 
             if ($this->withHeader && \count($headers) === 0) {
                 /** @var array<string> $headers */
-                $headers = \fgetcsv($stream->resource(), $this->charactersReadInLine, $this->separator, $this->enclosure, $this->escape);
+                $headers = \fgetcsv($stream->resource(), $this->charactersReadInLine, $separator, $enclosure, $escape);
             }
 
             /** @var array<mixed> $rowData */
-            $rowData = \fgetcsv($stream->resource(), $this->charactersReadInLine, $this->separator, $this->enclosure, $this->escape);
+            $rowData = \fgetcsv($stream->resource(), $this->charactersReadInLine, $separator, $enclosure, $escape);
 
             if (!\count($headers)) {
                 $headers = \array_map(fn (int $e) : string => 'e' . \str_pad((string) $e, 2, '0', STR_PAD_LEFT), \range(0, \count($rowData) - 1));
@@ -81,7 +87,7 @@ public function extract(FlowContext $context) : \Generator
                 }
 
                 if (\count($headers) !== \count($rowData)) {
-                    $rowData = \fgetcsv($stream->resource(), $this->charactersReadInLine, $this->separator, $this->enclosure, $this->escape);
+                    $rowData = \fgetcsv($stream->resource(), $this->charactersReadInLine, $separator, $enclosure, $escape);
 
                     continue;
                 }
@@ -101,7 +107,7 @@ public function extract(FlowContext $context) : \Generator
                     return;
                 }
 
-                $rowData = \fgetcsv($stream->resource(), $this->charactersReadInLine, $this->separator, $this->enclosure, $this->escape);
+                $rowData = \fgetcsv($stream->resource(), $this->charactersReadInLine, $separator, $enclosure, $escape);
             }
         }
 
diff --git a/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/Detector/Option.php b/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/Detector/Option.php
new file mode 100644
index 000000000..56906b504
--- /dev/null
+++ b/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/Detector/Option.php
@@ -0,0 +1,94 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Adapter\CSV\Detector;
+
+use Flow\ETL\Exception\InvalidArgumentException;
+
+final class Option
+{
+    private const COLUMN_SCORE_WEIGHT = 100_000;
+
+    private const COLUMNS_LENGTH_WEIGHT = 10_000;
+
+    /**
+     * @var array<mixed>
+     */
+    private array $rows;
+
+    public function __construct(
+        public string $separator,
+        public string $enclosure,
+        public string $escape = '\\'
+    ) {
+        if (\mb_strlen($this->separator) !== 1) {
+            throw new InvalidArgumentException('Separator must be a single character');
+        }
+
+        if (\mb_strlen($this->enclosure) !== 1) {
+            throw new InvalidArgumentException('Enclosure must be a single character');
+        }
+
+        $this->rows = [];
+    }
+
+    public function isValid() : bool
+    {
+        $columnsCount = null;
+
+        foreach ($this->rows as $row) {
+            if ($columnsCount === null) {
+                $columnsCount = \count($row);
+
+                continue;
+            }
+
+            if ($columnsCount !== \count($row)) {
+                return false;
+            }
+        }
+
+        if ($columnsCount === 1) {
+            return false;
+        }
+
+        return true;
+    }
+
+    public function parse(string $line) : void
+    {
+        $this->rows[] = \str_getcsv($line, $this->separator, $this->enclosure);
+    }
+
+    public function reset() : self
+    {
+        return new self($this->separator, $this->enclosure);
+    }
+
+    public function score() : int
+    {
+        if (!$this->isValid()) {
+            return 0;
+        }
+
+        if (!\count($this->rows)) {
+            return 0;
+        }
+
+        $columnScore = \count($this->rows[0]) * self::COLUMN_SCORE_WEIGHT;
+        $totalLength = \array_reduce(
+            $this->rows,
+            static fn (int $carry, array $row) : int => $carry + \array_reduce(
+                $row,
+                static fn (int $carry, $column) : int => $carry + (\is_string($column) ? \mb_strlen($column) : 0),
+                0
+            ),
+            0
+        );
+
+        $lengthScore = (int) \round((1 / ($totalLength + 1) * self::COLUMNS_LENGTH_WEIGHT));
+
+        return $columnScore + $lengthScore;
+    }
+}
diff --git a/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/Detector/Options.php b/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/Detector/Options.php
new file mode 100644
index 000000000..63eefb848
--- /dev/null
+++ b/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/Detector/Options.php
@@ -0,0 +1,85 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Adapter\CSV\Detector;
+
+use Flow\ETL\Adapter\CSV\Exception\CantDetectCSVOptions;
+
+final class Options
+{
+    /**
+     * @var array<Option>
+     */
+    private array $options;
+
+    /**
+     * @param array<Option> $options
+     */
+    public function __construct(array $options)
+    {
+        $this->options = $options;
+    }
+
+    public static function all() : self
+    {
+        $separators = [',', "\t", ';', '|', ' ', '_', '-', ':', '~', '@', '#', '$', '%', '^', '&', '*', '(', ')', '+', '=', '?', '!', '\\', '/', '.', '>', '<'];
+        $enclosures = ['"', "'"];
+
+        $options = [];
+
+        foreach ($separators as $separator) {
+            foreach ($enclosures as $enclosure) {
+                $options[] = new Option($separator, $enclosure);
+            }
+        }
+
+        return new self($options);
+    }
+
+    public function best() : Option
+    {
+        $best = null;
+
+        foreach ($this->options as $option) {
+            if ($best === null) {
+                $best = $option;
+
+                continue;
+            }
+
+            if ($option->score() > $best->score()) {
+                $best = $option;
+            }
+        }
+
+        if ($best === null) {
+            throw new CantDetectCSVOptions('No best option found');
+        }
+
+        return $best;
+    }
+
+    public function onlyValid() : self
+    {
+        return new self(\array_filter($this->options, fn (Option $option) : bool => $option->isValid()));
+    }
+
+    public function parse(string $line) : void
+    {
+        foreach ($this->options as $option) {
+            $option->parse($line);
+        }
+    }
+
+    public function reset() : self
+    {
+        $options = [];
+
+        foreach ($this->options as $option) {
+            $options[] = $option->reset();
+        }
+
+        return new self($options);
+    }
+}
diff --git a/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/Exception/CantDetectCSVOptions.php b/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/Exception/CantDetectCSVOptions.php
new file mode 100644
index 000000000..c0f44af0f
--- /dev/null
+++ b/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/Exception/CantDetectCSVOptions.php
@@ -0,0 +1,11 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Adapter\CSV\Exception;
+
+use Flow\ETL\Exception\RuntimeException;
+
+final class CantDetectCSVOptions extends RuntimeException
+{
+}
diff --git a/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/functions.php b/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/functions.php
index 6693bdc1e..7d0f05cc1 100644
--- a/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/functions.php
+++ b/src/adapter/etl-adapter-csv/src/Flow/ETL/Adapter/CSV/functions.php
@@ -5,6 +5,8 @@
 namespace Flow\ETL\Adapter\CSV;
 
 use function Flow\ETL\DSL\from_all;
+use Flow\ETL\Adapter\CSV\Detector\Option;
+use Flow\ETL\Adapter\CSV\Detector\Options;
 use Flow\ETL\Extractor;
 use Flow\ETL\Filesystem\Path;
 use Flow\ETL\Loader;
@@ -16,9 +18,9 @@ function from_csv(
     string|Path|array $path,
     bool $with_header = true,
     bool $empty_to_null = true,
-    string $delimiter = ',',
-    string $enclosure = '"',
-    string $escape = '\\',
+    string|null $delimiter = null,
+    string|null $enclosure = null,
+    string|null $escape = null,
     int $characters_read_in_line = 1000
 ) : Extractor {
     if (\is_array($path)) {
@@ -67,3 +69,14 @@ function to_csv(
         $new_line_separator
     );
 }
+
+/**
+ * @param resource $resource - valid resource to CSV file opened with 'r' mode
+ * @param int<1, max> $lines - number of lines to read from CSV file, default 5, more lines means more accurate detection but slower detection
+ * @param null|Option $fallback - fallback option to use when no best option can be detected, default is Option(',', '"', '\\')
+ * @param null|Options $options - options to use for detection, default is Options::all()
+ */
+function csv_detect_separator($resource, int $lines = 5, ?Option $fallback = new Option(',', '"', '\\'), ?Options $options = null) : Option
+{
+    return (new CSVDetector($resource, $fallback, $options))->detect($lines);
+}
diff --git a/src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Integration/CSVDetectorTest.php b/src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Integration/CSVDetectorTest.php
new file mode 100644
index 000000000..e332b64de
--- /dev/null
+++ b/src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Integration/CSVDetectorTest.php
@@ -0,0 +1,96 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Adapter\CSV\Tests\Integration;
+
+use Flow\ETL\Adapter\CSV\CSVDetector;
+use PHPUnit\Framework\Attributes\DataProvider;
+use PHPUnit\Framework\TestCase;
+
+final class CSVDetectorTest extends TestCase
+{
+    public static function enclosure_provider() : \Generator
+    {
+        yield ['double_quote' => '"'];
+        yield ['single_quote' => "'"];
+    }
+
+    public static function separator_provider() : \Generator
+    {
+        yield ['comma' => ','];
+        yield ['tab' => "\t"];
+        yield ['semicolon' => ';'];
+        yield ['pipe' => '|'];
+        yield ['space' => ' '];
+        yield ['underscore' => '_'];
+        yield ['dash' => '-'];
+        yield ['double_dot' => ':'];
+        yield ['tilde' => '~'];
+        yield ['at' => '@'];
+        yield ['hash' => '#'];
+        yield ['dollar' => '$'];
+        yield ['percent' => '%'];
+        yield ['caret' => '^'];
+        yield ['ampersand' => '&'];
+        yield ['asterisk' => '*'];
+        yield ['left_parenthesis' => '('];
+        yield ['right_parenthesis' => ')'];
+        yield ['plus' => '+'];
+        yield ['equal' => '='];
+        yield ['question_mark' => '?'];
+        yield ['exclamation_mark' => '!'];
+        yield ['backslash' => '\\'];
+        yield ['slash' => '/'];
+        yield ['dot' => '.'];
+        yield ['greater_than' => '>'];
+        yield ['less_than' => '<'];
+    }
+
+    #[DataProvider('enclosure_provider')]
+    public function test_detecting_enclosures(string $enclosure) : void
+    {
+        $detector = new CSVDetector($this->createResource(',', $enclosure));
+
+        $this->assertSame($enclosure, $detector->detect()->enclosure);
+    }
+
+    #[DataProvider('separator_provider')]
+    public function test_detecting_separators(string $separator) : void
+    {
+        $detector = new CSVDetector($this->createResource($separator));
+
+        $this->assertSame($separator, $detector->detect()->separator);
+    }
+
+    /**
+     * @return resource
+     */
+    private function createResource(string $separator = ',', string $enclosure = '"')
+    {
+        $data = [
+            ['id', 'name', 'email'],
+            ['1', 'John Doe', 'john@example.com'],
+            ['2', 'Jane Doe', 'jane@example.com'],
+            ['3', 'Mark', 'mark@example.com'],
+            ['4', 'Kate', 'kate@example.com'],
+            ['5', 'Peter', 'peter@example.com'],
+            ['6', 'Paul', 'paul@example.com'],
+            ['7', 'Mary', 'mary@example.com'],
+            ['8', 'Anna', 'anna@example.com'],
+            ['9', 'Robert', 'rober@example.com'],
+            ['10', 'Lucy', 'lucy@example.com'],
+            ['11', 'Ro\'bert', 'rob_ert@example.com'],
+        ];
+
+        $resource = \fopen('php://memory', 'rb+');
+
+        foreach ($data as $line) {
+            \fputcsv($resource, $line, $separator, $enclosure);
+        }
+
+        \rewind($resource);
+
+        return $resource;
+    }
+}
diff --git a/src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Unit/Detector/OptionTest.php b/src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Unit/Detector/OptionTest.php
new file mode 100644
index 000000000..0ca9f7a60
--- /dev/null
+++ b/src/adapter/etl-adapter-csv/tests/Flow/ETL/Adapter/CSV/Tests/Unit/Detector/OptionTest.php
@@ -0,0 +1,37 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Flow\ETL\Adapter\CSV\Tests\Unit\Detector;
+
+use Flow\ETL\Adapter\CSV\Detector\Option;
+use Flow\ETL\Exception\InvalidArgumentException;
+use PHPUnit\Framework\TestCase;
+
+final class OptionTest extends TestCase
+{
+    public function test_empty_enclosure() : void
+    {
+        $this->expectException(InvalidArgumentException::class);
+
+        new Option(',', '');
+    }
+
+    public function test_empty_separator() : void
+    {
+        $this->expectException(InvalidArgumentException::class);
+
+        new Option('', '"');
+    }
+
+    public function test_score() : void
+    {
+        $option = new Option(',', "'");
+
+        $option->parse('a,b,c');
+        $option->parse('a,b,c');
+        $option->parse('a,b,c');
+
+        $this->assertSame(301000, $option->score());
+    }
+}