252 行
8.3 KiB
Ruby
252 行
8.3 KiB
Ruby
class MaterialSyncRunner
|
|
Result = Struct.new(:imported, :updated, :unchanged, :suppressed, :failed,
|
|
:errors, keyword_init: true)
|
|
|
|
def self.sync_enabled!
|
|
results = MaterialSyncSource.enabled.order(:id).map { |source| new(source).sync! }
|
|
|
|
Result.new(imported: results.sum(&:imported),
|
|
updated: results.sum(&:updated),
|
|
unchanged: results.sum(&:unchanged),
|
|
suppressed: results.sum(&:suppressed),
|
|
failed: results.sum(&:failed),
|
|
errors: results.flat_map(&:errors))
|
|
end
|
|
|
|
def initialize source
|
|
@source = source
|
|
end
|
|
|
|
def sync!
|
|
result = Result.new(imported: 0, updated: 0, unchanged: 0,
|
|
suppressed: 0, failed: 0, errors: [])
|
|
|
|
if @source.source_kind == 'google_drive_path'
|
|
sync_google_drive_path!(result)
|
|
else
|
|
candidates.each do |candidate|
|
|
next if candidate.blank?
|
|
|
|
sync_candidate!(candidate, result)
|
|
end
|
|
end
|
|
|
|
@source.update!(last_synced_at: Time.current)
|
|
log_result(result)
|
|
result
|
|
rescue NotImplementedError, StandardError => e
|
|
result.failed += 1
|
|
result.errors << { source_id: @source.id, error: e.message }
|
|
log_result(result)
|
|
result
|
|
end
|
|
|
|
private
|
|
|
|
def candidates
|
|
case @source.source_kind
|
|
when 'uri'
|
|
[uri_candidate]
|
|
when 'google_drive_file'
|
|
[google_drive_file_candidate]
|
|
when 'legacy_drive_path'
|
|
raise NotImplementedError, 'legacy_drive_path material sync is not implemented'
|
|
else
|
|
raise NotImplementedError, "Unsupported material sync source_kind: #{ @source.source_kind }"
|
|
end
|
|
end
|
|
|
|
def uri_candidate
|
|
{ source_kind: @source.source_kind,
|
|
source_uri: @source.source_uri,
|
|
source_path: @source.source_path,
|
|
source_file_id: @source.source_file_id,
|
|
url: @source.source_uri,
|
|
tag: nil,
|
|
created_by_user: @source.created_by_user,
|
|
updated_by_user: @source.updated_by_user || @source.created_by_user }
|
|
end
|
|
|
|
def sync_candidate! candidate, result
|
|
block = MaterialImportBlockMatcher.match_for_sha256(candidate[:file_sha256])
|
|
if block
|
|
result.suppressed += 1
|
|
Rails.logger.info(
|
|
material_sync_log(action: 'suppressed',
|
|
reason: block.reason,
|
|
normalized_source_key: candidate_normalized_source_key(candidate)))
|
|
return
|
|
end
|
|
|
|
suppression = suppression_for(candidate)
|
|
|
|
if suppression
|
|
result.suppressed += 1
|
|
Rails.logger.info(
|
|
material_sync_log(action: 'suppressed',
|
|
normalized_source_key: suppression.normalized_source_key))
|
|
return
|
|
end
|
|
|
|
import = MaterialSyncImporter.import!(candidate)
|
|
result.public_send("#{ import.action }=", result.public_send(import.action) + 1)
|
|
rescue StandardError => e
|
|
result.failed += 1
|
|
result.errors << { source_id: @source.id,
|
|
normalized_source_key: candidate_normalized_source_key(candidate),
|
|
error: e.message }
|
|
Rails.logger.warn(
|
|
material_sync_log(action: 'failed',
|
|
error: e.message,
|
|
normalized_source_key: candidate_normalized_source_key(candidate)))
|
|
ensure
|
|
close_candidate_tempfile(candidate)
|
|
end
|
|
|
|
def google_drive_path_candidates
|
|
folder_id = google_drive_folder_id
|
|
Enumerator.new do |entries|
|
|
drive_client.each_material_file_under_folder(folder_id).each do |entry|
|
|
entries << build_google_drive_candidate(entry)
|
|
end
|
|
end
|
|
end
|
|
|
|
def sync_google_drive_path! result
|
|
folder_id = google_drive_folder_id
|
|
scanned_count = 0
|
|
|
|
drive_client.each_material_file_under_folder(folder_id) do |entry|
|
|
scanned_count += 1
|
|
sync_candidate!(build_google_drive_candidate(entry), result)
|
|
log_google_drive_progress(folder_id, scanned_count, result) if progress_log_scan_count?(scanned_count)
|
|
end
|
|
|
|
log_google_drive_progress(folder_id, scanned_count, result, summary: true)
|
|
end
|
|
|
|
def google_drive_file_candidate
|
|
entry = drive_client.fetch_material_file(google_drive_file_id)
|
|
return nil unless entry
|
|
|
|
build_google_drive_candidate(entry, single_file: true)
|
|
end
|
|
|
|
def build_google_drive_candidate entry, single_file: false
|
|
relative_path =
|
|
if single_file
|
|
nil
|
|
else
|
|
entry[:relative_path]
|
|
end
|
|
export_path =
|
|
MaterialSyncExportPath.build(prefix: @source.export_path_prefix,
|
|
relative_path:,
|
|
filename: entry[:name],
|
|
source_file_id: entry[:id])
|
|
|
|
{ source_kind: 'google_drive_file',
|
|
source_uri: entry[:web_view_link] || google_drive_file_url(entry[:id]),
|
|
source_path: relative_path.presence || MaterialSyncExportPath.normalize_path(entry[:name]),
|
|
source_file_id: entry[:id],
|
|
filename: entry[:name],
|
|
content_type: entry[:mime_type],
|
|
file_downloader: lambda {
|
|
drive_client.download_to_tempfile(entry[:id], filename: entry[:name])
|
|
},
|
|
file_sha256: entry[:sha256_checksum],
|
|
export_path:,
|
|
profile: @source.profile,
|
|
tag: nil,
|
|
url: nil,
|
|
created_by_user: @source.created_by_user,
|
|
updated_by_user: @source.updated_by_user || @source.created_by_user }
|
|
end
|
|
|
|
def candidate_normalized_source_key candidate
|
|
MaterialSyncSuppression.normalize_source_key(source_kind: candidate[:source_kind],
|
|
source_uri: candidate[:source_uri],
|
|
source_path: candidate[:source_path],
|
|
source_file_id: candidate[:source_file_id])
|
|
end
|
|
|
|
def suppression_for candidate
|
|
if candidate[:source_kind] == 'google_drive_file' && candidate[:source_path].present?
|
|
return MaterialSyncSuppressionMatcher.match_google_drive_candidate(
|
|
drive_file_id: candidate[:source_file_id],
|
|
relative_path: candidate[:source_path])
|
|
end
|
|
|
|
MaterialSyncSuppressionMatcher.match(source_kind: candidate[:source_kind],
|
|
source_uri: candidate[:source_uri],
|
|
source_path: candidate[:source_path],
|
|
source_file_id: candidate[:source_file_id])
|
|
end
|
|
|
|
def drive_client
|
|
@drive_client ||= GoogleDrive::ApiClient.new
|
|
end
|
|
|
|
def google_drive_folder_id
|
|
google_drive_target_id.tap do |id|
|
|
if id.blank?
|
|
raise ArgumentError, 'google_drive_path source_file_id or source_uri is required'
|
|
end
|
|
end
|
|
end
|
|
|
|
def google_drive_file_id
|
|
google_drive_target_id.tap do |id|
|
|
if id.blank?
|
|
raise ArgumentError, 'google_drive_file source_file_id or source_uri is required'
|
|
end
|
|
end
|
|
end
|
|
|
|
def google_drive_target_id
|
|
@source.source_file_id.presence || drive_client.extract_file_id(@source.source_uri)
|
|
end
|
|
|
|
def google_drive_file_url file_id
|
|
"https://drive.google.com/file/d/#{ file_id }/view"
|
|
end
|
|
|
|
def close_candidate_tempfile candidate
|
|
tempfile = candidate[:file_tempfile]
|
|
return unless tempfile
|
|
|
|
tempfile.close! unless tempfile.closed?
|
|
rescue StandardError
|
|
nil
|
|
end
|
|
|
|
def log_result result
|
|
Rails.logger.info(material_sync_log(imported: result.imported,
|
|
updated: result.updated,
|
|
unchanged: result.unchanged,
|
|
suppressed: result.suppressed,
|
|
failed: result.failed))
|
|
end
|
|
|
|
def progress_log_scan_count? scanned_count
|
|
scanned_count == 1 || (scanned_count % 50).zero?
|
|
end
|
|
|
|
def log_google_drive_progress folder_id, scanned_count, result, summary: false
|
|
Rails.logger.info(
|
|
material_sync_log(folder_id:,
|
|
scanned_count:,
|
|
imported: result.imported,
|
|
updated: result.updated,
|
|
unchanged: result.unchanged,
|
|
suppressed: result.suppressed,
|
|
failed: result.failed,
|
|
progress: summary ? 'summary' : 'scan'))
|
|
end
|
|
|
|
def material_sync_log fields
|
|
{ material_sync_source_id: @source.id,
|
|
material_sync_source_name: @source.name }.merge(fields).to_json
|
|
end
|
|
end
|