<commit_before># encoding: utf-8

require_relative './exceptions'

module CartoDB
  module Importer2
    class Connector
      # Requirements:
      #   * odbc_fdw extension must be installed in the user database

      class ConnectorError < StandardError
        attr_reader :user_name

        def initialize(message = 'General error', user = nil)
          @user_name = user && user.username
          message = message.to_s
          message << " User: #{@user_name}" if @user_name
          super(message)
        end

        def error_code
          ERRORS_MAP.fetch(self.class, UNKNOWN_ERROR_CODE)
        end
      end

      class InvalidParametersError < ConnectorError
      end

      class ConnectorsDisabledError < ConnectorError # ServiceDisabledError ?
        def initialize(message = 'CARTO-Connectors disabled', user = nil)
          super message: message, user: user
        end
      end

      attr_reader :results, :log, :job
      attr_accessor :stats

      def initialize(connector_source, options = {})
        @pg_options = options[:pg]
        @log        = options[:log] || new_logger
        @job        = options[:job] || new_job(@log, @pg_options)
        @user       = options[:user]

        @id = @job.id
        @unique_suffix = @id.delete('-')
        @json_params = connector_source
        extract_params
        validate_params!
        @schema = @user.database_schema
        @results = []
        @tracker = nil
        @stats = {}
      end

      def run(tracker = nil)
        @tracker = tracker
        @job.log "Connector #{@json_params}"
        # TODO: deal with schemas, org users, etc.,
        # TODO: logging with CartoDB::Logger
        table_name = @job.table_name
        qualified_table_name = %{"#{@job.schema}"."#{table_name}"}
        @job.log "Creating Server"
        execute_as_superuser create_server_command
        @job.log "Creating Foreign Table"
        execute_as_superuser create_foreign_table_command
        @job.log "Copying Foreign Table"
        execute "CREATE TABLE #{qualified_table_name} AS SELECT * FROM #{qualified_foreign_table_name};"
      rescue => error
        @job.log "Connector Error #{error}"
        @results.push result_for(@job.schema, table_name, error)
      else
        @job.log "Connector created table #{table_name}"
        @job.log "job schema: #{@job.schema}"
        @results.push result_for(@job.schema, table_name)
      ensure
        @log.append_and_store "Connector cleanup"
        execute_as_superuser drop_foreign_table_command
        execute_as_superuser drop_server_command
        @log.append_and_store "Connector cleaned-up"
      end

      def success?
        results.count(&:success?) > 0
      end

      def remote_data_updated?
        # TODO: can we detect if query results have changed?
        true
      end

      def tracker
        @tracker || lambda { |state| state }
      end

      def visualizations
        []
      end

      def warnings
        []
      end

      def self.check_availability!(user)
        unless user.has_feature_flag?('carto-connectors')
          raise ConnectorsDisabledError.new(user: user.username)
        end
      end

      def etag
      end

      def checksum
      end

      def last_modified
      end

      private

      ACCEPTED_PARAMETERS = %w(connection schema table sql_query sql_count encoding columns).freeze
      REQUIRED_PARAMETERS = %w(connection table).freeze

      SERVER_OPTIONS = %w(dsn driver host server address port database).freeze
      USER_OPTIONS   = %w(uid pwd user username password).freeze

      MAX_PG_IDENTIFIER_LEN = 60
      MIN_TAB_ID_LEN        = 10

      def fetch_ignoring_case(hash, key)
        if hash
          _k, v = hash.find { |k, _v| k.to_s.casecmp(key.to_s) == 0 }
          v
        end
      end

      def connection_options(options)
        # Prefix option names with "odbc_"
        Hash[options.map { |option_name, option_value| ["odbc_#{option_name}", option_value] }]
      end

      def server_params<commit_msg>Comments Connector empty methods needed for Runner compatibility<commit_after># encoding: utf-8

require_relative './exceptions'

module CartoDB
  module Importer2
    class Connector
      # Requirements:
      #   * odbc_fdw extension must be installed in the user database

      class ConnectorError < StandardError
        attr_reader :user_name

        def initialize(message = 'General error', user = nil)
          @user_name = user && user.username
          message = message.to_s
          message << " User: #{@user_name}" if @user_name
          super(message)
        end

        def error_code
          ERRORS_MAP.fetch(self.class, UNKNOWN_ERROR_CODE)
        end
      end

      class InvalidParametersError < ConnectorError
      end

      class ConnectorsDisabledError < ConnectorError # ServiceDisabledError ?
        def initialize(message = 'CARTO-Connectors disabled', user = nil)
          super message: message, user: user
        end
      end

      attr_reader :results, :log, :job
      attr_accessor :stats

      def initialize(connector_source, options = {})
        @pg_options = options[:pg]
        @log        = options[:log] || new_logger
        @job        = options[:job] || new_job(@log, @pg_options)
        @user       = options[:user]

        @id = @job.id
        @unique_suffix = @id.delete('-')
        @json_params = connector_source
        extract_params
        validate_params!
        @schema = @user.database_schema
        @results = []
        @tracker = nil
        @stats = {}
      end

      def run(tracker = nil)
        @tracker = tracker
        @job.log "Connector #{@json_params}"
        # TODO: deal with schemas, org users, etc.,
        # TODO: logging with CartoDB::Logger
        table_name = @job.table_name
        qualified_table_name = %{"#{@job.schema}"."#{table_name}"}
        @job.log "Creating Server"
        execute_as_superuser create_server_command
        @job.log "Creating Foreign Table"
        execute_as_superuser create_foreign_table_command
        @job.log "Copying Foreign Table"
        execute "CREATE TABLE #{qualified_table_name} AS SELECT * FROM #{qualified_foreign_table_name};"
      rescue => error
        @job.log "Connector Error #{error}"
        @results.push result_for(@job.schema, table_name, error)
      else
        @job.log "Connector created table #{table_name}"
        @job.log "job schema: #{@job.schema}"
        @results.push result_for(@job.schema, table_name)
      ensure
        @log.append_and_store "Connector cleanup"
        execute_as_superuser drop_foreign_table_command
        execute_as_superuser drop_server_command
        @log.append_and_store "Connector cleaned-up"
      end

      def success?
        results.count(&:success?) > 0
      end

      def remote_data_updated?
        # TODO: can we detect if query results have changed?
        true
      end

      def tracker
        @tracker || lambda { |state| state }
      end

      def visualizations
        # This method is needed to make the interface of Connector compatible with Runner
        []
      end

      def warnings
        # This method is needed to make the interface of Connector compatible with Runner
        []
      end

      def self.check_availability!(user)
        unless user.has_feature_flag?('carto-connectors')
          raise ConnectorsDisabledError.new(user: user.username)
        end
      end

      def etag
        # This method is needed to make the interface of Connector compatible with Runner,
        # but we have no meningful data to return here.
      end

      def checksum
        # This method is needed to make the interface of Connector compatible with Runner,
        # but we have no meningful data to return here.
      end

      def last_modified
        # This method is needed to make the interface of Connector compatible with Runner,
        # but we have no meningful data to return here.
      end

      private

      ACCEPTED_PARAMETERS = %w(connection schema table sql_query sql_count encoding columns).freeze
      REQUIRED_PARAMETERS = %w(connection table).freeze

      SERVER_OPTIONS = %w(dsn driver host server address port database).freeze
      USER_OPTIONS   = %w(uid pwd user username password).freeze

      MAX_PG_IDENTIFIER_LEN = 60
      MIN_TAB_ID_LEN        = 10

      def fetch_ignoring_case(hash, key)
        if hash
          _k, v = hash.find { |k, _v| k.to_s.casecmp(key.to_s) == 0 }
          v
        end
      end

      def connection_options(options)
        # Prefix option names with "odbc_"
        Hash[options.map { |option_name, option_value| ["odbc_#{option_name}", option_value] }]
      end

      def server_params
{'fraction_non_alphanumeric': 0.08092090266697059, 'fraction_numerical': 0.0018235696375655346, 'mean_word_length': 2.248796741947427, 'pattern_counts': {'":': 0, '<': 13, '<?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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2879365', 'n_tokens_mistral': 2539, 'n_tokens_neox': 2456, 'n_words': 690}