<commit_before># Copyright (c) 2012 SnowPlow Analytics Ltd. All rights reserved.
#
# This program is licensed to you under the Apache License Version 2.0,
# and you may not use this file except in compliance with the Apache License Version 2.0.
# You may obtain a copy of the Apache License Version 2.0 at http://www.apache.org/licenses/LICENSE-2.0.
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the Apache License Version 2.0 is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.

# Authors::    <NAME> (mailto:<EMAIL>), <NAME>
# Copyright:: Copyright (c) 2012 SnowPlow Analytics Ltd
# License::   Apache License Version 2.0

require 'fog'
require 'thread'

module Sluice
  module Storage
    module S3

      # TODO: figure out logging instead of puts (https://github.com/snowplow/sluice/issues/2)
      # TODO: consider moving to OO structure (https://github.com/snowplow/sluice/issues/3)

      # Constants
      CONCURRENCY = 10 # Threads
      RETRIES = 3      # Attempts
      RETRY_WAIT = 10  # Seconds

      # Class to describe an S3 location
      # TODO: if we are going to impose trailing line-breaks on
      # buckets, maybe we should make that clearer?
      class Location
        attr_reader :bucket, :dir, :s3location

        # Parameters:
        # +s3location+:: the s3 location config string e.g. "bucket/directory"
        def initialize(s3_location)
          @s3_location = s3_location

          s3_location_match = s3_location.match('^s3n?://([^/]+)/?(.*)/$')
          raise ArgumentError, 'Bad S3 location %s' % s3_location unless s3_location_match

          @bucket = s3_location_match[1]
          @dir = s3_location_match[2]
        end

        def dir_as_path
          if @dir.length > 0
            return @dir+'/'
          else
            return ''
          end
        end

        def to_s
          @s3_location
        end
      end

      # Helper function to instantiate a new Fog::Storage
      # for S3 based on our config options
      #
      # Parameters:
      # +region+:: Amazon S3 region we will be working with
      # +access_key_id+:: AWS access key ID
      # +secret_access_key+:: AWS secret access key
      def new_fog_s3_from(region, access_key_id, secret_access_key)
        Fog::Storage.new({
          :provider => 'AWS',
          :region => region,
          :aws_access_key_id => access_key_id,
          :aws_secret_access_key => secret_access_key
        })
      end
      module_function :new_fog_s3_from

      # Determine if a bucket is empty
      #
      # Parameters:
      # +s3+:: A Fog::Storage s3 connection
      # +location+:: The location to check
      def is_empty?(s3, location)
        s3.directories.get(location.bucket, :prefix => location.dir).files().length > 1
      end
      module_function :is_empty?

      # Download files from an S3 location to
      # local storage, concurrently
      #
      # Parameters:
      # +s3+:: A Fog::Storage s3 connection
      # +from_location+:: S3Location to delete files from      
      # +to_directory+:: Local directory to copy files to
      # +match_regex+:: a regex string to match the files to delete
      def download_files(s3, from_location, to_directory, match_regex='.+')

        puts "  downloading files from #{from_location} to #{to_directory}"
        process_files(:download, s3, from_location, match_regex, to_directory)
      end
      module_function :download_files    

      # Delete files from S3 locations concurrently
      #
      # Parameters:
      # +s3+:: A Fog::Storage s3 connection
      # +from_location+:: S3Location to delete files from
      # +match_regex+:: a regex string to match the files to delete
      def delete_files(s3, from_location, match_regex='.+')

        puts "  deleting files from #{from_location}"
        process_files(:delete, s3, from_location, match_regex)
      end
      module_function :delete_files

      # Copies files between S3 locations concurrently
      #
      # Parameters:
      # +s3+:: A Fog::Storage s3 connection
      # +from_location+:: S3Location to copy files from
      # +to_location+:: S3Location to copy files to
      # +match_regex+:: a regex string to match the files to copy
      # +alter_filename_lambda+:: lambda to alter the written filename
      def copy_files(s3, from_location, to_location, match_regex='.+', alter_filename_lambda=false)

        puts "  copying files from #{from_location} to #{to_location}"
        process_files(:copy, s3, from_location, match_regex, to_location, alter_filename_lambda)
      end
      module_function :copy_files

      # Moves files between S3 locations concurrently
      #
      # Parameters:
      # +s3+:: A Fog::Storage s3 connection
      # +from_location+:: S3Location to move files from
      # +to+:: S3Location to move files to
      # +match_regex+:: a regex string to match the files to move
      # +alter_filename_lambda+:: lambda to alter the written filename
      def move_files(s3, from_location, to_location, match_regex='.+', alter_filename_lambda=false)

        puts "  moving files from #{from_location} to #{to_location}"
        process_files(:move, s3, from_location, match_regex, to_location, alter_filename_lambda)
      end
      module_function :move_files

      # Download a single file to the exact path specified.
      # Has no intelligence around filenaming.
      # Makes sure to create the path as needed.
      #
      # Parameters:
      # +s3+:: A Fog::Storage s3 connection
      # +from_file:: A Fog::File to download
      # +to_file:: A local file path
      def download_file(s3, from_file, to_file)

        FileUtils.mkdir_p(File.dirname(to_file))
        local_file = File.open(to_file, "w")
        local_file.write(from_file.body)
        local_file.close
      end
      module_function :download_file

      private

      # Concurrent file operations between S3 locations. Supports:
      # - Copy
      # - Delete
      # - Move (= Copy + Delete)
      #
      # Parameters:
      # +operation+:: Operation to perform. :copy, :delete, :move supported
      # +s3+:: A Fog::Storage s3 connection
      # +from_location+:: S3Location to process files from
      # +match_regex+:: a regex string to match the files to process
      # +to_loc_or_dir+:: S3Location or local directory to process files to<commit_msg>Commented out delete step - should not be firing<commit_after># Copyright (c) 2012 SnowPlow Analytics Ltd. All rights reserved.
#
# This program is licensed to you under the Apache License Version 2.0,
# and you may not use this file except in compliance with the Apache License Version 2.0.
# You may obtain a copy of the Apache License Version 2.0 at http://www.apache.org/licenses/LICENSE-2.0.
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the Apache License Version 2.0 is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.

# Authors::    <NAME> (mailto:<EMAIL>), <NAME>
# Copyright:: Copyright (c) 2012 SnowPlow Analytics Ltd
# License::   Apache License Version 2.0

require 'fog'
require 'thread'

module Sluice
  module Storage
    module S3

      # TODO: figure out logging instead of puts (https://github.com/snowplow/sluice/issues/2)
      # TODO: consider moving to OO structure (https://github.com/snowplow/sluice/issues/3)

      # Constants
      CONCURRENCY = 10 # Threads
      RETRIES = 3      # Attempts
      RETRY_WAIT = 10  # Seconds

      # Class to describe an S3 location
      # TODO: if we are going to impose trailing line-breaks on
      # buckets, maybe we should make that clearer?
      class Location
        attr_reader :bucket, :dir, :s3location

        # Parameters:
        # +s3location+:: the s3 location config string e.g. "bucket/directory"
        def initialize(s3_location)
          @s3_location = s3_location

          s3_location_match = s3_location.match('^s3n?://([^/]+)/?(.*)/$')
          raise ArgumentError, 'Bad S3 location %s' % s3_location unless s3_location_match

          @bucket = s3_location_match[1]
          @dir = s3_location_match[2]
        end

        def dir_as_path
          if @dir.length > 0
            return @dir+'/'
          else
            return ''
          end
        end

        def to_s
          @s3_location
        end
      end

      # Helper function to instantiate a new Fog::Storage
      # for S3 based on our config options
      #
      # Parameters:
      # +region+:: Amazon S3 region we will be working with
      # +access_key_id+:: AWS access key ID
      # +secret_access_key+:: AWS secret access key
      def new_fog_s3_from(region, access_key_id, secret_access_key)
        Fog::Storage.new({
          :provider => 'AWS',
          :region => region,
          :aws_access_key_id => access_key_id,
          :aws_secret_access_key => secret_access_key
        })
      end
      module_function :new_fog_s3_from

      # Determine if a bucket is empty
      #
      # Parameters:
      # +s3+:: A Fog::Storage s3 connection
      # +location+:: The location to check
      def is_empty?(s3, location)
        s3.directories.get(location.bucket, :prefix => location.dir).files().length > 1
      end
      module_function :is_empty?

      # Download files from an S3 location to
      # local storage, concurrently
      #
      # Parameters:
      # +s3+:: A Fog::Storage s3 connection
      # +from_location+:: S3Location to delete files from      
      # +to_directory+:: Local directory to copy files to
      # +match_regex+:: a regex string to match the files to delete
      def download_files(s3, from_location, to_directory, match_regex='.+')

        puts "  downloading files from #{from_location} to #{to_directory}"
        process_files(:download, s3, from_location, match_regex, to_directory)
      end
      module_function :download_files    

      # Delete files from S3 locations concurrently
      #
      # Parameters:
      # +s3+:: A Fog::Storage s3 connection
      # +from_location+:: S3Location to delete files from
      # +match_regex+:: a regex string to match the files to delete
      def delete_files(s3, from_location, match_regex='.+')

        puts "  deleting files from #{from_location}"
        process_files(:delete, s3, from_location, match_regex)
      end
      module_function :delete_files

      # Copies files between S3 locations concurrently
      #
      # Parameters:
      # +s3+:: A Fog::Storage s3 connection
      # +from_location+:: S3Location to copy files from
      # +to_location+:: S3Location to copy files to
      # +match_regex+:: a regex string to match the files to copy
      # +alter_filename_lambda+:: lambda to alter the written filename
      def copy_files(s3, from_location, to_location, match_regex='.+', alter_filename_lambda=false)

        puts "  copying files from #{from_location} to #{to_location}"
        process_files(:copy, s3, from_location, match_regex, to_location, alter_filename_lambda)
      end
      module_function :copy_files

      # Moves files between S3 locations concurrently
      #
      # Parameters:
      # +s3+:: A Fog::Storage s3 connection
      # +from_location+:: S3Location to move files from
      # +to+:: S3Location to move files to
      # +match_regex+:: a regex string to match the files to move
      # +alter_filename_lambda+:: lambda to alter the written filename
      def move_files(s3, from_location, to_location, match_regex='.+', alter_filename_lambda=false)

        puts "  moving files from #{from_location} to #{to_location}"
        process_files(:move, s3, from_location, match_regex, to_location, alter_filename_lambda)
      end
      module_function :move_files

      # Download a single file to the exact path specified.
      # Has no intelligence around filenaming.
      # Makes sure to create the path as needed.
      #
      # Parameters:
      # +s3+:: A Fog::Storage s3 connection
      # +from_file:: A Fog::File to download
      # +to_file:: A local file path
      def download_file(s3, from_file, to_file)

        puts "from_file = #{from_file}"
        puts "to_file = #{to_file}"

        FileUtils.mkdir_p(File.dirname(to_file))
        local_file = File.open(to_file, "w")
        local_file.write(from_file.body)
        local_file.close
      end
      module_function :download_file

      private

      # Concurrent file operations between S3 locations. Supports:
      # - Copy
      # - Delete
      # - Move (= Copy + Delete)
      #
      # Parameters:
      # +operation+:: Operation to perform. :copy, :delete, :move supported
      # +s3+:: A Fog::Storage s3 connection
      # +from_location+:: S3Location to process files from
      # +match_regex+:: a regex string to match the files to process
      # +to_loc_or_dir+:: S3Location or local directory to process files to
{'fraction_non_alphanumeric': 0.08834643288996373, 'fraction_numerical': 0.013905683192261185, 'mean_word_length': 2.670735090152566, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 23, 'https://': 4, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 18, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2816500', 'n_tokens_mistral': 3933, 'n_tokens_neox': 3819, 'n_words': 1286}