lkqd.rb in embulk-input-lkqd-0.5.0

- old
+ new

@@ -14,10 +14,11 @@
       def self.transaction(config, &control)
         task = {
           "secret_key_id" => config.param("secret_key_id", :string),
           "secret_key" => config.param("secret_key", :string),
           "endpoint" => config.param("endpoint", :string, default: 'https://api.lkqd.com/reports'),
+          "try_convert" => config.param("try_convert", :string, default: "true"),
           "report_parameters" => config.param("report_parameters", :hash, default: {}),
         }
         task['authorization'] = Base64.urlsafe_encode64("#{task['secret_key_id']}:#{task['secret_key']}")
 
         response = request_lkqd({authorization: task['authorization'], endpoint: task['endpoint'], report_parameters: task['report_parameters']})
@@ -60,19 +61,19 @@
         else
           return {type: :string, name: column_name}
         end
       end
 
-      def self.try_convert(row)
+      def self.try_convert(row, options={})
         return row.map do |field|
           name, value = field
           column_name = name.gsub(/^\W/, '')
           column_option = DEFAULT_COLUMNS[column_name]
           if column_option.nil?
             next value
           elsif column_option['type'] == 'timestamp'
-            next Time.strptime(value, column_option['format']).to_i
+            next Time.strptime(value + " " + options[:timezone], column_option['format'] + " %Z").to_i
           elsif column_option['type'] == 'long'
             next value.gsub(',','').to_i
           elsif column_option['type'] == 'double' && value.match(/%$/)
             next value.gsub(',','').to_f / 100.0
           elsif column_option['type'] == 'double'
@@ -86,11 +87,13 @@
       def init
         @task = task
       end
 
       def run
+        convert_options = {timezone: @task['report_parameters']['timezone']}
         CSV.foreach(@task['tempfile_path'], {headers: true}).each do |row|
-          page_builder.add(Lkqd.try_convert(row))
+          row = @task['try_convert'] == "true" ? Lkqd.try_convert(row, convert_options) : row
+          page_builder.add(row)
         end
         page_builder.finish
         FileUtils.rm_rf(@task['tempfile_path'])
 
         task_report = {}