<commit_before>require "addressable/uri"

module Embulk
  module Filter

    class QueryStringRuby < FilterPlugin
      Plugin.register_filter("query_string_ruby", self)

      def self.transaction(config, in_schema, &control)
        task = {
          "target_column" => in_schema.find{|c| c.name == config.param("column", :string)},
          "schema" => config.param("schema", :array, :default => [])
        }

        out_columns = in_schema + task["schema"].map {|col| Column.new(nil, col["name"], col["type"].to_sym)}
        yield(task, out_columns)
      end

      def init
        @schema = task["schema"]
        @target_column = task["target_column"]
      end

      def close
      end

      def add(page)
        page.each do |record|
          q = query_parser(record[@target_column["index"]])
          add_record = make_record(@schema, q)
          page_builder.add(record + add_record)
        end
      end

      def finish
        page_builder.finish
      end

      private

      def query_parser(query_string)
        u = Addressable::URI.parse(query_string)
        uri = u.query ? u : Addressable::URI.parse("?#{query_string}")
        return uri.query_values(Hash)
      end

      def make_record(schema, query)
        return schema.map do |col|
          v = query[col["name"]]
          if !v.to_s.empty?
            begin
              case col["type"]
              when "long"
                Integer(v)
              when "double"
                Float(v)
              when "timestamp"
                Time.parse(v)
              else
                v.to_s
              end
            rescue => e
              Embulk.logger.warn("Cast failed '#{v}' as '#{col["type"]}' (query name is '#{col["name"]}')")
            end
          end
        end
      end

    end

  end
end
<commit_msg>query parse の際に exception の判定を追加 if end でネストさせていた箇所を next を使用してネストを解消<commit_after>require "addressable/uri"

module Embulk
  module Filter

    class QueryStringRuby < FilterPlugin
      Plugin.register_filter("query_string_ruby", self)

      def self.transaction(config, in_schema, &control)
        task = {
          "target_column" => in_schema.find{|c| c.name == config.param("column", :string)},
          "schema" => config.param("schema", :array, :default => [])
        }

        out_columns = in_schema + task["schema"].map {|col| Column.new(nil, col["name"], col["type"].to_sym)}
        yield(task, out_columns)
      end

      def init
        @schema = task["schema"]
        @target_column = task["target_column"]
      end

      def close
      end

      def add(page)
        page.each do |record|
          q = query_parser(record[@target_column["index"]])
          add_record = make_record(@schema, q)
          page_builder.add(record + add_record)
        end
      end

      def finish
        page_builder.finish
      end

      private

      def query_parser(query_string)
        begin
          u = Addressable::URI.parse(query_string)
          uri = u.query ? u : Addressable::URI.parse("?#{query_string}")
          uri.query_values(Hash)
        rescue => e
          Embulk.logger.warn("Parse failed '#{query_string}'")
        end
      end

      def make_record(schema, query)
        return schema.map do |col|
          v = query[col["name"]]

          if v.to_s.empty?
            next
          end

          begin
            case col["type"]
            when "long"
              Integer(v)
            when "double"
              Float(v)
            when "timestamp"
              Time.parse(v)
            else
              v.to_s
            end
          rescue => e
            Embulk.logger.warn("Cast failed '#{v}' as '#{col["type"]}' (query name is '#{col["name"]}')")
          end
        end
      end

    end

  end
end

{'fraction_non_alphanumeric': 0.11727748691099477, 'fraction_numerical': 0.0, 'mean_word_length': 1.7331902718168812, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 12, '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': '7542889', 'n_tokens_mistral': 1163, 'n_tokens_neox': 1123, 'n_words': 254}