<commit_before># -*- coding: utf-8 -*-
class Fluent::SendmailInput < Fluent::TailInput
  Fluent::Plugin.register_input("sendmail", self)

  config_param :lrucache_size, :integer, :default => (1024*1024)
  # sendmail default value of queuereturn is 5d (432000sec)
  config_param :queuereturn, :time, :default => 432000
  config_param :path_cache_file, :string, :default => nil

  require_relative "sendmailparser"
  require "lru_redux"

  def configure(conf)
    super
    @delivers = LruRedux::ThreadSafeCache.new(@lrucache_size)
    if @path_cache_file != nil
      if not File.exists?(@path_cache_file)
        File.open(@path_cache_file, "w+"){|cache_file|
          cache_file.puts('{}')
        }
      end
      if not File.readable?(@path_cache_file)
        raise ConfigError, "cache file exists but not readable."
      end
      if not File.writable?(@path_cache_file)
        raise Fluent::ConfigError, "cache file not writable."
      end
      File.open(@path_cache_file, "r") {|cache_file|
        line = cache_file.read()
        data = JSON.parse(line)
        data.each{|k, v|
          @delivers[k] = SendmailLog.new(v['time'], v['from_line'], v['nrcpts'])
        }
      }
    end
  end

  def configure_parser(conf)
    @parser = SendmailParser.new(conf)
  end

  def shutdown
    super
    if @path_cache_file != nil
      data = {}
      if @path_cache_file != nil
        @delivers.each{|k, v|
          data[k] = v.to_json
        }
      end
      File.open(@path_cache_file, "w+") {|cache_file|
        cache_file.puts(data)
      }
    end
  end

  def receive_lines(lines)
    es = Fluent::MultiEventStream.new
    lines.each {|line|
      begin
        line.chomp!  # remove \n
        record = parse_line(line)
        if record.nil?
          next
        end

        type   = record["type"]
        mta    = record["mta"]
        qid    = record["qid"]
        time   = record["time"]<commit_msg>fix bug to store cache in json.<commit_after># -*- coding: utf-8 -*-
class Fluent::SendmailInput < Fluent::TailInput
  Fluent::Plugin.register_input("sendmail", self)

  config_param :lrucache_size, :integer, :default => (1024*1024)
  # sendmail default value of queuereturn is 5d (432000sec)
  config_param :queuereturn, :time, :default => 432000
  config_param :path_cache_file, :string, :default => nil

  require_relative "sendmailparser"
  require "lru_redux"
  require "json"

  def configure(conf)
    super
    @delivers = LruRedux::ThreadSafeCache.new(@lrucache_size)
    if @path_cache_file != nil
      if not File.exists?(@path_cache_file)
        File.open(@path_cache_file, "w+"){|cache_file|
          cache_file.puts('{}')
        }
      end
      if not File.readable?(@path_cache_file)
        raise ConfigError, "cache file exists but not readable."
      end
      if not File.writable?(@path_cache_file)
        raise Fluent::ConfigError, "cache file not writable."
      end
      File.open(@path_cache_file, "r") {|cache_file|
        line = cache_file.read()
        data = JSON.parse(line)
        data.each{|k, v|
          @delivers[k] = SendmailLog.new(v['time'], v['from_line'], v['nrcpts'])
        }
      }
    end
  end

  def configure_parser(conf)
    @parser = SendmailParser.new(conf)
  end

  def shutdown
    super
    if @path_cache_file != nil
      data = {}
      @delivers.each{|k, v|
        data[k] = v.to_json
      }
      File.open(@path_cache_file, "w+") {|cache_file|
        cache_file.puts(data.to_json)
      }
    end
  end

  def receive_lines(lines)
    es = Fluent::MultiEventStream.new
    lines.each {|line|
      begin
        line.chomp!  # remove \n
        record = parse_line(line)
        if record.nil?
          next
        end

        type   = record["type"]
        mta    = record["mta"]
        qid    = record["qid"]
        time   = record["time"]
{'fraction_non_alphanumeric': 0.12480539699014011, 'fraction_numerical': 0.011416709911779968, 'mean_word_length': 2.4918478260869565, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 9, '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': '2209008', 'n_tokens_mistral': 1347, 'n_tokens_neox': 1284, 'n_words': 277}