from . import parser from . block import Block from typing import NamedTuple class Patch(NamedTuple): """ Patch represents chunk of chat data which is fetched by asyncdl.download_patch._fetch(). """ chats : list = [] continuation : str = None first : int = None last : int = None def fill(block:Block, patch:Patch): block_end = block.end if patch.last < block_end or block.is_last: set_patch(block, patch) return for line in reversed(patch.chats): line_offset = parser.get_offset(line) if line_offset < block_end: break patch.chats.pop() set_patch(block, patch._replace( continuation = None, last = line_offset ) ) block.remaining=0 block.done=True def split(parent_block:Block, child_block:Block, patch:Patch): parent_block.during_split = False if patch.first <= parent_block.last: ''' When patch overlaps with parent_block, discard this block. ''' child_block.continuation = None ''' Leave child_block.during_split == True to exclude from during_split sequence. ''' return child_block.during_split = False child_block.first = patch.first parent_block.end = patch.first fill(child_block, patch) def set_patch(block:Block, patch:Patch): block.continuation = patch.continuation block.chat_data.extend(patch.chats) block.last = patch.last block.remaining = block.end-block.last