<commit_before>require 'avro2kafka/version'
require 'avro2kafka/avro_reader'
require 'avro2kafka/kafka_publisher'

require 'logr'

class Avro2Kafka
  attr_reader :options

  def self.logger
    @logger ||= Logr::Logger.new('avro2kafka')
  end

  def initialize(options)
    @input_path = ARGV.first
    @options = options
  end

  def reader
    ARGF.tap { |argf| argf.rewind }
  end

  def publish
    Avro2Kafka.logger.event('started_publishing', filename: input_filename, topic: topic, data: extra_data)
                     .monitored("Avro2Kafka started publishing #{input_filename}", '')
                     .info("Started publishing #{filename}")

    records = AvroReader.new(reader).read
    KafkaPublisher.new(**kafka_options).publish(records)

    Avro2Kafka.logger.event('finished_publishing', filename: input_filename, topic: topic, data: extra_data)
                     .monitored("Avro2Kafka finished publishing #{input_filename}", '')
                     .metric('lines_processed', records.count)
                     .info("Finished publishing #{filename}")
  end

  def input_filename
    File.basename(@input_path)
  end
<commit_msg>Rename to path and filename<commit_after>require 'avro2kafka/version'
require 'avro2kafka/avro_reader'
require 'avro2kafka/kafka_publisher'

require 'logr'

class Avro2Kafka
  attr_reader :options

  def self.logger
    @logger ||= Logr::Logger.new('avro2kafka')
  end

  def initialize(options)
    @path = ARGV.first
    @options = options
  end

  def reader
    ARGF.tap { |argf| argf.rewind }
  end

  def publish
    Avro2Kafka.logger.event('started_publishing', filename: input_filename, topic: topic, data: extra_data)
                     .monitored("Started publishing #{filename}", "Started publishing #{filename} to the #{topic} Kafka topic.")
                     .info("Started publishing #{filename}")

    records = AvroReader.new(reader).read
    KafkaPublisher.new(**kafka_options).publish(records)

    Avro2Kafka.logger.event('finished_publishing', filename: input_filename, topic: topic, data: extra_data)
                     .monitored("Finished publishing #{filename}", "Finished publishing #{filename} to the #{topic} Kafka topic.")
                     .metric('lines_processed', records.count)
                     .info("Finished publishing #{filename}")
  end

  def filename
    File.basename(@path)
  end

{'fraction_non_alphanumeric': 0.11125052279381012, 'fraction_numerical': 0.006691760769552488, 'mean_word_length': 3.2336283185840706, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '2593909', 'n_tokens_mistral': 750, 'n_tokens_neox': 711, 'n_words': 158}