convert affiliations in activities. #324

app/jobs/activity_convert_affiliation_by_id_job.rb

@@ -0,0 +1,12 @@
+class ActivityConvertAffiliationByIdJob < ActiveJob::Base
+  queue_as :lupo_background
+
+  rescue_from ActiveJob::DeserializationError, Elasticsearch::Transport::Transport::Errors::BadRequest do |error|
+    logger = Logger.new(STDOUT)
+    logger.error error.message
+  end
+
+  def perform(options={})
+    Activity.convert_affiliation_by_id(options)
+  end
+end

app/jobs/activity_import_by_id_job.rb

@@ -1,7 +1,12 @@
 class ActivityImportByIdJob < ActiveJob::Base
   queue_as :lupo_background
 
+  rescue_from ActiveJob::DeserializationError, Elasticsearch::Transport::Transport::Errors::BadRequest do |error|
+    logger = Logger.new(STDOUT)
+    logger.error error.message
+  end
+
   def perform(options={})
     Activity.import_by_id(options)
   end
-end
+end

app/models/activity.rb

@@ -250,6 +250,82 @@ def self.import_by_id(options={})
     count
   end
 
+  def self.convert_affiliations(options={})
+    from_id = (options[:from_id] || Doi.minimum(:id)).to_i
+    until_id = (options[:until_id] || Doi.maximum(:id)).to_i
+
+    # get every id between from_id and end_id
+    (from_id..until_id).step(500).each do |id|
+      ActivityConvertAffiliationByIdJob.perform_later(options.merge(id: id))
+      puts "Queued converting affiliations for activities with IDs starting with #{id}." unless Rails.env.test?
+    end
+
+    (from_id..until_id).to_a.length
+  end
+
+  def self.convert_affiliation_by_id(options={})
+    return nil unless options[:id].present?
+
+    id = options[:id].to_i
+    count = 0
+
+    logger = Logger.new(STDOUT)
+
+    Activity.where(id: id..(id + 499)).find_each do |activity|
+      should_update = false
+      audited_changes = activity.audited_changes
+      creators = Array.wrap(audited_changes["creators"]).map do |c|
+        # c is an array if there are changes
+        return [] if c.blank?
+        c = c.last if c.is_a?(Array)
+
+        if c["affiliation"].nil?
+          c["affiliation"] = []
+          should_update = true
+        elsif c["affiliation"].is_a?(String)
+          c["affiliation"] = [{ "name" => c["affiliation"] }] 
+          should_update = true
+        else c["affiliation"].is_a?(Hash)
+          c["affiliation"] = Array.wrap(c["affiliation"])
+          should_update = true
+        end
+
+        c
+      end
+      contributors = Array.wrap(audited_changes["contributors"]).map do |c|
+        # c is an array if there are changes
+        return [] if c.blank?
+        c = c.last if c.is_a?(Array)
+
+        if c["affiliation"].nil?
+          c["affiliation"] = []
+          should_update = true
+        elsif c["affiliation"].is_a?(String)
+          c["affiliation"] = [{ "name" => c["affiliation"] }] 
+          should_update = true
+        else c["affiliation"].is_a?(Hash)
+          c["affiliation"] = Array.wrap(c["affiliation"])
+          should_update = true
+        end
+
+        c
+      end
+
+      if should_update
+        audited_changes["creators"] = creators
+        audited_changes["contributors"] = contributors
+        activity.update_attributes(audited_changes: audited_changes)
+        count += 1
+      end
+    end
+
+    logger.info "[Elasticsearch] Converted affiliations for #{count} activities with IDs #{id} - #{(id + 499)}." if count > 0
+
+    count
+  rescue Elasticsearch::Transport::Transport::Errors::RequestEntityTooLarge, Faraday::ConnectionFailed, ActiveRecord::LockWaitTimeout => error
+    logger.info "[Elasticsearch] Error #{error.message} converting affiliations for DOIs with IDs #{id} - #{(id + 499)}."
+  end
+
   def uid
     doi.present? ? doi.uid : changes.to_h['doi']
   end

app/models/concerns/indexable.rb

@@ -396,9 +396,13 @@ def upgrade_index
       inactive_index ||= self.inactive_index
 
       self.__elasticsearch__.delete_index!(index: inactive_index) if self.__elasticsearch__.index_exists?(index: inactive_index)
-      self.__elasticsearch__.create_index!(index: inactive_index)
 
-      "Upgraded inactive index #{inactive_index}."
+      if self.__elasticsearch__.index_exists?(index: inactive_index)
+        "Error: inactive index #{inactive_index} could not be upgraded."
+      else
+        self.__elasticsearch__.create_index!(index: inactive_index)
+        "Upgraded inactive index #{inactive_index}."
+      end
     end
 
     # show stats for both indexes

app/models/doi.rb

@@ -584,6 +584,7 @@ def self.convert_affiliation_by_id(options={})
       creators = Array.wrap(doi.creators).map do |c|
         if c["affiliation"].nil?
           c["affiliation"] = []
+          should_update = true
         elsif c["affiliation"].is_a?(String)
           c["affiliation"] = [{ "name" => c["affiliation"] }] 
           should_update = true
@@ -597,6 +598,7 @@ def self.convert_affiliation_by_id(options={})
       contributors = Array.wrap(doi.contributors).map do |c|
         if c["affiliation"].nil?
           c["affiliation"] = []
+          should_update = true
         elsif c["affiliation"].is_a?(String)
           c["affiliation"] = [{ "name" => c["affiliation"] }] 
           should_update = true

lib/tasks/activity.rake

@@ -51,4 +51,12 @@ namespace :activity do
 
     Activity.import_by_ids(from_id: from_id, until_id: until_id)
   end
+
+  desc 'Convert affiliations to new format'
+  task :convert_affiliations => :environment do
+    from_id = (ENV['FROM_ID'] || Doi.minimum(:id)).to_i
+    until_id = (ENV['UNTIL_ID'] || Doi.maximum(:id)).to_i
+
+    Activity.convert_affiliations(from_id: from_id, until_id: until_id)
+  end
 end