<commit_before>module Melinis
  class Task
    attr_reader :last_run, :failures, :logger, :current_run

    # Returns a hash with the following keys:
    #
    # * name: Name of the task - Required
    # * description: Task Description
    # * file_path: File Path
    # * command: Command that starts the execution of the task
    # * individual_retries_limit: Maximum number of times each individual failed entry must be retried
    # * bulk_retries_limit: Maximum number of consecutive failed entries that will halt the task execution
    def self.properties
      raise NotImplementedError
    end

    def self.setup(first_run_data = {})
      options = {
        :description => '',
        :file_path => '',
        :command => '',
        :individual_retries_limit => 1,
        :bulk_retries_limit => 1
      }.merge(self.properties)
      task_name = options.delete(:name)
      raise NoNameError unless task_name
      task = Melinis::TaskList.find_or_initialize_by(name: task_name)
      task.update_attributes(options)

      unless first_run_data.blank?
        current_run = Melinis::TaskProcessing.create!({
          :task_id => task.id,
          :processed_details => first_run_data.to_yaml
        })
      end
    end

    def initialize
      task_name = self.class.properties[:name]
      raise NoNameError unless task_name

      @task = Melinis::TaskList.find_by_name(task_name)
      raise SetupError unless @task

      @logger = Logger.new("log/%s.log" % [task_name.snakecase])
      @last_run = @task.task_processings.last

      # 'failures' returns the list of records that are still in failed state and have individual retries left
      @failures = @task.task_failures.to_be_processed(@task.individual_retries_limit)
    end

    def prepare
      []
    end

    def execute(unit)
      raise NotImplementedError
    end

    def execution_failure(unit)
      {}
    end

    def wrapup
      {}
    end

    def run
      success, total = 0, 0
      begin
        @current_run = Melinis::TaskProcessing.create!({:task_id => @task.id})
        logger.info { "Starting run #%d" % [@current_run.id] }
        data = prepare
        total = data.size
        data.each do |unit|
          begin
            execute(unit)
            success += 1
          rescue Exception => e
            failure(execution_failure(unit), {:exception => e})
            logger.error { e }
          end
        end
      rescue Exception => e
        failure({}, {:exception => e})
        logger.error { e }
      ensure
        success_info = { :success_count => success,
                         :total => total }
        wrapup_info = wrapup
        final_wrapup = wrapup_info.is_a?(Hash) ? wrapup_info.merge(success_info) : success_info
        @current_run.processed_details = final_wrapup.to_yaml
        @current_run.save!
      end
    end

    # * failure_details: A hash that will keep track of the records being processed
    #
    # Optional Parameters that can be passed:
    # * task_failure_id: Id of the failed record in case an earlier failed record is being processed again
    # * status: 'failure' in case a record fails either for the first time or in subsequent attempts.
    #            Or 'success' in case an earlier failed record succeeds in the next attempt.
    def failure(failure_details, args = {})<commit_msg>Added logic to prevent creation of multiple melinis_task_processings if we are running the task again and again for same object<commit_after>module Melinis
  class Task
    attr_reader :last_run, :failures, :logger, :current_run

    # Returns a hash with the following keys:
    #
    # * name: Name of the task - Required
    # * description: Task Description
    # * file_path: File Path
    # * command: Command that starts the execution of the task
    # * individual_retries_limit: Maximum number of times each individual failed entry must be retried
    # * bulk_retries_limit: Maximum number of consecutive failed entries that will halt the task execution
    def self.properties
      raise NotImplementedError
    end

    def self.setup(first_run_data = {})
      options = {
        :description => '',
        :file_path => '',
        :command => '',
        :individual_retries_limit => 1,
        :bulk_retries_limit => 1
      }.merge(self.properties)
      task_name = options.delete(:name)
      raise NoNameError unless task_name
      task = Melinis::TaskList.find_or_initialize_by(name: task_name)
      task.update_attributes(options)

      unless first_run_data.blank?
        current_run = Melinis::TaskProcessing.create!({
          :task_id => task.id,
          :processed_details => first_run_data.to_yaml
        })
      end
    end

    def initialize
      task_name = self.class.properties[:name]
      raise NoNameError unless task_name

      @task = Melinis::TaskList.find_by_name(task_name)
      raise SetupError unless @task

      @logger = Logger.new("log/%s.log" % [task_name.snakecase])
      @last_run = @task.task_processings.last

      # 'failures' returns the list of records that are still in failed state and have individual retries left
      @failures = @task.task_failures.to_be_processed(@task.individual_retries_limit)

      # 'For multiple operation on any individual object,
      # if we want just a single TaskProcessing entry, we need to populate it in overwrite
      @melinis_task_processing_id = nil
    end

    def prepare
      []
    end

    def execute(unit)
      raise NotImplementedError
    end

    def execution_failure(unit)
      {}
    end

    def wrapup
      {}
    end

    def run
      success, total = 0, 0
      begin
        @current_run = Melinis::TaskProcessing.find_by(id: @melinis_task_processing_id)
        @current_run ||= Melinis::TaskProcessing.create!({ task_id: @task.id })

        logger.info { "Starting run #%d" % [@current_run.id] }
        data = prepare
        total = data.size
        data.each do |unit|
          begin
            execute(unit)
            success += 1
          rescue Exception => e
            failure(execution_failure(unit), { exception: e})
            logger.error { e }
          end
        end
      rescue Exception => e
        failure(execution_failure(nil), { exception: e})
        logger.error { e }
      ensure
        success_info = { success_count: success,
                         total: total }
        wrapup_info = wrapup
        final_wrapup = wrapup_info.is_a?(Hash) ? wrapup_info.merge(success_info) : success_info
        @current_run.processed_details = final_wrapup.to_yaml
        @current_run.save!
      end
    end

    # * failure_details: A hash that will keep track of the records being processed
    #
    # Optional Parameters that can be passed:
    # * task_failure_id: Id of the failed record in case an earlier failed record is being processed again
    # * status: 'failure' in case a record fails either for the first time or in subsequent attempts.
    #            Or 'success' in case an earlier failed record succeeds in the next attempt.
    def failure(failure_details, args = {})
{'fraction_non_alphanumeric': 0.08089887640449438, 'fraction_numerical': 0.0014044943820224719, 'mean_word_length': 2.4584749878581835, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 26, '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': '2314690', 'n_tokens_mistral': 2026, 'n_tokens_neox': 1962, 'n_words': 603}