Class: Wp2txt::FtsIndexBuilder

Inherits:
Object
  • Object
show all
Defined in:
lib/wp2txt/fts_index.rb

Overview

Builds an FtsIndex by scanning all streams in parallel: workers decompress, parse, and render cleaned section text; the parent owns the sole SQLite connection and inserts from the Parallel finish hook.

Constant Summary collapse

STREAMS_PER_BATCH =

Smaller batches than the metadata builder: rendering dominates, so this keeps progress granular and per-batch IPC payloads moderate

10

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(multistream_path, stream_offsets, db_path:, meta_db_path:, tokenizer: nil, num_processes: 4, optimize: true) ⇒ FtsIndexBuilder

Returns a new instance of FtsIndexBuilder.



369
370
371
372
373
374
375
376
377
378
 
# File 'lib/wp2txt/fts_index.rb', line 369

def initialize(multistream_path, stream_offsets, db_path:, meta_db_path:,
               tokenizer: nil, num_processes: 4, optimize: true)
  @multistream_path = multistream_path
  @stream_offsets = stream_offsets
  @db_path = db_path
  @meta_db_path = meta_db_path
  @tokenizer = tokenizer || FtsIndex.default_tokenizer(multistream_path)
  @num_processes = num_processes
  @optimize = optimize
end

Class Method Details

.scan_batch(multistream_path, offset_pairs) ⇒ Object

Runs in worker processes: must not touch SQLite



406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
 
# File 'lib/wp2txt/fts_index.rb', line 406

def self.scan_batch(multistream_path, offset_pairs)
  renderer = SectionRenderer.new
  rows = []
  File.open(multistream_path, "rb") do |f|
    offset_pairs.each do |offset, next_offset|
      f.seek(offset)
      data = next_offset ? f.read(next_offset - offset) : f.read
      xml = MetadataIndexBuilder.decompress_bz2(data)
      xml.scan(MetadataIndexBuilder::PAGE_BLOCK_REGEX) do
        scan_page(::Regexp.last_match(1), renderer, rows)
      end
    end
  end
  rows
end

.scan_page(block, renderer, rows) ⇒ Object 



422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
 
# File 'lib/wp2txt/fts_index.rb', line 422

def self.scan_page(block, renderer, rows)
  title = block[MetadataIndexBuilder::TITLE_REGEX, 1]
  return unless title && !title.empty?

  ns = (block[MetadataIndexBuilder::NS_REGEX, 1] || "0").to_i
  return unless ns.zero?

  page_id = block[MetadataIndexBuilder::ID_REGEX, 1]&.to_i
  return unless page_id

  title = MetadataIndexBuilder.unescape_xml(title)
  text = block[MetadataIndexBuilder::TEXT_REGEX, 1] || ""
  text = MetadataIndexBuilder.unescape_xml(text)
  return if REDIRECT_REGEX.match?(text)

  renderer.render_sections(title, text).each do |heading, ord, section_text|
    rows << [page_id, heading, ord, section_text]
  end
rescue StandardError
  # A single malformed page must not abort the whole batch
  nil
end

Instance Method Details

#build(&progress) ⇒ Object 



380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
 
# File 'lib/wp2txt/fts_index.rb', line 380

def build(&progress)
  index = FtsIndex.new(@db_path, @meta_db_path)
  index.prepare_build!(tokenizer: @tokenizer)
  index.close

  pairs = @stream_offsets.zip(@stream_offsets[1..].to_a + [nil])
  batches = pairs.each_slice(STREAMS_PER_BATCH).to_a
  done = 0

  Parallel.map(
    batches,
    in_processes: @num_processes,
    finish: lambda { |_item, _idx, rows|
      index.insert_batch(rows)
      done += 1
      progress&.call(done, batches.size)
    }
  ) do |batch|
    self.class.scan_batch(@multistream_path, batch)
  end

  index.finalize_build!(@multistream_path, optimize: @optimize)
  index
end