data_pipelines-1.x-dev/src/Plugin/DatasetSource/CsvSource.php

<?php

declare(strict_types=1);

namespace Drupal\data_pipelines\Plugin\DatasetSource;

use Drupal\Core\Field\BaseFieldDefinition;
use Drupal\Core\StringTranslation\TranslatableMarkup;
use Drupal\data_pipelines\Attribute\DatasetSource;
use Drupal\data_pipelines\DatasetData;
use Drupal\data_pipelines\Entity\DatasetInterface;
use Drupal\data_pipelines\Source\DatasetSourceBase;
use Drupal\data_pipelines\Traits\NonPrintingCharsTrait;

/**
 * A class for providing csv as a source.
 */
#[DatasetSource(
  id: 'csv',
  label: new TranslatableMarkup('CSV'),
  deriver: SourceDeriver::class,
)]
class CsvSource extends DatasetSourceBase {

  use NonPrintingCharsTrait;

  /**
   * {@inheritdoc}
   */
  public function extractDataFromDataSet(DatasetInterface $dataset): \Generator {
    try {
      $resource = $this->getResource($dataset);
      $delimiter = ',';
      if ($dataset->hasField('csv_delimiter')) {
        $delimiter_selected = $dataset->get('csv_delimiter')->value;
        switch ($delimiter_selected) {
          case 'semicolon':
            $delimiter = ';';
            break;

          case 'tab':
            $delimiter = "\t";
            break;

          case 'pipe':
            $delimiter = '|';
            break;

          case 'colon':
            $delimiter = ':';
            break;

          default:
            break;
        }
      }
      $this->removeBom($resource);
      $file_rows = 0;
      while (($file_row = fgetcsv($resource, NULL, $delimiter)) !== FALSE) {
        $file_row = array_map(function ($value) {
          return $this->removeDosChars($value);
        }, $file_row);
        if (0 === $file_rows) {
          $file_header_row = $file_row;
          $file_header_row_length = count($file_header_row);
          $file_header_row = array_map('trim', $file_header_row);
          $file_rows++;
          continue;
        }
        // Ensure the header is the same length as the row.
        $normalized_row = array_pad(array_slice($file_row, 0, $file_header_row_length), $file_header_row_length, NULL);
        yield new DatasetData(array_combine($file_header_row, $normalized_row));
        $file_rows++;
      }
      fclose($resource);
    }
    catch (\Exception $e) {
      $this->loggerChannel->critical($e->getMessage());
      yield from [];
    }
  }

  /**
   * {@inheritdoc}
   */
  public function buildFieldDefinitions(): array {
    $fields = parent::buildFieldDefinitions();
    $fields['csv_delimiter'] = BaseFieldDefinition::create('list_string')
      ->setLabel(new TranslatableMarkup('Delimiter'))
      ->setDefaultValue('comma')
      ->setSetting('allowed_values', [
        'comma'     => 'Comma (,)',
        'semicolon' => 'Semicolon (;)',
        'tab'         => 'Tab (\t)',
        'pipe'         => 'Pipe (|)',
        'colon'     => 'Colon (:)',
      ])
      ->setDisplayOptions('form', [
        'type' => 'options_select',
      ])
      ->setRequired(TRUE);
    return $fields;
  }

}