Class: Wp2txt::MultistreamReader

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

Overview

Reads articles from multistream bz2 files

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(multistream_path, index_or_path, use_cache: true, cache_dir: nil) ⇒ MultistreamReader

Initialize reader with multistream file and index

Parameters:

  • multistream_path (String) —

    Path to the multistream bz2 file

  • index_or_path (MultistreamIndex, String) —

    Either an existing index instance or path to index file

  • use_cache (Boolean) (defaults to: true) —

    Whether to use SQLite cache for index (default: true, only used if index_or_path is a path)

  • cache_dir (String, nil) (defaults to: nil) —

    Directory for SQLite cache (only used if index_or_path is a path)



233
234
235
236
237
238
239
240
241
242
243
 
# File 'lib/wp2txt/multistream.rb', line 233

def initialize(multistream_path, index_or_path, use_cache: true, cache_dir: nil)
  @multistream_path = multistream_path

  # Accept an existing index (or any object with the same lookup interface,
  # e.g. a lazy SQLite-backed one) or a path to create one
  if index_or_path.respond_to?(:find_by_title)
    @index = index_or_path
  else
    @index = MultistreamIndex.new(index_or_path, use_cache: use_cache, cache_dir: cache_dir)
  end
end

Instance Attribute Details

#index ⇒ Object (readonly)

Returns the value of attribute index.



226
227
228
 
# File 'lib/wp2txt/multistream.rb', line 226

def index
  @index
end

#multistream_path ⇒ Object (readonly)

Returns the value of attribute multistream_path.



226
227
228
 
# File 'lib/wp2txt/multistream.rb', line 226

def multistream_path
  @multistream_path
end

Instance Method Details

#each_article_in_first_streams(stream_count, &block) ⇒ Object

Iterate through first N streams



341
342
343
344
345
 
# File 'lib/wp2txt/multistream.rb', line 341

def each_article_in_first_streams(stream_count, &block)
  @index.stream_offsets.first(stream_count).each do |offset|
    each_article_in_stream(offset, &block)
  end
end

#each_article_in_stream(offset, &block) ⇒ Object

Iterate through all articles in a stream



335
336
337
338
 
# File 'lib/wp2txt/multistream.rb', line 335

def each_article_in_stream(offset, &block)
  stream_content = read_stream_at(offset)
  extract_all_pages_from_xml(stream_content, &block)
end

#each_article_parallel(entries, num_processes: 4) {|Hash| ... } ⇒ Object

Iterate through articles in parallel, yielding each page Groups articles by stream and processes streams in parallel

Parameters:

  • entries (Array<Hash>) —

    Array of index entries with :title and :offset

  • num_processes (Integer) (defaults to: 4) —

    Number of parallel processes

Yields:

  • (Hash) —

    Page data for each article



310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
 
# File 'lib/wp2txt/multistream.rb', line 310

def each_article_parallel(entries, num_processes: 4)
  return enum_for(:each_article_parallel, entries, num_processes: num_processes) unless block_given?

  # Group by stream offset
  grouped = entries.group_by { |e| e[:offset] }

  # Process streams in parallel, collecting all pages
  all_pages = Parallel.flat_map(grouped.keys, in_processes: num_processes) do |offset|
    entries_in_stream = grouped[offset]
    stream_content = read_stream_at(offset)

    pages = []
    entries_in_stream.each do |entry|
      page = extract_page_from_xml(stream_content, entry[:title])
      pages << page if page
    end

    pages
  end

  # Yield each page (sequential, as yielding must happen in main process)
  all_pages.each { |page| yield page }
end

#extract_article(title) ⇒ Object

Extract a single article by title



246
247
248
249
250
251
252
 
# File 'lib/wp2txt/multistream.rb', line 246

def extract_article(title)
  entry = @index.find_by_title(title)
  return nil unless entry

  stream_content = read_stream_at(entry[:offset])
  extract_page_from_xml(stream_content, title)
end

#extract_articles(titles) ⇒ Object

Extract multiple articles (sequential)



255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
 
# File 'lib/wp2txt/multistream.rb', line 255

def extract_articles(titles)
  # Group by stream offset for efficiency
  grouped = titles.group_by { |t| @index.stream_offset_for(t) }

  results = {}
  grouped.each do |offset, titles_in_stream|
    next unless offset

    stream_content = read_stream_at(offset)
    titles_in_stream.each do |title|
      page = extract_page_from_xml(stream_content, title)
      results[title] = page if page
    end
  end
  results
end

#extract_articles_parallel(titles, num_processes: 4, &progress_callback) ⇒ Hash

Extract multiple articles in parallel (by stream)

Parameters:

  • titles (Array<String>) —

    Article titles to extract

  • num_processes (Integer) (defaults to: 4) —

    Number of parallel processes (default: 4)

  • progress_callback (Proc, nil) —

    Optional callback for progress updates

Returns:

  • (Hash) —

    Map of title => page data



277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
 
# File 'lib/wp2txt/multistream.rb', line 277

def extract_articles_parallel(titles, num_processes: 4, &progress_callback)
  # Group titles by stream offset
  grouped = titles.group_by { |t| @index.stream_offset_for(t) }
  grouped.delete(nil) # Remove titles not found in index

  # Process streams in parallel
  stream_results = Parallel.map(grouped.keys, in_processes: num_processes) do |offset|
    titles_in_stream = grouped[offset]
    stream_content = read_stream_at(offset)

    stream_pages = {}
    titles_in_stream.each do |title|
      page = extract_page_from_xml(stream_content, title)
      stream_pages[title] = page if page
    end

    stream_pages
  end

  # Merge results from all streams
  results = {}
  stream_results.each do |stream_pages|
    results.merge!(stream_pages)
  end

  results
end