<commit_before># vim:fileencoding=utf-8

require 'facter'
require 'fileutils'
require 'uri'

module DBEE
  module Job
    class Download
      extend Job

      attr_accessor :request
      @queue = :all_worker

      def self.queue_prefix
        ""
      end

      def self.perform(request_id, running_job, args)
        request = Request.new(request_id)
        worker = Facter.value(:fqdn)
        request.start_job(:worker => worker, :running_job => running_job)

        request_data = request.get.body

        # URLからダウンロードして保存する
        download_dir = "#{DBEE::Config::Encode::OUTPUT_DIR}/download"
        filename = request_data["program"]["filename"]
        url = URI.encode("#{args["base_url"]}#{filename}")
        FileUtils.mkdir_p(download_dir) unless File.exists?(download_dir)
        download_file = "#{download_dir}/#{filename}"

        puts "start downloading metadata from #{url}.json.."
        # まずメタデータを取得する (ファイルには保存せずメモリ上へ展開)
        response = request.http.get("#{url}.json")
        if response.status != 200
          request_data["running_job"] = nil
          request.put(request_data)
          raise "failed to download from #{url}.json. got #{response.status}"
        end
        metadata = JSON.parse(response.content)

        puts "start downloading material from #{url} to #{download_file}...."
        f = File.open(download_file, "wb")
        response = request.http.get(url) do |data|
          f << data
        end
        f.close

        if response.status != 200
          File.unlink(download_file)
          request_data["running_job"] = nil
          request.put(request_data)
          raise "failed to download from #{url}. got #{response.status}"
        end

        puts "Calculating SHA256 for #{download_file}...."
        digest = Digest::SHA256.new
        File.open(download_file, 'r') do |f|
          buf = String.new
          while f.read(1024 * 8, buf)
            digest << buf
          end
        end

        # SHA256でダウンロードした素材を確かめる
        if metadata["SHA256"] != digest.hexdigest
          File.unlink(download_file)
          request_data["running_job"] = nil
          request.put(request_data)
          raise "downloaded file #{download_file} does not match SHA256 checksums." +
                "expected: #{metadata["SHA256"]}, got #{digest.hexdigest}"
        end

        puts "....finished"

        request_data["run_list"][0]["output"]["file"] = download_file
        request_data["run_list"][0]["output"]["worker"] = worker

        # 次のジョブも同一ノードで実行してほしい
        request_data["run_list"][0]["output"]["next_same_node"] = true
        request.put(request_data)

        # 最後にrunning_jobをDELETEしてジョブの正常終了を通知
        request.delete("running_job")
        puts "Downloading job for request##{request.request_id} sucessfully finished."
      end
    end
  end
end<commit_msg>download: get refactored<commit_after># vim:fileencoding=utf-8

require 'facter'
require 'fileutils'
require 'uri'

module DBEE
  module Job
    class Download
      extend Job

      attr_accessor :request
      @queue = :all_worker

      def self.queue_prefix
        ""
      end

      def self.perform(request_id, running_job, args)
        request = Request.new(request_id)
        worker = Facter.value(:fqdn)
        request.start_job(:worker => worker, :running_job => running_job)

        request_data = request.get.body

        # URLからダウンロードして保存する
        download_dir = "#{DBEE::Config::Encode::OUTPUT_DIR}/download"
        filename = request_data["program"]["filename"]
        url = URI.encode("#{args["base_url"]}#{filename}")
        FileUtils.mkdir_p(download_dir) unless File.exists?(download_dir)
        download_file = "#{download_dir}/#{filename}"

        puts "start downloading metadata from #{url}.json.."
        # まずメタデータを取得する (ファイルには保存せずメモリ上へ展開)
        response = request.http.get("#{url}.json")
        if response.status != 200
          request_data["running_job"] = nil
          request.put(request_data)
          raise "failed to download from #{url}.json. got #{response.status}"
        end
        metadata = JSON.parse(response.content)

        # 終了処理
        closer = proc {
          puts "....finished"

          request_data["run_list"][0]["output"]["file"] = download_file
          request_data["run_list"][0]["output"]["worker"] = worker

          # 次のジョブも同一ノードで実行してほしい
          request_data["run_list"][0]["output"]["next_same_node"] = true
          request.put(request_data)

          # 最後にrunning_jobをDELETEしてジョブの正常終了を通知
          request.delete("running_job")
          puts "Downloading job for request##{request.request_id} sucessfully finished."
        }

        # material_node == workerなら同一マシンなのでダウンロードしない
        if request_data["material_node"] == worker
          closer.call
        end

        # ファイルが存在していて、かつハッシュ値が同じならダウンロードしない
        if File.exist?(download_file)
          digest = calc_digest(download_file)
          if metadata["SHA256"] == digest.hexdigest
            closer.call
          end
        end

        puts "start downloading material from #{url} to #{download_file}...."
        f = File.open(download_file, "wb")
        response = request.http.get(url) do |data|
          f << data
        end
        f.close

        if response.status != 200
          File.unlink(download_file)
          request_data["running_job"] = nil
          request.put(request_data)
          raise "failed to download from #{url}. got #{response.status}"
        end

        # SHA256でダウンロードした素材を確かめる
        if metadata["SHA256"] != digest.hexdigest
          File.unlink(download_file)
          request_data["running_job"] = nil
          request.put(request_data)
          raise "downloaded file #{download_file} does not match SHA256 checksums." +
                "expected: #{metadata["SHA256"]}, got #{digest.hexdigest}"
        end
        closer.call
      end

      def calc_digest(download_file)
        puts "Calculating SHA256 for #{download_file}...."
        digest = Digest::SHA256.new
        File.open(download_file, 'r') do |f|
          buf = String.new
          while f.read(1024 * 8, buf)
            digest << buf
          end
        end
        digest
      end
    end
  end
end
{'fraction_non_alphanumeric': 0.116, 'fraction_numerical': 0.01104, 'mean_word_length': 2.3004223864836324, 'pattern_counts': {'":': 0, '<': 11, '<?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': 46, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '481634', 'n_tokens_mistral': 2026, 'n_tokens_neox': 1957, 'n_words': 381}