<commit_before>module Metacrunch
  class Job::Context

    def initialize(job, args: nil)
      @_job = job
      @_args = args
    end

    def run
      @_job.run
    end

    def source(source)
      @_job.sources << source if source
    end

    def destination(destination)
      @_job.destinations << destination if destination
    end

    def pre_process(callable = nil, &block)
      add_callable_or_block(@_job.pre_processes, callable, &block)
    end

    def post_process(callable = nil, &block)
      add_callable_or_block(@_job.post_processes, callable, &block)
    end

    def transformation(callable = nil, &block)
      add_callable_or_block(@_job.transformations, callable, &block)
    end

    def register_options(&block)
      registry = OptionRegistry.new
      yield(registry)

      OptionParser.new do |parser|
        parser.banner = "Job specific options:"

        registry.each do |key, opt_def|
          options[key] = opt_def[:default]
          parser.on(*opt_def[:args]) do |value|
            options[key] = value
          end
        end
      end.parse(@_args)
    end

    def options
      @_options ||= {}
    end

  private

    def add_callable_or_block(array, callable, &block)
      if block_given?
        array << block
      elsif callable
        array << callable
      end
    end

    class OptionRegistry

      def add(name, *args, default:)
        options[name.to_sym] = {
          args: args,
          default: default
        }
      end

      def each(&block)
        options.each(&block)
      end

    private

      def options
        @options ||= {}<commit_msg>Add row_buffer to DSL.<commit_after>module Metacrunch
  class Job::Context

    def initialize(job, args: nil)
      @_job = job
      @_args = args
    end

    def run
      @_job.run
    end

    def source(source)
      @_job.sources << source if source
    end

    def destination(destination)
      @_job.destinations << destination if destination
    end

    def pre_process(callable = nil, &block)
      add_callable_or_block(@_job.pre_processes, callable, &block)
    end

    def post_process(callable = nil, &block)
      add_callable_or_block(@_job.post_processes, callable, &block)
    end

    def transformation(callable = nil, &block)
      add_callable_or_block(@_job.transformations, callable, &block)
    end

    def register_options(&block)
      registry = OptionRegistry.new
      yield(registry)

      OptionParser.new do |parser|
        parser.banner = "Job specific options:"

        registry.each do |key, opt_def|
          options[key] = opt_def[:default]
          parser.on(*opt_def[:args]) do |value|
            options[key] = value
          end
        end
      end.parse(@_args)
    end

    def options
      @_options ||= {}
    end

    def row_buffer(id, row, size: 1)
      buffer = row_buffers[id] ||= []
      buffer << row

      if buffer.count >= size
        row_buffers[id] = []
        buffer
      end
    end

    def row_buffers
      @__row_buffers ||= {}
    end

  private

    def add_callable_or_block(array, callable, &block)
      if block_given?
        array << block
      elsif callable
        array << callable
      end
    end

    class OptionRegistry

      def add(name, *args, default:)
        options[name.to_sym] = {
          args: args,
          default: default
        }
      end

      def each(&block)
        options.each(&block)
      end

    private

      def options
        @options ||= {}
{'fraction_non_alphanumeric': 0.0927981781952747, 'fraction_numerical': 0.0002846569883290635, 'mean_word_length': 2.0609756097560976, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4768594', 'n_tokens_mistral': 1110, 'n_tokens_neox': 1064, 'n_words': 265}