<commit_before>class Kumogata::PostProcessing
  TRIGGER_TIMING = [:create, :update]

  def initialize(options)
    @options = options
    @commands = {}

    @command_options = {
      :undent    => true,
      :trim_mode => nil,
    }
  end

  def fetch!(template)
    _post = template.delete(:_post)
    return unless _post

    options = _post[:options] || {}
    @command_options.merge(options)

    outputs = template['Outputs'] || {}

    _post.fetch(:commands).each do |name, attrs|
      unless attrs.kind_of?(Hash) and attrs['command']
        raise "Invalid post processing: #{name} => #{attrs.inspect}"
      end

      timing = [(attrs['after'] || TRIGGER_TIMING)].flatten.map {|i| i.to_sym }
      command = attrs['command']

      validate_timing(name, timing)
      validate_command_template(name, command, outputs)

      @commands[name] = {
        :after   => timing,
        :command => command,
      }

      if (ssh = attrs['ssh'])
        validate_ssh(name, ssh, outputs)
        @commands[name][:ssh] = ssh
      end
    end
  end

  def run(timing, outputs)
    results = []

    @commands.each do |name, attrs|
      next unless attrs[:after].include?(timing)

      print_command(name)
      out, err, status = run_command(attrs, outputs)
      print_command_result(out, err, status)

      results << {
        name => {
          'ExitStatus' => status.to_i,
          'StdOut' => out.force_encoding('UTF-8'),
          'StdErr' => err.force_encoding('UTF-8'),
        }
      }
    end

    save_command_results(results) unless results.empty?
  end

  private

  def validate_timing(name, timing)
    timing.each do |t|
      unless TRIGGER_TIMING.include?(t)
        raise "Unknown post processing timing: #{name} => #{timing.inspect}"
      end
    end
  end

  def validate_ssh(name, ssh, outputs)
    host, user, options = ssh.values_at('host', 'user', 'options')

    unless host and user
      raise "`host` and `user` is required for post processing ssh: #{name}"
    end

    if host.kind_of?(Hash)
      if host.keys != ['Key']
        raise "Invalid post processing ssh host: #{name} => #{host.inspect}"
      end

      host_key, host_value = host.first
      ssh['host'] = "<%= #{host_key} #{host_value.to_s.inspect} %>"
    else
      ssh['host'] = host.to_s
    end

    validate_command_template(name, ssh['host'], outputs)

    if user.kind_of?(Hash)
      if user.keys != ['Key']
        raise "Invalid post processing ssh user: #{name} => #{user.inspect}"
      end

      user_key, user_value = user.first
      ssh['user'] = "<%= #{user_key} #{user_value.to_s.inspect} %>"
    else
      ssh['user'] = user.to_s
    end

    validate_command_template(name, ssh['user'], outputs)

    if options and not options.kind_of?(Hash)
      raise "Invalid post processing ssh options: #{name} => #{options.inspect}"
    end
  end

  def run_command(attrs, outputs)
    command, ssh = attrs.values_at(:command, :ssh)

    if ssh
      run_ssh_command(ssh, command, outputs)
    else
      run_shell_command(command, outputs)
    end
  end

  def run_ssh_command(ssh, command, outputs)
    host, user, options = ssh.values_at('host', 'user', 'options')
    host = evaluate_command_template(host, outputs)
    user = evaluate_command_template(user, outputs)
    args = [host, user]
    args << ssh['options'] if ssh['options']

    command = evaluate_command_template(command, outputs)

    connect_tries = (ssh['connect_tries'] || 36).to_i
    retry_interval = (ssh['retry_interval'] || 5).to_i

    begin
      retryable(:tries => connect_tries, :on => Net::SSH::Disconnect, :sleep => retry_interval) do
        Net::SSH.start(*args) {|ssh| ssh_exec!(ssh, command) }
      end
    rescue Net::SSH::HostKeyMismatch => e
      e.remember_host!
      retry
    end
  end

  def ssh_exec!(ssh, command)
    stdout_data = ''
    stderr_data = ''
    exit_code = nil
    #exit_signal = nil

    stdout_stream = create_stdout_stream
    stderr_stream = create_stderr_stream

    ssh.open_channel do |channel|
      channel.exec(command) do |ch, success|
        unless success
          raise "Couldn't execute command #{command.inspect} (ssh.channel.exec)"
        end

        channel.on_data do |ch, data|
          stdout_stream.push data
          stdout_data << data
        end

        channel.on_extended_data do |ch, type, data|
          stderr_stream.push data
          stderr_data << data
        end
<commit_msg>disable stderr while ssh<commit_after>class Kumogata::PostProcessing
  TRIGGER_TIMING = [:create, :update]

  def initialize(options)
    @options = options
    @commands = {}

    @command_options = {
      :undent    => true,
      :trim_mode => nil,
    }
  end

  def fetch!(template)
    _post = template.delete(:_post)
    return unless _post

    options = _post[:options] || {}
    @command_options.merge(options)

    outputs = template['Outputs'] || {}

    _post.fetch(:commands).each do |name, attrs|
      unless attrs.kind_of?(Hash) and attrs['command']
        raise "Invalid post processing: #{name} => #{attrs.inspect}"
      end

      timing = [(attrs['after'] || TRIGGER_TIMING)].flatten.map {|i| i.to_sym }
      command = attrs['command']

      validate_timing(name, timing)
      validate_command_template(name, command, outputs)

      @commands[name] = {
        :after   => timing,
        :command => command,
      }

      if (ssh = attrs['ssh'])
        validate_ssh(name, ssh, outputs)
        @commands[name][:ssh] = ssh
      end
    end
  end

  def run(timing, outputs)
    results = []

    @commands.each do |name, attrs|
      next unless attrs[:after].include?(timing)

      print_command(name)
      out, err, status = run_command(attrs, outputs)
      print_command_result(out, err, status)

      results << {
        name => {
          'ExitStatus' => status.to_i,
          'StdOut' => out.force_encoding('UTF-8'),
          'StdErr' => err.force_encoding('UTF-8'),
        }
      }
    end

    save_command_results(results) unless results.empty?
  end

  private

  def validate_timing(name, timing)
    timing.each do |t|
      unless TRIGGER_TIMING.include?(t)
        raise "Unknown post processing timing: #{name} => #{timing.inspect}"
      end
    end
  end

  def validate_ssh(name, ssh, outputs)
    host, user, options = ssh.values_at('host', 'user', 'options')

    unless host and user
      raise "`host` and `user` is required for post processing ssh: #{name}"
    end

    if host.kind_of?(Hash)
      if host.keys != ['Key']
        raise "Invalid post processing ssh host: #{name} => #{host.inspect}"
      end

      host_key, host_value = host.first
      ssh['host'] = "<%= #{host_key} #{host_value.to_s.inspect} %>"
    else
      ssh['host'] = host.to_s
    end

    validate_command_template(name, ssh['host'], outputs)

    if user.kind_of?(Hash)
      if user.keys != ['Key']
        raise "Invalid post processing ssh user: #{name} => #{user.inspect}"
      end

      user_key, user_value = user.first
      ssh['user'] = "<%= #{user_key} #{user_value.to_s.inspect} %>"
    else
      ssh['user'] = user.to_s
    end

    validate_command_template(name, ssh['user'], outputs)

    if options and not options.kind_of?(Hash)
      raise "Invalid post processing ssh options: #{name} => #{options.inspect}"
    end
  end

  def run_command(attrs, outputs)
    command, ssh = attrs.values_at(:command, :ssh)

    if ssh
      run_ssh_command(ssh, command, outputs)
    else
      run_shell_command(command, outputs)
    end
  end

  def run_ssh_command(ssh, command, outputs)
    host, user, options = ssh.values_at('host', 'user', 'options')
    host = evaluate_command_template(host, outputs)
    user = evaluate_command_template(user, outputs)
    args = [host, user]
    args << ssh['options'] if ssh['options']

    command = evaluate_command_template(command, outputs)

    connect_tries = (ssh['connect_tries'] || 36).to_i
    retry_interval = (ssh['retry_interval'] || 5).to_i

    stderr_orig = nil

    begin
      stderr_orig = STDERR.dup
      STDERR.reopen('/dev/null', 'w')

      begin
        retryable(:tries => connect_tries, :on => Net::SSH::Disconnect, :sleep => retry_interval) do
          Net::SSH.start(*args) {|ssh| ssh_exec!(ssh, command) }
        end
      rescue Net::SSH::HostKeyMismatch => e
        e.remember_host!
        retry
      end
    ensure
      STDERR.reopen(stderr_orig)
    end
  end

  def ssh_exec!(ssh, command)
    stdout_data = ''
    stderr_data = ''
    exit_code = nil
    #exit_signal = nil

    stdout_stream = create_stdout_stream
    stderr_stream = create_stderr_stream

    ssh.open_channel do |channel|
      channel.exec(command) do |ch, success|
        unless success
          raise "Couldn't execute command #{command.inspect} (ssh.channel.exec)"
        end

        channel.on_data do |ch, data|
          stdout_stream.push data
          stdout_data << data
        end

        channel.on_extended_data do |ch, type, data|
          stderr_stream.push data
          stderr_data << data
        end

{'fraction_non_alphanumeric': 0.12854317732366513, 'fraction_numerical': 0.001098659635245001, 'mean_word_length': 2.6690850463522775, 'pattern_counts': {'":': 0, '<': 23, '<?xml version=': 0, '>': 41, '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': '7636106', 'n_tokens_mistral': 3020, 'n_tokens_neox': 2866, 'n_words': 683}