#
# Eigenes Plugin: liest alle CSV-Dateien eines Verzeichnisses als eine Quelle
#
#   bin/etl run examples/eigenes_plugin.rb [--out=tmp]
#
require 'fileutils'
require 'datapipeline'

DATA = File.expand_path('data', __dir__)
out = DataPipeline.params.fetch(:out, 'tmp')
FileUtils.mkdir_p(out)

class VerzeichnisStorage < DataPipeline::StoragePlugin
  def initialize dir:, pattern: '*.csv', **csv_options
    @dir = dir
    @pattern = pattern
    @csv_options = csv_options
  end

  # jede Zeile bekommt den Dateinamen dazu; mit Block zeilenweise (für streamed)
  def read(&block)
    rows = Dir.glob(File.join(@dir, @pattern)).sort.flat_map do |file|
      DataPipeline::CsvStorage.new(file:, **@csv_options).read.map { |row| row.merge('datei' => File.basename(file, '.*')) }
    end
    return rows unless block

    rows.each(&block)
    nil
  end
end

DataPipeline.register :verzeichnis, VerzeichnisStorage

DataPipeline.new do
  read container(:verzeichnis, dir: "#{DATA}/filialen")
  map { |row| row.merge('menge' => row['menge'].to_i) }
  write container(:csv, file: "#{out}/alle_filialen.csv", columns: %w[datei artikel menge])
end.run