Class: Wp2txt::MetadataIndexBuilder

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

Overview

Builds a MetadataIndex by scanning all streams of a multistream dump in parallel. Workers decompress and regex-scan streams; the parent process owns the sole SQLite connection and inserts each batch from the Parallel finish hook.

Constant Summary collapse

STREAMS_PER_BATCH = 
50
PAGE_BLOCK_REGEX = 
%r{<page>(.*?)</page>}m
TITLE_REGEX = 
%r{<title>([^<]*)</title>}
NS_REGEX = 
%r{<ns>(-?\d+)</ns>}
ID_REGEX = 
%r{<id>(\d+)</id>}
TEXT_REGEX = 
%r{<text[^>]*>(.*)</text>}m
HEADING_REGEX = 
/\A(={2,6})[ \t]*(.+?)[ \t]*={2,6}[ \t]*\z/
HTML_COMMENT_REGEX = 
/<!--.*?-->/m

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(multistream_path, stream_offsets, db_path:, num_processes: 4) ⇒ MetadataIndexBuilder

Returns a new instance of MetadataIndexBuilder.



622
623
624
625
626
627
 
# File 'lib/wp2txt/metadata_index.rb', line 622

def initialize(multistream_path, stream_offsets, db_path:, num_processes: 4)
  @multistream_path = multistream_path
  @stream_offsets = stream_offsets
  @db_path = db_path
  @num_processes = num_processes
end

Class Method Details

.decompress_bz2(data) ⇒ Object 



674
675
676
677
678
679
 
# File 'lib/wp2txt/metadata_index.rb', line 674

def self.decompress_bz2(data)
  stdout, status = Open3.capture2("bzcat", stdin_data: data)
  raise "bzcat failed (exit #{status.exitstatus})" unless status.success?

  stdout.force_encoding(Encoding::UTF_8)
end

.scan_batch(multistream_path, offset_pairs) ⇒ Object

Scan a batch of [offset, next_offset] stream pairs. Runs inside worker processes: must not touch SQLite.



661
662
663
664
665
666
667
668
669
670
671
672
 
# File 'lib/wp2txt/metadata_index.rb', line 661

def self.scan_batch(multistream_path, offset_pairs)
  out = { pages: [], categories: [], sections: [], hierarchy: [] }
  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 = decompress_bz2(data)
      xml.scan(PAGE_BLOCK_REGEX) { scan_page(::Regexp.last_match(1), out) }
    end
  end
  out
end

.scan_page(block, out) ⇒ Object 



681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
 
# File 'lib/wp2txt/metadata_index.rb', line 681

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

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

  title = unescape_xml(title)
  ns = (block[NS_REGEX, 1] || "0").to_i
  text = block[TEXT_REGEX, 1] || ""
  text = unescape_xml(text)
  # Strip HTML comments before scanning, matching what the Article parser
  # does for the FTS index: a trailing comment after a heading's closing
  # `==` must not hide the heading, and commented-out [[Category:]] links
  # must not be indexed as real categories
  text = text.gsub(HTML_COMMENT_REGEX, "")

  redirect_to = nil
  if (m = REDIRECT_REGEX.match(text))
    redirect_to = m[1].split(/[#|]/).first.to_s.strip
    redirect_to = nil if redirect_to.empty?
  end

  out[:pages] << [page_id, title, ns, redirect_to, text.length]

  categories = text.scan(CATEGORY_REGEX)
                   .map { |c| MetadataIndex.normalize_category(c[0]) }
                   .reject(&:empty?).uniq
  if ns == MetadataIndex::NS_CATEGORY
    child = MetadataIndex.normalize_category(title.sub(/\A[^:]+:/, ""))
    categories.each { |c| out[:hierarchy] << [child, c] } unless child.empty?
  else
    categories.each { |c| out[:categories] << [page_id, c] }
  end

  # ord 0 is the lead text (not stored here — it has no heading); headings
  # start at 1 so ord aligns with fts_map.ord in the full-text DB
  ord = 0
  text.each_line do |line|
    l = line.chomp
    next unless l.start_with?("==")

    hm = HEADING_REGEX.match(l)
    next unless hm

    ord += 1
    heading = MetadataIndex.clean_heading(hm[2])
    next if heading.empty?

    out[:sections] << [page_id, heading, hm[1].length, ord]
  end
end

.unescape_xml(str) ⇒ Object 



734
735
736
 
# File 'lib/wp2txt/metadata_index.rb', line 734

def self.unescape_xml(str)
  str.gsub("&lt;", "<").gsub("&gt;", ">").gsub("&quot;", '"').gsub("&amp;", "&")
end

Instance Method Details

#build(&progress) ⇒ MetadataIndex

Build the index. Yields (batches_done, batches_total) after each batch.

Returns:



631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
 
# File 'lib/wp2txt/metadata_index.rb', line 631

def build(&progress)
  index = MetadataIndex.new(@db_path)
  index.prepare_build!
  # Close before Parallel forks workers so children do not inherit a
  # writable SQLite connection; the finish hook reopens it lazily in
  # the parent, which is the only process that ever writes.
  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, result|
      index.insert_batch(result)
      done += 1
      progress&.call(done, batches.size)
    }
  ) do |batch|
    self.class.scan_batch(@multistream_path, batch)
  end

  index.finalize_build!(@multistream_path)
  index
end