Add a job to harvest resources from a resourcesync resource list into a stream.

Author: cbeer

app/jobs/harvest_resource_list_to_stream_job.rb

@@ -0,0 +1,79 @@
+# frozen_string_literal: true
+
+##
+# Background job to harvest a resource sync resourcelist into a stream
+class HarvestResourceListToStreamJob < ApplicationJob
+  RS_MD_HASH_KEY = 'rs_md_hash'
+  XMLNS = { sitemap: 'http://www.sitemaps.org/schemas/sitemap/0.9', rs: 'http://www.openarchives.org/rs/terms/' }.freeze
+
+  # rubocop:disable Metrics/AbcSize,Metrics/MethodLength,Metrics/CyclomaticComplexity,Metrics/PerceivedComplexity
+  def perform(stream, url:, access_token: Settings.resource_sync_harvest_token)
+    auth_headers = access_token.present? ? { 'Authorization' => "Bearer #{access_token}" } : {}
+    response = HTTP.get(url, headers: {}.merge(auth_headers))
+    Rails.logger.info("HarvestResourceListToStreamJob: Fetched resource list from #{url} with status #{response.status}")
+
+    raise "Failed to fetch resource list from #{url}: #{response.status}" unless response.status.success?
+
+    resource_list = Nokogiri::XML(response.body.to_s)
+    existing_files_hash = stream.uploads.with_attached_files.flat_map do |u|
+      ["md5:#{Base64.decode64(u.blob.checksum).unpack1('H*')}", u.blob.dig('metadata', RS_MD_HASH_KEY)]
+    end.to_set
+
+    resource_list.xpath('//sitemap:url', **XMLNS).each do |url|
+      loc = url.at_xpath('sitemap:loc', **XMLNS)&.text
+      next if loc.blank?
+
+      filename = File.basename(URI.parse(loc))
+      hash = url.at_xpath('rs:md/@hash', **XMLNS)&.text
+
+      next if existing_files_hash.include?(hash)
+
+      resource = HTTP.get(loc, headers: {}.merge(auth_headers))
+      Rails.logger.info("HarvestResourceListToStreamJob: Fetched resource from #{loc} with status #{resource.status}")
+
+      tmpfile = create_tmpfile_for_http_response(resource.body)
+
+      test_checksum(tmpfile, hash)
+
+      upload = stream.uploads.build
+      upload.files.attach(io: tmpfile, filename: filename, content_type: resource.headers['Content-Type'],
+                          metadata: { RS_MD_HASH_KEY => hash })
+
+      upload.save!
+
+      sleep 1
+    end
+    # rubocop:enable Metrics/AbcSize,Metrics/MethodLength,Metrics/CyclomaticComplexity,Metrics/PerceivedComplexity
+  end
+
+  def create_tmpfile_for_http_response(body)
+    tmpfile = Tempfile.new binmode: true
+
+    body.each do |chunk|
+      tmpfile.write(chunk)
+    end
+    tmpfile.rewind
+    tmpfile
+  end
+
+  def test_checksum(file, expected_checksum)
+    return if expected_checksum.blank?
+
+    algorithm, expected_hash = expected_checksum.split(':', 2)
+    actual_hash = case algorithm
+                  when 'md5'
+                    Digest::MD5.file(file.path).hexdigest
+                  when 'sha1'
+                    Digest::SHA1.file(file.path).hexdigest
+                  when 'sha256'
+                    Digest::SHA256.file(file.path).hexdigest
+                  else
+                    Rails.logger.warn "Unsupported checksum algorithm: #{algorithm}"
+                    return
+                  end
+
+    return if actual_hash == expected_hash
+
+    raise "Checksum mismatch: expected #{expected_checksum}, got #{algorithm}:#{actual_hash}"
+  end
+end

config/settings.yml

@@ -60,3 +60,5 @@ oai_max_page_size: 5000
 oai_repository_id: 'pod.stanford.edu'
 
 marc_record_writer_tmpdir: ~
+
+resource_sync_harvest_token: ~