<commit_before>require "couchchanges"
require "em-elasticsearch"

module Couchlastic
  class Indexer
    def initialize
      @indices = {}
      @mappings = {}
      yield self if block_given?
    end

    attr_accessor :couch

    attr_reader :elastic

    def elastic= url
      @elastic = EventMachine::ElasticSearch::Client.new(url)
    end

    def map name, mapping
      @mappings[name] = mapping
    end

    def index name, &block
      @indices[name] = block
    end

    def start
      put_mappings = lambda {|hash, iter|
        name, type = hash[0].split("/")
        mapping    = hash[1]
        index      = elastic.index(name)

        Couchlastic.logger.info "Mapping #{name}/#{type}."

        index.create {
          index.type(type).map(mapping) {iter.next}
        }
      }

      start_changes = lambda {
        changes = CouchChanges.new
        changes.update {|change|
          Couchlastic.logger.info "Indexing update sequence #{change["seq"]}"

          doc = change.delete("doc")
          index_change change, doc
        }
        changes.listen :url => @couch, :include_docs => true
      }

      EM::Iterator.new(@mappings).each(put_mappings, start_changes)
    end

    private

    def index_change change, doc
      @indices.each {|name, block|
        if doc = block.call(change, doc)
          type = elastic.index(name).type(doc[:type])
          type.index(doc[:id], doc[:doc])
        end
      }
    end
  end
end<commit_msg>more logging<commit_after>require "couchchanges"
require "em-elasticsearch"

module Couchlastic
  class Indexer
    def initialize
      @indices = {}
      @mappings = {}
      yield self if block_given?
    end

    attr_accessor :couch

    attr_reader :elastic

    def elastic= url
      @elastic = EventMachine::ElasticSearch::Client.new(url)
    end

    def map name, mapping
      @mappings[name] = mapping
    end

    def index name, &block
      @indices[name] = block
    end

    def start
      put_mappings = lambda {|hash, iter|
        name, type = hash[0].split("/")
        mapping    = hash[1]
        index      = elastic.index(name)

        Couchlastic.logger.info "Mapping #{name}/#{type}."

        index.create {
          index.type(type).map(mapping) {iter.next}
        }
      }

      start_changes = lambda {
        Couchlastic.logger.info "Listening to changes from #{@couch}"

        changes = CouchChanges.new
        changes.update {|change|
          Couchlastic.logger.info "Indexing update sequence #{change["seq"]}"

          doc = change.delete("doc")
          index_change change, doc
        }
        changes.listen :url => @couch, :include_docs => true
      }

      EM::Iterator.new(@mappings).each(put_mappings, start_changes)
    end

    private

    def index_change change, doc
      @indices.each {|name, block|
        if doc = block.call(change, doc)
          type = elastic.index(name).type(doc[:type])
          type.index(doc[:id], doc[:doc])
        end
      }
    end
  end
end
{'fraction_non_alphanumeric': 0.11332007952286283, 'fraction_numerical': 0.0013253810470510272, 'mean_word_length': 2.1220268872802484, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, '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': '3365124', 'n_tokens_mistral': 940, 'n_tokens_neox': 904, 'n_words': 214}